Merge "Add "supercedes" pipeline option"

doc/source/user/config.rst

@@ -293,6 +293,15 @@ success, the pipeline reports back to Gerrit with ``Verified`` vote of
       type of the connection will dictate which options are available.
       See :ref:`drivers`.
 
+   .. attr:: supercedes
+
+      The name of a pipeline, or a list of names, that this pipeline
+      supercedes.  When a change is enqueued in this pipeline, it will
+      be removed from the pipelines listed here.  For example, a
+      :term:`gate` pipeline may supercede a :term:`check` pipeline so
+      that test resources are not spent running near-duplicate jobs
+      simultaneously.
+
    .. attr:: dequeue-on-new-patchset
       :default: true
 

releasenotes/notes/pipeline-supercedes-ba622ac28df61ffd.yaml

@@ -0,0 +1,11 @@
+---
+features:
+  - |
+    Pipelines may now indicate that they supercede other pipelines with the
+    :attr:`pipeline.supercedes` attribute.
+
+    When a change is enqueued in a pipeline which supercedes others,
+    it will be removed from the other pipelines.  For example, a
+    :term:`gate` pipeline may supercede a :term:`check` pipeline so
+    that test resources are not spent running near-duplicate jobs
+    simultaneously.

tests/fixtures/layouts/pipeline-supercedes.yaml

@@ -0,0 +1,51 @@
+- pipeline:
+    name: check
+    manager: independent
+    trigger:
+      gerrit:
+        - event: patchset-created
+    success:
+      gerrit:
+        Verified: 1
+    failure:
+      gerrit:
+        Verified: -1
+
+- pipeline:
+    name: gate
+    manager: dependent
+    supercedes: check
+    success-message: Build succeeded (gate).
+    trigger:
+      gerrit:
+        - event: comment-added
+          approval:
+            - Approved: 1
+    success:
+      gerrit:
+        Verified: 2
+        submit: true
+    failure:
+      gerrit:
+        Verified: -2
+    start:
+      gerrit:
+        Verified: 0
+    precedence: high
+
+- job:
+    name: base
+    parent: null
+    run: playbooks/base.yaml
+
+- job:
+    name: test-job
+
+- project:
+    name: org/project
+    check:
+      jobs:
+        - test-job
+    gate:
+      jobs:
+        - test-job

tests/unit/test_scheduler.py

@@ -7528,3 +7528,41 @@ class TestSchedulerFailFast(ZuulTestCase):
             dict(name='project-test5', result='SUCCESS', changes='1,1'),
             dict(name='project-test6', result='FAILURE', changes='1,1'),
         ], ordered=False)
+
+
+class TestPipelineSupersedes(ZuulTestCase):
+
+    @simple_layout('layouts/pipeline-supercedes.yaml')
+    def test_supercedes(self):
+        """
+        Tests that a pipeline that is flagged with fail-fast
+        aborts jobs early.
+        """
+        self.executor_server.hold_jobs_in_build = True
+
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+
+        self.assertEqual(len(self.builds), 1)
+        self.assertEqual(self.builds[0].name, 'test-job')
+
+        A.addApproval('Code-Review', 2)
+        self.fake_gerrit.addEvent(A.addApproval('Approved', 1))
+        self.waitUntilSettled()
+
+        self.assertEqual(len(self.builds), 1)
+        self.assertEqual(self.builds[0].name, 'test-job')
+        self.assertEqual(self.builds[0].pipeline, 'gate')
+
+        self.executor_server.hold_jobs_in_build = False
+        self.executor_server.release()
+        self.waitUntilSettled()
+
+        self.assertEqual(len(self.builds), 0)
+        self.assertEqual(A.reported, 2)
+        self.assertEqual(A.data['status'], 'MERGED')
+        self.assertHistory([
+            dict(name='test-job', result='ABORTED', changes='1,1'),
+            dict(name='test-job', result='SUCCESS', changes='1,1'),
+        ], ordered=False)

tests/unit/test_v3.py

@@ -2138,6 +2138,34 @@ class TestInRepoConfig(ZuulTestCase):
                       A.messages[0],
                       "A should have an error reported")
 
+    def test_pipeline_supercedes_error(self):
+        with open(os.path.join(FIXTURE_DIR,
+                               'config/in-repo/git/',
+                               'common-config/zuul.yaml')) as f:
+            base_common_config = f.read()
+
+        in_repo_conf_A = textwrap.dedent(
+            """
+            - pipeline:
+                name: periodic
+                manager: independent
+                supercedes: doesnotexist
+                trigger: {}
+            """)
+
+        file_dict = {'zuul.yaml': None,
+                     'zuul.d/main.yaml': base_common_config,
+                     'zuul.d/test1.yaml': in_repo_conf_A}
+        A = self.fake_gerrit.addFakeChange('common-config', 'master', 'A',
+                                           files=file_dict)
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+        self.assertEqual(A.reported, 1,
+                         "A should report failure")
+        self.assertIn('supercedes an unknown',
+                      A.messages[0],
+                      "A should have an error reported")
+
     def test_change_series_error(self):
         with open(os.path.join(FIXTURE_DIR,
                                'config/in-repo/git/',

zuul/configloader.py

@@ -1164,6 +1164,7 @@ class PipelineParser(object):
         pipeline = {vs.Required('name'): str,
                     vs.Required('manager'): manager,
                     'precedence': precedence,
+                    'supercedes': to_list(str),
                     'description': str,
                     'success-message': str,
                     'failure-message': str,
@@ -1198,6 +1199,7 @@ class PipelineParser(object):
         pipeline.source_context = conf['_source_context']
         pipeline.start_mark = conf['_start_mark']
         pipeline.description = conf.get('description')
+        pipeline.supercedes = as_list(conf.get('supercedes', []))
 
         precedence = model.PRECEDENCE_MAP[conf.get('precedence')]
         pipeline.precedence = precedence
@@ -1959,6 +1961,10 @@ class TenantParser(object):
             for job in jobs:
                 with reference_exceptions('job', job, layout.loading_errors):
                     job.validateReferences(layout)
+        for pipeline in layout.pipelines.values():
+            with reference_exceptions(
+                    'pipeline', pipeline, layout.loading_errors):
+                pipeline.validateReferences(layout)
 
         if skip_semaphores:
             # We should not actually update the layout with new

zuul/manager/__init__.py

@@ -334,6 +334,7 @@ class PipelineManager(object):
             tenant = self.pipeline.tenant
             zuul_driver.onChangeEnqueued(
                 tenant, item.change, self.pipeline, event)
+            self.dequeueSupercededItems(item)
             return True
 
     def dequeueItem(self, item):
@@ -351,6 +352,23 @@ class PipelineManager(object):
         self.dequeueItem(item)
         self.reportStats(item)
 
+    def dequeueSupercededItems(self, item):
+        for other_name in self.pipeline.supercedes:
+            other_pipeline = self.pipeline.tenant.layout.pipelines.get(
+                other_name)
+            if not other_pipeline:
+                continue
+
+            found = None
+            for other_item in other_pipeline.getAllItems():
+                if other_item.live and other_item.change.equals(item.change):
+                    found = other_item
+                    break
+            if found:
+                self.log.info("Item %s is superceded by %s, removing" %
+                              (found, item))
+                other_pipeline.manager.removeItem(found)
+
     def updateCommitDependencies(self, change, change_queue, event):
         log = get_annotated_logger(self.log, event)
 

zuul/model.py

@@ -294,6 +294,18 @@ class Pipeline(object):
     def getSafeAttributes(self):
         return Attributes(name=self.name)
 
+    def validateReferences(self, layout):
+        # Verify that references to other objects in the layout are
+        # valid.
+
+        for pipeline in self.supercedes:
+            if not layout.pipelines.get(pipeline):
+                raise Exception(
+                    'The pipeline "{this}" supercedes an unknown pipeline '
+                    '{other}.'.format(
+                        this=self.name,
+                        other=pipeline))
+
     def setManager(self, manager):
         self.manager = manager