Add Zuul ref replication

Zuul refs may now be pushed to an arbitrary number of git URLs.

Change-Id: Icae2fc94eb73b63adbf6f6b799f2be166e951b55

doc/source/zuul.rst

@@ -160,6 +160,16 @@ smtp
   This can be overridden by individual pipelines.
   ``default_to=you@example.com``
 
+replication
+"""""""""""
+
+Zuul can push the refs it creates to any number of servers.  To do so,
+list the git push URLs in this section, one per line as follows::
+
+  [replication]
+    url1=ssh://user@host1.example.com:port/path/to/repo
+    url2=ssh://user@host2.example.com:port/path/to/repo
+
 layout.yaml
 ~~~~~~~~~~~
 

tests/test_scheduler.py

@@ -2920,6 +2920,39 @@ class TestScheduler(testtools.TestCase):
         self.assertEqual(B.data['status'], 'MERGED')
         self.assertEqual(B.reported, 2)
 
+    def test_push_urls(self):
+        "Test that Zuul can push refs to multiple URLs"
+        upstream_path = os.path.join(self.upstream_root, 'org/project')
+        replica1 = os.path.join(self.upstream_root, 'replica1')
+        replica2 = os.path.join(self.upstream_root, 'replica2')
+
+        self.config.add_section('replication')
+        self.config.set('replication', 'url1', 'file://%s' % replica1)
+        self.config.set('replication', 'url2', 'file://%s' % replica2)
+        self.sched.reconfigure(self.config)
+
+        r1 = git.Repo.clone_from(upstream_path, replica1 + '/org/project.git')
+        r2 = git.Repo.clone_from(upstream_path, replica2 + '/org/project.git')
+
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
+        A.addApproval('CRVW', 2)
+        self.fake_gerrit.addEvent(A.addApproval('APRV', 1))
+        B = self.fake_gerrit.addFakeChange('org/project', 'mp', 'B')
+        B.addApproval('CRVW', 2)
+        self.fake_gerrit.addEvent(B.addApproval('APRV', 1))
+        self.waitUntilSettled()
+        count = 0
+        for ref in r1.refs:
+            if ref.path.startswith('refs/zuul'):
+                count += 1
+        self.assertEqual(count, 3)
+
+        count = 0
+        for ref in r2.refs:
+            if ref.path.startswith('refs/zuul'):
+                count += 1
+        self.assertEqual(count, 3)
+
     def test_timer(self):
         "Test that a periodic job is triggered"
         self.worker.hold_jobs_in_build = True

zuul/merger.py

@@ -16,6 +16,7 @@ import git
 import os
 import logging
 import model
+import threading
 
 
 class ZuulReference(git.Reference):
@@ -131,6 +132,11 @@ class Repo(object):
                                                  self.remote_url))
         repo.remotes.origin.push('%s:%s' % (local, remote))
 
+    def push_url(self, url, refspecs):
+        repo = self.createRepoObject()
+        self.log.debug("Pushing %s to %s" % (refspecs, url))
+        repo.git.push([url] + refspecs)
+
     def update(self):
         repo = self.createRepoObject()
         self.log.debug("Updating repository %s" % self.local_path)
@@ -142,7 +148,7 @@ class Merger(object):
     log = logging.getLogger("zuul.Merger")
 
     def __init__(self, trigger, working_root, push_refs, sshkey, email,
-                 username):
+                 username, replicate_urls):
         self.trigger = trigger
         self.repos = {}
         self.working_root = working_root
@@ -153,6 +159,7 @@ class Merger(object):
             self._makeSSHWrapper(sshkey)
         self.email = email
         self.username = username
+        self.replicate_urls = replicate_urls
 
     def _makeSSHWrapper(self, key):
         name = os.path.join(self.working_root, '.ssh_wrapper')
@@ -219,6 +226,25 @@ class Merger(object):
             return False
         return commit
 
+    def replicateRefspecs(self, refspecs):
+        threads = []
+        for url in self.replicate_urls:
+            t = threading.Thread(target=self._replicate,
+                                 args=(url, refspecs))
+            t.start()
+            threads.append(t)
+        for t in threads:
+            t.join()
+
+    def _replicate(self, url, project_refspecs):
+        try:
+            for project, refspecs in project_refspecs.items():
+                repo = self.getRepo(project)
+                repo.push_url(os.path.join(url, project.name + '.git'),
+                              refspecs)
+        except Exception:
+            self.log.exception("Exception pushing to %s" % url)
+
     def mergeChanges(self, items, target_ref=None):
         # Merge shortcuts:
         # if this is the only change just merge it against its branch.
@@ -257,6 +283,7 @@ class Merger(object):
             return commit
 
         project_branches = []
+        replicate_refspecs = {}
         for i in reversed(items):
             # Here we create all of the necessary zuul refs and potentially
             # push them back to Gerrit.
@@ -276,10 +303,13 @@ class Merger(object):
                     self.log.exception("Unable to set zuul ref %s for "
                                        "change %s" % (zuul_ref, i.change))
                     return False
+            ref = 'refs/zuul/' + i.change.branch + '/' + target_ref
+            refspecs = replicate_refspecs.get(i.change.project, [])
+            refspecs.append('%s:%s' % (ref, ref))
+            replicate_refspecs[i.change.project] = refspecs
             if self.push_refs:
                 # Push the results upstream to the zuul ref after
                 # they are created.
-                ref = 'refs/zuul/' + i.change.branch + '/' + target_ref
                 try:
                     repo.push(ref, ref)
                     complete = self.trigger.waitForRefSha(i.change.project,
@@ -291,5 +321,5 @@ class Merger(object):
                     self.log.error("Ref %s did not show up in repo" % ref)
                     return False
             project_branches.append((i.change.project, i.change.branch))
-
+        self.replicateRefspecs(replicate_refspecs)
         return commit

zuul/scheduler.py

@@ -353,6 +353,11 @@ class Scheduler(threading.Thread):
         else:
             push_refs = False
 
+        replicate_urls = []
+        if self.config.has_section('replication'):
+            for k, v in self.config.items('replication'):
+                replicate_urls.append(v)
+
         if self.config.has_option('gerrit', 'sshkey'):
             sshkey = self.config.get('gerrit', 'sshkey')
         else:
@@ -363,7 +368,8 @@ class Scheduler(threading.Thread):
         # location.
         self.merger = merger.Merger(self.triggers['gerrit'],
                                     merge_root, push_refs,
-                                    sshkey, merge_email, merge_name)
+                                    sshkey, merge_email, merge_name,
+                                    replicate_urls)
         for project in self.layout.projects.values():
             url = self.triggers['gerrit'].getGitUrl(project)
             self.merger.addProject(project, url)