remote: Add new retrier with support for circuit breaking

Add a generic retrier implementation (Retrier2) that can be
configured by plugging in a backoff strategy, a function to
decide on retriable errors and a circuit breaker. A concrete
implementation is added via RemoteRetrier that mostly is a
copy of the code of the existing Retrier.

Retrier2 adds support for circuit breaking [1]. It allows the
retrier to reject execution when failure rates are high. The
remote execution code will use this to gently switch between
local and remote execution/caching if the latter experiences
lots of failures.

Retrier2 is also useful when not used with gRPC. We need
retriers for the HTTP caching interface too.

All the code added in this CL is unused, to keep reviews
managable. In a follow up CL, I will switch the code to use
the new Retrier and delete the old retrier.

[1] https://martinfowler.com/bliki/CircuitBreaker.html

PiperOrigin-RevId: 168355597

2 files changed, 446 insertions, 0 deletions

diff --git a/src/test/java/com/google/devtools/build/lib/remote/RemoteRetrierTest.java b/src/test/java/com/google/devtools/build/lib/remote/RemoteRetrierTest.java
new file mode 100644
index 0000000000..66de467e55
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/remote/RemoteRetrierTest.java
@@ -0,0 +1,160 @@
+// Copyright 2017 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;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.Range;
+import com.google.devtools.build.lib.remote.RemoteRetrier.ExponentialBackoff;
+import com.google.devtools.build.lib.remote.Retrier2.Backoff;
+import com.google.devtools.build.lib.remote.Retrier2.RetryException2;
+import com.google.devtools.build.lib.remote.Retrier2.Sleeper;
+import com.google.devtools.common.options.Options;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mockito;
+
+/**
+ * Tests for {@link RemoteRetrier}.
+ */
+@RunWith(JUnit4.class)
+public class RemoteRetrierTest {
+
+  interface Foo {
+    public String foo();
+  }
+
+  private RemoteRetrierTest.Foo fooMock;
+
+  @Before
+  public void setUp() {
+    fooMock = Mockito.mock(RemoteRetrierTest.Foo.class);
+  }
+
+  @Test
+  public void testExponentialBackoff() throws Exception {
+    Retrier2.Backoff backoff =
+        new ExponentialBackoff(Duration.ofSeconds(1), Duration.ofSeconds(10), 2, 0, 6);
+    assertThat(backoff.nextDelayMillis()).isEqualTo(1000);
+    assertThat(backoff.nextDelayMillis()).isEqualTo(2000);
+    assertThat(backoff.nextDelayMillis()).isEqualTo(4000);
+    assertThat(backoff.nextDelayMillis()).isEqualTo(8000);
+    assertThat(backoff.nextDelayMillis()).isEqualTo(10000);
+    assertThat(backoff.nextDelayMillis()).isEqualTo(10000);
+    assertThat(backoff.nextDelayMillis()).isLessThan(0L);
+  }
+
+  @Test
+  public void testExponentialBackoffJittered() throws Exception {
+    Retrier2.Backoff backoff =
+        new ExponentialBackoff(Duration.ofSeconds(1), Duration.ofSeconds(10), 2, 0.1, 6);
+    assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(900L, 1100L));
+    assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(1800L, 2200L));
+    assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(3600L, 4400L));
+    assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(7200L, 8800L));
+    assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(9000L, 11000L));
+    assertThat(backoff.nextDelayMillis()).isIn(Range.closedOpen(9000L, 11000L));
+    assertThat(backoff.nextDelayMillis()).isLessThan(0L);
+  }
+
+  private void assertThrows(RemoteRetrier retrier, int attempts) throws Exception {
+    try {
+      retrier.execute(() -> fooMock.foo());
+      fail();
+    } catch (RetryException2 e) {
+      assertThat(e.getAttempts()).isEqualTo(attempts);
+    }
+  }
+
+  @Test
+  public void testNoRetries() throws Exception {
+    RemoteOptions options = Options.getDefaults(RemoteOptions.class);
+    options.experimentalRemoteRetry = false;
+
+    RemoteRetrier retrier = Mockito.spy(new RemoteRetrier(options,
+        RemoteRetrier.RETRIABLE_GRPC_ERRORS, Retrier2.ALLOW_ALL_CALLS));
+    when(fooMock.foo())
+        .thenReturn("bla")
+        .thenThrow(Status.Code.UNKNOWN.toStatus().asRuntimeException());
+    assertThat(retrier.execute(() -> fooMock.foo())).isEqualTo("bla");
+    assertThrows(retrier, 1);
+    Mockito.verify(fooMock, Mockito.times(2)).foo();
+  }
+
+  @Test
+  public void testRepeatedRetriesReset() throws Exception {
+    Supplier<Backoff> s =
+        () -> new ExponentialBackoff(Duration.ofSeconds(1), Duration.ofSeconds(10), 2.0, 0.0, 2);
+    Sleeper sleeper = Mockito.mock(Sleeper.class);
+    RemoteRetrier retrier = Mockito.spy(new RemoteRetrier(s, (e) -> true,
+        Retrier2.ALLOW_ALL_CALLS, sleeper));
+
+    when(fooMock.foo()).thenThrow(Status.Code.UNKNOWN.toStatus().asRuntimeException());
+    assertThrows(retrier, 3);
+    assertThrows(retrier, 3);
+    Mockito.verify(sleeper, Mockito.times(2)).sleep(1000);
+    Mockito.verify(sleeper, Mockito.times(2)).sleep(2000);
+    Mockito.verify(fooMock, Mockito.times(6)).foo();
+  }
+
+  @Test
+  public void testInterruptedExceptionIsPassedThrough() throws Exception {
+    InterruptedException thrown = new InterruptedException();
+
+    RemoteOptions options = Options.getDefaults(RemoteOptions.class);
+    options.experimentalRemoteRetry = false;
+    RemoteRetrier retrier = new RemoteRetrier(options, RemoteRetrier.RETRIABLE_GRPC_ERRORS,
+        Retrier2.ALLOW_ALL_CALLS);
+    try {
+      retrier.execute(() -> {
+        throw thrown;
+      });
+      fail();
+    } catch (InterruptedException expected) {
+      assertThat(expected).isSameAs(thrown);
+    }
+  }
+
+  @Test
+  public void testPassThroughException() throws Exception {
+    StatusRuntimeException thrown = Status.Code.UNKNOWN.toStatus().asRuntimeException();
+
+    RemoteOptions options = Options.getDefaults(RemoteOptions.class);
+    RemoteRetrier retrier = new RemoteRetrier(options, RemoteRetrier.RETRIABLE_GRPC_ERRORS,
+        Retrier2.ALLOW_ALL_CALLS);
+
+    AtomicInteger numCalls = new AtomicInteger();
+    try {
+      retrier.execute(() -> {
+        numCalls.incrementAndGet();
+        throw new RemoteRetrier.PassThroughException(thrown);
+      });
+      fail();
+    } catch (RetryException2 expected) {
+      assertThat(expected).hasCauseThat().isSameAs(thrown);
+    }
+
+    assertThat(numCalls.get()).isEqualTo(1);
+  }
+}
diff --git a/src/test/java/com/google/devtools/build/lib/remote/Retrier2Test.java b/src/test/java/com/google/devtools/build/lib/remote/Retrier2Test.java
new file mode 100644
index 0000000000..8e377f2cea
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/remote/Retrier2Test.java
@@ -0,0 +1,286 @@
+// Copyright 2016 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;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.devtools.build.lib.remote.Retrier2.Backoff;
+import com.google.devtools.build.lib.remote.Retrier2.CircuitBreaker;
+import com.google.devtools.build.lib.remote.Retrier2.CircuitBreaker.State;
+import com.google.devtools.build.lib.remote.Retrier2.CircuitBreakerException;
+import com.google.devtools.build.lib.remote.Retrier2.RetryException2;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import javax.annotation.concurrent.ThreadSafe;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+/**
+ * Tests for {@link Retrier2}.
+ */
+@RunWith(JUnit4.class)
+public class Retrier2Test {
+
+  @Mock
+  private CircuitBreaker alwaysOpen;
+
+  private static final Predicate<Exception> RETRY_ALL = (e) -> true;
+
+  @Before
+  public void setup() {
+    MockitoAnnotations.initMocks(this);
+    when(alwaysOpen.state()).thenReturn(State.ACCEPT_CALLS);
+  }
+
+  @Test
+  public void retryShouldWork_failure() throws Exception {
+    // Test that a call is retried according to the backoff.
+    // All calls fail.
+
+    Supplier<Backoff> s  = () -> new ZeroBackoff(/*maxRetries=*/2);
+    Retrier2 r = new Retrier2(s, RETRY_ALL, alwaysOpen);
+    try {
+      r.execute(() -> {
+        throw new Exception("call failed");
+      });
+      fail("exception expected.");
+    } catch (RetryException2 e) {
+      assertThat(e.getAttempts()).isEqualTo(3);
+    }
+
+    verify(alwaysOpen, times(3)).recordFailure();
+    verify(alwaysOpen, never()).recordSuccess();
+  }
+
+  @Test
+  public void retryShouldWork_success() throws Exception {
+    // Test that a call is retried according to the backoff.
+    // The last call succeeds.
+
+    Supplier<Backoff> s  = () -> new ZeroBackoff(/*maxRetries=*/2);
+    Retrier2 r = new Retrier2(s, RETRY_ALL, alwaysOpen);
+    AtomicInteger numCalls = new AtomicInteger();
+    int val = r.execute(() -> {
+      numCalls.incrementAndGet();
+      if (numCalls.get() == 3) {
+        return 1;
+      }
+      throw new Exception("call failed");
+    });
+    assertThat(val).isEqualTo(1);
+
+    verify(alwaysOpen, times(2)).recordFailure();
+    verify(alwaysOpen, times(1)).recordSuccess();
+  }
+
+  @Test
+  public void nestedRetriesShouldWork() throws Exception {
+    // Test that nested calls using retries compose as expected.
+
+    Supplier<Backoff> s  = () -> new ZeroBackoff(/*maxRetries=*/1);
+    Retrier2 r = new Retrier2(s, RETRY_ALL, alwaysOpen);
+
+    AtomicInteger attemptsLvl0 = new AtomicInteger();
+    AtomicInteger attemptsLvl1 = new AtomicInteger();
+    AtomicInteger attemptsLvl2 = new AtomicInteger();
+    try {
+      r.execute(() -> {
+        attemptsLvl0.incrementAndGet();
+        return r.execute(() -> {
+          attemptsLvl1.incrementAndGet();
+          return r.execute(() -> {
+            attemptsLvl2.incrementAndGet();
+            throw new Exception("call failed");
+          });
+        });
+      });
+    } catch (RetryException2 outer) {
+      assertThat(outer.getAttempts()).isEqualTo(2);
+      assertThat(outer).hasCauseThat().hasMessageThat().isEqualTo("call failed");
+      assertThat(attemptsLvl0.get()).isEqualTo(2);
+      assertThat(attemptsLvl1.get()).isEqualTo(4);
+      assertThat(attemptsLvl2.get()).isEqualTo(8);
+    }
+  }
+
+  @Test
+  public void circuitBreakerShouldTrip() throws Exception {
+    // Test that a circuit breaker can trip.
+
+    Supplier<Backoff> s  = () -> new ZeroBackoff(/*maxRetries=*/3);
+    TripAfterNCircuitBreaker cb = new TripAfterNCircuitBreaker(/*maxConsecutiveFailures=*/2);
+    Retrier2 r = new Retrier2(s, RETRY_ALL, cb);
+
+    try {
+      r.execute(() -> {
+        throw new Exception("call failed");
+      });
+      fail ("exception expected");
+    } catch (CircuitBreakerException expected) {
+      // Intentionally left empty.
+    }
+
+    assertThat(cb.state()).isEqualTo(State.REJECT_CALLS);
+    assertThat(cb.consecutiveFailures).isEqualTo(2);
+  }
+
+  @Test
+  public void circuitBreakerCanRecover() throws Exception {
+    // Test that a circuit breaker can recover from REJECT_CALLS to ACCEPT_CALLS by
+    // utilizing the TRIAL_CALL state.
+
+    Supplier<Backoff> s  = () -> new ZeroBackoff(/*maxRetries=*/3);
+    TripAfterNCircuitBreaker cb = new TripAfterNCircuitBreaker(/*maxConsecutiveFailures=*/2);
+    Retrier2 r = new Retrier2(s, RETRY_ALL, cb);
+
+    cb.trialCall();
+
+    assertThat(cb.state()).isEqualTo(State.TRIAL_CALL);
+
+    int val = r.execute(() -> 10);
+    assertThat(val).isEqualTo(10);
+    assertThat(cb.state()).isEqualTo(State.ACCEPT_CALLS);
+  }
+
+  @Test
+  public void circuitBreakerHalfOpenIsNotRetried() throws Exception {
+    // Test that a call executed in TRIAL_CALL state is not retried
+    // in case of failure.
+
+    Supplier<Backoff> s  = () -> new ZeroBackoff(/*maxRetries=*/3);
+    TripAfterNCircuitBreaker cb = new TripAfterNCircuitBreaker(/*maxConsecutiveFailures=*/2);
+    Retrier2 r = new Retrier2(s, RETRY_ALL, cb);
+
+    cb.trialCall();
+
+    try {
+      r.execute(() -> {
+        throw new Exception("call failed");
+      });
+    } catch (RetryException2 expected) {
+      // Intentionally left empty.
+    }
+
+    assertThat(cb.consecutiveFailures).isEqualTo(1);
+  }
+
+  @Test
+  public void interruptsShouldNotBeRetried_flag() throws Exception {
+    // Test that a call is not executed / retried if the current thread
+    // is interrupted.
+
+    Supplier<Backoff> s  = () -> new ZeroBackoff(/*maxRetries=*/3);
+    TripAfterNCircuitBreaker cb = new TripAfterNCircuitBreaker(/*maxConsecutiveFailures=*/2);
+    Retrier2 r = new Retrier2(s, RETRY_ALL, cb);
+
+    try {
+      Thread.currentThread().interrupt();
+      r.execute(() -> 10);
+    } catch (InterruptedException expected) {
+      // Intentionally left empty.
+    }
+  }
+
+  @Test
+  public void interruptsShouldNotBeRetried_exception() throws Exception {
+    // Test that a call is not retried if an InterruptedException is thrown.
+
+    Supplier<Backoff> s  = () -> new ZeroBackoff(/*maxRetries=*/3);
+    TripAfterNCircuitBreaker cb = new TripAfterNCircuitBreaker(/*maxConsecutiveFailures=*/2);
+    Retrier2 r = new Retrier2(s, RETRY_ALL, cb);
+
+    try {
+      Thread.currentThread().interrupt();
+      r.execute(() -> {
+        throw new InterruptedException();
+      });
+    } catch (InterruptedException expected) {
+      // Intentionally left empty.
+    }
+  }
+
+  /**
+   * Simple circuit breaker that trips after N consecutive failures.
+   */
+  @ThreadSafe
+  private static class TripAfterNCircuitBreaker implements CircuitBreaker {
+
+    private final int maxConsecutiveFailures;
+
+    private State state = State.ACCEPT_CALLS;
+    private int consecutiveFailures;
+
+    TripAfterNCircuitBreaker(int maxConsecutiveFailures) {
+      this.maxConsecutiveFailures = maxConsecutiveFailures;
+    }
+
+    @Override
+    public synchronized State state() {
+      return state;
+    }
+
+    @Override
+    public synchronized void recordFailure() {
+      consecutiveFailures++;
+      if (consecutiveFailures >= maxConsecutiveFailures) {
+        state = State.REJECT_CALLS;
+      }
+    }
+
+    @Override
+    public synchronized void recordSuccess() {
+      consecutiveFailures = 0;
+      state = State.ACCEPT_CALLS;
+    }
+
+    void trialCall() {
+      state = State.TRIAL_CALL;
+    }
+  }
+
+  private static class ZeroBackoff implements Backoff {
+
+    private final int maxRetries;
+    private int retries;
+
+    public ZeroBackoff(int maxRetries) {
+      this.maxRetries = maxRetries;
+    }
+
+    @Override
+    public long nextDelayMillis() {
+      if (retries >= maxRetries) {
+        return -1;
+      }
+      retries++;
+      return 0;
+    }
+
+    @Override
+    public int getRetryAttempts() {
+      return retries;
+    }
+  }
+}