Merge "Reduce redundant Gerrit queries"

3 files changed, 27 insertions, 7 deletions

diff --git a/zuul/driver/gerrit/gerritconnection.py b/zuul/driver/gerrit/gerritconnection.py
index 1ec334915..2213293f8 100644
--- a/zuul/driver/gerrit/gerritconnection.py
+++ b/zuul/driver/gerrit/gerritconnection.py
@@ -103,10 +103,12 @@ class GerritChangeData(object):
     SSH = 1
     HTTP = 2
 
-    def __init__(self, fmt, data, related=None, files=None):
+    def __init__(self, fmt, data, related=None, files=None,
+                 zuul_query_ltime=None):
         self.format = fmt
         self.data = data
         self.files = files
+        self.zuul_query_ltime = zuul_query_ltime
 
         if fmt == self.SSH:
             self.parseSSH(data)
@@ -329,19 +331,20 @@ class GerritEventConnector(threading.Thread):
 
             self.connection.clearConnectionCacheOnBranchEvent(event)
 
-        self._getChange(event)
+        self._getChange(event, connection_event.zuul_event_ltime)
         self.connection.logEvent(event)
         self.connection.sched.addTriggerEvent(
             self.connection.driver_name, event
         )
 
-    def _getChange(self, event):
+    def _getChange(self, event, connection_event_ltime):
         # Grab the change if we are managing the project or if it exists in the
         # cache as it may be a dependency
         if event.change_number:
             refresh = True
             change_key = self.connection.source.getChangeKey(event)
-            if self.connection._change_cache.get(change_key) is None:
+            change = self.connection._change_cache.get(change_key)
+            if change is None:
                 refresh = False
                 for tenant in self.connection.sched.abide.tenants.values():
                     # TODO(fungi): it would be better to have some simple means
@@ -353,6 +356,13 @@ class GerritEventConnector(threading.Thread):
                             event.project_name))):
                         refresh = True
                         break
+            else:
+                # We have a cache entry for this change Get the
+                # query ltime for the cache entry; if it's after the
+                # event ltime, we don't need to refresh.
+                if (change.zuul_query_ltime and
+                    change.zuul_query_ltime > connection_event_ltime):
+                    refresh = False
 
             if refresh:
                 # Call _getChange for the side effect of updating the
@@ -1418,15 +1428,20 @@ class GerritConnection(ZKChangeCacheMixin, ZKBranchCacheMixin, BaseConnection):
 
     def queryChange(self, number, event=None):
         for attempt in range(3):
+            # Get a query ltime -- any events before this point should be
+            # included in our change data.
+            zuul_query_ltime = self.sched.zk_client.getCurrentLtime()
             try:
                 if self.session:
                     data, related, files = self.queryChangeHTTP(
                         number, event=event)
                     return GerritChangeData(GerritChangeData.HTTP,
-                                            data, related, files)
+                                            data, related, files,
+                                            zuul_query_ltime=zuul_query_ltime)
                 else:
                     data = self.queryChangeSSH(number, event=event)
-                    return GerritChangeData(GerritChangeData.SSH, data)
+                    return GerritChangeData(GerritChangeData.SSH, data,
+                                            zuul_query_ltime=zuul_query_ltime)
             except Exception:
                 if attempt >= 3:
                     raise
diff --git a/zuul/driver/gerrit/gerritmodel.py b/zuul/driver/gerrit/gerritmodel.py
index f0ec32f77..0ac3e7f9d 100644
--- a/zuul/driver/gerrit/gerritmodel.py
+++ b/zuul/driver/gerrit/gerritmodel.py
@@ -35,8 +35,10 @@ class GerritChange(Change):
         self.approvals = []
         self.missing_labels = set()
         self.commit = None
+        self.zuul_query_ltime = None
 
     def update(self, data, connection):
+        self.zuul_query_ltime = data.zuul_query_ltime
         if data.format == data.SSH:
             self.updateFromSSH(data.data, connection)
         else:
@@ -51,6 +53,7 @@ class GerritChange(Change):
             "approvals": self.approvals,
             "missing_labels": list(self.missing_labels),
             "commit": self.commit,
+            "zuul_query_ltime": self.zuul_query_ltime,
         })
         return d
 
@@ -62,6 +65,7 @@ class GerritChange(Change):
         self.approvals = data["approvals"]
         self.missing_labels = set(data["missing_labels"])
         self.commit = data.get("commit")
+        self.zuul_query_ltime = data.get("zuul_query_ltime")
 
     def updateFromSSH(self, data, connection):
         if self.patchset is None:
diff --git a/zuul/zk/event_queues.py b/zuul/zk/event_queues.py
index 52ffd582e..ad7529791 100644
--- a/zuul/zk/event_queues.py
+++ b/zuul/zk/event_queues.py
@@ -909,7 +909,7 @@ class ConnectionEventQueue(ZooKeeperEventQueue):
         self._put({'event_data': data})
 
     def __iter__(self):
-        for data, ack_ref, _ in self._iterEvents():
+        for data, ack_ref, zstat in self._iterEvents():
             if not data:
                 self.log.warning("Malformed event found: %s", data)
                 self._remove(ack_ref.path)
@@ -918,6 +918,7 @@ class ConnectionEventQueue(ZooKeeperEventQueue):
             event = model.ConnectionEvent.fromDict(
                 data.get('event_data', data))
             event.ack_ref = ack_ref
+            event.zuul_event_ltime = zstat.creation_transaction_id
             yield event