Repository: Lightning-AI/pytorch-lightning
Pull request: Fix Google Cloud Storage checkpointing
## What does this PR do?

This PR fixes an issue when using Google Storage as the Trainer's root directory, which failed because `fs.makedirs` is a no-op with GCS.

Fixes #18044

<!-- Does your PR introduce any breaking changes? If yes, please list them. -->

<details>
  <summary><b>Before submitting</b></summary>

- Was this **discussed/agreed** via a GitHub issue? (not for typos and docs)
- [x] Did you read the [contributor guideline](https://github.com/Lightning-AI/lightning/blob/master/.github/CONTRIBUTING.md), **Pull Request** section?
- [x] Did you make sure your **PR does only one thing**, instead of bundling different changes together?
- Did you make sure to **update the documentation** with your changes? (if necessary)
- Did you write any **new necessary tests**? (not for typos and docs)
- [x] Did you verify new and **existing tests pass** locally with your changes?
- Did you list all the **breaking changes** introduced by this pull request?
- [x] Did you **update the CHANGELOG**? (not for typos, docs, test updates, or minor internal changes/refactors)

<!-- In the CHANGELOG, separate each item in the unreleased section by a blank line to reduce collisions -->

</details>

## PR review

Anyone in the community is welcome to review the PR.
Before you start reviewing, make sure you have read the [review guidelines](https://github.com/Lightning-AI/lightning/wiki/Review-guidelines). In short, see the following bullet-list:

<details>
  <summary>Reviewer checklist</summary>

- [ ] Is this pull request ready for review? (if not, please submit in draft mode)
- [ ] Check that all items from **Before submitting** are resolved
- [ ] Make sure the title is self-explanatory and the description concisely explains the PR
- [ ] Add labels and milestones (and optionally projects) to the PR so it can be classified

</details>

<!--

Did you have fun?

Make sure you had fun coding 🙃

-->


Problem statement:
Missing folder exception with Google Cloud Storage checkpointing
### Bug description

The training fails when a GCP bucket is chosen as the Trainer's `default_root_dir`. I am properly logged in using `gcloud auth application-default login` and the correct GCP project is set using `gcloud config set project <my-project>`. I have no problem with reading or writing to the same bucket using the console or other applications.

### What version are you seeing the problem on?

v2.0

### How to reproduce the bug

```python
import pytorch_lightning as pl
from pytorch_lightning.demos.boring_classes import BoringDataModule, BoringModel

if __name__ == "__main__":
    trainer = pl.Trainer(
        max_epochs=1,
        default_root_dir="gs://llm-bucket/checkpoints/debug/",
        limit_train_batches=8,
        limit_val_batches=8,
    )
    trainer.fit(BoringModel(), datamodule=BoringDataModule())
```


### Error messages and logs

```
[2023-07-10 15:30:07,988] [INFO] [real_accelerator.py:110:get_accelerator] Setting ds_accelerator to cuda (auto detect)
GPU available: False, used: False
TPU available: False, using: 0 TPU cores
IPU available: False, using: 0 IPUs
HPU available: False, using: 0 HPUs
/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/trainer/connectors/logger_connector/logger_connector.py:67: UserWarning: Starting from v1.9.0, `tensorboardX` has been removed as a dependency of the `pytorch_lightning` package, due to potential conflicts with other packages in the ML ecosystem. For this reason, `logger=True` will use `CSVLogger` as the default logger, unless the `tensorboard` or `tensorboardX` packages are found. Please `pip install lightning[extra]` or one of them to enable TensorBoard support by default
  warning_cache.warn(
Missing logger folder: gs://llm-bucket/checkpoints/debug/lightning_logs

  | Name  | Type   | Params
---------------------------------
0 | layer | Linear | 66    
---------------------------------
66        Trainable params
0         Non-trainable params
66        Total params
0.000     Total estimated model params size (MB)
Traceback (most recent call last):
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/trainer/call.py", line 42, in _call_and_handle_interrupt
    return trainer_fn(*args, **kwargs)
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/trainer/trainer.py", line 570, in _fit_impl
    self._run(model, ckpt_path=ckpt_path)
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/trainer/trainer.py", line 958, in _run
    _log_hyperparams(self)
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/loggers/utilities.py", line 94, in _log_hyperparams
    logger.save()
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/lightning_utilities/core/rank_zero.py", line 32, in wrapped_fn
    return fn(*args, **kwargs)
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/lightning_fabric/loggers/csv_logs.py", line 141, in save
    self.experiment.save()
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/loggers/csv_logs.py", line 61, in save
    save_hparams_to_yaml(hparams_file, self.hparams)
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/core/saving.py", line 292, in save_hparams_to_yaml
    raise RuntimeError(f"Missing folder: {os.path.dirname(config_yaml)}.")
RuntimeError: Missing folder: gs://llm-bucket/checkpoints/debug/lightning_logs/version_0.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/mre_pt_gcp.py", line 11, in <module>
    trainer.fit(BoringModel(), datamodule=BoringDataModule())
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/trainer/trainer.py", line 531, in fit
    call._call_and_handle_interrupt(
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/trainer/call.py", line 65, in _call_and_handle_interrupt
    logger.finalize("failed")
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/lightning_utilities/core/rank_zero.py", line 32, in wrapped_fn
    return fn(*args, **kwargs)
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/lightning_fabric/loggers/csv_logs.py", line 149, in finalize
    self.save()
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/lightning_utilities/core/rank_zero.py", line 32, in wrapped_fn
    return fn(*args, **kwargs)
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/lightning_fabric/loggers/csv_logs.py", line 141, in save
    self.experiment.save()
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/loggers/csv_logs.py", line 61, in save
    save_hparams_to_yaml(hparams_file, self.hparams)
  File "/Users/bilelomrani/Documents/ILLUIN.nosync/instructions-finetuning/.venv/lib/python3.9/site-packages/pytorch_lightning/core/saving.py", line 292, in save_hparams_to_yaml
    raise RuntimeError(f"Missing folder: {os.path.dirname(config_yaml)}.")
RuntimeError: Missing folder: gs://llm-bucket/checkpoints/debug/lightning_logs/version_0.
```

### Environment

<details>
  <summary>Current environment</summary>

* CUDA:
        - GPU:               None
        - available:         False
        - version:           None
* Lightning:
        - lightning-utilities: 0.9.0
        - pytorch-lightning: 2.0.4
        - torch:             2.0.1
        - torchmetrics:      0.11.4
* Packages:
        - accelerate:        0.20.3
        - aiohttp:           3.8.4
        - aiosignal:         1.3.1
        - appdirs:           1.4.4
        - async-timeout:     4.0.2
        - attrs:             23.1.0
        - bitsandbytes:      0.39.1
        - black:             23.3.0
        - cachetools:        5.3.1
        - certifi:           2023.5.7
        - chardet:           5.1.0
        - charset-normalizer: 3.1.0
        - click:             8.1.3
        - configue:          4.2.0
        - coolname:          2.2.0
        - coverage:          7.2.7
        - datasets:          2.13.1
        - decorator:         5.1.1
        - deepspeed:         0.9.5
        - deptry:            0.12.0
        - dill:              0.3.6
        - docker:            6.1.3
        - docker-pycreds:    0.4.0
        - einops:            0.6.1
        - exceptiongroup:    1.1.2
        - filelock:          3.12.2
        - fire:              0.5.0
        - frozenlist:        1.3.3
        - fsspec:            2023.6.0
        - gcsfs:             2023.6.0
        - gitdb:             4.0.10
        - gitpython:         3.1.31
        - google-api-core:   2.11.1
        - google-auth:       2.21.0
        - google-auth-oauthlib: 1.0.0
        - google-cloud-core: 2.3.2
        - google-cloud-storage: 2.10.0
        - google-crc32c:     1.5.0
        - google-resumable-media: 2.5.0
        - googleapis-common-protos: 1.59.1
        - greenlet:          2.0.2
        - hjson:             3.1.0
        - huggingface-hub:   0.15.1
        - idna:              3.4
        - iniconfig:         2.0.0
        - instruction-finetuning: 0.1.dev89+g6a38a51.d20230710
        - instructions-finetuning: 0.1.dev75+gee76c64.d20230630
        - jinja2:            3.1.2
        - jiwer:             3.0.2
        - joblib:            1.3.1
        - lightning-utilities: 0.9.0
        - markupsafe:        2.1.3
        - mpmath:            1.3.0
        - multidict:         6.0.4
        - multiprocess:      0.70.14
        - mypy:              1.4.1
        - mypy-extensions:   1.0.0
        - networkx:          3.1
        - ninja:             1.11.1
        - numpy:             1.25.0
        - oauthlib:          3.2.2
        - packaging:         23.1
        - pandas:            2.0.3
        - pandasql:          0.7.3
        - pathspec:          0.11.1
        - pathtools:         0.1.2
        - peft:              0.4.0.dev0
        - platformdirs:      3.8.1
        - pluggy:            1.2.0
        - protobuf:          3.20.3
        - psutil:            5.9.5
        - py-cpuinfo:        9.0.0
        - pyarrow:           12.0.1
        - pyasn1:            0.5.0
        - pyasn1-modules:    0.3.0
        - pydantic:          1.10.11
        - pytest:            7.4.0
        - pytest-mock:       3.11.1
        - python-dateutil:   2.8.2
        - pytorch-lightning: 2.0.4
        - pytz:              2023.3
        - pyyaml:            5.4.1
        - rapidfuzz:         2.13.7
        - regex:             2023.6.3
        - requests:          2.31.0
        - requests-oauthlib: 1.3.1
        - rsa:               4.9
        - ruff:              0.0.277
        - safetensors:       0.3.1
        - scikit-learn:      1.3.0
        - scipy:             1.11.1
        - sentencepiece:     0.1.99
        - sentry-sdk:        1.26.0
        - setproctitle:      1.3.2
        - setuptools:        68.0.0
        - six:               1.16.0
        - smmap:             5.0.0
        - sqlalchemy:        2.0.17
        - sympy:             1.12
        - termcolor:         2.3.0
        - threadpoolctl:     3.1.0
        - tiktoken:          0.4.0
        - tokenizers:        0.13.3
        - tomli:             2.0.1
        - torch:             2.0.1
        - torchmetrics:      0.11.4
        - tqdm:              4.65.0
        - transformers:      4.30.2
        - types-google-cloud-ndb: 2.1.0.7
        - types-tqdm:        4.65.0.1
        - typing-extensions: 4.6.3
        - tzdata:            2023.3
        - urllib3:           1.26.16
        - wandb:             0.15.4
        - websocket-client:  1.6.1
        - xxhash:            3.2.0
        - yarl:              1.9.2
* System:
        - OS:                Darwin
        - architecture:
                - 64bit
                - 
        - processor:         i386
        - python:            3.9.17
        - release:           22.3.0
        - version:           Darwin Kernel Version 22.3.0: Mon Jan 30 20:42:11 PST 2023; root:xnu-8792.81.3~2/RELEASE_X86_64

</details>

### More info

_No response_

cc @borda

Hints from issue discussion:
I managed to trace it back to [this issue](https://github.com/fsspec/gcsfs/issues/312) in gcsfs. Essentially, the first call to `fs.isdir` returns `False` even when the 'folder' exists, while the second call properly returns `True`. I also noticed that a similar behavior was observed and fixed in Lightning back in 2021 in #7889.
@awaelchli I have a fix for this particular test case. I am not extremely confident in the correctness of my fix, I am quite confused by how `gcsfs` handles directories, not only is `fs.isdir` not idempotent, but `fs.makedirs` is a no-op, hence breaking the `CSVLogger` (and probably other things). Still if you are interested I would be happy to contribute a pull request for this.
@bilelomrani1 I know absolutely nothing about gcsfs and won't be able to test it myself most likely. Feel free to send a PR. If the tests pass, it is probabaly a good sign that your fix can work. If not, we need to see how much effort it would be. No promises, but thanks for looking into it! It is much appreciated.

Commit message: [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Patch to review:
diff --git a/src/lightning/fabric/loggers/csv_logs.py b/src/lightning/fabric/loggers/csv_logs.py
index 31756502e95f7..fd0c69c3ab439 100644
--- a/src/lightning/fabric/loggers/csv_logs.py
+++ b/src/lightning/fabric/loggers/csv_logs.py
@@ -21,7 +21,7 @@
 from torch import Tensor
 
 from lightning.fabric.loggers.logger import Logger, rank_zero_experiment
-from lightning.fabric.utilities.cloud_io import get_filesystem
+from lightning.fabric.utilities.cloud_io import _is_dir, get_filesystem
 from lightning.fabric.utilities.logger import _add_prefix
 from lightning.fabric.utilities.rank_zero import rank_zero_only, rank_zero_warn
 from lightning.fabric.utilities.types import _PATH
@@ -155,7 +155,7 @@ def finalize(self, status: str) -> None:
     def _get_next_version(self) -> int:
         versions_root = os.path.join(self._root_dir, self.name)
 
-        if not self._fs.isdir(versions_root):
+        if not _is_dir(self._fs, versions_root, strict=True):
             log.warning("Missing logger folder: %s", versions_root)
             return 0
 
@@ -163,7 +163,7 @@ def _get_next_version(self) -> int:
         for d in self._fs.listdir(versions_root):
             full_path = d["name"]
             name = os.path.basename(full_path)
-            if self._fs.isdir(full_path) and name.startswith("version_"):
+            if _is_dir(self._fs, full_path) and name.startswith("version_"):
                 existing_versions.append(int(name.split("_")[1]))
 
         if len(existing_versions) == 0:
diff --git a/src/lightning/fabric/loggers/tensorboard.py b/src/lightning/fabric/loggers/tensorboard.py
index a3f84315c9185..3708ac9b73d1b 100644
--- a/src/lightning/fabric/loggers/tensorboard.py
+++ b/src/lightning/fabric/loggers/tensorboard.py
@@ -22,7 +22,7 @@
 from torch.nn import Module
 
 from lightning.fabric.loggers.logger import Logger, rank_zero_experiment
-from lightning.fabric.utilities.cloud_io import get_filesystem
+from lightning.fabric.utilities.cloud_io import _is_dir, get_filesystem
 from lightning.fabric.utilities.logger import _add_prefix, _convert_params, _flatten_dict
 from lightning.fabric.utilities.logger import _sanitize_params as _utils_sanitize_params
 from lightning.fabric.utilities.rank_zero import rank_zero_only, rank_zero_warn
@@ -294,7 +294,7 @@ def _get_next_version(self) -> int:
         for listing in listdir_info:
             d = listing["name"]
             bn = os.path.basename(d)
-            if self._fs.isdir(d) and bn.startswith("version_"):
+            if _is_dir(self._fs, d) and bn.startswith("version_"):
                 dir_ver = bn.split("_")[1].replace("/", "")
                 existing_versions.append(int(dir_ver))
         if len(existing_versions) == 0:
diff --git a/src/lightning/fabric/utilities/cloud_io.py b/src/lightning/fabric/utilities/cloud_io.py
index ecabbbdb6bb74..8a9cd54574dd1 100644
--- a/src/lightning/fabric/utilities/cloud_io.py
+++ b/src/lightning/fabric/utilities/cloud_io.py
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Utilities related to data saving/loading."""
-
+import importlib.util
 import io
 from pathlib import Path
 from typing import Any, Dict, IO, Union
@@ -21,6 +21,7 @@
 import torch
 from fsspec.core import url_to_fs
 from fsspec.implementations.local import AbstractFileSystem
+from lightning_utilities.core.imports import lazy_import
 
 from lightning.fabric.utilities.types import _MAP_LOCATION_TYPE, _PATH
 
@@ -70,3 +71,51 @@ def _atomic_save(checkpoint: Dict[str, Any], filepath: Union[str, Path]) -> None
     torch.save(checkpoint, bytesbuffer)
     with fsspec.open(filepath, "wb") as f:
         f.write(bytesbuffer.getvalue())
+
+
+def _is_object_storage(fs: AbstractFileSystem) -> bool:
+    if importlib.util.find_spec("adlfs") is not None:
+        adlfs = lazy_import("adlfs")
+        if isinstance(fs, adlfs.AzureBlobFileSystem):
+            return True
+
+    if importlib.util.find_spec("gcsfs") is not None:
+        gcsfs = lazy_import("gcsfs")
+        if isinstance(fs, gcsfs.GCSFileSystem):
+            return True
+
+    if importlib.util.find_spec("s3fs") is not None:
+        s3fs = lazy_import("s3fs")
+        if isinstance(fs, s3fs.S3FileSystem):
+            return True
+
+    return False
+
+
+def _is_dir(fs: AbstractFileSystem, path: Union[str, Path], strict: bool = False) -> bool:
+    """Check if a path is directory-like.
+
+    This function determines if a given path is considered directory-like, taking into account the behavior
+    specific to object storage platforms. For other filesystems, it behaves similarly to the standard `fs.isdir`
+    method.
+
+    Args:
+        fs: The filesystem to check the path against.
+        path: The path or URL to be checked.
+        strict: A flag specific to Object Storage platforms. If set to ``False``, any non-existing path is considered
+            as a valid directory-like path. In such cases, the directory (and any non-existing parent directories)
+            will be created on the fly. Defaults to False.
+    """
+    # Object storage fsspec's are inconsistent with other file systems because they do not have real directories,
+    # see for instance https://gcsfs.readthedocs.io/en/latest/api.html?highlight=makedirs#gcsfs.core.GCSFileSystem.mkdir
+    # In particular, `fs.makedirs` is a no-op so we use `strict=False` to consider any path as valid, except if the
+    # path already exists but is a file
+    if _is_object_storage(fs):
+        if strict:
+            return fs.isdir(path)
+
+        # Check if the path is not already taken by a file. If not, it is considered a valid directory-like path
+        # because the directory (and all non-existing parent directories) will be created on the fly.
+        return not fs.isfile(path)
+
+    return fs.isdir(path)
diff --git a/src/lightning/pytorch/CHANGELOG.md b/src/lightning/pytorch/CHANGELOG.md
index 5d8f9627a2119..4665b96cd7a3e 100644
--- a/src/lightning/pytorch/CHANGELOG.md
+++ b/src/lightning/pytorch/CHANGELOG.md
@@ -165,6 +165,7 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 - `LightningCLI` not saving correctly `seed_everything` when `run=True` and `seed_everything=True` ([#18056](https://github.com/Lightning-AI/lightning/pull/18056))
 
+- Fixed a `Missing folder` exception when using a Google Storage URL as a `default_root_dir` ([#18088](https://github.com/Lightning-AI/lightning/pull/18088))
 
 - Fixed an issue that prevented the use of custom logger classes without an `experiment` property defined ([#18093](https://github.com/Lightning-AI/lightning/pull/18093))
 
diff --git a/src/lightning/pytorch/callbacks/model_checkpoint.py b/src/lightning/pytorch/callbacks/model_checkpoint.py
index 01f05841c9b83..8a75e9e137653 100644
--- a/src/lightning/pytorch/callbacks/model_checkpoint.py
+++ b/src/lightning/pytorch/callbacks/model_checkpoint.py
@@ -32,7 +32,7 @@
 from torch import Tensor
 
 import lightning.pytorch as pl
-from lightning.fabric.utilities.cloud_io import get_filesystem
+from lightning.fabric.utilities.cloud_io import _is_dir, get_filesystem
 from lightning.fabric.utilities.types import _PATH
 from lightning.pytorch.callbacks import Checkpoint
 from lightning.pytorch.utilities.exceptions import MisconfigurationException
@@ -620,7 +620,7 @@ def _find_last_checkpoints(self, trainer: "pl.Trainer") -> Set[str]:
         return set()
 
     def __warn_if_dir_not_empty(self, dirpath: _PATH) -> None:
-        if self.save_top_k != 0 and self._fs.isdir(dirpath) and len(self._fs.ls(dirpath)) > 0:
+        if self.save_top_k != 0 and _is_dir(self._fs, dirpath, strict=True) and len(self._fs.ls(dirpath)) > 0:
             rank_zero_warn(f"Checkpoint directory {dirpath} exists and is not empty.")
 
     def _get_metric_interpolated_filepath_name(
diff --git a/src/lightning/pytorch/core/saving.py b/src/lightning/pytorch/core/saving.py
index 13c1bde84ca3e..03a3fd353488d 100644
--- a/src/lightning/pytorch/core/saving.py
+++ b/src/lightning/pytorch/core/saving.py
@@ -30,6 +30,7 @@
 from lightning_utilities.core.apply_func import apply_to_collection
 
 import lightning.pytorch as pl
+from lightning.fabric.utilities.cloud_io import _is_dir
 from lightning.fabric.utilities.cloud_io import _load as pl_load
 from lightning.fabric.utilities.cloud_io import get_filesystem
 from lightning.fabric.utilities.types import _MAP_LOCATION_TYPE, _PATH
@@ -234,7 +235,7 @@ def load_hparams_from_tags_csv(tags_csv: _PATH) -> Dict[str, Any]:
 
 def save_hparams_to_tags_csv(tags_csv: _PATH, hparams: Union[dict, Namespace]) -> None:
     fs = get_filesystem(tags_csv)
-    if not fs.isdir(os.path.dirname(tags_csv)):
+    if not _is_dir(fs, os.path.dirname(tags_csv)):
         raise RuntimeError(f"Missing folder: {os.path.dirname(tags_csv)}.")
 
     if isinstance(hparams, Namespace):
@@ -288,7 +289,7 @@ def save_hparams_to_yaml(config_yaml: _PATH, hparams: Union[dict, Namespace], us
 
     """
     fs = get_filesystem(config_yaml)
-    if not fs.isdir(os.path.dirname(config_yaml)):
+    if not _is_dir(fs, os.path.dirname(config_yaml)):
         raise RuntimeError(f"Missing folder: {os.path.dirname(config_yaml)}.")
 
     # convert Namespace or AD to dict
diff --git a/src/lightning/pytorch/loggers/tensorboard.py b/src/lightning/pytorch/loggers/tensorboard.py
index c0d9c5ba0d5a2..c99ca2ad02e2a 100644
--- a/src/lightning/pytorch/loggers/tensorboard.py
+++ b/src/lightning/pytorch/loggers/tensorboard.py
@@ -26,6 +26,7 @@
 import lightning.pytorch as pl
 from lightning.fabric.loggers.tensorboard import _TENSORBOARD_AVAILABLE, _TENSORBOARDX_AVAILABLE
 from lightning.fabric.loggers.tensorboard import TensorBoardLogger as FabricTensorBoardLogger
+from lightning.fabric.utilities.cloud_io import _is_dir
 from lightning.fabric.utilities.logger import _convert_params
 from lightning.fabric.utilities.types import _PATH
 from lightning.pytorch.callbacks import ModelCheckpoint
@@ -217,7 +218,7 @@ def save(self) -> None:
         hparams_file = os.path.join(dir_path, self.NAME_HPARAMS_FILE)
 
         # save the metatags file if it doesn't exist and the log directory exists
-        if self._fs.isdir(dir_path) and not self._fs.isfile(hparams_file):
+        if _is_dir(self._fs, dir_path) and not self._fs.isfile(hparams_file):
             save_hparams_to_yaml(hparams_file, self.hparams)
 
     @rank_zero_only
@@ -248,7 +249,7 @@ def _get_next_version(self) -> int:
         for listing in listdir_info:
             d = listing["name"]
             bn = os.path.basename(d)
-            if self._fs.isdir(d) and bn.startswith("version_"):
+            if _is_dir(self._fs, d) and bn.startswith("version_"):
                 dir_ver = bn.split("_")[1].replace("/", "")
                 existing_versions.append(int(dir_ver))
         if len(existing_versions) == 0:
diff --git a/src/lightning/pytorch/trainer/connectors/checkpoint_connector.py b/src/lightning/pytorch/trainer/connectors/checkpoint_connector.py
index a27ebc5c212b1..ed54a9b4ebb77 100644
--- a/src/lightning/pytorch/trainer/connectors/checkpoint_connector.py
+++ b/src/lightning/pytorch/trainer/connectors/checkpoint_connector.py
@@ -23,7 +23,7 @@
 
 import lightning.pytorch as pl
 from lightning.fabric.plugins.environments.slurm import SLURMEnvironment
-from lightning.fabric.utilities.cloud_io import get_filesystem
+from lightning.fabric.utilities.cloud_io import _is_dir, get_filesystem
 from lightning.fabric.utilities.types import _PATH
 from lightning.pytorch.callbacks import ModelCheckpoint
 from lightning.pytorch.plugins.precision import MixedPrecisionPlugin
@@ -55,7 +55,7 @@ def _hpc_resume_path(self) -> Optional[str]:
         dir_path_hpc = self.trainer.default_root_dir
         dir_path_hpc = str(dir_path_hpc)
         fs, path = url_to_fs(dir_path_hpc)
-        if not fs.isdir(path):
+        if not _is_dir(fs, path):
             return None
         max_version = self.__max_ckpt_version_in_folder(dir_path_hpc, "hpc_ckpt_")
         if max_version is not None:
diff --git a/tests/tests_fabric/utilities/test_cloud_io.py b/tests/tests_fabric/utilities/test_cloud_io.py
index 9cdc52acd4c8d..cd600eadf7e00 100644
--- a/tests/tests_fabric/utilities/test_cloud_io.py
+++ b/tests/tests_fabric/utilities/test_cloud_io.py
@@ -15,8 +15,9 @@
 
 import fsspec
 from fsspec.implementations.local import LocalFileSystem
+from fsspec.spec import AbstractFileSystem
 
-from lightning.fabric.utilities.cloud_io import get_filesystem
+from lightning.fabric.utilities.cloud_io import _is_dir, get_filesystem
 
 
 def test_get_filesystem_custom_filesystem():
@@ -32,3 +33,61 @@ class DummyFileSystem(LocalFileSystem):
 
 def test_get_filesystem_local_filesystem():
     assert isinstance(get_filesystem("tmpdir/tmp_file"), LocalFileSystem)
+
+
+def test_is_dir_with_local_filesystem(tmp_path):
+    fs = LocalFileSystem()
+    tmp_existing_directory = tmp_path
+    tmp_non_existing_directory = tmp_path / "non_existing"
+
+    assert _is_dir(fs, tmp_existing_directory)
+    assert not _is_dir(fs, tmp_non_existing_directory)
+
+
+def test_is_dir_with_object_storage_filesystem():
+    class MockAzureBlobFileSystem(AbstractFileSystem):
+        def isdir(self, path):
+            return path.startswith("azure://") and not path.endswith(".txt")
+
+        def isfile(self, path):
+            return path.startswith("azure://") and path.endswith(".txt")
+
+    class MockGCSFileSystem(AbstractFileSystem):
+        def isdir(self, path):
+            return path.startswith("gcs://") and not path.endswith(".txt")
+
+        def isfile(self, path):
+            return path.startswith("gcs://") and path.endswith(".txt")
+
+    class MockS3FileSystem(AbstractFileSystem):
+        def isdir(self, path):
+            return path.startswith("s3://") and not path.endswith(".txt")
+
+        def isfile(self, path):
+            return path.startswith("s3://") and path.endswith(".txt")
+
+    fsspec.register_implementation("azure", MockAzureBlobFileSystem, clobber=True)
+    fsspec.register_implementation("gcs", MockGCSFileSystem, clobber=True)
+    fsspec.register_implementation("s3", MockS3FileSystem, clobber=True)
+
+    azure_directory = "azure://container/directory/"
+    azure_file = "azure://container/file.txt"
+    gcs_directory = "gcs://bucket/directory/"
+    gcs_file = "gcs://bucket/file.txt"
+    s3_directory = "s3://bucket/directory/"
+    s3_file = "s3://bucket/file.txt"
+
+    assert _is_dir(get_filesystem(azure_directory), azure_directory)
+    assert _is_dir(get_filesystem(azure_directory), azure_directory, strict=True)
+    assert not _is_dir(get_filesystem(azure_directory), azure_file)
+    assert not _is_dir(get_filesystem(azure_directory), azure_file, strict=True)
+
+    assert _is_dir(get_filesystem(gcs_directory), gcs_directory)
+    assert _is_dir(get_filesystem(gcs_directory), gcs_directory, strict=True)
+    assert not _is_dir(get_filesystem(gcs_directory), gcs_file)
+    assert not _is_dir(get_filesystem(gcs_directory), gcs_file, strict=True)
+
+    assert _is_dir(get_filesystem(s3_directory), s3_directory)
+    assert _is_dir(get_filesystem(s3_directory), s3_directory, strict=True)
+    assert not _is_dir(get_filesystem(s3_directory), s3_file)
+    assert not _is_dir(get_filesystem(s3_directory), s3_file, strict=True)


Code review comments:
diff_hunk: @@ -21,6 +21,7 @@
 import torch
 from fsspec.core import url_to_fs
 from fsspec.implementations.local import AbstractFileSystem
+from lightning_utilities.core.imports import lazy_import
path: src/lightning/fabric/utilities/cloud_io.py
text: @user1:
```suggestion
from lightning_utilities.core.imports import module_available
```

diff_hunk: @@ -70,3 +71,51 @@ def _atomic_save(checkpoint: Dict[str, Any], filepath: Union[str, Path]) -> None
     torch.save(checkpoint, bytesbuffer)
     with fsspec.open(filepath, "wb") as f:
         f.write(bytesbuffer.getvalue())
+
+
+def _is_object_storage(fs: AbstractFileSystem) -> bool:
+    if importlib.util.find_spec("adlfs") is not None:
+        adlfs = lazy_import("adlfs")
+        if isinstance(fs, adlfs.AzureBlobFileSystem):
+            return True
+
+    if importlib.util.find_spec("gcsfs") is not None:
+        gcsfs = lazy_import("gcsfs")
+        if isinstance(fs, gcsfs.GCSFileSystem):
+            return True
+
+    if importlib.util.find_spec("s3fs") is not None:
+        s3fs = lazy_import("s3fs")
+        if isinstance(fs, s3fs.S3FileSystem):
path: src/lightning/fabric/utilities/cloud_io.py
text: @user1:
```suggestion
    if module_available("adlfs"):
        from adlfs import AzureBlobFileSystem
        if isinstance(fs, AzureBlobFileSystem):
            return True

    if module_available("gcsfs"):
        from adlfs import GCSFileSystem
        if isinstance(fs, GCSFileSystem):
            return True

    if module_available("s3fs"):
        from s3fs import S3FileSystem
        if isinstance(fs, S3FileSystem):
```