feat(core): error-resilience in workflow migrations (#2481)

Author: mohammad-alisafaee

renku/core/management/migrations/m_0005__2_cwl.py

@@ -19,6 +19,7 @@
 
 import glob
 import os
+import traceback
 import uuid
 from collections import defaultdict
 from functools import cmp_to_key
@@ -58,10 +59,10 @@ def migrate(migration_context):
     """Migration function."""
     if MigrationType.WORKFLOWS not in migration_context.options.type:
         return
-    _migrate_old_workflows(migration_context.client)
+    _migrate_old_workflows(client=migration_context.client, strict=migration_context.options.strict)
 
 
-def _migrate_old_workflows(client):
+def _migrate_old_workflows(client, strict):
     """Migrates old cwl workflows to new jsonld format."""
 
     def sort_cwl_commits(e1, e2):
@@ -86,17 +87,25 @@ def sort_cwl_commits(e1, e2):
         communication.echo(f"Processing commit {n}/{len(cwl_paths)}", end="\r")
 
         cwl_file, commit = element
-        if not Path(cwl_file).exists():
-            continue
-        path = _migrate_cwl(client, cwl_file, commit)
-        os.remove(cwl_file)
-
-        client.repository.add(cwl_file, path)
 
-        if client.repository.is_dirty():
-            commit_msg = "renku migrate: committing migrated workflow"
-            committer = Actor(name=f"renku {__version__}", email=version_url)
-            client.repository.commit(commit_msg, committer=committer, no_verify=True)
+        try:
+            if not Path(cwl_file).exists():
+                continue
+
+            path = _migrate_cwl(client, cwl_file, commit)
+            os.remove(cwl_file)
+
+            client.repository.add(cwl_file, path)
+
+            if client.repository.is_dirty():
+                commit_msg = "renku migrate: committing migrated workflow"
+                committer = Actor(name=f"renku {__version__}", email=version_url)
+                client.repository.commit(commit_msg, committer=committer, no_verify=True)
+        except Exception:
+            if strict:
+                raise
+            communication.echo("")
+            communication.warn(f"Cannot process commit '{commit.hexsha}' - Exception: {traceback.format_exc()}")
 
 
 def _migrate_cwl(client, path, commit):

renku/core/management/migrations/m_0009__new_metadata_storage.py

@@ -183,7 +183,7 @@ def generate_new_metadata(
     n_commits = len(commits)
 
     for n, commit in enumerate(commits, start=1):
-        communication.echo(f"Processing commits {n}/{n_commits} {commit.hexsha}", end="\n")
+        communication.echo(f"Processing commits {n}/{n_commits} {commit.hexsha}", end="\r")
 
         # NOTE: Treat the last commit differently if it was done by this migration
         is_last_commit = committed and n == n_commits

renku/core/management/migrations/models/v9.py

@@ -316,7 +316,7 @@ def from_revision(cls, client, path, revision="HEAD", parent=None, find_previous
                             client, member_path, commit, parent=entity, find_previous=find_previous, **kwargs
                         )
                     )
-                except KeyError:
+                except errors.GitCommitNotFoundError:
                     pass
 
         else:

renku/core/metadata/repository.py

@@ -438,6 +438,19 @@ def get_object_hash(self, path: Union[Path, str], revision: str = None) -> Optio
                 # NOTE: If object does not exist anymore, hash-object doesn't work, fall back to rev-parse
                 revision = "HEAD"
 
+        def get_staged_directory_hash() -> Optional[str]:
+            if not os.path.isdir(absolute_path):
+                return
+
+            stashed_revision = self.run_git_command("stash", "create")
+            if not stashed_revision:
+                return
+
+            try:
+                return self.run_git_command("rev-parse", f"{stashed_revision}:{relative_path}")
+            except errors.GitCommandError:
+                return
+
         def get_object_hash_from_submodules() -> Optional[str]:
             for submodule in self.submodules:
                 try:
@@ -452,8 +465,13 @@ def get_object_hash_from_submodules() -> Optional[str]:
         try:
             return self.run_git_command("rev-parse", f"{revision}:{relative_path}")
         except errors.GitCommandError:
-            # NOTE: The file can be in a submodule or it was not there when the command ran but was there when workflows
-            # were migrated (this can happen only for Usage); the project might be broken too.
+            # NOTE: The file can be in a submodule or it can be a directory which is staged but not committed yet.
+            # It's also possible that the file was not there when the command ran but was there when workflows were
+            # migrated (this can happen only for Usage); the project might be broken too.
+            staged_directory_hash = get_staged_directory_hash()
+            if staged_directory_hash:
+                return staged_directory_hash
+
             return get_object_hash_from_submodules()
 
     def get_user(self) -> "Actor":

renku/core/utils/git.py

@@ -228,15 +228,12 @@ def get_directory_members(absolute_path: Path) -> List[Entity]:
 
             member_path = member.relative_to(repository.path)
 
-            try:
-                assert all(member_path != m.path for m in members)
+            assert all(member_path != m.path for m in members)
 
-                entity = get_entity_from_revision(repository, member_path, revision)
-                # NOTE: If a path is not found at a revision we assume that it didn't exist at that revision
-                if entity:
-                    members.append(entity)
-            except KeyError:
-                pass
+            entity = get_entity_from_revision(repository, member_path, revision)
+            # NOTE: If a path is not found at a revision we assume that it didn't exist at that revision
+            if entity:
+                members.append(entity)
 
         return members
 
@@ -246,12 +243,11 @@ def get_directory_members(absolute_path: Path) -> List[Entity]:
     if cached_entry:
         return cached_entry
 
-    # TODO: What checksum we get at "HEAD" if object is staged but not committed
+    # NOTE: For untracked directory the hash is None; make sure to stage them first before calling this function.
     checksum = repository.get_object_hash(revision=revision, path=path)
     # NOTE: If object was not found at a revision it's either removed or exists in a different revision; keep the
     # entity and use revision as checksum
     checksum = checksum or revision
-    # TODO: What would be checksum for a directory if it's not committed yet.
     id = Entity.generate_id(checksum=checksum, path=path)
 
     absolute_path = repository.path / path

setup.py

@@ -137,7 +137,7 @@ def run(self):
         "Jinja2>=2.11.3,<3.0.2",
         "renku-sphinx-theme>=0.2.0",
         "sphinx-rtd-theme>=0.5.0,<1.1",
-        "sphinxcontrib-spelling==7.*"
+        "sphinxcontrib-spelling==7.*",
     ],
     "runner": ["cwlref-runner==1.0"],
     "notebook": [],

tests/core/metadata/test_repository.py

@@ -198,3 +198,32 @@ def test_hash_deleted_objects(git_repository):
 
     with pytest.raises(errors.GitCommandError):
         Repository.hash_object("B")
+
+
+def test_hash_directories(git_repository):
+    """Test hashing tree objects."""
+    (git_repository.path / "X").mkdir()
+    (git_repository.path / "X" / "A").write_text("modified")
+
+    assert git_repository.get_object_hash("X", revision="HEAD") is None
+    assert git_repository.get_object_hash("X") is None
+
+    with pytest.raises(errors.GitCommandError):
+        Repository.hash_object("X")
+
+    # NOTE: When staging a directory then the hash can be calculated
+    git_repository.add("X")
+
+    directory_hash = git_repository.get_object_hash("X", revision="HEAD")
+
+    assert directory_hash is not None
+    assert directory_hash == git_repository.get_object_hash("X")
+
+    # NOTE: Hash of the committed directory is the same as the staged hash
+    git_repository.commit("Committed X")
+
+    assert directory_hash == git_repository.get_object_hash("X", revision="HEAD")
+    assert directory_hash == git_repository.get_object_hash("X")
+
+    with pytest.raises(errors.GitCommandError):
+        Repository.hash_object("X")