[6.3.0] Implement failure circuit breaker

src/main/java/com/google/devtools/build/lib/remote/BUILD

@@ -7,6 +7,7 @@ package(
 filegroup(
     name = "srcs",
     srcs = glob(["*"]) + [
+        "//src/main/java/com/google/devtools/build/lib/remote/circuitbreaker:srcs",
         "//src/main/java/com/google/devtools/build/lib/remote/common:srcs",
         "//src/main/java/com/google/devtools/build/lib/remote/downloader:srcs",
         "//src/main/java/com/google/devtools/build/lib/remote/disk:srcs",
@@ -84,6 +85,7 @@ java_library(
         "//src/main/java/com/google/devtools/build/lib/exec/local",
         "//src/main/java/com/google/devtools/build/lib/packages/semantics",
         "//src/main/java/com/google/devtools/build/lib/profiler",
+        "//src/main/java/com/google/devtools/build/lib/remote/circuitbreaker",
         "//src/main/java/com/google/devtools/build/lib/remote/common",
         "//src/main/java/com/google/devtools/build/lib/remote/common:cache_not_found_exception",
         "//src/main/java/com/google/devtools/build/lib/remote/disk",

src/main/java/com/google/devtools/build/lib/remote/GrpcCacheClient.java

@@ -495,4 +495,8 @@ ListenableFuture<Void> uploadChunker(
         MoreExecutors.directExecutor());
     return f;
   }
+
+  Retrier getRetrier() {
+    return this.retrier;
+  }
 }

src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutor.java

@@ -243,4 +243,8 @@ public void close() {
     }
     channel.release();
   }
+
+  RemoteRetrier getRetrier() {
+    return this.retrier;
+  }
 }

src/main/java/com/google/devtools/build/lib/remote/RemoteModule.java

@@ -62,6 +62,7 @@
 import com.google.devtools.build.lib.exec.SpawnStrategyRegistry;
 import com.google.devtools.build.lib.remote.RemoteServerCapabilities.ServerCapabilitiesRequirement;
 import com.google.devtools.build.lib.remote.ToplevelArtifactsDownloader.PathToMetadataConverter;
+import com.google.devtools.build.lib.remote.circuitbreaker.CircuitBreakerFactory;
 import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
 import com.google.devtools.build.lib.remote.common.RemoteExecutionClient;
 import com.google.devtools.build.lib.remote.downloader.GrpcRemoteDownloader;
@@ -475,12 +476,11 @@ public void beforeCommand(CommandEnvironment env) throws AbruptExitException {
         GoogleAuthUtils.newCallCredentialsProvider(credentials);
     CallCredentials callCredentials = callCredentialsProvider.getCallCredentials();
 
+    Retrier.CircuitBreaker circuitBreaker =
+        CircuitBreakerFactory.createCircuitBreaker(remoteOptions);
     RemoteRetrier retrier =
         new RemoteRetrier(
-            remoteOptions,
-            RemoteRetrier.RETRIABLE_GRPC_ERRORS,
-            retryScheduler,
-            Retrier.ALLOW_ALL_CALLS);
+            remoteOptions, RemoteRetrier.RETRIABLE_GRPC_ERRORS, retryScheduler, circuitBreaker);
 
     // We only check required capabilities for a given endpoint.
     //
@@ -598,7 +598,7 @@ public void beforeCommand(CommandEnvironment env) throws AbruptExitException {
                 remoteOptions,
                 RemoteRetrier.RETRIABLE_GRPC_ERRORS, // Handle NOT_FOUND internally
                 retryScheduler,
-                Retrier.ALLOW_ALL_CALLS);
+                circuitBreaker);
         remoteExecutor =
             new ExperimentalGrpcRemoteExecutor(
                 remoteOptions, execChannel.retain(), callCredentialsProvider, execRetrier);
@@ -608,7 +608,7 @@ public void beforeCommand(CommandEnvironment env) throws AbruptExitException {
                 remoteOptions,
                 RemoteRetrier.RETRIABLE_GRPC_EXEC_ERRORS,
                 retryScheduler,
-                Retrier.ALLOW_ALL_CALLS);
+                circuitBreaker);
         remoteExecutor =
             new GrpcRemoteExecutor(execChannel.retain(), callCredentialsProvider, execRetrier);
       }

src/main/java/com/google/devtools/build/lib/remote/RemoteSpawnRunner.java

@@ -58,6 +58,7 @@
 import com.google.devtools.build.lib.profiler.SilentCloseable;
 import com.google.devtools.build.lib.remote.RemoteExecutionService.RemoteActionResult;
 import com.google.devtools.build.lib.remote.RemoteExecutionService.ServerLogs;
+import com.google.devtools.build.lib.remote.circuitbreaker.CircuitBreakerFactory;
 import com.google.devtools.build.lib.remote.common.BulkTransferException;
 import com.google.devtools.build.lib.remote.common.OperationObserver;
 import com.google.devtools.build.lib.remote.options.RemoteOptions;
@@ -660,6 +661,8 @@ private void report(Event evt) {
   private static RemoteRetrier createExecuteRetrier(
       RemoteOptions options, ListeningScheduledExecutorService retryService) {
     return new ExecuteRetrier(
-        options.remoteMaxRetryAttempts, retryService, Retrier.ALLOW_ALL_CALLS);
+        options.remoteMaxRetryAttempts,
+        retryService,
+        CircuitBreakerFactory.createCircuitBreaker(options));
   }
 }

src/main/java/com/google/devtools/build/lib/remote/Retrier.java

@@ -100,7 +100,7 @@ enum State {
     State state();
 
     /** Called after an execution failed. */
-    void recordFailure();
+    void recordFailure(Exception e);
 
     /** Called after an execution succeeded. */
     void recordSuccess();
@@ -130,7 +130,7 @@ public State state() {
         }
 
         @Override
-        public void recordFailure() {}
+        public void recordFailure(Exception e) {}
 
         @Override
         public void recordSuccess() {}
@@ -245,7 +245,7 @@ public <T> T execute(Callable<T> call, Backoff backoff) throws Exception {
         circuitBreaker.recordSuccess();
         return r;
       } catch (Exception e) {
-        circuitBreaker.recordFailure();
+        circuitBreaker.recordFailure(e);
         Throwables.throwIfInstanceOf(e, InterruptedException.class);
         if (State.TRIAL_CALL.equals(circuitState)) {
           throw e;
@@ -272,19 +272,35 @@ public <T> ListenableFuture<T> executeAsync(AsyncCallable<T> call) {
    * backoff.
    */
   public <T> ListenableFuture<T> executeAsync(AsyncCallable<T> call, Backoff backoff) {
+    final State circuitState = circuitBreaker.state();
+    if (State.REJECT_CALLS.equals(circuitState)) {
+      return Futures.immediateFailedFuture(new CircuitBreakerException());
+    }
     try {
+      ListenableFuture<T> future =
+          Futures.transformAsync(
+              call.call(),
+              (f) -> {
+                circuitBreaker.recordSuccess();
+                return Futures.immediateFuture(f);
+              },
+              MoreExecutors.directExecutor());
       return Futures.catchingAsync(
-          call.call(),
+          future,
           Exception.class,
-          t -> onExecuteAsyncFailure(t, call, backoff),
+          t -> onExecuteAsyncFailure(t, call, backoff, circuitState),
           MoreExecutors.directExecutor());
     } catch (Exception e) {
-      return onExecuteAsyncFailure(e, call, backoff);
+      return onExecuteAsyncFailure(e, call, backoff, circuitState);
     }
   }
 
   private <T> ListenableFuture<T> onExecuteAsyncFailure(
-      Exception t, AsyncCallable<T> call, Backoff backoff) {
+      Exception t, AsyncCallable<T> call, Backoff backoff, State circuitState) {
+    circuitBreaker.recordFailure(t);
+    if (circuitState.equals(State.TRIAL_CALL)) {
+      return Futures.immediateFailedFuture(t);
+    }
     if (isRetriable(t)) {
       long waitMillis = backoff.nextDelayMillis(t);
       if (waitMillis >= 0) {
@@ -310,4 +326,8 @@ public Backoff newBackoff() {
   public boolean isRetriable(Exception e) {
     return shouldRetry.test(e);
   }
+
+  CircuitBreaker getCircuitBreaker() {
+    return this.circuitBreaker;
+  }
 }

src/main/java/com/google/devtools/build/lib/remote/circuitbreaker/BUILD

@@ -0,0 +1,23 @@
+load("@rules_java//java:defs.bzl", "java_library")
+
+package(
+    default_applicable_licenses = ["//:license"],
+    default_visibility = ["//src:__subpackages__"],
+)
+
+filegroup(
+    name = "srcs",
+    srcs = glob(["*"]),
+    visibility = ["//src:__subpackages__"],
+)
+
+java_library(
+    name = "circuitbreaker",
+    srcs = glob(["*.java"]),
+    deps = [
+        "//src/main/java/com/google/devtools/build/lib/remote:Retrier",
+        "//src/main/java/com/google/devtools/build/lib/remote/common:cache_not_found_exception",
+        "//src/main/java/com/google/devtools/build/lib/remote/options",
+        "//third_party:guava",
+    ],
+)

src/main/java/com/google/devtools/build/lib/remote/circuitbreaker/CircuitBreakerFactory.java

@@ -0,0 +1,45 @@
+// Copyright 2023 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.remote.circuitbreaker;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.devtools.build.lib.remote.Retrier;
+import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
+import com.google.devtools.build.lib.remote.options.RemoteOptions;
+
+/** Factory for {@link Retrier.CircuitBreaker} */
+public class CircuitBreakerFactory {
+
+  public static final ImmutableSet<Class<? extends Exception>> DEFAULT_IGNORED_ERRORS =
+      ImmutableSet.of(CacheNotFoundException.class);
+
+  private CircuitBreakerFactory() {}
+
+  /**
+   * Creates the instance of the {@link Retrier.CircuitBreaker} as per the strategy defined in
+   * {@link RemoteOptions}. In case of undefined strategy defaults to {@link
+   * Retrier.ALLOW_ALL_CALLS} implementation.
+   *
+   * @param remoteOptions The configuration for the CircuitBreaker implementation.
+   * @return an instance of CircuitBreaker.
+   */
+  public static Retrier.CircuitBreaker createCircuitBreaker(final RemoteOptions remoteOptions) {
+    if (remoteOptions.circuitBreakerStrategy == RemoteOptions.CircuitBreakerStrategy.FAILURE) {
+      return new FailureCircuitBreaker(
+          remoteOptions.remoteFailureThreshold,
+          (int) remoteOptions.remoteFailureWindowInterval.toMillis());
+    }
+    return Retrier.ALLOW_ALL_CALLS;
+  }
+}

src/main/java/com/google/devtools/build/lib/remote/circuitbreaker/FailureCircuitBreaker.java

@@ -0,0 +1,83 @@
+// Copyright 2023 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.remote.circuitbreaker;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.devtools.build.lib.remote.Retrier;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * The {@link FailureCircuitBreaker} implementation of the {@link Retrier.CircuitBreaker} prevents
+ * further calls to a remote cache once the number of failures within a given window exceeds a
+ * specified threshold for a build. In the context of Bazel, a new instance of {@link
+ * Retrier.CircuitBreaker} is created for each build. Therefore, if the circuit breaker trips during
+ * a build, the remote cache will be disabled for that build. However, it will be enabled again for
+ * the next build as a new instance of {@link Retrier.CircuitBreaker} will be created.
+ */
+public class FailureCircuitBreaker implements Retrier.CircuitBreaker {
+
+  private State state;
+  private final AtomicInteger failures;
+  private final int failureThreshold;
+  private final int slidingWindowSize;
+  private final ScheduledExecutorService scheduledExecutor;
+  private final ImmutableSet<Class<? extends Exception>> ignoredErrors;
+
+  /**
+   * Creates a {@link FailureCircuitBreaker}.
+   *
+   * @param failureThreshold is used to set the number of failures required to trip the circuit
+   *     breaker in given time window.
+   * @param slidingWindowSize the size of the sliding window in milliseconds to calculate the number
+   *     of failures.
+   */
+  public FailureCircuitBreaker(int failureThreshold, int slidingWindowSize) {
+    this.failureThreshold = failureThreshold;
+    this.failures = new AtomicInteger(0);
+    this.slidingWindowSize = slidingWindowSize;
+    this.state = State.ACCEPT_CALLS;
+    this.scheduledExecutor =
+        slidingWindowSize > 0 ? Executors.newSingleThreadScheduledExecutor() : null;
+    this.ignoredErrors = CircuitBreakerFactory.DEFAULT_IGNORED_ERRORS;
+  }
+
+  @Override
+  public State state() {
+    return this.state;
+  }
+
+  @Override
+  public void recordFailure(Exception e) {
+    if (!ignoredErrors.contains(e.getClass())) {
+      int failureCount = failures.incrementAndGet();
+      if (slidingWindowSize > 0) {
+        var unused =
+            scheduledExecutor.schedule(
+                failures::decrementAndGet, slidingWindowSize, TimeUnit.MILLISECONDS);
+      }
+      // Since the state can only be changed to the open state, synchronization is not required.
+      if (failureCount > this.failureThreshold) {
+        this.state = State.REJECT_CALLS;
+      }
+    }
+  }
+
+  @Override
+  public void recordSuccess() {
+    // do nothing, implement if we need to set threshold on failure rate instead of count.
+  }
+}

src/main/java/com/google/devtools/build/lib/remote/options/CommonRemoteOptions.java

@@ -13,11 +13,16 @@
 // limitations under the License.
 package com.google.devtools.build.lib.remote.options;
 
+import com.google.devtools.common.options.Converter;
+import com.google.devtools.common.options.Converters;
 import com.google.devtools.common.options.Option;
 import com.google.devtools.common.options.OptionDocumentationCategory;
 import com.google.devtools.common.options.OptionEffectTag;
 import com.google.devtools.common.options.OptionsBase;
+import com.google.devtools.common.options.OptionsParsingException;
+import java.time.Duration;
 import java.util.List;
+import java.util.regex.Pattern;
 
 /** Options for remote execution and distributed caching that shared between Bazel and Blaze. */
 public class CommonRemoteOptions extends OptionsBase {
@@ -33,4 +38,23 @@ public class CommonRemoteOptions extends OptionsBase {
               + " the client to request certain artifacts that might be needed locally (e.g. IDE"
               + " support)")
   public List<String> remoteDownloadRegex;
+
+  /** Returns the specified duration. Assumes seconds if unitless. */
+  public static class RemoteDurationConverter extends Converter.Contextless<Duration> {
+
+    private static final Pattern UNITLESS_REGEX = Pattern.compile("^[0-9]+$");
+
+    @Override
+    public Duration convert(String input) throws OptionsParsingException {
+      if (UNITLESS_REGEX.matcher(input).matches()) {
+        input += "s";
+      }
+      return new Converters.DurationConverter().convert(input, /* conversionContext= */ null);
+    }
+
+    @Override
+    public String getTypeDescription() {
+      return "An immutable length of time.";
+    }
+  }
 }