Cause the enqueue command to ignore pipeline reqs
The enqueue command typically is used after a Zuul restart. It
may also be useful when the normal method of triggering a change
is not possible, or perhaps too time consuming. In all of those
cases, it may be desirable to bypass pipeline requirements (and
in the first case, it may be necessary since the act of enqueing
a change in a pipeline may cause it, as a side effect, to no longer
satisfy the requirements for that pipeline).
To resolve, this, ignore the pipeline requirements when a change
is enqueued into a pipeline as the result of an administrative
enqueue command.
Change-Id: Ic672452cd99355e3ee91df3c649f944ef75958a9
diff --git a/zuul/rpclistener.py b/zuul/rpclistener.py
index 8719c76..fcf1161 100644
--- a/zuul/rpclistener.py
+++ b/zuul/rpclistener.py
@@ -116,7 +116,7 @@
if errors:
job.sendWorkException(errors.encode('utf8'))
else:
- self.sched.addEvent(event)
+ self.sched.enqueue(event)
job.sendWorkComplete()
def handle_promote(self, job):
diff --git a/zuul/scheduler.py b/zuul/scheduler.py
index 0c653e7..15520ac 100644
--- a/zuul/scheduler.py
+++ b/zuul/scheduler.py
@@ -109,6 +109,18 @@
self.change_ids = change_ids
+class EnqueueEvent(ManagementEvent):
+ """Enqueue a change into a pipeline
+
+ :arg TriggerEvent trigger_event: a TriggerEvent describing the
+ trigger, pipeline, and change to enqueue
+ """
+
+ def __init__(self, trigger_event):
+ super(EnqueueEvent, self).__init__()
+ self.trigger_event = trigger_event
+
+
class ResultEvent(object):
"""An event that needs to modify the pipeline state due to a
result from an external system."""
@@ -522,6 +534,14 @@
event.wait()
self.log.debug("Promotion complete")
+ def enqueue(self, trigger_event):
+ event = EnqueueEvent(trigger_event)
+ self.management_event_queue.put(event)
+ self.wake_event.set()
+ self.log.debug("Waiting for enqueue")
+ event.wait()
+ self.log.debug("Enqueue complete")
+
def exit(self):
self.log.debug("Prepare to exit")
self._pause = True
@@ -691,6 +711,17 @@
quiet=True,
ignore_requirements=True)
+ def _doEnqueueEvent(self, event):
+ project = self.layout.projects.get(event.project_name)
+ pipeline = self.layout.pipelines[event.forced_pipeline]
+ trigger = self.triggers.get(event.trigger_name)
+ change = event.getChange(project, trigger)
+ self.log.debug("Event %s for change %s was directly assigned "
+ "to pipeline %s" % (event, change, self))
+ self.log.info("Adding %s, %s to %s" %
+ (project, change, pipeline))
+ pipeline.manager.addChange(change, ignore_requirements=True)
+
def _areAllBuildsComplete(self):
self.log.debug("Checking if all builds are complete")
waiting = False
@@ -800,6 +831,8 @@
self._doReconfigureEvent(event)
elif isinstance(event, PromoteEvent):
self._doPromoteEvent(event)
+ elif isinstance(event, EnqueueEvent):
+ self._doEnqueueEvent(event.trigger_event)
else:
self.log.error("Unable to handle event %s" % event)
event.done()