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

// Copyright (C) 2012 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 static com.google.gerrit.server.quota.QuotaGroupDefinitions.REPOSITORY_SIZE_GROUP;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

import com.google.common.flogger.FluentLogger;
import com.google.gerrit.common.Nullable;
import com.google.gerrit.common.UsedAt;
import com.google.gerrit.common.data.Capable;
import com.google.gerrit.entities.Change;
import com.google.gerrit.entities.Project;
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.server.IdentifiedUser;
import com.google.gerrit.server.config.AllUsersName;
import com.google.gerrit.server.config.ConfigUtil;
import com.google.gerrit.server.config.GerritServerConfig;
import com.google.gerrit.server.config.ReceiveCommitsExecutor;
import com.google.gerrit.server.git.MultiProgressMonitor;
import com.google.gerrit.server.git.PermissionAwareRepositoryManager;
import com.google.gerrit.server.git.ProjectRunnable;
import com.google.gerrit.server.git.TransferConfig;
import com.google.gerrit.server.git.UsersSelfAdvertiseRefsHook;
import com.google.gerrit.server.logging.Metadata;
import com.google.gerrit.server.permissions.PermissionBackend;
import com.google.gerrit.server.permissions.PermissionBackendException;
import com.google.gerrit.server.permissions.ProjectPermission;
import com.google.gerrit.server.project.ContributorAgreementsChecker;
import com.google.gerrit.server.project.ProjectState;
import com.google.gerrit.server.query.change.InternalChangeQuery;
import com.google.gerrit.server.quota.QuotaBackend;
import com.google.gerrit.server.quota.QuotaException;
import com.google.gerrit.server.quota.QuotaResponse;
import com.google.gerrit.server.util.MagicBranch;
import com.google.gerrit.server.util.RequestScopePropagator;
import com.google.inject.Inject;
import com.google.inject.PrivateModule;
import com.google.inject.Provider;
import com.google.inject.Provides;
import com.google.inject.Singleton;
import com.google.inject.assistedinject.Assisted;
import com.google.inject.assistedinject.FactoryModuleBuilder;
import com.google.inject.name.Named;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import org.eclipse.jgit.lib.Config;
import org.eclipse.jgit.lib.Repository;
import org.eclipse.jgit.transport.PreReceiveHook;
import org.eclipse.jgit.transport.ReceiveCommand;
import org.eclipse.jgit.transport.ReceiveCommand.Result;
import org.eclipse.jgit.transport.ReceivePack;

/**
 * Hook that delegates to {@link ReceiveCommits} in a worker thread.
 *
 * <p>Since the work that {@link ReceiveCommits} does may take a long, potentially unbounded amount
 * of time, it runs in the background so it can be monitored for timeouts and cancelled, and have
 * stalls reported to the user from the main thread.
 */
public class AsyncReceiveCommits implements PreReceiveHook {
  private static final FluentLogger logger = FluentLogger.forEnclosingClass();

  private static final String TIMEOUT_NAME = "ReceiveCommitsOverallTimeout";

  public interface Factory {
    AsyncReceiveCommits create(
        ProjectState projectState,
        IdentifiedUser user,
        Repository repository,
        @Nullable MessageSender messageSender);
  }

  public static class Module extends PrivateModule {
    @Override
    public void configure() {
      install(new FactoryModuleBuilder().build(LazyPostReceiveHookChain.Factory.class));
      install(new FactoryModuleBuilder().build(AsyncReceiveCommits.Factory.class));
      expose(AsyncReceiveCommits.Factory.class);
      // Don't expose the binding for ReceiveCommits.Factory. All callers should
      // be using AsyncReceiveCommits.Factory instead.
      install(new FactoryModuleBuilder().build(ReceiveCommits.Factory.class));
      install(new FactoryModuleBuilder().build(BranchCommitValidator.Factory.class));
    }

    @Provides
    @Singleton
    @Named(TIMEOUT_NAME)
    long getTimeoutMillis(@GerritServerConfig Config cfg) {
      return ConfigUtil.getTimeUnit(
          cfg, "receive", null, "timeout", TimeUnit.MINUTES.toMillis(4), TimeUnit.MILLISECONDS);
    }
  }

  private class Worker implements ProjectRunnable {
    final MultiProgressMonitor progress;
    final String name;

    private final Collection<ReceiveCommand> commands;

    private Worker(Collection<ReceiveCommand> commands, String name) {
      this.commands = commands;
      this.name = name;
      progress = new MultiProgressMonitor(new MessageSenderOutputStream(), "Processing changes");
    }

    @Override
    public void run() {
      String oldName = Thread.currentThread().getName();
      Thread.currentThread().setName(oldName + "-for-" + name);
      try {
        receiveCommits.processCommands(commands, progress);
      } finally {
        Thread.currentThread().setName(oldName);
      }
    }

    @Override
    public Project.NameKey getProjectNameKey() {
      return receiveCommits.getProject().getNameKey();
    }

    @Override
    public String getRemoteName() {
      return null;
    }

    @Override
    public boolean hasCustomizedPrint() {
      return true;
    }

    @Override
    public String toString() {
      return "receive-commits";
    }

    void sendMessages() {
      receiveCommits.sendMessages();
    }

    private class MessageSenderOutputStream extends OutputStream {
      @Override
      public void write(int b) {
        receiveCommits.getMessageSender().sendBytes(new byte[] {(byte) b});
      }

      @Override
      public void write(byte[] what, int off, int len) {
        receiveCommits.getMessageSender().sendBytes(what, off, len);
      }

      @Override
      public void write(byte[] what) {
        receiveCommits.getMessageSender().sendBytes(what);
      }

      @Override
      public void flush() {
        receiveCommits.getMessageSender().flush();
      }
    }
  }

  private enum PushType {
    CREATE_REPLACE,
    NORMAL,
    AUTOCLOSE,
  }

  @Singleton
  private static class Metrics {
    private final Histogram1<PushType> changes;
    private final Timer1<PushType> latencyPerChange;
    private final Timer1<PushType> latencyPerPush;
    private final Counter0 timeouts;

    @Inject
    Metrics(MetricMaker metricMaker) {
      // For the changes metric the push type field is never set to PushType.NORMAL, hence it is not
      // mentioned in the field description.
      changes =
          metricMaker.newHistogram(
              "receivecommits/changes_per_push",
              new Description("number of changes uploaded in a single push.").setCumulative(),
              Field.ofEnum(PushType.class, "type", Metadata.Builder::pushType)
                  .description("type of push (create/replace, autoclose)")
                  .build());

      Field<PushType> pushTypeField =
          Field.ofEnum(PushType.class, "type", Metadata.Builder::pushType)
              .description("type of push (create/replace, autoclose, normal)")
              .build();

      latencyPerChange =
          metricMaker.newTimer(
              "receivecommits/latency_per_push_per_change",
              new Description(
                      "Processing delay per push divided by the number of changes in said push. "
                          + "(Only includes pushes which contain changes.)")
                  .setUnit(Units.MILLISECONDS)
                  .setCumulative(),
              pushTypeField);

      latencyPerPush =
          metricMaker.newTimer(
              "receivecommits/latency_per_push",
              new Description("processing delay for a processing single push")
                  .setUnit(Units.MILLISECONDS)
                  .setCumulative(),
              pushTypeField);

      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;
  private final RequestScopePropagator scopePropagator;
  private final ReceiveConfig receiveConfig;
  private final ContributorAgreementsChecker contributorAgreements;
  private final long timeoutMillis;
  private final ProjectState projectState;
  private final IdentifiedUser user;
  private final Repository repo;
  private final AllRefsWatcher allRefsWatcher;

  @Inject
  AsyncReceiveCommits(
      ReceiveCommits.Factory factory,
      PermissionBackend permissionBackend,
      Provider<InternalChangeQuery> queryProvider,
      @ReceiveCommitsExecutor ExecutorService executor,
      RequestScopePropagator scopePropagator,
      ReceiveConfig receiveConfig,
      TransferConfig transferConfig,
      LazyPostReceiveHookChain.Factory lazyPostReceive,
      ContributorAgreementsChecker contributorAgreements,
      Metrics metrics,
      QuotaBackend quotaBackend,
      UsersSelfAdvertiseRefsHook usersSelfAdvertiseRefsHook,
      AllUsersName allUsersName,
      @Named(TIMEOUT_NAME) long timeoutMillis,
      @Assisted ProjectState projectState,
      @Assisted IdentifiedUser user,
      @Assisted Repository repo,
      @Assisted @Nullable MessageSender messageSender)
      throws PermissionBackendException {
    this.executor = executor;
    this.scopePropagator = scopePropagator;
    this.receiveConfig = receiveConfig;
    this.contributorAgreements = contributorAgreements;
    this.timeoutMillis = timeoutMillis;
    this.projectState = projectState;
    this.user = user;
    this.repo = repo;
    this.metrics = metrics;
    // If the user lacks READ permission, some references may be filtered and hidden from view.
    // Check objects mentioned inside the incoming pack file are reachable from visible refs.
    Project.NameKey projectName = projectState.getNameKey();
    this.perm = permissionBackend.user(user).project(projectName);

    receivePack = new ReceivePack(PermissionAwareRepositoryManager.wrap(repo, perm));
    receivePack.setAllowCreates(true);
    receivePack.setAllowDeletes(true);
    receivePack.setAllowNonFastForwards(true);
    receivePack.setRefLogIdent(user.newRefLogIdent());
    receivePack.setTimeout(transferConfig.getTimeout());
    receivePack.setMaxObjectSizeLimit(projectState.getEffectiveMaxObjectSizeLimit().value);
    receivePack.setCheckReceivedObjects(projectState.getConfig().getCheckReceivedObjects());
    receivePack.setRefFilter(new ReceiveRefFilter());
    receivePack.setAllowPushOptions(true);
    receivePack.setPreReceiveHook(this);
    receivePack.setPostReceiveHook(lazyPostReceive.create(user, projectName));

    try {
      projectState.checkStatePermitsRead();
      this.perm.check(ProjectPermission.READ);
    } catch (AuthException | ResourceConflictException e) {
      receivePack.setCheckReferencedObjectsAreReachable(
          receiveConfig.checkReferencedObjectsAreReachable);
    }

    allRefsWatcher = new AllRefsWatcher();
    receivePack.setAdvertiseRefsHook(
        ReceiveCommitsAdvertiseRefsHookChain.create(
            allRefsWatcher, usersSelfAdvertiseRefsHook, allUsersName, queryProvider, projectName));
    resultChangeIds = new ResultChangeIds();
    receiveCommits =
        factory.create(
            projectState, user, receivePack, repo, allRefsWatcher, messageSender, resultChangeIds);
    receiveCommits.init();
    QuotaResponse.Aggregated availableTokens =
        quotaBackend.user(user).project(projectName).availableTokens(REPOSITORY_SIZE_GROUP);
    try {
      availableTokens.throwOnError();
    } catch (QuotaException e) {
      logger.atWarning().withCause(e).log(
          "Quota %s availableTokens request failed for project %s",
          REPOSITORY_SIZE_GROUP, projectName);
      throw new RuntimeException(e);
    }
    availableTokens.availableTokens().ifPresent(v -> receivePack.setMaxObjectSizeLimit(v));
  }

  /** Determine if the user can upload commits. */
  public Capable canUpload() throws IOException, PermissionBackendException {
    try {
      perm.check(ProjectPermission.PUSH_AT_LEAST_ONE_REF);
    } catch (AuthException e) {
      return new Capable("Upload denied for project '" + projectState.getName() + "'");
    }

    try {
      contributorAgreements.check(projectState.getNameKey(), user);
    } catch (AuthException e) {
      return new Capable(e.getMessage());
    }

    if (receiveConfig.checkMagicRefs) {
      return MagicBranch.checkMagicBranchRefs(repo, projectState.getProject());
    }
    return Capable.OK;
  }

  @Override
  public void onPreReceive(ReceivePack rp, Collection<ReceiveCommand> commands) {
    if (commands.stream().anyMatch(c -> c.getResult() != Result.NOT_ATTEMPTED)) {
      // Stop processing when command was already processed by previously invoked
      // pre-receive hooks
      return;
    }

    long startNanos = System.nanoTime();
    Worker w = new Worker(commands, Thread.currentThread().getName());
    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());
      rp.sendError("internal error while processing changes");
      // ReceiveCommits has tried its best to catch errors, so anything at this
      // point is very bad.
      for (ReceiveCommand c : commands) {
        if (c.getResult() == Result.NOT_ATTEMPTED) {
          c.setResult(Result.REJECTED_OTHER_REASON, "internal error");
        }
      }
    } finally {
      w.sendMessages();
    }

    long deltaNanos = System.nanoTime() - startNanos;
    int totalChanges = 0;

    PushType pushType;
    if (resultChangeIds.isMagicPush()) {
      pushType = PushType.CREATE_REPLACE;
      List<Change.Id> created = resultChangeIds.get(ResultChangeIds.Key.CREATED);
      List<Change.Id> replaced = resultChangeIds.get(ResultChangeIds.Key.REPLACED);
      metrics.changes.record(pushType, created.size() + replaced.size());
      totalChanges = replaced.size() + created.size();
    } else {
      List<Change.Id> autoclosed = resultChangeIds.get(ResultChangeIds.Key.AUTOCLOSED);
      if (!autoclosed.isEmpty()) {
        pushType = PushType.AUTOCLOSE;
        metrics.changes.record(pushType, autoclosed.size());
        totalChanges = autoclosed.size();
      } else {
        pushType = PushType.NORMAL;
      }
    }

    if (totalChanges > 0) {
      metrics.latencyPerChange.record(pushType, deltaNanos / totalChanges, NANOSECONDS);
    }

    metrics.latencyPerPush.record(pushType, deltaNanos, NANOSECONDS);
  }

  /** Returns the Change.Ids that were processed in onPreReceive */
  @UsedAt(UsedAt.Project.GOOGLE)
  public ResultChangeIds getResultChangeIds() {
    return resultChangeIds;
  }

  public ReceivePack getReceivePack() {
    return receivePack;
  }
}