Make repo state buildset global

Store repo state globally for whole buildset
including inherited and required projects.
This is necessary to avoid inconsistencies in case,
e.g., a required projects HEAD changes between two
dependent jobs executions in the same buildset.

Change-Id: I872d4272d8a594b2a40dee0c627f14c990399dd5

tests/fixtures/config/global-repo-state/git/common-config/playbooks/pre.yaml

@@ -0,0 +1,3 @@
+- hosts: localhost
+  roles:
+    - test-role

tests/fixtures/config/global-repo-state/git/common-config/playbooks/require-test.yaml

@@ -0,0 +1,4 @@
+- hosts: localhost
+  tasks:
+    - name: Execute script
+      shell: "bash {{ ansible_env['HOME'] }}/{{ zuul.projects['review.example.com/org/requiredproject'].src_dir }}/script.sh"

tests/fixtures/config/global-repo-state/git/common-config/playbooks/test1.yaml

@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []

tests/fixtures/config/global-repo-state/git/common-config/playbooks/test2.yaml

@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []

tests/fixtures/config/global-repo-state/git/common-config/zuul.yaml

@@ -0,0 +1,75 @@
+- pipeline:
+    name: gate
+    manager: dependent
+    trigger:
+      gerrit:
+        - event: comment-added
+          approval:
+            - Code-Review: 2
+    success:
+      gerrit:
+        Verified: 2
+        submit: true
+    failure:
+      gerrit:
+        Verified: -2
+    start:
+      gerrit:
+        Verified: 0
+    precedence: high
+
+- pipeline:
+    name: check
+    manager: independent
+    trigger:
+      gerrit:
+        - event: comment-added
+          approval:
+            - Code-Review: 2
+    success:
+      gerrit:
+        Verified: 1
+        submit: true
+    failure:
+      gerrit:
+        Verified: -1
+    start:
+      gerrit:
+        Verified: 0
+    precedence: high
+
+- job:
+    name: base
+    pre-run:
+      - playbooks/pre.yaml
+    roles:
+      - zuul: org/roles
+    parent: null
+
+- job:
+    name: test1
+    run: playbooks/test1.yaml
+
+- job:
+    name: test2
+    run: playbooks/test2.yaml
+
+- job:
+    name: require-test1
+    run: playbooks/require-test.yaml
+    required-projects:
+      - name: org/requiredproject
+
+- job:
+    name: require-test2
+    run: playbooks/require-test.yaml
+    required-projects:
+      - name: org/requiredproject
+
+- job:
+    name: dependent-test1
+    run: playbooks/require-test.yaml
+
+- job:
+    name: dependent-test2
+    run: playbooks/require-test.yaml

tests/fixtures/config/global-repo-state/git/org_dependentproject/.zuul.yaml

@@ -0,0 +1,8 @@
+- project:
+    name: org/dependentproject
+    check:
+      jobs:
+        - dependent-test1
+        - dependent-test2:
+            dependencies:
+              - dependent-test1

tests/fixtures/config/global-repo-state/git/org_project/.zuul.yaml

@@ -0,0 +1,8 @@
+- project:
+    name: org/project
+    gate:
+      jobs:
+        - test1
+        - test2:
+            dependencies:
+              - test1

tests/fixtures/config/global-repo-state/git/org_project/README

@@ -0,0 +1 @@
+test

tests/fixtures/config/global-repo-state/git/org_requiredproject/script.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+exit 0

tests/fixtures/config/global-repo-state/git/org_requiringproject/.zuul.yaml

@@ -0,0 +1,8 @@
+- project:
+    name: org/requiringproject
+    gate:
+      jobs:
+        - require-test1
+        - require-test2:
+            dependencies:
+              - require-test1

tests/fixtures/config/global-repo-state/git/org_roles/roles/test-role/tasks/main.yaml

@@ -0,0 +1,3 @@
+- name: Test
+  debug:
+    msg: test-role

tests/fixtures/config/global-repo-state/main.yaml

@@ -0,0 +1,12 @@
+- tenant:
+    name: tenant-one
+    source:
+      gerrit:
+        config-projects:
+          - common-config
+        untrusted-projects:
+          - org/dependentproject
+          - org/project
+          - org/requiredproject
+          - org/requiringproject
+          - org/roles

tests/unit/test_v3.py

@@ -2435,6 +2435,127 @@ class TestInRepoConfig(ZuulTestCase):
                       A.messages[0], "A should have debug info")
 
 
+class TestGlobalRepoState(AnsibleZuulTestCase):
+    tenant_config_file = 'config/global-repo-state/main.yaml'
+
+    def test_inherited_playbooks(self):
+        # Test that the repo state is restored globally for the whole buildset
+        # including inherited projects not in the dependency chain.
+        self.executor_server.hold_jobs_in_build = True
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
+        A.addApproval('Approved', 1)
+        self.fake_gerrit.addEvent(A.addApproval('Code-Review', 2))
+        self.waitUntilSettled()
+
+        # The build test1 is running while test2 is waiting for test1.
+        self.assertEqual(len(self.builds), 1)
+
+        # Now merge a change to the playbook out of band. This will break test2
+        # if it updates common-config to latest master. However due to the
+        # buildset-global repo state test2 must not be broken afterwards.
+        playbook = textwrap.dedent(
+            """
+            - hosts: localhost
+              tasks:
+                - name: fail
+                  fail:
+                    msg: foobar
+            """)
+
+        file_dict = {'playbooks/test2.yaml': playbook}
+        B = self.fake_gerrit.addFakeChange('common-config', 'master', 'A',
+                                           files=file_dict)
+        self.log.info('Merge test change on common-config')
+        B.setMerged()
+
+        self.executor_server.hold_jobs_in_build = False
+        self.executor_server.release()
+        self.waitUntilSettled()
+        self.assertHistory([
+            dict(name='test1', result='SUCCESS', changes='1,1'),
+            dict(name='test2', result='SUCCESS', changes='1,1'),
+        ])
+
+    def test_required_projects(self):
+        # Test that the repo state is restored globally for the whole buildset
+        # including required projects not in the dependency chain.
+        self.executor_server.hold_jobs_in_build = True
+        A = self.fake_gerrit.addFakeChange('org/requiringproject', 'master',
+                                           'A')
+        A.addApproval('Approved', 1)
+        self.fake_gerrit.addEvent(A.addApproval('Code-Review', 2))
+        self.waitUntilSettled()
+
+        # The build require-test1 is running,
+        # require-test2 is waiting for require-test1.
+        self.assertEqual(len(self.builds), 1)
+
+        # Now merge a change to the test script out of band.
+        # This will break required-test2 if it updates requiredproject
+        # to latest master. However, due to the buildset-global repo state,
+        # required-test2 must not be broken afterwards.
+        runscript = textwrap.dedent(
+            """
+            #!/bin/bash
+            exit 1
+            """)
+
+        file_dict = {'script.sh': runscript}
+        B = self.fake_gerrit.addFakeChange('org/requiredproject', 'master',
+                                           'A', files=file_dict)
+        self.log.info('Merge test change on common-config')
+        B.setMerged()
+
+        self.executor_server.hold_jobs_in_build = False
+        self.executor_server.release()
+        self.waitUntilSettled()
+        self.assertHistory([
+            dict(name='require-test1', result='SUCCESS', changes='1,1'),
+            dict(name='require-test2', result='SUCCESS', changes='1,1'),
+        ])
+
+    def test_dependent_project(self):
+        # Test that the repo state is restored globally for the whole buildset
+        # including dependent projects.
+        self.executor_server.hold_jobs_in_build = True
+        B = self.fake_gerrit.addFakeChange('org/requiredproject', 'master',
+                                           'B')
+        A = self.fake_gerrit.addFakeChange('org/dependentproject', 'master',
+                                           'A')
+        A.setDependsOn(B, 1)
+        A.addApproval('Approved', 1)
+        self.fake_gerrit.addEvent(A.addApproval('Code-Review', 2))
+        self.waitUntilSettled()
+
+        # The build dependent-test1 is running,
+        # dependent-test2 is waiting for dependent-test1.
+        self.assertEqual(len(self.builds), 1)
+
+        # Now merge a change to the test script out of band.
+        # This will break dependent-test2 if it updates requiredproject
+        # to latest master. However, due to the buildset-global repo state,
+        # dependent-test2 must not be broken afterwards.
+        runscript = textwrap.dedent(
+            """
+            #!/bin/bash
+            exit 1
+            """)
+
+        file_dict = {'script.sh': runscript}
+        C = self.fake_gerrit.addFakeChange('org/requiredproject', 'master',
+                                           'C', files=file_dict)
+        self.log.info('Merge test change on common-config')
+        C.setMerged()
+
+        self.executor_server.hold_jobs_in_build = False
+        self.executor_server.release()
+        self.waitUntilSettled()
+        self.assertHistory([
+            dict(name='dependent-test1', result='SUCCESS', changes='1,1 2,1'),
+            dict(name='dependent-test2', result='SUCCESS', changes='1,1 2,1'),
+        ])
+
+
 class TestNonLiveMerges(ZuulTestCase):
 
     config_file = 'zuul-connections-gerrit-and-github.conf'

zuul/manager/__init__.py

@@ -20,6 +20,7 @@ from zuul import model
 from zuul.lib.dependson import find_dependency_headers
 from zuul.lib.logutil import get_annotated_logger
 from zuul.lib.tarjan import strongly_connected_components
+from zuul.model import QueueItem
 
 
 class DynamicChangeQueueContextManager(object):
@@ -853,6 +854,24 @@ class PipelineManager(metaclass=ABCMeta):
         self.log.debug("Preparing dynamic layout for: %s" % item.change)
         return self._loadDynamicLayout(item)
 
+    def _branchesForRepoState(self, projects, tenant, items=None):
+        if all(tenant.getExcludeUnprotectedBranches(project)
+               for project in projects):
+            branches = set()
+
+            # Add all protected branches of all involved projects
+            for project in projects:
+                branches.update(tenant.getProjectBranches(project))
+
+            # Additionally add all target branches of all involved items.
+            if items is not None:
+                branches.update(item.change.branch for item in items
+                                if hasattr(item.change, 'branch'))
+            branches = list(branches)
+        else:
+            branches = None
+        return branches
+
     def scheduleMerge(self, item, files=None, dirs=None):
         log = item.annotateLogger(self.log)
         log.debug("Scheduling merge for item %s (files: %s, dirs: %s)" %
@@ -871,20 +890,8 @@ class PipelineManager(metaclass=ABCMeta):
             item.change.project for item in items
             if tenant.getProject(item.change.project.canonical_name)[1]
         }
-        if all(tenant.getExcludeUnprotectedBranches(project)
-               for project in projects):
-            branches = set()
-
-            # Add all protected branches of all involved projects
-            for project in projects:
-                branches.update(tenant.getProjectBranches(project))
-
-            # Additionally add all target branches of all involved items.
-            branches.update(item.change.branch for item in items
-                            if hasattr(item.change, 'branch'))
-            branches = list(branches)
-        else:
-            branches = None
+        branches = self._branchesForRepoState(projects=projects, tenant=tenant,
+                                              items=items)
 
         if isinstance(item.change, model.Change):
             self.sched.merger.mergeChanges(build_set.merger_items,
@@ -912,7 +919,54 @@ class PipelineManager(metaclass=ABCMeta):
             event=item.event)
         return False
 
-    def prepareItem(self, item):
+    def scheduleGlobalRepoState(self, item: QueueItem) -> bool:
+        log = item.annotateLogger(self.log)
+        log.info('Scheduling global repo state for item %s', item)
+
+        tenant = item.pipeline.tenant
+        jobs = item.job_graph.getJobs()
+        projects = set()
+        for job in jobs:
+            log.debug('Processing job %s', job.name)
+            projects.update(job.getAffectedProjects(tenant))
+            log.debug('Needed projects: %s', projects)
+
+        # Filter projects for ones that are already in repo state
+        repo_state = item.current_build_set.repo_state
+        connections = self.sched.connections.connections
+        projects_to_remove = set()
+        for connection in repo_state.keys():
+            canonical_hostname = connections[connection].canonical_hostname
+            for project in repo_state[connection].keys():
+                canonical_project_name = canonical_hostname + '/' + project
+                for affected_project in projects:
+                    if canonical_project_name ==\
+                            affected_project.canonical_name:
+                        projects_to_remove.add(affected_project)
+        projects -= projects_to_remove
+
+        if not projects:
+            item.current_build_set.repo_state_state =\
+                item.current_build_set.COMPLETE
+            return True
+
+        branches = self._branchesForRepoState(projects=projects, tenant=tenant)
+
+        new_items = list()
+        for project in projects:
+            new_item = dict()
+            new_item['project'] = project.name
+            new_item['connection'] = project.connection_name
+            new_items.append(new_item)
+
+        # Get state for not yet tracked projects
+        self.sched.merger.getRepoState(items=new_items,
+                                       build_set=item.current_build_set,
+                                       event=item.event,
+                                       branches=branches)
+        return True
+
+    def prepareItem(self, item: QueueItem) -> bool:
         build_set = item.current_build_set
         tenant = item.pipeline.tenant
         # We always need to set the configuration of the item if it
@@ -1000,6 +1054,15 @@ class PipelineManager(metaclass=ABCMeta):
                 item.setConfigError("Unable to freeze job graph: %s" %
                                     (str(e)))
                 return False
+
+        # At this point we know all frozen jobs and their repos so update the
+        # repo state with all missing repos.
+        if build_set.repo_state_state == build_set.NEW:
+            build_set.repo_state_state = build_set.PENDING
+            self.scheduleGlobalRepoState(item)
+        if build_set.repo_state_state == build_set.PENDING:
+            return False
+
         return True
 
     def _processOneItem(self, item, nnfi):
@@ -1262,6 +1325,13 @@ class PipelineManager(metaclass=ABCMeta):
         build_set.files_state = build_set.COMPLETE
 
     def onMergeCompleted(self, event):
+        build_set = event.build_set
+        if build_set.merge_state == build_set.COMPLETE:
+            self._onGlobalRepoStateCompleted(event)
+        else:
+            self._onMergeCompleted(event)
+
+    def _onMergeCompleted(self, event):
         build_set = event.build_set
         item = build_set.item
         item.change.containing_branches = event.item_in_branches
@@ -1281,6 +1351,21 @@ class PipelineManager(metaclass=ABCMeta):
             self.log.info("Unable to merge change %s" % item.change)
             item.setUnableToMerge()
 
+    def _onGlobalRepoStateCompleted(self, event):
+        if not event.updated:
+            item = event.build_set.item
+            self.log.info("Unable to get global repo state for change %s"
+                          % item.change)
+            item.setUnableToMerge()
+        else:
+            repo_state = event.build_set.repo_state
+            for connection in event.repo_state.keys():
+                if connection in repo_state:
+                    repo_state[connection].update(event.repo_state[connection])
+                else:
+                    repo_state[connection] = event.repo_state[connection]
+            event.build_set.repo_state_state = event.build_set.COMPLETE
+
     def onNodesProvisioned(self, event):
         # TODOv3(jeblair): handle provisioning failure here
         request = event.request

zuul/merger/merger.py

@@ -1039,7 +1039,7 @@ class Merger(object):
             self._restoreRepoState(item['connection'], item['project'], repo,
                                    repo_state, zuul_event_id)
 
-    def getRepoState(self, items, branches=None, repo_locks=None):
+    def getRepoState(self, items, repo_locks, branches=None):
         # Gets the repo state for items.  Generally this will be
         # called in any non-change pipeline.  We will return the repo
         # state for each item, but manipulated with any information in
@@ -1051,19 +1051,16 @@ class Merger(object):
         # A list of branch names the last item appears in.
         item_in_branches = []
         for item in items:
-            # If we're in the executor context we have the repo_locks object
-            # and perform per repo locking.
-            if repo_locks is not None:
-                lock = repo_locks.getRepoLock(
-                    item['connection'], item['project'])
-            else:
-                lock = nullcontext()
+            # If we're in the executor context we need to lock the repo.
+            # If not repo_locks will give us a fake lock.
+            lock = repo_locks.getRepoLock(item['connection'], item['project'])
             with lock:
                 repo = self.getRepo(item['connection'], item['project'])
-                key = (item['connection'], item['project'], item['branch'])
+                key = (item['connection'], item['project'])
                 if key not in seen:
                     try:
                         repo.reset()
+                        seen.add(key)
                     except Exception:
                         self.log.exception("Unable to reset repo %s" % repo)
                         return (False, {}, [])
@@ -1079,7 +1076,9 @@ class Merger(object):
                         item['ref'], item['newrev'])
         item = items[-1]
         repo = self.getRepo(item['connection'], item['project'])
-        item_in_branches = repo.contains(item['newrev'])
+        item_in_branches = False
+        if item.get('newrev'):
+            item_in_branches = repo.contains(item['newrev'])
         return (True, repo_state, item_in_branches)
 
     def getFiles(self, connection_name, project_name, branch, files, dirs=[]):

zuul/merger/server.py

@@ -199,8 +199,7 @@ class BaseMergeServer(metaclass=ABCMeta):
         zuul_event_id = args.get('zuul_event_id')
         success, repo_state, item_in_branches = \
             self.merger.getRepoState(
-                args['items'], branches=args.get('branches'),
-                repo_locks=self.repo_locks)
+                args['items'], self.repo_locks, branches=args.get('branches'))
         result = dict(updated=success,
                       repo_state=repo_state,
                       item_in_branches=item_in_branches)

zuul/model.py

@@ -19,6 +19,8 @@ import copy
 import json
 import logging
 import os
+from itertools import chain
+
 import re2
 import struct
 import time
@@ -1744,6 +1746,32 @@ class Job(ConfigObject):
 
         return True
 
+    def _projectsFromPlaybooks(self, playbooks):
+        for playbook in playbooks:
+            # noop job does not have source_context
+            if playbook.source_context:
+                yield playbook.source_context.project.canonical_name
+            for role in playbook.roles:
+                if role.implicit:
+                    continue
+                yield role.project_canonical_name
+
+    def getAffectedProjects(self, tenant):
+        """
+        Gets all projects that are required to run this job. This includes
+        required_projects, referenced playbooks, roles and dependent changes.
+        """
+        project_canonical_names = set()
+        project_canonical_names.update(self.required_projects.keys())
+        project_canonical_names.update(self._projectsFromPlaybooks(
+            chain(self.pre_run, [self.run[0]], self.post_run,
+                  self.cleanup_run)))
+
+        projects = list()
+        for project_canonical_name in project_canonical_names:
+            projects.append(tenant.getProject(project_canonical_name)[1])
+        return projects
+
 
 class JobProject(ConfigObject):
     """ A reference to a project from a job. """
@@ -2067,6 +2095,7 @@ class BuildSet(object):
             self.files_state = self.COMPLETE
         else:
             self.files_state = self.NEW
+        self.repo_state_state = self.NEW
 
     @property
     def ref(self):