Reduce redundant Gerrit queries

Sometimes Gerrit events may arrive in batches (for example, an
automated process modifies several related changes nearly
simultaneously). Because of our inbuilt delay (10 seconds by
default), it's possible that in these cases, many or all of the
updates represented by these events will have settled on the Gerrit
server before we even start processing the first event.  In these
cases, we don't need to query the same changes multiple times.

Take for example a stack of 10 changes.  Someone approves all 10
simultaneously.  That would produce (at least) 10 events for Zuul
to process.  Each event would cause Zuul to query all 10 changes in
the series (since they are related).  That's 100 change queries
(and each change query requires 2 or 3 HTTP queries).

But if we know that all the event arrived before our first set of
change queries, we can reduce that set of 100 queries to 10 by
suppressing any queries after the first.

This change generates a logical timestamp (ltime) immediately
before querying Gerrit for a change, and stores that ltime in the
change cache.  Whenever an event arrives for processing with an
ltime later than the query ltime, we assume the change is already
up to date with that event and do not perform another query.

Change-Id: Ib1b9245cc84ab3f5a0624697f4e3fc73bc8e03bd

1 files changed, 56 insertions, 0 deletions

diff --git a/tests/unit/test_gerrit.py b/tests/unit/test_gerrit.py
index f0f9027bd..0f129ae7a 100644
--- a/tests/unit/test_gerrit.py
+++ b/tests/unit/test_gerrit.py
@@ -13,6 +13,7 @@
 # under the License.
 
 import os
+import threading
 import textwrap
 from unittest import mock
 
@@ -867,3 +868,58 @@ class TestGerritFake(ZuulTestCase):
         # The Gerrit connection method filters out the queried change
         ret = self.fake_gerrit._getSubmittedTogether(C1, None)
         self.assertEqual(ret, [(4, 1)])
+
+
+class TestGerritConnection(ZuulTestCase):
+    config_file = 'zuul-gerrit-web.conf'
+    tenant_config_file = 'config/single-tenant/main.yaml'
+
+    def test_zuul_query_ltime(self):
+        # Add a lock around the event queue iterator so that we can
+        # ensure that multiple events arrive before the first is
+        # processed.
+        lock = threading.Lock()
+
+        orig_iterEvents = self.fake_gerrit.gerrit_event_connector.\
+            event_queue._iterEvents
+
+        def _iterEvents(*args, **kw):
+            with lock:
+                return orig_iterEvents(*args, **kw)
+
+        self.patch(self.fake_gerrit.gerrit_event_connector.event_queue,
+                   '_iterEvents', _iterEvents)
+
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
+        B = self.fake_gerrit.addFakeChange('org/project', 'master', 'B')
+        B.setDependsOn(A, 1)
+        # Hold the connection queue processing so these events get
+        # processed together
+        with lock:
+            self.fake_gerrit.addEvent(A.addApproval('Code-Review', 2))
+            self.fake_gerrit.addEvent(B.addApproval('Approved', 1))
+            self.fake_gerrit.addEvent(B.addApproval('Code-Review', 2))
+        self.waitUntilSettled()
+        self.assertHistory([])
+        # One query for each change in the above cluster of events.
+        self.assertEqual(A.queried, 1)
+        self.assertEqual(B.queried, 1)
+        self.fake_gerrit.addEvent(A.addApproval('Approved', 1))
+        self.waitUntilSettled()
+        self.assertHistory([
+            dict(name="project-merge", result="SUCCESS", changes="1,1"),
+            dict(name="project-test1", result="SUCCESS", changes="1,1"),
+            dict(name="project-test2", result="SUCCESS", changes="1,1"),
+            dict(name="project-merge", result="SUCCESS", changes="1,1 2,1"),
+            dict(name="project-test1", result="SUCCESS", changes="1,1 2,1"),
+            dict(name="project-test2", result="SUCCESS", changes="1,1 2,1"),
+        ], ordered=False)
+        # One query due to the event on change A, followed by a query
+        # to verify the merge.
+        self.assertEqual(A.queried, 3)
+        # No query for change B necessary since our cache is up to
+        # date with respect for the triggering event.  One query to
+        # verify the merge.
+        self.assertEqual(B.queried, 2)
+        self.assertEqual(A.data['status'], 'MERGED')
+        self.assertEqual(B.data['status'], 'MERGED')