ENH: orchestrators(datalad-pair): Do a more targeted update

datalad-pair's fetch() fetches the job ref from the resource and then
relies on 'datalad update' to bring down changes from the remote.  If
the merged in branch doesn't contain the job ref, then we temporarily
switch to the job ref, get the outputs, and then let the caller know
that the changes weren't brought in.

This approach is problematic with submodules.  One issue is that the
head switch mentioned above does not check out submodules, and in
general DataLad doesn't yet offer a way to set/reload a dataset
hierarchy to a particular state.

Another issue is that update() selects a merge target from the
submodule remote that doesn't necessarily contain the registered
commit.  See DataLad's 94efa8637a (NF: update: Add option to merge in
recorded submodule commit, 2020-02-17) for more information.  In what
will be the 0.13 release, update() learned a --follow=parentds mode
that makes update() take the registered commit as the merge target
when updating submodules.  Use it.

However, this still doesn't deal with bringing a non-mainline job ref
into the top-level dataset because update() unconditionally merges in
a branch from the remote.  Try harder to bring in the job ref by doing
an explicit 'git merge' of the ref before the update() call.

Author: kyleam

reproman/support/jobs/orchestrators.py

@@ -35,6 +35,7 @@
 
 import reproman
 from reproman.dochelpers import borrowdoc
+from reproman.dochelpers import exc_str
 from reproman.utils import cached_property
 from reproman.utils import chpwd
 from reproman.utils import write_update
@@ -922,6 +923,8 @@ def fetch(self, on_remote_finish=None):
             will be passed two arguments, the resource and the failed subjobs
             (list of ints).
         """
+        from datalad.support.exceptions import CommandError as DCError
+
         lgr.info("Fetching results for %s", self.jobid)
         failed = self.get_failed_subjobs()
         resource_name = self.resource.name
@@ -930,10 +933,28 @@ def fetch(self, on_remote_finish=None):
                  resource_name)
         self.ds.repo.fetch(resource_name, "{0}:{0}".format(ref),
                            recurse_submodules="no")
-        self.ds.update(sibling=resource_name,
-                       merge=True, recursive=True)
-        lgr.info("Getting outputs from '%s'", resource_name)
-        with head_at(self.ds, ref):
+        failure = False
+        try:
+            self.ds.repo.merge(ref)
+        except DCError as exc:
+            lgr.warning(
+                "Failed to merge in changes from %s. "
+                "Check %s for merge conflicts. %s",
+                ref, self.ds.path, exc_str(exc))
+        else:
+            # Handle any subdataset updates. We could avoid this if we knew
+            # there were no subdataset changes, but it's probably simplest to
+            # just unconditionally call update().
+            for res in self.ds.update(
+                    sibling=resource_name,
+                    merge=True, follow="parentds", recursive=True,
+                    on_failure="ignore"):
+                if res["status"] == "error":
+                    # DataLad will log failure.
+                    failure = True
+
+        if not failure:
+            lgr.info("Getting outputs from '%s'", resource_name)
             outputs = list(self.get_outputs())
             if outputs:
                 self.ds.get(path=outputs)
@@ -944,11 +965,6 @@ def fetch(self, on_remote_finish=None):
         lgr.info("Finished with remote resource '%s'", resource_name)
         if on_remote_finish:
             on_remote_finish(self.resource, failed)
-        if not self.ds.repo.is_ancestor(ref, "HEAD"):
-            lgr.info("Results stored on %s. "
-                     "Bring them into this branch with "
-                     "'git merge %s'",
-                     ref, ref)
 
 
 class FetchDataladRunMixin(object):

reproman/support/jobs/tests/test_orchestrators.py

@@ -19,6 +19,7 @@
 from reproman.consts import TEST_SSH_DOCKER_DIGEST
 from reproman.utils import chpwd
 from reproman.utils import swallow_logs
+from reproman.utils import swallow_outputs
 from reproman.resource.shell import Shell
 from reproman.support.exceptions import MissingExternalDependency
 from reproman.support.exceptions import OrchestratorError
@@ -665,3 +666,98 @@ def test_orc_datalad_pair_submodule(job_spec, dataset, shell):
         orc.submit()
         orc.follow()
         orc.fetch()
+
+
+def test_orc_datalad_pair_need_follow_parent(job_spec, dataset, shell):
+    # An example of a scenario that fails without DataLad's --follow=parentds
+    with chpwd(dataset.path):
+        dataset.create("sub")
+        dataset.save()
+
+        job_spec["_resolved_command_str"] = "sh -c 'echo baz >baz'"
+        job_spec["inputs"] = []
+        job_spec["outputs"] = []
+
+        orc0 = orcs.DataladPairOrchestrator(
+            shell, submission_type="local", job_spec=job_spec)
+        orc0.prepare_remote()
+        orc0.submit()
+        orc0.follow()
+
+        job_spec["_resolved_command_str"] = "sh -c 'echo bar >sub/bar'"
+        output = op.join("sub", "bar")
+        job_spec["outputs"] = [output]
+        orc1 = orcs.DataladPairOrchestrator(
+            shell, submission_type="local", job_spec=job_spec)
+        orc1.prepare_remote()
+        orc1.submit()
+        orc1.follow()
+        orc1.fetch()
+        assert op.exists(output)
+
+
+def test_orc_datalad_pair_submodule_conflict(caplog, job_spec, dataset, shell):
+    # In this scenario, one job modifies a submodule, and before that change,
+    # another job is launched that modifies the same submodule. This creates a
+    # change that can't be brought in with `datalad update` because, even with
+    # --follow=parentds, the top-level repo still brings in changes from the
+    # remote, whose master branch points to the first job. In a diagram, the
+    # remote state is:
+    #
+    #         ---- job 1 (master)
+    #  base --|
+    #         ---- job 2 (detached)
+    #
+    # On fetch of job 2, we merge the job 2 ref. The `datalad update` call
+    # fails trying to merge in master.
+    #
+    # If this scenario ends up being common enough, we could consider modifying
+    # `datalad update` to optionally not try to merge the remote state of the
+    # top-level repo.
+    with chpwd(dataset.path):
+        dataset.create("sub")
+        dataset.save()
+
+        job_spec["_resolved_command_str"] = "sh -c 'echo baz >sub/baz'"
+        job_spec["inputs"] = []
+        job_spec["outputs"] = []
+
+        orc0 = orcs.DataladPairOrchestrator(
+            shell, submission_type="local", job_spec=job_spec)
+        orc0.prepare_remote()
+        orc0.submit()
+        orc0.follow()
+
+        job_spec["_resolved_command_str"] = "sh -c 'echo bar >sub/bar'"
+        orc1 = orcs.DataladPairOrchestrator(
+            shell, submission_type="local", job_spec=job_spec)
+        orc1.prepare_remote()
+        orc1.submit()
+        orc1.follow()
+        # swallow_logs() won't work here because it hard codes the logger and
+        # the log message being checked is bubbled up by DataLad.
+        caplog.clear()
+        with caplog.at_level(logging.ERROR):
+            orc1.fetch()
+        assert "CONFLICT" in caplog.text
+        assert dataset.repo.call_git(["ls-files", "--unmerged"]).strip()
+
+
+def test_orc_datalad_pair_merge_conflict(job_spec, dataset, shell):
+    with chpwd(dataset.path):
+        job_spec["_resolved_command_str"] = "sh -c 'echo baz >baz'"
+        job_spec["inputs"] = []
+        job_spec["outputs"] = []
+
+        orc0 = orcs.DataladPairOrchestrator(
+            shell, submission_type="local", job_spec=job_spec)
+        orc0.prepare_remote()
+        orc0.submit()
+        orc0.follow()
+        # Introduce a conflict.
+        (dataset.pathobj / "baz").write_text("different")
+        dataset.save()
+        with swallow_logs(new_level=logging.WARNING) as logs:
+            orc0.fetch()
+            assert "Failed to merge in changes" in logs.out
+        assert dataset.repo.call_git(["ls-files", "--unmerged"]).strip()