add: warn on cache link errors instead of failing (#4801)

* add: warn on cache link failure instead of error

* add test case

* fix linted string

Co-authored-by: Paweł Redzyński <[email protected]>

Co-authored-by: Paweł Redzyński <[email protected]>

dvc/cache/base.py

@@ -8,6 +8,7 @@ from shortuuid import uuid
 
 import dvc.prompt as prompt
 from dvc.exceptions import (
+    CacheLinkError,
     CheckoutError,
     ConfirmRemoveError,
     DvcException,
@@ -192,7 +193,7 @@ class CloudCache:
                 )
                 del link_types[0]
 
-        raise DvcException("no possible cache types left to try out.")
+        raise CacheLinkError([to_info])
 
     def _do_link(self, from_info, to_info, link_method):
         if self.tree.exists(to_info):

dvc/exceptions.py

@@ -338,3 +338,18 @@ class NoOutputOrStageError(DvcException):
 
 class MergeError(DvcException):
     pass
+
+
+class CacheLinkError(DvcException):
+    SUPPORT_LINK = "See {} for more information.".format(
+        format_link(
+            "https://dvc.org/doc/user-guide/troubleshooting#cache-types"
+        )
+    )
+
+    def __init__(self, path_infos):
+        msg = "No possible cache link types for '{}'. {}".format(
+            ", ".join([str(path) for path in path_infos]), self.SUPPORT_LINK,
+        )
+        super().__init__(msg)
+        self.path_infos = path_infos

dvc/repo/add.py

@@ -6,6 +6,7 @@ import colorama
 from dvc.dvcfile import Dvcfile, is_dvc_file
 
 from ..exceptions import (
+    CacheLinkError,
     OutputDuplicationError,
     OverlappingOutputPathsError,
     RecursiveAddingWhileUsingFilename,
@@ -30,6 +31,7 @@ def add(
     if isinstance(targets, str):
         targets = [targets]
 
+    link_failures = []
     stages_list = []
     num_targets = len(targets)
     with Tqdm(total=num_targets, desc="Add", unit="file", leave=True) as pbar:
@@ -79,33 +81,56 @@ def add(
                     exc.output, set(exc.stages) - set(stages)
                 )
 
-            with Tqdm(
-                total=len(stages),
-                desc="Processing",
-                unit="file",
-                disable=len(stages) == 1,
-            ) as pbar_stages:
-                for stage in stages:
-                    try:
-                        stage.save()
-                    except OutputDoesNotExistError:
-                        pbar.n -= 1
-                        raise
-
-                    if not no_commit:
-                        stage.commit()
-
-                    Dvcfile(repo, stage.path).dump(stage)
-                    pbar_stages.update()
-
+            link_failures.extend(
+                _process_stages(repo, stages, no_commit, pbar)
+            )
             stages_list += stages
 
         if num_targets == 1:  # restore bar format for stats
             pbar.bar_format = pbar.BAR_FMT_DEFAULT
 
+    if link_failures:
+        msg = (
+            "Some targets could not be linked from cache to workspace.\n{}\n"
+            "To re-link these targets, reconfigure cache types and then run:\n"
+            "\n\tdvc checkout {}"
+        ).format(
+            CacheLinkError.SUPPORT_LINK,
+            " ".join([str(stage.relpath) for stage in link_failures]),
+        )
+        logger.warning(msg)
+
     return stages_list
 
 
+def _process_stages(repo, stages, no_commit, pbar):
+    link_failures = []
+
+    with Tqdm(
+        total=len(stages),
+        desc="Processing",
+        unit="file",
+        disable=len(stages) == 1,
+    ) as pbar_stages:
+        for stage in stages:
+            try:
+                stage.save()
+            except OutputDoesNotExistError:
+                pbar.n -= 1
+                raise
+
+            try:
+                if not no_commit:
+                    stage.commit()
+            except CacheLinkError:
+                link_failures.append(stage)
+
+            Dvcfile(repo, stage.path).dump(stage)
+            pbar_stages.update()
+
+    return link_failures
+
+
 def _find_all_targets(repo, target, recursive):
     if os.path.isdir(target) and recursive:
         return [

dvc/stage/__init__.py

@@ -7,7 +7,12 @@ from funcy import cached_property, project
 
 import dvc.dependency as dependency
 import dvc.prompt as prompt
-from dvc.exceptions import CheckoutError, DvcException, MergeError
+from dvc.exceptions import (
+    CacheLinkError,
+    CheckoutError,
+    DvcException,
+    MergeError,
+)
 from dvc.output.base import OutputDoesNotExistError
 from dvc.utils import relpath
 
@@ -422,12 +427,17 @@ class Stage(params.StageParams):
 
     @rwlocked(write=["outs"])
     def commit(self, allow_missing=False):
+        link_failures = []
         for out in self.outs:
             try:
                 out.commit()
             except OutputDoesNotExistError:
                 if not allow_missing:
                     raise
+            except CacheLinkError:
+                link_failures.append(out.path_info)
+        if link_failures:
+            raise CacheLinkError(link_failures)
 
     @rwlocked(read=["deps"], write=["outs"])
     def run(

tests/func/test_add.py

@@ -820,3 +820,20 @@ def test_add_file_in_symlink_dir(make_tmp_dir, tmp_dir, dvc, external):
 
     with pytest.raises(DvcException):
         dvc.add(os.path.join("dir", "foo"))
+
+
+def test_add_with_cache_link_error(tmp_dir, dvc, mocker, caplog):
+    tmp_dir.gen("foo", "foo")
+
+    mocker.patch.object(
+        dvc.cache.local, "_do_link", side_effect=DvcException("link failed")
+    )
+    with caplog.at_level(logging.WARNING, logger="dvc"):
+        dvc.add("foo")
+        assert "reconfigure cache types" in caplog.text
+
+    assert not (tmp_dir / "foo").exists()
+    assert (tmp_dir / "foo.dvc").exists()
+    assert (tmp_dir / ".dvc" / "cache").read_text() == {
+        "ac": {"bd18db4cc2f85cedef654fccc4a4d8": "foo"}
+    }