Merge "Reuse queue items after reconfiguration"

4 files changed, 33 insertions, 18 deletions

diff --git a/zuul/manager/__init__.py b/zuul/manager/__init__.py
index c1f61e5fa..65fa58d5b 100644
--- a/zuul/manager/__init__.py
+++ b/zuul/manager/__init__.py
@@ -100,11 +100,14 @@ class PipelineManager(metaclass=ABCMeta):
             # Make sure we have state and change list objects, and
             # ensure that they exist in ZK.  We don't hold the
             # pipeline lock, but if they don't exist, that means they
-            # are new, so no one else will either.  These will not
-            # automatically refresh now, so they will be out of date
-            # until they are refreshed later.
+            # are new, so no one else will either, so the write on
+            # create is okay.  If they do exist and we have an old
+            # object, we'll just reuse it.  If it does exist and we
+            # don't have an old object, we'll get a new empty one.
+            # Regardless, these will not automatically refresh now, so
+            # they will be out of date until they are refreshed later.
             self.pipeline.state = PipelineState.create(
-                self.pipeline, layout.uuid)
+                self.pipeline, layout.uuid, self.pipeline.state)
             self.pipeline.change_list = PipelineChangeList.create(
                 self.pipeline)
 
diff --git a/zuul/model.py b/zuul/model.py
index 0daa0d434..e339b1ffa 100644
--- a/zuul/model.py
+++ b/zuul/model.py
@@ -626,18 +626,27 @@ class PipelineState(zkobject.ZKObject):
         return obj
 
     @classmethod
-    def create(cls, pipeline, layout_uuid):
+    def create(cls, pipeline, layout_uuid, old_state=None):
         # If the object does not exist in ZK, create it with the
         # default attributes and the supplied layout UUID.  Otherwise,
-        # return an initialized object without loading any data so
-        # that data can be loaded on the next refresh.
+        # return an initialized object (or the old object for reuse)
+        # without loading any data so that data can be loaded on the
+        # next refresh.
         ctx = pipeline.manager.current_context
         state = cls()
         state._set(pipeline=pipeline)
         if state.exists(ctx):
+            if old_state:
+                old_state._resetObjectRefs()
+                return old_state
             return state
         return cls.new(ctx, pipeline=pipeline, layout_uuid=layout_uuid)
 
+    def _resetObjectRefs(self):
+        # Update the pipeline references on the queue objects.
+        for queue in self.queues + self.old_queues:
+            queue.pipeline = self.pipeline
+
     def getPath(self):
         if hasattr(self, '_path'):
             return self._path
@@ -896,15 +905,17 @@ class PipelineChangeList(zkobject.ShardedZKObject):
         return pipeline_path + '/change_list'
 
     @classmethod
-    def create(cls, pipeline):
+    def create(cls, pipeline, old_list=None):
         # If the object does not exist in ZK, create it with the
         # default attributes.  Otherwise, return an initialized object
-        # without loading any data so that data can be loaded on the
-        # next refresh.
+        # (or the old object for reuse) without loading any data so
+        # that data can be loaded on the next refresh.
         ctx = pipeline.manager.current_context
         change_list = cls()
         change_list._set(pipeline=pipeline)
         if change_list.exists(ctx):
+            if old_list:
+                return old_list
             return change_list
         return cls.new(ctx, pipeline=pipeline)
 
@@ -1057,7 +1068,7 @@ class ChangeQueue(zkobject.ZKObject):
             else:
                 tpe_jobs.append(('item', tpe.submit(
                     QueueItem.fromZK, context, item_path,
-                    pipeline=self.pipeline, queue=self)))
+                    queue=self)))
 
         for (kind, future) in tpe_jobs:
             result = future.result()
@@ -1132,7 +1143,6 @@ class ChangeQueue(zkobject.ZKObject):
             enqueue_time = time.time()
         item = QueueItem.new(self.zk_context,
                              queue=self,
-                             pipeline=self.pipeline,
                              change=change,
                              event=event,
                              span_info=span_info,
@@ -1141,8 +1151,7 @@ class ChangeQueue(zkobject.ZKObject):
         return item
 
     def enqueueItem(self, item):
-        # FIXME: the pipeline should not change
-        item._set(pipeline=self.pipeline, queue=self)
+        item._set(queue=self)
         if self.queue:
             item.updateAttributes(self.zk_context, item_ahead=self.queue[-1])
             with item.item_ahead.activeContext(self.zk_context):
@@ -4508,7 +4517,6 @@ class QueueItem(zkobject.ZKObject):
         super().__init__()
         self._set(
             uuid=uuid4().hex,
-            pipeline=None,
             queue=None,
             change=None,  # a ref
             dequeued_needing_change=None,
@@ -4540,6 +4548,12 @@ class QueueItem(zkobject.ZKObject):
             dequeued_bundle_failing=False
         )
 
+    @property
+    def pipeline(self):
+        if self.queue:
+            return self.queue.pipeline
+        return None
+
     @classmethod
     def new(klass, context, **kw):
         obj = klass()
diff --git a/zuul/scheduler.py b/zuul/scheduler.py
index a748bf494..e30495d20 100644
--- a/zuul/scheduler.py
+++ b/zuul/scheduler.py
@@ -1614,7 +1614,6 @@ class Scheduler(threading.Thread):
                     reenqueued = False
                     if new_project:
                         item.change.project = new_project
-                        item.pipeline = None
                         item.queue = None
                         if not old_item_ahead or not last_head:
                             last_head = item
diff --git a/zuul/web/__init__.py b/zuul/web/__init__.py
index de8e91c2c..7f27cd970 100755
--- a/zuul/web/__init__.py
+++ b/zuul/web/__init__.py
@@ -1698,8 +1698,7 @@ class ZuulWebAPI(object):
         change.branch = branch_name or "master"
         with LocalZKContext(self.log) as context:
             queue = ChangeQueue.new(context, pipeline=pipeline)
-            item = QueueItem.new(context, queue=queue, change=change,
-                                 pipeline=queue.pipeline)
+            item = QueueItem.new(context, queue=queue, change=change)
             item.freezeJobGraph(tenant.layout, context,
                                 skip_file_matcher=True,
                                 redact_secrets_and_keys=True)