Store initial repo state in the merger

When we ask a merger to speculatively merge changes, record the
complete starting state of each repo (defined as all of the refs
other than Zuul refs) and return that at the completion of all
of the merges.

This will later be used so that when a pipeline manager asks a
merger to speculatively merge a change, the process can later
be repeated by the (potentially multiple) executors which will
end up running jobs for the change.  Between the time that the
merger runs and the jobs run, the underlying repos may have changed.
This ensures a consistent state throughout.

The facility which used saved zuul refs within the merger repo
to short-cut the merge sequence for an additional change added to
a previously completed merge sequence is removed, because in that
case, we would not be able to know the original repo state for the
earlier merge sequence.  This is slightly less efficient, however,
we are proposing removing zuul refs anyway due to the maintenance
burden they cause.

Change-Id: If0215d53c3b08877ded7276955a55fc5e617b244

zuul/executor/server.py

@@ -484,16 +484,14 @@ class ExecutorServer(object):
 
     def merge(self, job):
         args = json.loads(job.arguments)
-        ret = self.merger.mergeChanges(args['items'], args.get('files'))
+        ret = self.merger.mergeChanges(args['items'], args.get('files'),
+                                       args.get('repo_state'))
         result = dict(merged=(ret is not None),
                       zuul_url=self.zuul_url)
-        if args.get('files'):
+        if ret is None:
-            if ret:
+            result['commit'] = result['files'] = result['repo_state'] = None
-                result['commit'], result['files'] = ret
-            else:
-                result['commit'], result['files'] = (None, None)
         else:
-            result['commit'] = ret
+            result['commit'], result['files'], result['repo_state'] = ret
         job.sendWorkComplete(json.dumps(result))
 
 
@@ -588,13 +586,10 @@ class AnsibleJob(object):
 
         merge_items = [i for i in args['items'] if i.get('refspec')]
         if merge_items:
-            commit = self.doMergeChanges(merge_items)
+            if not self.doMergeChanges(merge_items):
-            if not commit:
                 # There was a merge conflict and we have already sent
                 # a work complete result, don't run any jobs
                 return
-        else:
-            commit = args['items'][-1]['newrev']  # noqa
 
         # Delete the origin remote from each repo we set up since
         # it will not be valid within the jobs.
@@ -640,11 +635,12 @@ class AnsibleJob(object):
     def doMergeChanges(self, items):
         # Get a merger in order to update the repos involved in this job.
         merger = self.executor_server._getMerger(self.jobdir.src_root)
-        commit = merger.mergeChanges(items)  # noqa
+        ret = merger.mergeChanges(items)  # noqa
-        if not commit:  # merge conflict
+        if not ret:  # merge conflict
             result = dict(result='MERGER_FAILURE')
             self.job.sendWorkComplete(json.dumps(result))
-        return commit
+            return False
+        return True
 
     def runPlaybooks(self, args):
         result = None

zuul/manager/__init__.py

@@ -509,9 +509,8 @@ class PipelineManager(object):
         build_set = item.current_build_set
         build_set.merge_state = build_set.PENDING
         self.sched.merger.mergeChanges(merger_items,
-                                       item.current_build_set,
+                                       item.current_build_set, files,
-                                       files,
+                                       precedence=self.pipeline.precedence)
-                                       self.pipeline.precedence)
         return False
 
     def prepareItem(self, item):

zuul/merger/client.py

@@ -107,10 +107,11 @@ class MergeClient(object):
                                timeout=300)
         return job
 
-    def mergeChanges(self, items, build_set, files=None,
+    def mergeChanges(self, items, build_set, files=None, repo_state=None,
                      precedence=zuul.model.PRECEDENCE_NORMAL):
         data = dict(items=items,
-                    files=files)
+                    files=files,
+                    repo_state=repo_state)
         self.submitJob('merger:merge', data, build_set, precedence)
 
     def getFiles(self, connection_name, project_name, branch, files,

zuul/merger/merger.py

@@ -124,6 +124,10 @@ class Repo(object):
         ref = repo.refs[refname]
         return ref.commit
 
+    def getRefs(self):
+        repo = self.createRepoObject()
+        return repo.refs
+
     def checkout(self, ref):
         repo = self.createRepoObject()
         self.log.debug("Checking out %s" % ref)
@@ -285,6 +289,18 @@ class Merger(object):
             raise Exception("Project %s/%s does not have branch %s" %
                             (connection_name, project_name, branch))
 
+    def _saveRepoState(self, connection_name, project_name, repo,
+                       repo_state):
+        projects = repo_state.setdefault(connection_name, {})
+        project = projects.setdefault(project_name, {})
+        if project:
+            # We already have a state for this project.
+            return
+        for ref in repo.getRefs():
+            if ref.path.startswith('refs/zuul'):
+                continue
+            project[ref.path] = ref.object.hexsha
+
     def _mergeChange(self, item, ref):
         repo = self.getRepo(item['connection'], item['project'])
         try:
@@ -314,27 +330,13 @@ class Merger(object):
 
         return commit
 
-    def _mergeItem(self, item, recent):
+    def _mergeItem(self, item, recent, repo_state):
         self.log.debug("Processing refspec %s for project %s/%s / %s ref %s" %
                        (item['refspec'], item['connection'],
                         item['project'], item['branch'], item['ref']))
         repo = self.getRepo(item['connection'], item['project'])
         key = (item['connection'], item['project'], item['branch'])
 
-        # See if we have a commit for this change already in this repo
-        zuul_ref = item['branch'] + '/' + item['ref']
-        with repo.createRepoObject().git.custom_environment(
-            GIT_SSH_COMMAND=self._get_ssh_cmd(item['connection'])):
-            commit = repo.getCommitFromRef(zuul_ref)
-            if commit:
-                self.log.debug(
-                    "Found commit %s for ref %s" % (commit, zuul_ref))
-                # Store this as the most recent commit for this
-                # project-branch
-                recent[key] = commit
-                return commit
-
-        self.log.debug("Unable to find commit for ref %s" % (zuul_ref,))
         # We need to merge the change
         # Get the most recent commit for this project-branch
         base = recent.get(key)
@@ -348,6 +350,10 @@ class Merger(object):
                 self.log.exception("Unable to reset repo %s" % repo)
                 return None
             base = repo.getBranchHead(item['branch'])
+            # Save the repo state so that later mergers can repeat
+            # this process.
+            self._saveRepoState(item['connection'], item['project'], repo,
+                                repo_state)
         else:
             self.log.debug("Found base commit %s for %s" % (base, key,))
         # Merge the change
@@ -365,17 +371,22 @@ class Merger(object):
                 try:
                     repo = self.getRepo(connection, project)
                     zuul_ref = branch + '/' + item['ref']
-                    repo.createZuulRef(zuul_ref, mrc)
+                    if not repo.getCommitFromRef(zuul_ref):
+                        repo.createZuulRef(zuul_ref, mrc)
                 except Exception:
                     self.log.exception("Unable to set zuul ref %s for "
                                        "item %s" % (zuul_ref, item))
                     return None
             return commit
 
-    def mergeChanges(self, items, files=None):
+    def mergeChanges(self, items, files=None, repo_state=None):
+        # connection+project+branch -> commit
         recent = {}
         commit = None
         read_files = []
+        # connection -> project -> ref -> commit
+        if repo_state is None:
+            repo_state = {}
         for item in items:
             if item.get("number") and item.get("patchset"):
                 self.log.debug("Merging for change %s,%s." %
@@ -383,7 +394,7 @@ class Merger(object):
             elif item.get("newrev") and item.get("oldrev"):
                 self.log.debug("Merging for rev %s with oldrev %s." %
                                (item["newrev"], item["oldrev"]))
-            commit = self._mergeItem(item, recent)
+            commit = self._mergeItem(item, recent, repo_state)
             if not commit:
                 return None
             if files:
@@ -394,9 +405,7 @@ class Merger(object):
                     project=item['project'],
                     branch=item['branch'],
                     files=repo_files))
-        if files:
+        return commit.hexsha, read_files, repo_state
-            return commit.hexsha, read_files
-        return commit.hexsha
 
     def getFiles(self, connection_name, project_name, branch, files):
         repo = self.getRepo(connection_name, project_name)

zuul/merger/server.py

@@ -103,16 +103,14 @@ class MergeServer(object):
 
     def merge(self, job):
         args = json.loads(job.arguments)
-        ret = self.merger.mergeChanges(args['items'], args.get('files'))
+        ret = self.merger.mergeChanges(args['items'], args.get('files'),
+                                       args.get('repo_state'))
         result = dict(merged=(ret is not None),
                       zuul_url=self.zuul_url)
-        if args.get('files'):
+        if ret is None:
-            if ret:
+            result['commit'] = result['files'] = result['repo_state'] = None
-                result['commit'], result['files'] = ret
-            else:
-                result['commit'], result['files'] = (None, None)
         else:
-            result['commit'] = ret
+            result['commit'], result['files'], result['repo_state'] = ret
         job.sendWorkComplete(json.dumps(result))
 
     def cat(self, job):