ReceiveCommits: expose change IDs touched; add metrics

ReceiveCommits now populates the change IDs that were successfully
updated.

AsyncReceiveCommits uses this to record the following metrics:

 receivecommits/changes - number of changes uploaded in a single push.
 receivecommits/latency - average delay per updated change
 receivecommits/timeout - rate of push timeouts

Change-Id: Ibbee943ec87141e8d06cf549f40a98f0111353b0

Documentation/metrics.txt

@@ -20,6 +20,15 @@ by RetryHelper to execute an action (1 == single attempt, no retry)
 * `action/retry_timeout_count`: Number of action executions of RetryHelper
 that ultimately timed out
 
+=== Pushes
+
+* `receivecommits/changes`: histogram of number of changes processed
+in a single upload, split up by update type (new change created,
+existing changed updated, change autoclosed).
+* `receivecommits/latency`: latency per change for processing a push,
+split up by update type (create+replace, and autoclose)
+* `receivecommits/timeout`: number of timeouts during push processing.
+
 === Process
 
 * `proc/birth_timestamp`: Time at which the Gerrit process started.

java/com/google/gerrit/server/git/receive/AsyncReceiveCommits.java

@@ -14,11 +14,21 @@
 
 package com.google.gerrit.server.git.receive;
 
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
 import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.common.Nullable;
 import com.google.gerrit.common.data.Capable;
 import com.google.gerrit.extensions.restapi.AuthException;
 import com.google.gerrit.extensions.restapi.ResourceConflictException;
+import com.google.gerrit.metrics.Counter0;
+import com.google.gerrit.metrics.Description;
+import com.google.gerrit.metrics.Description.Units;
+import com.google.gerrit.metrics.Field;
+import com.google.gerrit.metrics.Histogram1;
+import com.google.gerrit.metrics.MetricMaker;
+import com.google.gerrit.metrics.Timer1;
+import com.google.gerrit.reviewdb.client.Change;
 import com.google.gerrit.reviewdb.client.Project;
 import com.google.gerrit.server.IdentifiedUser;
 import com.google.gerrit.server.config.ConfigUtil;
@@ -164,7 +174,39 @@ public class AsyncReceiveCommits implements PreReceiveHook {
     }
   }
 
+  @Singleton
+  private static class Metrics {
+    private final Histogram1<ResultChangeIds.Key> changes;
+    private final Timer1<String> latencyPerChange;
+    private final Counter0 timeouts;
+
+    @Inject
+    Metrics(MetricMaker metricMaker) {
+      changes =
+          metricMaker.newHistogram(
+              "receivecommits/changes",
+              new Description("number of changes uploaded in a single push.").setCumulative(),
+              Field.ofEnum(
+                  ResultChangeIds.Key.class,
+                  "type",
+                  "type of update (replace, create, autoclose)"));
+      latencyPerChange =
+          metricMaker.newTimer(
+              "receivecommits/latency",
+              new Description("average delay per updated change")
+                  .setUnit(Units.MILLISECONDS)
+                  .setCumulative(),
+              Field.ofString("type", "type of update (create/replace, autoclose)"));
+
+      timeouts =
+          metricMaker.newCounter(
+              "receivecommits/timeout", new Description("rate of push timeouts").setRate());
+    }
+  }
+
+  private final Metrics metrics;
   private final ReceiveCommits receiveCommits;
+  private final ResultChangeIds resultChangeIds;
   private final PermissionBackend.ForProject perm;
   private final ReceivePack receivePack;
   private final ExecutorService executor;
@@ -188,6 +230,7 @@ public class AsyncReceiveCommits implements PreReceiveHook {
       TransferConfig transferConfig,
       Provider<LazyPostReceiveHookChain> lazyPostReceive,
       ContributorAgreementsChecker contributorAgreements,
+      Metrics metrics,
       @Named(TIMEOUT_NAME) long timeoutMillis,
       @Assisted ProjectState projectState,
       @Assisted IdentifiedUser user,
@@ -202,6 +245,7 @@ public class AsyncReceiveCommits implements PreReceiveHook {
     this.projectState = projectState;
     this.user = user;
     this.repo = repo;
+    this.metrics = metrics;
 
     Project.NameKey projectName = projectState.getNameKey();
     receivePack = new ReceivePack(repo);
@@ -237,7 +281,10 @@ public class AsyncReceiveCommits implements PreReceiveHook {
     advHooks.add(new HackPushNegotiateHook());
     receivePack.setAdvertiseRefsHook(AdvertiseRefsHookChain.newChain(advHooks));
 
-    receiveCommits = factory.create(projectState, user, receivePack, allRefsWatcher, messageSender);
+    resultChangeIds = new ResultChangeIds();
+    receiveCommits =
+        factory.create(
+            projectState, user, receivePack, allRefsWatcher, messageSender, resultChangeIds);
     receiveCommits.init();
   }
 
@@ -268,11 +315,14 @@ public class AsyncReceiveCommits implements PreReceiveHook {
       // pre-receive hooks
       return;
     }
+
+    long startNanos = System.nanoTime();
     Worker w = new Worker(commands);
     try {
       w.progress.waitFor(
           executor.submit(scopePropagator.wrap(w)), timeoutMillis, TimeUnit.MILLISECONDS);
     } catch (ExecutionException e) {
+      metrics.timeouts.increment();
       logger.atWarning().withCause(e).log(
           "Error in ReceiveCommits while processing changes for project %s",
           projectState.getName());
@@ -287,6 +337,23 @@ public class AsyncReceiveCommits implements PreReceiveHook {
     } finally {
       w.sendMessages();
     }
+
+    long deltaNanos = System.nanoTime() - startNanos;
+    int totalChanges = 0;
+    for (ResultChangeIds.Key key : ResultChangeIds.Key.values()) {
+      List<Change.Id> ids = resultChangeIds.get(key);
+      metrics.changes.record(key, ids.size());
+      totalChanges += ids.size();
+    }
+
+    if (totalChanges > 0) {
+      metrics.latencyPerChange.record(
+          resultChangeIds.get(ResultChangeIds.Key.AUTOCLOSED).isEmpty()
+              ? "CREATE_REPLACE"
+              : ResultChangeIds.Key.AUTOCLOSED.name(),
+          deltaNanos / totalChanges,
+          NANOSECONDS);
+    }
   }
 
   public ReceivePack getReceivePack() {

java/com/google/gerrit/server/git/receive/BUILD

@@ -6,6 +6,7 @@ java_library(
         "//java/com/google/gerrit/common:annotations",
         "//java/com/google/gerrit/common:server",
         "//java/com/google/gerrit/extensions:api",
+        "//java/com/google/gerrit/metrics",
         "//java/com/google/gerrit/reviewdb:server",
         "//java/com/google/gerrit/server",
         "//java/com/google/gerrit/server/logging",

java/com/google/gerrit/server/git/receive/ReceiveCommits.java

@@ -107,6 +107,7 @@ import com.google.gerrit.server.git.MultiProgressMonitor.Task;
 import com.google.gerrit.server.git.ReceivePackInitializer;
 import com.google.gerrit.server.git.TagCache;
 import com.google.gerrit.server.git.ValidationError;
+import com.google.gerrit.server.git.receive.ResultChangeIds.Key;
 import com.google.gerrit.server.git.validators.CommitValidationMessage;
 import com.google.gerrit.server.git.validators.RefOperationValidationException;
 import com.google.gerrit.server.git.validators.RefOperationValidators;
@@ -228,7 +229,8 @@ class ReceiveCommits {
         IdentifiedUser user,
         ReceivePack receivePack,
         AllRefsWatcher allRefsWatcher,
-        MessageSender messageSender);
+        MessageSender messageSender,
+        ResultChangeIds resultChangeIds);
   }
 
   private class ReceivePackMessageSender implements MessageSender {
@@ -352,6 +354,7 @@ class ReceiveCommits {
   private Optional<String> tracePushOption;
 
   private MessageSender messageSender;
+  private ResultChangeIds resultChangeIds;
 
   @Inject
   ReceiveCommits(
@@ -393,7 +396,8 @@ class ReceiveCommits {
       @Assisted IdentifiedUser user,
       @Assisted ReceivePack rp,
       @Assisted AllRefsWatcher allRefsWatcher,
-      @Nullable @Assisted MessageSender messageSender)
+      @Nullable @Assisted MessageSender messageSender,
+      @Assisted ResultChangeIds resultChangeIds)
       throws IOException {
     // Injected fields.
     this.accountResolver = accountResolver;
@@ -460,6 +464,7 @@ class ReceiveCommits {
 
     // Handles for outputting back over the wire to the end user.
     this.messageSender = messageSender != null ? messageSender : new ReceivePackMessageSender();
+    this.resultChangeIds = resultChangeIds;
   }
 
   void init() {
@@ -810,6 +815,13 @@ class ReceiveCommits {
       } catch (UpdateException e) {
         throw INSERT_EXCEPTION.apply(e);
       }
+
+      replaceByChange
+          .values()
+          .stream()
+          .forEach(req -> resultChangeIds.add(Key.REPLACED, req.ontoChange));
+      newChanges.stream().forEach(req -> resultChangeIds.add(Key.CREATED, req.changeId));
+
       if (magicBranchCmd != null) {
         magicBranchCmd.setResult(OK);
       }
@@ -3031,6 +3043,7 @@ class ReceiveCommits {
   private void autoCloseChanges(ReceiveCommand cmd, Task progress) {
     logger.atFine().log("Starting auto-closing of changes");
     String refName = cmd.getRefName();
+    Set<Change.Id> ids = new HashSet<>();
 
     // TODO(dborowitz): Combine this BatchUpdate with the main one in
     // handleRegularCommands
@@ -3107,6 +3120,7 @@ class ReceiveCommits {
                         .create(requestScopePropagator, req.psId, refName)
                         .setPatchSetProvider(req.replaceOp::getPatchSet));
                 bu.addOp(id, new ChangeProgressOp(progress));
+                ids.add(id);
               }
 
               logger.atFine().log(
@@ -3115,7 +3129,14 @@ class ReceiveCommits {
               bu.execute();
             } catch (IOException | OrmException | PermissionBackendException e) {
               logger.atSevere().withCause(e).log("Failed to auto-close changes");
+              return null;
             }
+
+            // If we are here, we didn't throw UpdateException. Record the result.
+            // The ordering is indeterminate due to the HashSet; unfortunately, Change.Id doesn't
+            // fit into TreeSet.
+            ids.stream().forEach(id -> resultChangeIds.add(Key.AUTOCLOSED, id));
+
             return null;
           },
           // Use a multiple of the default timeout to account for inner retries that may otherwise

java/com/google/gerrit/server/git/receive/ResultChangeIds.java

@@ -0,0 +1,58 @@
+// Copyright (C) 2018 The Android Open Source Project
+//
+// 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.server.git.receive;
+
+import com.google.common.collect.ImmutableList;
+import com.google.gerrit.reviewdb.client.Change;
+import java.util.ArrayList;
+import java.util.EnumMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Keeps track of the change IDs thus far updated by ReceiveCommit.
+ *
+ * <p>This class is thread-safe.
+ */
+public class ResultChangeIds {
+  enum Key {
+    CREATED,
+    REPLACED,
+    AUTOCLOSED,
+  }
+
+  private Map<Key, List<Change.Id>> ids;
+
+  ResultChangeIds() {
+    ids = new EnumMap<>(Key.class);
+    for (Key k : Key.values()) {
+      ids.put(k, new ArrayList<>());
+    }
+  }
+
+  /** Record a change ID update as having completed. Thread-safe. */
+  public void add(Key key, Change.Id id) {
+    synchronized (this) {
+      ids.get(key).add(id);
+    }
+  }
+
+  /** Returns change IDs of the given type for which the BatchUpdate succeeded. Thread-safe. */
+  public List<Change.Id> get(Key key) {
+    synchronized (this) {
+      return ImmutableList.copyOf(ids.get(key));
+    }
+  }
+}