zuul/zuul
Code Issues Proposed changes

Only store trigger event info on queue item

Browse Source 
The event that's currently stored as part of the queue item is not
sharded. This means that we can see Zookeeper disconnects when the queue
item data exceeds the max. Znode size of 1MB.

Since we only need the event's timestamp and the Zuul event-id after an
item is enqueued, we can reduce the amount of data we store in Zookeeper
and also avoid sharding the event.

Change-Id: I13577498e55fd4bb189679836219dea4dc5729fc
This commit is contained in:
Simon Westphahl
2023-02-21 10:39:03 +01:00
parent d57942a41b
commit 8ca99e0154
5 changed files with 101 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/source/developer/model-changelog.rst
+7
View File
@@ -112,3 +112,10 @@ Version 12
:Prior Zuul version: 8.0.1
:Description: Adds job_versions and build_versions to BuildSet.
Affects schedulers.
Version 13
----------
:Prior Zuul version: 8.2.0
:Description: Stores only the necessary event info as part of a queue item
instead of the full trigger event.
Affects schedulers.
tests/unit/test_model_upgrade.py
+27
View File
@@ -293,6 +293,33 @@ class TestModelUpgrade(ZuulTestCase):
result='SUCCESS', changes='1,1'),
], ordered=False)
@model_version(12)
def test_model_12_13(self):
# Initially queue items will still have the full trigger event
# stored in Zookeeper. The trigger event will be converted to
# an event info object after the model API update.
self.executor_server.hold_jobs_in_build = True
A = self.fake_gerrit.addFakeChange('org/project1', 'master', 'A')
self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
self.waitUntilSettled()
self.assertEqual(len(self.builds), 1)
# Upgrade our component
self.model_test_component_info.model_api = 13
self.executor_server.hold_jobs_in_build = False
self.executor_server.release()
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='project1-project2-integration',
result='SUCCESS', changes='1,1'),
], ordered=False)
class TestGithubModelUpgrade(ZuulTestCase):
config_file = 'zuul-github-driver.conf'
zuul/manager/__init__.py
+9 -6
View File
@@ -625,7 +625,7 @@ class PipelineManager(metaclass=ABCMeta):
if enqueue_time:
item.enqueue_time = enqueue_time
item.live = live
self.reportStats(item, added=True)
self.reportStats(item, trigger_event=event)
item.quiet = quiet
if item.live:
@@ -2197,7 +2197,7 @@ class PipelineManager(metaclass=ABCMeta):
log.error("Reporting item %s received: %s", item, ret)
return action, (not ret)
def reportStats(self, item, added=False):
def reportStats(self, item, trigger_event=None):
if not self.sched.statsd:
return
try:
@@ -2236,18 +2236,21 @@ class PipelineManager(metaclass=ABCMeta):
if dt:
self.sched.statsd.timing(key + '.resident_time', dt)
self.sched.statsd.incr(key + '.total_changes')
if added and hasattr(item.event, 'arrived_at_scheduler_timestamp'):
if (
trigger_event
and hasattr(trigger_event, 'arrived_at_scheduler_timestamp')
):
now = time.time()
arrived = item.event.arrived_at_scheduler_timestamp
arrived = trigger_event.arrived_at_scheduler_timestamp
processing = (now - arrived) * 1000
elapsed = (now - item.event.timestamp) * 1000
elapsed = (now - trigger_event.timestamp) * 1000
self.sched.statsd.timing(
basekey + '.event_enqueue_processing_time',
processing)
self.sched.statsd.timing(
basekey + '.event_enqueue_time', elapsed)
self.reportPipelineTiming('event_enqueue_time',
item.event.timestamp)
trigger_event.timestamp)
except Exception:
self.log.exception("Exception reporting pipeline stats")
zuul/model.py
+57 -11
View File
@@ -4666,6 +4666,37 @@ class BuildSet(zkobject.ZKObject):
return Attributes(uuid=self.uuid)
class EventInfo:
def __init__(self):
self.zuul_event_id = None
self.timestamp = time.time()
self.span_context = None
@classmethod
def fromEvent(cls, event):
tinfo = cls()
tinfo.zuul_event_id = event.zuul_event_id
tinfo.timestamp = event.timestamp
tinfo.span_context = event.span_context
return tinfo
@classmethod
def fromDict(cls, d):
tinfo = cls()
tinfo.zuul_event_id = d["zuul_event_id"]
tinfo.timestamp = d["timestamp"]
tinfo.span_context = d["span_context"]
return tinfo
def toDict(self):
return {
"zuul_event_id": self.zuul_event_id,
"timestamp": self.timestamp,
"span_context": self.span_context,
}
class QueueItem(zkobject.ZKObject):
"""Represents the position of a Change in a ChangeQueue.
@@ -4700,7 +4731,7 @@ class QueueItem(zkobject.ZKObject):
live=True, # Whether an item is intended to be processed at all
layout_uuid=None,
_cached_sql_results={},
event=None, # The trigger event that lead to this queue item
event=None, # Info about the event that lead to this queue item
# Additional container for connection specifig information to be
# used by reporters throughout the lifecycle
@@ -4722,6 +4753,9 @@ class QueueItem(zkobject.ZKObject):
def new(klass, context, **kw):
obj = klass()
obj._set(**kw)
if COMPONENT_REGISTRY.model_api >= 13:
obj._set(event=obj.event and EventInfo.fromEvent(obj.event))
data = obj._trySerialize(context)
obj._save(context, data, create=True)
files_state = (BuildSet.COMPLETE if obj.change.files is not None
@@ -4750,10 +4784,18 @@ class QueueItem(zkobject.ZKObject):
return (tenant, pipeline, uuid)
def serialize(self, context):
if isinstance(self.event, TriggerEvent):
event_type = "TriggerEvent"
if COMPONENT_REGISTRY.model_api < 13:
if isinstance(self.event, TriggerEvent):
event_type = "TriggerEvent"
else:
event_type = self.event.__class__.__name__
else:
event_type = self.event.__class__.__name__
event_type = "EventInfo"
if not isinstance(self.event, EventInfo):
# Convert our local trigger event to a trigger info
# object. This will only happen on the transition to
# model API version 13.
self._set(event=EventInfo.fromEvent(self.event))
data = {
"uuid": self.uuid,
@@ -4795,14 +4837,18 @@ class QueueItem(zkobject.ZKObject):
# child objects.
self._set(uuid=data["uuid"])
event_type = data["event"]["type"]
if event_type == "TriggerEvent":
event_class = (
self.pipeline.manager.sched.connections.getTriggerEventClass(
data["event"]["data"]["driver_name"])
)
if COMPONENT_REGISTRY.model_api < 13:
event_type = data["event"]["type"]
if event_type == "TriggerEvent":
event_class = (
self.pipeline.manager.sched.connections
.getTriggerEventClass(
data["event"]["data"]["driver_name"])
)
else:
event_class = EventTypeIndex.event_type_mapping.get(event_type)
else:
event_class = EventTypeIndex.event_type_mapping.get(event_type)
event_class = EventInfo
if event_class is None:
raise NotImplementedError(
zuul/model_api.py
+1 -1
View File
@@ -14,4 +14,4 @@
# When making ZK schema changes, increment this and add a record to
# doc/source/developer/model-changelog.rst
MODEL_API = 12
MODEL_API = 13