Reference change dependencies by key

In order to cache changes in Zookeeper we need to make change objects
JSON serializable. This means that we can no longer reference other
change objects directly. Instead we will use a cache key consisting of
the connection name and a connection specific cache key.

Those cache keys can be resolved by getting the source instance using
the connection name and then retrieving the change instance via the new
`getChangeByKey()` interface.

The pipeline manager provides a helper method for resolving a list of
cache keys. Cache keys that where resolved once are also cached by the
manager as long as the reference is needed by any change in the
pipeline. The cache will be cleaned up at the end of a run of the queue
processor.

Until we can run multiple schedulers the change cache in the pipeline
manager will avoid hitting Zookeeper every time we resolve a cache key.

Later on when we have the pipeline state in Zookeeper we probably want
to clear the change cache in the pipeline manager at the end of the
queue processor. This way we make sure the change is recent enough when
we start processing a pipeline.

Change-Id: I09845d65864edc0e54af5f24d3c7be8fe2f7a919

1 files changed, 18 insertions, 9 deletions

diff --git a/zuul/driver/gerrit/gerritconnection.py b/zuul/driver/gerrit/gerritconnection.py
index f7e509b34..8502940c7 100644
--- a/zuul/driver/gerrit/gerritconnection.py
+++ b/zuul/driver/gerrit/gerritconnection.py
@@ -41,7 +41,7 @@ from zuul.driver.gerrit.gcloudauth import GCloudAuth
 from zuul.driver.gerrit.gerritmodel import GerritChange, GerritTriggerEvent
 from zuul.driver.git.gitwatcher import GitWatcher
 from zuul.lib.logutil import get_annotated_logger
-from zuul.model import Ref, Tag, Branch, Project
+from zuul.model import Ref, Tag, Branch, Project, CacheStat
 from zuul.zk.event_queues import ConnectionEventQueue, EventReceiverElection
 
 # HTTP timeout in seconds
@@ -798,6 +798,8 @@ class GerritConnection(BaseConnection):
             change = GerritChange(None)
             change.number = number
             change.patchset = patchset
+        change.cache_stat = CacheStat((change.number, change.patchset),
+                                      None, None)
         self._change_cache.setdefault(change.number, {})
         self._change_cache[change.number][change.patchset] = change
         try:
@@ -902,8 +904,8 @@ class GerritConnection(BaseConnection):
             # already merged. So even if it is "ABANDONED", we should not
             # ignore it.
             if (not dep.is_merged) and dep not in needs_changes:
-                git_needs_changes.append(dep)
-                needs_changes.add(dep)
+                git_needs_changes.append(dep.cache_key)
+                needs_changes.add(dep.cache_key)
         change.git_needs_changes = git_needs_changes
 
         compat_needs_changes = []
@@ -914,8 +916,8 @@ class GerritConnection(BaseConnection):
             dep = self._getChange(dep_num, dep_ps, history=history,
                                   event=event)
             if dep.open and dep not in needs_changes:
-                compat_needs_changes.append(dep)
-                needs_changes.add(dep)
+                compat_needs_changes.append(dep.cache_key)
+                needs_changes.add(dep.cache_key)
         change.compat_needs_changes = compat_needs_changes
 
         needed_by_changes = set()
@@ -928,8 +930,8 @@ class GerritConnection(BaseConnection):
                                       event=event)
                 if (dep.open and dep.is_current_patchset and
                     dep not in needed_by_changes):
-                    git_needed_by_changes.append(dep)
-                    needed_by_changes.add(dep)
+                    git_needed_by_changes.append(dep.cache_key)
+                    needed_by_changes.add(dep.cache_key)
             except Exception:
                 log.exception("Failed to get git-needed change %s,%s",
                               dep_num, dep_ps)
@@ -952,13 +954,20 @@ class GerritConnection(BaseConnection):
                     event=event)
                 if (dep.open and dep.is_current_patchset
                     and dep not in needed_by_changes):
-                    compat_needed_by_changes.append(dep)
-                    needed_by_changes.add(dep)
+                    compat_needed_by_changes.append(dep.cache_key)
+                    needed_by_changes.add(dep.cache_key)
             except Exception:
                 log.exception("Failed to get commit-needed change %s,%s",
                               dep_num, dep_ps)
         change.compat_needed_by_changes = compat_needed_by_changes
 
+    def getChangeByKey(self, key):
+        try:
+            number, patchset = key
+            return self._change_cache[number][patchset]
+        except (KeyError, ValueError):
+            return None
+
     def isMerged(self, change, head=None):
         self.log.debug("Checking if change %s is merged" % change)
         if not change.number: