Automatically push changes to refs as they are applied locally

If configured, $site_path/replication.config is a git style config
file listing remotes which can be pushed to automatically in order
to update mirror sites in the background.

An example replication.config might say:

  [remote "mirror"]
    other.system:/base/path/${name}.git

At runtime the ${name} placeholder is replaced with the current project
name in Gerrit.  This implies that the remote mirror must use the same
project layout on disk.

Only modified refs are actually pushed, slightly reducing the load on
the Gerrit side of the connection.

The merge worker queue is used to handle the replication, with one
job per destination/project pair, to allow parallel replications.

Signed-off-by: Shawn O. Pearce <sop@google.com>

Documentation/config-replication.txt

@@ -0,0 +1,85 @@
+Gerrit2 Git Replication
+=======================
+
+Gerrit can automatically push any changes it makes to its local Git
+repositories to another system.  Usually this would be configured
+to provide mirroring of changes, for warm-standby backups or a
+load-balanced public mirror farm.
+
+Typically replication should be done over SSH, with a passwordless
+public/private key pair.  On a trusted network it is also possible to
+use replication over the insecure (but much faster) git:// protocol,
+by enabling the `receive-pack` service on the receiving system.
+
+To enable push replication, create `$\{site_path\}/replication.config`
+as a Git-style config file, then restart Gerrit.
+
+Currently the replication runs on a 15 second delay.  This gives
+Gerrit a short time window to batch updates going to the same
+project, such as when a user uploads multiple changes at once.
+
+An example configuration file to replicate in parallel to four
+different hosts follows:
+
+====
+  [remote "host-one"]
+    url = gerrit2@host-one.example.com:/some/path/${name}.git
+    # push defaults to +refs/*:refs/*
+
+  [remote "pubmirror"]
+    url = mirror1.us.some.org:/pub/git/${name}.git
+    url = mirror2.us.some.org:/pub/git/${name}.git
+    url = mirror3.us.some.org:/pub/git/${name}.git
+    push = +refs/heads/*
+    push = +refs/tags/*
+====
+
+Within each url value the magic placeholder `$\{name}` is replaced
+with the Gerrit project name.  This is a Gerrit specific extension
+to the Git URL syntax.
+
+Multiple URLs may be specified within a single remote block,
+listing different destinations which share the same settings.
+
+Most standard Git remote configuration settings can be used within
+a remote block.  The `receivepack` option can be used to specify
+the remote path to the `git-receive-pack` executable, if it is
+not in the remote user's `PATH` by default.
+
+One or more `push` entries may also be specified within a remote
+block to select a subset of the refs for replication.  If no push
+values are configured, a default of `+refs/\*:refs/*` is used.
+
+As the above configuration uses SSH for both connections it also
+relies upon having a `~/.ssh/config` file setup for the user
+running Gerrit.  A matching configuration file might be:
+
+====
+  Host host-one.example.com:
+    IdentityFile ~/.ssh/id_hostone
+    PreferredAuthentications publickey
+    BatchMode yes
+
+  Host mirror*.us.some.org:
+    User mirror-updater
+    IdentityFile ~/.ssh/id_pubmirror
+    PreferredAuthentications publickey
+    BatchMode yes
+====
+
+Most (but not all) SSH options are honored:
+
+ * Host
+ * Hostname
+ * User
+ * Port
+ * IdentityFile
+ * PreferredAuthentications
+ * BatchMode
+
+SSH authentication must be by passwordless public key, as there is
+no facility to read passphases on startup or passwords during the
+SSH connection setup.
+
+Make sure you are replicating to a standard Git daemon or SSH
+daemon, and not to another Gerrit's internal SSHD.

Documentation/index.txt

@@ -16,6 +16,7 @@ Configuration
 -------------
 
 * link:config-gerrit.html[system_config Settings]
+* link:config-replication.html[Git Replication/Mirroring]
 * link:config-gitweb.html[Gitweb Integration]
 * link:config-headerfooter.html[Site Header/Footer]
 * link:config-sso.html[Single Sign-On Systems]

Documentation/install.txt

@@ -146,6 +146,7 @@ Gerrit2 supports some site-specific customizations.  These are
 optional and are not required to run a server, but may be desired.
 
 * link:config-sso.html[Single Sign-On Systems]
+* link:config-replication.html[Git Replication/Mirroring]
 * link:config-headerfooter.html[Site Header/Footer]
 * link:config-gitweb.html[Gitweb Integration]
 * link:config-gerrit.html[Other system_config Settings]

appjar/src/main/java/com/google/gerrit/git/MergeOp.java

@@ -370,6 +370,8 @@ public class MergeOp {
         switch (branchUpdate.update(rw)) {
           case NEW:
           case FAST_FORWARD:
+            PushQueue.scheduleUpdate(destBranch.getParentKey(), branchUpdate
+                .getName());
             break;
 
           default:

appjar/src/main/java/com/google/gerrit/git/PushQueue.java

@@ -0,0 +1,263 @@
+// Copyright 2009 Google Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.gerrit.git;
+
+import com.google.gerrit.client.reviewdb.Project;
+import com.google.gerrit.server.GerritServer;
+import com.google.gwtjsonrpc.server.XsrfException;
+import com.google.gwtorm.client.OrmException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.spearce.jgit.errors.NotSupportedException;
+import org.spearce.jgit.errors.TransportException;
+import org.spearce.jgit.lib.NullProgressMonitor;
+import org.spearce.jgit.lib.Repository;
+import org.spearce.jgit.lib.RepositoryConfig;
+import org.spearce.jgit.transport.PushResult;
+import org.spearce.jgit.transport.RefSpec;
+import org.spearce.jgit.transport.RemoteConfig;
+import org.spearce.jgit.transport.RemoteRefUpdate;
+import org.spearce.jgit.transport.Transport;
+import org.spearce.jgit.transport.URIish;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+public class PushQueue {
+  private static final Logger log = LoggerFactory.getLogger(PushQueue.class);
+  private static final int startDelay = 15; // seconds
+  private static List<RemoteConfig> configs;
+  private static final Map<URIish, PushOp> active =
+      new HashMap<URIish, PushOp>();
+
+  public static void scheduleUpdate(final Project.NameKey project,
+      final String ref) {
+    for (final RemoteConfig srcConf : allConfigs()) {
+      RefSpec spec = null;
+      for (final RefSpec s : srcConf.getPushRefSpecs()) {
+        if (s.matchSource(ref)) {
+          spec = s;
+          break;
+        }
+      }
+      if (spec == null) {
+        continue;
+      }
+
+      for (URIish uri : srcConf.getURIs()) {
+        uri = uri.setPath(replace(uri.getPath(), "name", project.get()));
+        scheduleImp(project, ref, srcConf, uri);
+      }
+    }
+  }
+
+  private static synchronized void scheduleImp(final Project.NameKey project,
+      final String ref, final RemoteConfig srcConf, final URIish uri) {
+    PushOp e = active.get(uri);
+    if (e == null) {
+      final PushOp newOp = new PushOp(project.get(), srcConf, uri);
+      WorkQueue.schedule(new Runnable() {
+        public void run() {
+          try {
+            pushImpl(newOp);
+          } catch (RuntimeException e) {
+            log.error("Unexpected error during replication", e);
+          } catch (Error e) {
+            log.error("Unexpected error during replication", e);
+          }
+        }
+      }, startDelay, TimeUnit.SECONDS);
+      active.put(uri, newOp);
+      e = newOp;
+    }
+    e.delta.add(ref);
+  }
+
+  private static void pushImpl(final PushOp op) {
+    removeFromActive(op);
+    final Repository db;
+    try {
+      db = GerritServer.getInstance().getRepositoryCache().get(op.projectName);
+    } catch (OrmException e) {
+      log.error("Cannot open repository cache", e);
+      return;
+    } catch (XsrfException e) {
+      log.error("Cannot open repository cache", e);
+      return;
+    } catch (InvalidRepositoryException e) {
+      log.error("Cannot replicate " + op.projectName, e);
+      return;
+    }
+
+    final ArrayList<RemoteRefUpdate> cmds = new ArrayList<RemoteRefUpdate>();
+    try {
+      for (final String ref : op.delta) {
+        final String src = ref;
+        RefSpec spec = null;
+        for (final RefSpec s : op.config.getPushRefSpecs()) {
+          if (s.matchSource(src)) {
+            spec = s.expandFromSource(src);
+            break;
+          }
+        }
+        if (spec == null) {
+          continue;
+        }
+
+        final String dst = spec.getDestination();
+        final boolean force = spec.isForceUpdate();
+        cmds.add(new RemoteRefUpdate(db, src, dst, force, null, null));
+      }
+    } catch (IOException e) {
+      log.error("Cannot replicate " + op.projectName, e);
+      return;
+    }
+
+    final Transport tn;
+    try {
+      tn = Transport.open(db, op.uri);
+      tn.applyConfig(op.config);
+    } catch (NotSupportedException e) {
+      log.error("Cannot replicate to " + op.uri, e);
+      return;
+    }
+
+    final PushResult res;
+    try {
+      res = tn.push(NullProgressMonitor.INSTANCE, cmds);
+    } catch (NotSupportedException e) {
+      log.error("Cannot replicate to " + op.uri, e);
+      return;
+    } catch (TransportException e) {
+      log.error("Cannot replicate to " + op.uri, e);
+      return;
+    } finally {
+      tn.close();
+    }
+
+    for (final RemoteRefUpdate u : res.getRemoteUpdates()) {
+      switch (u.getStatus()) {
+        case OK:
+        case UP_TO_DATE:
+        case NON_EXISTING:
+          break;
+
+        case NOT_ATTEMPTED:
+        case AWAITING_REPORT:
+        case REJECTED_NODELETE:
+        case REJECTED_NONFASTFORWARD:
+        case REJECTED_REMOTE_CHANGED:
+          log.error("Failed replicate of " + u.getRemoteName() + " to "
+              + op.uri + ": status " + u.getStatus().name());
+          break;
+
+        case REJECTED_OTHER_REASON:
+          log.error("Failed replicate of " + u.getRemoteName() + " to "
+              + op.uri + ", reason: " + u.getMessage());
+          break;
+      }
+    }
+  }
+
+  private static synchronized void removeFromActive(final PushOp op) {
+    active.remove(op.uri);
+  }
+
+  private static String replace(final String pat, final String key,
+      final String val) {
+    final int n = pat.indexOf("${" + key + "}");
+    return pat.substring(0, n) + val + pat.substring(n + 3 + key.length());
+  }
+
+  private static synchronized List<RemoteConfig> allConfigs() {
+    if (configs == null) {
+      final File path;
+      try {
+        final GerritServer gs = GerritServer.getInstance();
+        path = gs.getSitePath();
+        if (path == null || gs.getRepositoryCache() == null) {
+          return Collections.emptyList();
+        }
+      } catch (OrmException e) {
+        return Collections.emptyList();
+      } catch (XsrfException e) {
+        return Collections.emptyList();
+      }
+
+      final File cfgFile = new File(path, "replication.config");
+      final RepositoryConfig cfg = new RepositoryConfig(null, cfgFile);
+      try {
+        cfg.load();
+        final ArrayList<RemoteConfig> r = new ArrayList<RemoteConfig>();
+        for (final RemoteConfig c : RemoteConfig.getAllRemoteConfigs(cfg)) {
+          if (c.getURIs().isEmpty()) {
+            continue;
+          }
+
+          for (final URIish u : c.getURIs()) {
+            if (u.getPath() == null || !u.getPath().contains("${name}")) {
+              final String s = u.toString();
+              throw new URISyntaxException(s, "No ${name}");
+            }
+          }
+
+          if (c.getPushRefSpecs().isEmpty()) {
+            RefSpec spec = new RefSpec();
+            spec = spec.setSourceDestination("refs/*", "refs/*");
+            spec = spec.setForceUpdate(true);
+            c.addPushRefSpec(spec);
+          }
+
+          r.add(c);
+        }
+        configs = Collections.unmodifiableList(r);
+      } catch (FileNotFoundException e) {
+        log.warn("No " + cfgFile + "; not replicating");
+        configs = Collections.emptyList();
+      } catch (IOException e) {
+        log.error("Can't read " + cfgFile, e);
+        return Collections.emptyList();
+      } catch (URISyntaxException e) {
+        log.error("Invalid URI in " + cfgFile, e);
+        return Collections.emptyList();
+      }
+    }
+    return configs;
+  }
+
+  private static class PushOp {
+    final Set<String> delta = new HashSet<String>();
+    final String projectName;
+    final RemoteConfig config;
+    final URIish uri;
+
+    PushOp(final String d, final RemoteConfig c, final URIish u) {
+      projectName = d;
+      config = c;
+      uri = u;
+    }
+  }
+}

appjar/src/main/java/com/google/gerrit/server/ssh/Receive.java

@@ -31,6 +31,7 @@ import com.google.gerrit.client.reviewdb.PatchSet;
 import com.google.gerrit.client.reviewdb.ReviewDb;
 import com.google.gerrit.client.rpc.Common;
 import com.google.gerrit.git.PatchSetImporter;
+import com.google.gerrit.git.PushQueue;
 import com.google.gerrit.server.ChangeMail;
 import com.google.gerrit.server.ChangeUtil;
 import com.google.gerrit.server.GerritServer;
@@ -529,6 +530,7 @@ class Receive extends AbstractGitCommand {
     ru.setForceUpdate(true);
     ru.setNewObjectId(c);
     ru.update(walk);
+    PushQueue.scheduleUpdate(proj.getNameKey(), ru.getName());
 
     allNewChanges.add(change.getId());
 
@@ -687,6 +689,7 @@ class Receive extends AbstractGitCommand {
       ru.setForceUpdate(true);
       ru.setNewObjectId(c);
       ru.update(rp.getRevWalk());
+      PushQueue.scheduleUpdate(proj.getNameKey(), ru.getName());
       cmd.setResult(ReceiveCommand.Result.OK);
     }
   }