Add queue.dependencies-by-topic

This adds a pipeline queue setting to emulate the Gerrit behavior
of submitWholeTopic without needing to enable it site-wide in Gerrit.

Change-Id: Icb33a1e87d15229e6fb3aa1e4b1ad14a60623a29

6 files changed, 102 insertions, 14 deletions

diff --git a/zuul/configloader.py b/zuul/configloader.py
index e0cf88d12..161b29f25 100644
--- a/zuul/configloader.py
+++ b/zuul/configloader.py
@@ -1379,6 +1379,7 @@ class QueueParser:
         queue = {vs.Required('name'): str,
                  'per-branch': bool,
                  'allow-circular-dependencies': bool,
+                 'dependencies-by-topic': bool,
                  '_source_context': model.SourceContext,
                  '_start_mark': model.ZuulMark,
                  }
@@ -1390,7 +1391,12 @@ class QueueParser:
             conf['name'],
             conf.get('per-branch', False),
             conf.get('allow-circular-dependencies', False),
+            conf.get('dependencies-by-topic', False),
         )
+        if (queue.dependencies_by_topic and not
+            queue.allow_circular_dependencies):
+            raise Exception("The 'allow-circular-dependencies' setting must be"
+                            "enabled in order to use dependencies-by-topic")
         queue.source_context = conf.get('_source_context')
         queue.start_mark = conf.get('_start_mark')
         queue.freeze()
diff --git a/zuul/driver/gerrit/gerritmodel.py b/zuul/driver/gerrit/gerritmodel.py
index 1cdf77ccc..a7833fe3a 100644
--- a/zuul/driver/gerrit/gerritmodel.py
+++ b/zuul/driver/gerrit/gerritmodel.py
@@ -114,6 +114,7 @@ class GerritChange(Change):
         self.wip = data.get('wip', False)
         self.owner = data['owner'].get('username')
         self.message = data['commitMessage']
+        self.topic = data.get('topic')
 
         self.missing_labels = set()
         for sr in data.get('submitRecords', []):
@@ -189,6 +190,7 @@ class GerritChange(Change):
         self.wip = data.get('work_in_progress', False)
         self.owner = data['owner'].get('username')
         self.message = current_revision['commit']['message']
+        self.topic = data.get('topic')
 
 
 class GerritTriggerEvent(TriggerEvent):
diff --git a/zuul/driver/gerrit/gerritsource.py b/zuul/driver/gerrit/gerritsource.py
index 086323e3c..5e6b82f05 100644
--- a/zuul/driver/gerrit/gerritsource.py
+++ b/zuul/driver/gerrit/gerritsource.py
@@ -143,6 +143,36 @@ class GerritSource(BaseSource):
                 changes.append(change)
         return changes
 
+    def getChangesByTopic(self, topic, changes=None):
+        if not topic:
+            return []
+
+        if changes is None:
+            changes = {}
+
+        query = 'status:open topic:%s' % topic
+        results = self.connection.simpleQuery(query)
+        for result in results:
+            change_key = ChangeKey(self.connection.connection_name, None,
+                                   'GerritChange',
+                                   str(result.number),
+                                   str(result.current_patchset))
+            if change_key in changes:
+                continue
+
+            change = self.connection._getChange(change_key)
+            changes[change_key] = change
+
+        for change in changes.values():
+            for git_key in change.git_needs_changes:
+                if git_key in changes:
+                    continue
+                git_change = self.getChange(git_key)
+                if not git_change.topic or git_change.topic == topic:
+                    continue
+                self.getChangesByTopic(git_change.topic, changes)
+        return list(changes.values())
+
     def getCachedChanges(self):
         yield from self.connection._change_cache
 
diff --git a/zuul/manager/__init__.py b/zuul/manager/__init__.py
index 63614b4c4..fdf7d33bc 100644
--- a/zuul/manager/__init__.py
+++ b/zuul/manager/__init__.py
@@ -627,7 +627,7 @@ class PipelineManager(metaclass=ABCMeta):
                 return scc
         return []
 
-    def canProcessCycle(self, project):
+    def getQueueConfig(self, project):
         layout = self.pipeline.tenant.layout
         pipeline_queue_name = None
         project_queue_name = None
@@ -651,13 +651,23 @@ class PipelineManager(metaclass=ABCMeta):
         # project while project has precedence.
         queue_name = project_queue_name or pipeline_queue_name
         if queue_name is None:
+            return None
+
+        return layout.queues.get(queue_name)
+
+    def canProcessCycle(self, project):
+        queue_config = self.getQueueConfig(project)
+        if queue_config is None:
             return False
 
-        queue_config = layout.queues.get(queue_name)
-        return (
-            queue_config is not None and
-            queue_config.allow_circular_dependencies
-        )
+        return queue_config.allow_circular_dependencies
+
+    def useDependenciesByTopic(self, project):
+        queue_config = self.getQueueConfig(project)
+        if queue_config is None:
+            return False
+
+        return queue_config.dependencies_by_topic
 
     def canMergeCycle(self, bundle):
         """Check if the cycle still fulfills the pipeline's ready criteria."""
@@ -809,12 +819,24 @@ class PipelineManager(metaclass=ABCMeta):
                 log.debug("  Adding dependency: %s", dep)
                 dependencies.append(dep)
         new_commit_needs_changes = [d.cache_key for d in dependencies]
+
+        update_attrs = dict(commit_needs_changes=new_commit_needs_changes)
+
+        # Ask the source for any tenant-specific changes (this allows
+        # drivers to implement their own way of collecting deps):
+        source = self.sched.connections.getSource(
+            change.project.connection_name)
+        if self.useDependenciesByTopic(change.project):
+            log.debug("  Updating topic dependencies for %s", change)
+            new_topic_needs_changes = []
+            for dep in source.getChangesByTopic(change.topic):
+                if dep and (not dep.is_merged):
+                    log.debug("  Adding dependency: %s", dep)
+                    new_topic_needs_changes.append(dep.cache_key)
+            update_attrs['topic_needs_changes'] = new_topic_needs_changes
+
         if change.commit_needs_changes != new_commit_needs_changes:
-            source = self.sched.connections.getSource(
-                change.project.connection_name)
-            source.setChangeAttributes(
-                change,
-                commit_needs_changes=new_commit_needs_changes)
+            source.setChangeAttributes(change, **update_attrs)
 
     def provisionNodes(self, item):
         log = item.annotateLogger(self.log)
diff --git a/zuul/model.py b/zuul/model.py
index a9c0f3a86..3035fbcc8 100644
--- a/zuul/model.py
+++ b/zuul/model.py
@@ -5399,12 +5399,18 @@ class Change(Branch):
         # to Depends-On headers (all drivers):
         self.commit_needs_changes = None
 
+        # Needed changes by topic (all
+        # drivers in theory, but Gerrit only in practice for
+        # emulate-submit-whole-topic):
+        self.topic_needs_changes = None
+
         self.is_current_patchset = True
         self.can_merge = False
         self.is_merged = False
         self.failed_to_merge = False
         self.open = None
         self.owner = None
+        self.topic = None
 
         # This may be the commit message, or it may be a cover message
         # in the case of a PR.  Either way, it's the place where we
@@ -5428,6 +5434,7 @@ class Change(Branch):
             None if data.get("commit_needs_changes") is None
             else data.get("commit_needs_changes", [])
         )
+        self.topic_needs_changes = data.get("topic_needs_changes")
         self.is_current_patchset = data.get("is_current_patchset", True)
         self.can_merge = data.get("can_merge", False)
         self.is_merged = data.get("is_merged", False)
@@ -5448,6 +5455,7 @@ class Change(Branch):
             "compat_needs_changes": self.compat_needs_changes,
             "compat_needed_by_changes": self.git_needed_by_changes,
             "commit_needs_changes": self.commit_needs_changes,
+            "topic_needs_changes": self.topic_needs_changes,
             "is_current_patchset": self.is_current_patchset,
             "can_merge": self.can_merge,
             "is_merged": self.is_merged,
@@ -5478,8 +5486,9 @@ class Change(Branch):
     @property
     def needs_changes(self):
         commit_needs_changes = self.commit_needs_changes or []
+        topic_needs_changes = self.topic_needs_changes or []
         return (self.git_needs_changes + self.compat_needs_changes +
-                commit_needs_changes)
+                commit_needs_changes + topic_needs_changes)
 
     @property
     def needed_by_changes(self):
@@ -7265,11 +7274,13 @@ class Semaphore(ConfigObject):
 
 class Queue(ConfigObject):
     def __init__(self, name, per_branch=False,
-                 allow_circular_dependencies=False):
+                 allow_circular_dependencies=False,
+                 dependencies_by_topic=False):
         super().__init__()
         self.name = name
         self.per_branch = per_branch
         self.allow_circular_dependencies = allow_circular_dependencies
+        self.dependencies_by_topic = dependencies_by_topic
 
     def __ne__(self, other):
         return not self.__eq__(other)
@@ -7281,7 +7292,9 @@ class Queue(ConfigObject):
             self.name == other.name and
             self.per_branch == other.per_branch and
             self.allow_circular_dependencies ==
-            other.allow_circular_dependencies
+            other.allow_circular_dependencies and
+            self.dependencies_by_topic ==
+            other.dependencies_by_topic
         )
 
 
diff --git a/zuul/source/__init__.py b/zuul/source/__init__.py
index a66107eef..b1b15a6ef 100644
--- a/zuul/source/__init__.py
+++ b/zuul/source/__init__.py
@@ -126,8 +126,23 @@ class BaseSource(object, metaclass=abc.ABCMeta):
         search scope.
         """
 
+    def getChangesByTopic(self, topic):
+        """Return changes in the same topic.
+
+        This should return changes under the same topic, as well as
+        changes under the same topic of any git-dependent changes,
+        recursively.
+
+        This is only implemented by the Gerrit driver, however if
+        other systems have a similar "topic" functionality, it could
+        be added to other drivers.
+        """
+
+        return []
+
     @abc.abstractmethod
     def getProjectOpenChanges(self, project):
+
         """Get the open changes for a project."""
 
     @abc.abstractmethod