zuul/zuul
Code Issues Proposed changes

Do not update gerrit change cache with old data

Browse Source 
An issue was observed with a change series in gerrit when using
cherry-pick.  The sequence of events on two schedulers was:

[2] tell gerrit to merge 1,1
[1] receive a comment-added event for 1,1
[1] perform a gerrit query for 1,1 and receive state=NEW
[2] perform the isMerged query for 1,1 and receive state=MERGED
[2] update 1,1 in the change cache with is_merged=True
[1] update 1,1 in the change cache with is_merged=False

This race requires that scheduler #1 perform a gerrit query and store
the information locally, while scheduler #2 merges the change and
performs a query on the same change and saves it to the change cache,
all before scheduler #1 actually saves its data to the change cache.
This means that scheduler #1 writes old data to the change cache and
flips the is_merged flag back to false.

This can occur easily if there are multiple related changes.  In this
case, scheduler #1 may spend quite some time querying other changes
while scheduler #2 only has to query a single change.

Once queue processing resumes after the isMerged query completes,
scheduler #2 will remove the change at the head, then consider the
next change.  It will see a change that depends on the first change,
and it will see that is_merged on the first change is false, therefore
it will remove the dependency.

So far, we have described a sequence of events that could happen with
any merge strategy, so why does this only appear to be a problem with
cherry-pick?  That's because right after the comment-added event is
sent, gerrit sends the change-merged event.  In most merge-modes, that
will cause us to query the change a third time and then set the
is_merged flag back to true.  So there is only a very small window
where this race condition can have an effect with most merge modes.

But with the cherry-pick mode, the change-merged event is emitted for
a new patchset, and it is the change cache entry for this new patchset
where is_merged is set to true.  The original patchset is left with
is_merged=false.

To correct this, we update the gerrit driver so that when it is about
to update the change cache with data from a query, it compares the
ltime of the query data to the ltime already set on the change object.
If another scheduler has updated the change cache and that has
propagated to our change object, and our data are older than what is
on the change object, we will skip the update.

The change cache itself uses the zookeeper object version system to
ensure that the local change object has the most up-to-date data
before writing it to zk.  The check described above will work together
with this to ensure that, first, if any other scheduler has written
the change to zk, we have that data and the query ltime for it; then
second, that our query ltime is newer than that before we update our
data.

Change-Id: Ic206a93ab79242d2cd9d76c006bed67b1f669b31
This commit is contained in:
James E. Blair
2025-05-31 16:33:43 -07:00
parent dab7b06fc0
commit 06e8c5daba
2 changed files with 170 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
tests/unit/test_sos.py
View File

@ -25,6 +25,7 @@ from tests.base import (
okay_tracebacks,
simple_layout,
)
from zuul.zk.change_cache import ChangeKey
from zuul.zk.locks import SessionAwareWriteLock, TENANT_LOCK_ROOT
from zuul.scheduler import PendingReconfiguration
@ -456,6 +457,161 @@ class TestScaleOutScheduler(ZuulTestCase):
dict(name='project-test2', result='SUCCESS', changes='1,1 2,1'),
], ordered=False)
@simple_layout('layouts/cherry-pick.yaml')
def test_change_cache_ismerged(self):
# Test that we don't bump a depending change from the pipeline
# due to a race with isMerged.
self.executor_server.hold_jobs_in_build = True
A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
A.cherry_pick = True
B = self.fake_gerrit.addFakeChange('org/project', 'master', 'B')
B.cherry_pick = True
B.setDependsOn(A, 1)
A.addApproval('Code-Review', 2)
B.addApproval('Code-Review', 2)
self.fake_gerrit.addEvent(A.addApproval('Approved', 1))
self.waitUntilSettled("Change A enqueued")
self.fake_gerrit.addEvent(B.addApproval('Approved', 1))
self.waitUntilSettled("Change B enqueued")
app = self.createScheduler()
app.start()
self.assertEqual(len(self.scheds), 2)
first = self.scheds.first
second = app
# Wait until they're both ready
self.waitUntilSettled("Second scheduler ready")
# Here's the sequence we need to happen (scheduler id in []):
# [2] tell gerrit to merge 1,1
# [1] receive a comment-added event for 1,1
# [1] perform a gerrit query for 1,1 and receive state=NEW
# [1] stop processing right before updating the change cache
# [2] perform the isMerged query for 1,1 and receive state=MERGED
# [2] update 1,1 in the change cache with is_merged=True
# [1] resume processing the change cache update
# Then we make sure is_merged does not flip back to false
# These controls whether scheduler 1 can write updates to the
# change cache.
update_permitted_event = threading.Event()
update_permitted_event.set()
waiting_to_update_event = threading.Event()
# These allow us to stop scheduler 2 between the two queue
# items.
queue_item_finished = threading.Event()
queue_item_resume = threading.Event()
queue_item_resume.set()
# Insert our control into the scheduler 1 change cache
s1_cache = first.sched.connections.connections['gerrit']._change_cache
orig_s1_set = s1_cache.updateChangeWithRetry
def s1_set(*args, **kw):
if not update_permitted_event.isSet():
# We have performed the comment-added query and are
# about to write the results.
self.log.debug("Waiting to update change cache")
# Tell scheduler 2 that we're waiting; we pause so
# that it can proceed with its correct data.
waiting_to_update_event.set()
update_permitted_event.wait()
# Now we're going to write old data.
self.log.debug("Resuming update to change cache")
return orig_s1_set(*args, **kw)
s1_cache.updateChangeWithRetry = s1_set
# Gain control of the review method.
s2_conn = second.sched.connections.connections['gerrit']
orig_s2_review = s2_conn.review
def s2_review(*args, **kw):
self.log.debug("Review start")
# We don't normally automatically emit any post-merge
# events in tests in order to keep them simple. But we
# need to in this case so that we can trigger a change
# query that races the isMerged query. Emit the
# comment-added event which is the first thing that gerrit
# emits as we merge a change. Normally a change-merged
# event would follow, but is not necessary for this test.
self.fake_gerrit.addEvent(A.getChangeCommentEvent(1))
# Wait until query is done
self.log.debug("Wait for change-merged query to finish")
waiting_to_update_event.wait()
# We know the change-merged query has completed with state=NEW
self.log.debug("Review resume")
# Allow the report to happen, which will cause the state
# to change to MERGED for subsequent queries.
ret = orig_s2_review(*args, **kw)
self.log.debug("Review done")
return ret
s2_conn.review = s2_review
# Gain control of the pipeline manager so that we can pause
# between processing queue items.
s2_gate = second.sched.abide.tenants[
'tenant-one'].layout.pipeline_managers['gate']
orig_s2_processOneItem = s2_gate._processOneItem
def s2_processOneItem(*args, **kw):
ret = orig_s2_processOneItem(*args, **kw)
self.log.debug("Finished queue item")
queue_item_finished.set()
self.log.debug("Wait for queue item")
queue_item_resume.wait()
return ret
s2_gate._processOneItem = s2_processOneItem
# Hold the lock on the first scheduler, so the queue
# processing happens on the second.
with first.sched.run_handler_lock:
key = ChangeKey('gerrit', None, 'GerritChange', '1', '1')
c1 = s1_cache.get(key)
c2 = s1_cache.get(key)
# Clear events for the start of the interesting part of
# the test.
queue_item_finished.clear()
queue_item_resume.clear()
update_permitted_event.clear()
self.log.debug("Release held build")
self.builds[0].release()
self.log.debug("Wait for first queue item to finish")
queue_item_finished.wait()
self.log.debug("First queue item is finished")
# Allow the delayed update to take effect (with the old,
# pre-merge data) before we process the next queue item.
old_uuid = c2.cache_stat.uuid
update_permitted_event.set()
# We wait for the caches to sync to make sure that we
# don't trip the conflict detection in the change cache.
# If we do, then the change cache will re-query the data.
self.log.debug("Wait for caches to sync")
for x in iterate_timeout(30, 'caches to sync'):
if (c2.cache_stat.uuid != old_uuid and
c1.cache_stat.uuid == c2.cache_stat.uuid):
break
# Allow the next item to process
self.log.debug("Resume processing second queue item")
queue_item_resume.set()
self.waitUntilSettled("Merge change A", matcher=[second])
# Release everything now
self.executor_server.hold_jobs_in_build = False
self.executor_server.release()
self.waitUntilSettled("End")
self.assertHistory([
dict(name='check-job', result='SUCCESS', changes='1,1'),
dict(name='check-job', result='SUCCESS', changes='1,1 2,1'),
])
self.assertEqual(A.data['status'], 'MERGED')
self.assertEqual(B.data['status'], 'MERGED')
@okay_tracebacks('Unterminated string starting at')
def test_pipeline_summary(self):
# Test that we can deal with a truncated pipeline summary

22
zuul/driver/gerrit/gerritmodel.py
View File

@ -43,14 +43,20 @@ class GerritChange(Change):
self.zuul_query_ltime = None
def update(self, data, extra, connection):
self.zuul_query_ltime = data.zuul_query_ltime
if data.format == data.SSH:
self.updateFromSSH(data.data, connection)
else:
self.updateFromHTTP(data.data, data.files, data.commentable_files,
connection)
for k, v in extra.items():
setattr(self, k, v)
if not (self.zuul_query_ltime and
self.zuul_query_ltime > data.zuul_query_ltime):
# If our query is older than the existing change data, it
# means that another scheduler raced us and won. Skip the
# update to avoid having is_merged flap.
self.zuul_query_ltime = data.zuul_query_ltime
if data.format == data.SSH:
self.updateFromSSH(data.data, connection)
else:
self.updateFromHTTP(data.data, data.files,
data.commentable_files,
connection)
for k, v in extra.items():
setattr(self, k, v)
key = ChangeKey(connection.connection_name, None,
'GerritChange', str(self.number), str(self.patchset))
return key