Add CallCredentialsProvider and refresh credentials in ByteStreamUploader Users may get authentication error in the mid of a long remote build due to credentials timeout. This PR: 1. Add `CallCredentialsProvider` which can be used by gRPC clients to refresh credentials. 2. Update `ByteStreamUploader.java` to use the provider and refresh credentials on authentication error. The next step would be updating other places where using `CallCrendentials` currently (e.g. `RemoteCacheClient.java`) to use this provider and refresh credentials when necessary. Closes #12106. PiperOrigin-RevId: 332394306
diff --git a/src/main/java/com/google/devtools/build/lib/authandtls/BUILD b/src/main/java/com/google/devtools/build/lib/authandtls/BUILD index 23b7178..55b2ebb 100644 --- a/src/main/java/com/google/devtools/build/lib/authandtls/BUILD +++ b/src/main/java/com/google/devtools/build/lib/authandtls/BUILD
@@ -12,6 +12,7 @@ name = "authandtls", srcs = glob(["*.java"]), deps = [ + "//src/main/java/com/google/devtools/build/lib/concurrent", "//src/main/java/com/google/devtools/common/options", "//third_party:auth", "//third_party:guava",
diff --git a/src/main/java/com/google/devtools/build/lib/authandtls/CallCredentialsProvider.java b/src/main/java/com/google/devtools/build/lib/authandtls/CallCredentialsProvider.java new file mode 100644 index 0000000..d2dc5d4 --- /dev/null +++ b/src/main/java/com/google/devtools/build/lib/authandtls/CallCredentialsProvider.java
@@ -0,0 +1,54 @@ +// Copyright 2020 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.authandtls; + +import io.grpc.CallCredentials; +import java.io.IOException; +import javax.annotation.Nullable; + +/** Interface for providing {@link CallCredentials}. Implementations must be thread-safe. */ +public interface CallCredentialsProvider { + /** + * Returns the current {@link CallCredentials}. May be {@code null}, in which case no + * authentication is required. + */ + @Nullable + CallCredentials getCallCredentials(); + + /** + * Refresh the authorization data, discarding any cached state. + * + * <p>For use by the transport to allow retry after getting an error indicating there may be + * invalid tokens or other cached state. + * + * @throws IOException if there was an error getting up-to-date access. + */ + void refresh() throws IOException; + + /** A no-op implementation that has no credentials and performs no refreshes. */ + public static class NoCredentials implements CallCredentialsProvider { + + @Nullable + @Override + public CallCredentials getCallCredentials() { + return null; + } + + @Override + public void refresh() throws IOException {} + } + + public static CallCredentialsProvider NO_CREDENTIALS = new NoCredentials(); +}
diff --git a/src/main/java/com/google/devtools/build/lib/authandtls/GoogleAuthCallCredentialsProvider.java b/src/main/java/com/google/devtools/build/lib/authandtls/GoogleAuthCallCredentialsProvider.java new file mode 100644 index 0000000..6c833f7 --- /dev/null +++ b/src/main/java/com/google/devtools/build/lib/authandtls/GoogleAuthCallCredentialsProvider.java
@@ -0,0 +1,60 @@ +// Copyright 2020 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.authandtls; + +import com.google.auth.Credentials; +import com.google.common.base.Preconditions; +import com.google.common.base.Stopwatch; +import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe; +import io.grpc.CallCredentials; +import io.grpc.auth.MoreCallCredentials; +import java.io.IOException; +import java.time.Duration; + +/** A {@link CallCredentialsProvider} implementation which uses {@link Credentials} */ +@ThreadSafe +public class GoogleAuthCallCredentialsProvider implements CallCredentialsProvider { + + private final Credentials credentials; + private final CallCredentials callCredentials; + private final Stopwatch refreshStopwatch; + + public GoogleAuthCallCredentialsProvider(Credentials credentials) { + Preconditions.checkNotNull(credentials, "credentials"); + this.credentials = credentials; + + callCredentials = MoreCallCredentials.from(credentials); + refreshStopwatch = Stopwatch.createStarted(); + } + + @Override + public CallCredentials getCallCredentials() { + return callCredentials; + } + + @Override + public void refresh() throws IOException { + synchronized (this) { + // Call credentials.refresh() at most once per second. The one second was arbitrarily chosen, + // as a small enough value that we don't expect to interfere with actual token lifetimes, but + // it should just make sure that potentially hundreds of threads don't call this method + // at the same time. + if (refreshStopwatch.elapsed().compareTo(Duration.ofSeconds(1)) > 0) { + credentials.refresh(); + refreshStopwatch.reset().start(); + } + } + } +}
diff --git a/src/main/java/com/google/devtools/build/lib/authandtls/GoogleAuthUtils.java b/src/main/java/com/google/devtools/build/lib/authandtls/GoogleAuthUtils.java index c3ec90e..c3bd004 100644 --- a/src/main/java/com/google/devtools/build/lib/authandtls/GoogleAuthUtils.java +++ b/src/main/java/com/google/devtools/build/lib/authandtls/GoogleAuthUtils.java
@@ -195,14 +195,23 @@ return null; } + public static CallCredentialsProvider newCallCredentialsProvider(AuthAndTLSOptions options) + throws IOException { + Credentials creds = newCredentials(options); + if (creds != null) { + return new GoogleAuthCallCredentialsProvider(creds); + } + return CallCredentialsProvider.NO_CREDENTIALS; + } + @VisibleForTesting - public static CallCredentials newCallCredentials( + public static CallCredentialsProvider newCallCredentialsProvider( @Nullable InputStream credentialsFile, List<String> authScope) throws IOException { Credentials creds = newCredentials(credentialsFile, authScope); if (creds != null) { - return MoreCallCredentials.from(creds); + return new GoogleAuthCallCredentialsProvider(creds); } - return null; + return CallCredentialsProvider.NO_CREDENTIALS; } /**
diff --git a/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java b/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java index 711cdd1..a9d5808 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java +++ b/src/main/java/com/google/devtools/build/lib/remote/ByteStreamUploader.java
@@ -33,9 +33,9 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.SettableFuture; +import com.google.devtools.build.lib.authandtls.CallCredentialsProvider; import com.google.devtools.build.lib.remote.RemoteRetrier.ProgressiveBackoff; import com.google.devtools.build.lib.remote.util.TracingMetadataUtils; -import io.grpc.CallCredentials; import io.grpc.CallOptions; import io.grpc.Channel; import io.grpc.ClientCall; @@ -74,7 +74,7 @@ private final String instanceName; private final ReferenceCountedChannel channel; - private final CallCredentials callCredentials; + private final CallCredentialsProvider callCredentialsProvider; private final long callTimeoutSecs; private final RemoteRetrier retrier; @@ -96,8 +96,7 @@ * @param instanceName the instance name to be prepended to resource name of the {@code Write} * call. See the {@code ByteStream} service definition for details * @param channel the {@link io.grpc.Channel} to use for calls - * @param callCredentials the credentials to use for authentication. May be {@code null}, in which - * case no authentication is performed + * @param callCredentialsProvider the credentials provider to use for authentication. * @param callTimeoutSecs the timeout in seconds after which a {@code Write} gRPC call must be * complete. The timeout resets between retries * @param retrier the {@link RemoteRetrier} whose backoff strategy to use for retry timings. @@ -105,14 +104,14 @@ public ByteStreamUploader( @Nullable String instanceName, ReferenceCountedChannel channel, - @Nullable CallCredentials callCredentials, + CallCredentialsProvider callCredentialsProvider, long callTimeoutSecs, RemoteRetrier retrier) { checkArgument(callTimeoutSecs > 0, "callTimeoutSecs must be gt 0."); this.instanceName = instanceName; this.channel = channel; - this.callCredentials = callCredentials; + this.callCredentialsProvider = callCredentialsProvider; this.callTimeoutSecs = callTimeoutSecs; this.retrier = retrier; } @@ -267,7 +266,7 @@ } } - private static String uploadResourceName( + private static String buildUploadResourceName( String instanceName, UUID uuid, HashCode hash, long size) { String resourceName = format("uploads/%s/blobs/%s/%d", uuid, hash, size); if (!Strings.isNullOrEmpty(instanceName)) { @@ -276,7 +275,7 @@ return resourceName; } - /** Starts a file upload an returns a future representing the upload. */ + /** Starts a file upload and returns a future representing the upload. */ private ListenableFuture<Void> startAsyncUpload(HashCode hash, Chunker chunker) { try { chunker.reset(); @@ -285,9 +284,10 @@ } UUID uploadId = UUID.randomUUID(); - String resourceName = uploadResourceName(instanceName, uploadId, hash, chunker.getSize()); + String resourceName = buildUploadResourceName(instanceName, uploadId, hash, chunker.getSize()); AsyncUpload newUpload = - new AsyncUpload(channel, callCredentials, callTimeoutSecs, retrier, resourceName, chunker); + new AsyncUpload( + channel, callCredentialsProvider, callTimeoutSecs, retrier, resourceName, chunker); ListenableFuture<Void> currUpload = newUpload.start(); currUpload.addListener( () -> { @@ -323,7 +323,7 @@ private static class AsyncUpload { private final Channel channel; - private final CallCredentials callCredentials; + private final CallCredentialsProvider callCredentialsProvider; private final long callTimeoutSecs; private final Retrier retrier; private final String resourceName; @@ -333,13 +333,13 @@ AsyncUpload( Channel channel, - CallCredentials callCredentials, + CallCredentialsProvider callCredentialsProvider, long callTimeoutSecs, Retrier retrier, String resourceName, Chunker chunker) { this.channel = channel; - this.callCredentials = callCredentials; + this.callCredentialsProvider = callCredentialsProvider; this.callTimeoutSecs = callTimeoutSecs; this.retrier = retrier; this.resourceName = resourceName; @@ -350,15 +350,34 @@ Context ctx = Context.current(); ProgressiveBackoff progressiveBackoff = new ProgressiveBackoff(retrier::newBackoff); AtomicLong committedOffset = new AtomicLong(0); - return Futures.transformAsync( - retrier.executeAsync( - () -> { - if (committedOffset.get() < chunker.getSize()) { - return ctx.call(() -> callAndQueryOnFailure(committedOffset, progressiveBackoff)); + + ListenableFuture<Void> callFuture = + callAndQueryOnFailureWithRetrier(ctx, committedOffset, progressiveBackoff); + + callFuture = + Futures.catchingAsync( + callFuture, + Exception.class, + (e) -> { + Status status = Status.fromThrowable(e); + if (status != null + && (status.getCode() == Code.UNAUTHENTICATED + || status.getCode() == Code.PERMISSION_DENIED)) { + try { + callCredentialsProvider.refresh(); + } catch (IOException ioe) { + e.addSuppressed(ioe); + return Futures.immediateFailedFuture(e); + } + return callAndQueryOnFailureWithRetrier(ctx, committedOffset, progressiveBackoff); + } else { + return Futures.immediateFailedFuture(e); } - return Futures.immediateFuture(null); }, - progressiveBackoff), + MoreExecutors.directExecutor()); + + return Futures.transformAsync( + callFuture, (result) -> { long committedSize = committedOffset.get(); long expected = chunker.getSize(); @@ -376,10 +395,22 @@ private ByteStreamFutureStub bsFutureStub() { return ByteStreamGrpc.newFutureStub(channel) .withInterceptors(TracingMetadataUtils.attachMetadataFromContextInterceptor()) - .withCallCredentials(callCredentials) + .withCallCredentials(callCredentialsProvider.getCallCredentials()) .withDeadlineAfter(callTimeoutSecs, SECONDS); } + private ListenableFuture<Void> callAndQueryOnFailureWithRetrier( + Context ctx, AtomicLong committedOffset, ProgressiveBackoff progressiveBackoff) { + return retrier.executeAsync( + () -> { + if (committedOffset.get() < chunker.getSize()) { + return ctx.call(() -> callAndQueryOnFailure(committedOffset, progressiveBackoff)); + } + return Futures.immediateFuture(null); + }, + progressiveBackoff); + } + private ListenableFuture<Void> callAndQueryOnFailure( AtomicLong committedOffset, ProgressiveBackoff progressiveBackoff) { return Futures.catchingAsync( @@ -456,7 +487,7 @@ private ListenableFuture<Void> call(AtomicLong committedOffset) { CallOptions callOptions = CallOptions.DEFAULT - .withCallCredentials(callCredentials) + .withCallCredentials(callCredentialsProvider.getCallCredentials()) .withDeadlineAfter(callTimeoutSecs, SECONDS); call = channel.newCall(ByteStreamGrpc.getWriteMethod(), callOptions);
diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteModule.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteModule.java index 2f35be1..5629ce2 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/RemoteModule.java +++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteModule.java
@@ -38,6 +38,7 @@ import com.google.devtools.build.lib.analysis.configuredtargets.RuleConfiguredTarget; import com.google.devtools.build.lib.analysis.test.TestProvider; import com.google.devtools.build.lib.authandtls.AuthAndTLSOptions; +import com.google.devtools.build.lib.authandtls.CallCredentialsProvider; import com.google.devtools.build.lib.authandtls.GoogleAuthUtils; import com.google.devtools.build.lib.bazel.repository.downloader.Downloader; import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader; @@ -374,13 +375,16 @@ } } - CallCredentials credentials; + CallCredentialsProvider callCredentialsProvider; try { - credentials = GoogleAuthUtils.newCallCredentials(authAndTlsOptions); + callCredentialsProvider = GoogleAuthUtils.newCallCredentialsProvider(authAndTlsOptions); } catch (IOException e) { handleInitFailure(env, e, Code.CREDENTIALS_INIT_FAILURE); return; } + + CallCredentials credentials = callCredentialsProvider.getCallCredentials(); + RemoteRetrier retrier = new RemoteRetrier( remoteOptions, @@ -456,7 +460,7 @@ new ByteStreamUploader( remoteOptions.remoteInstanceName, cacheChannel.retain(), - credentials, + callCredentialsProvider, remoteOptions.remoteTimeout.getSeconds(), retrier);
diff --git a/src/main/java/com/google/devtools/build/lib/remote/Retrier.java b/src/main/java/com/google/devtools/build/lib/remote/Retrier.java index 9fa0208..19ee40b 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/Retrier.java +++ b/src/main/java/com/google/devtools/build/lib/remote/Retrier.java
@@ -278,14 +278,18 @@ private <T> ListenableFuture<T> onExecuteAsyncFailure( Exception t, AsyncCallable<T> call, Backoff backoff) { - long waitMillis = backoff.nextDelayMillis(); - if (waitMillis >= 0 && isRetriable(t)) { - try { - return Futures.scheduleAsync( - () -> executeAsync(call, backoff), waitMillis, TimeUnit.MILLISECONDS, retryService); - } catch (RejectedExecutionException e) { - // May be thrown by .scheduleAsync(...) if i.e. the executor is shutdown. - return Futures.immediateFailedFuture(new IOException(e)); + if (isRetriable(t)) { + long waitMillis = backoff.nextDelayMillis(); + if (waitMillis >= 0) { + try { + return Futures.scheduleAsync( + () -> executeAsync(call, backoff), waitMillis, TimeUnit.MILLISECONDS, retryService); + } catch (RejectedExecutionException e) { + // May be thrown by .scheduleAsync(...) if i.e. the executor is shutdown. + return Futures.immediateFailedFuture(new IOException(e)); + } + } else { + return Futures.immediateFailedFuture(t); } } else { return Futures.immediateFailedFuture(t);
diff --git a/src/test/java/com/google/devtools/build/lib/remote/ByteStreamBuildEventArtifactUploaderTest.java b/src/test/java/com/google/devtools/build/lib/remote/ByteStreamBuildEventArtifactUploaderTest.java index 5188192..ce3072e 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/ByteStreamBuildEventArtifactUploaderTest.java +++ b/src/test/java/com/google/devtools/build/lib/remote/ByteStreamBuildEventArtifactUploaderTest.java
@@ -40,6 +40,7 @@ import com.google.devtools.build.lib.actions.FileArtifactValue; import com.google.devtools.build.lib.actions.FileArtifactValue.RemoteFileArtifactValue; import com.google.devtools.build.lib.actions.util.ActionsTestUtil; +import com.google.devtools.build.lib.authandtls.CallCredentialsProvider; import com.google.devtools.build.lib.buildeventstream.BuildEvent.LocalFile; import com.google.devtools.build.lib.buildeventstream.BuildEvent.LocalFile.LocalFileType; import com.google.devtools.build.lib.buildeventstream.PathConverter; @@ -167,7 +168,8 @@ TestUtils.newRemoteRetrier(() -> new FixedBackoff(1, 0), (e) -> true, retryService); ReferenceCountedChannel refCntChannel = new ReferenceCountedChannel(channel); ByteStreamUploader uploader = - new ByteStreamUploader("instance", refCntChannel, null, 3, retrier); + new ByteStreamUploader( + "instance", refCntChannel, CallCredentialsProvider.NO_CREDENTIALS, 3, retrier); ByteStreamBuildEventArtifactUploader artifactUploader = newArtifactUploader(uploader); PathConverter pathConverter = artifactUploader.upload(filesToUpload).get(); @@ -252,7 +254,8 @@ TestUtils.newRemoteRetrier(() -> new FixedBackoff(1, 0), (e) -> true, retryService); ReferenceCountedChannel refCntChannel = new ReferenceCountedChannel(channel); ByteStreamUploader uploader = - new ByteStreamUploader("instance", refCntChannel, null, 3, retrier); + new ByteStreamUploader( + "instance", refCntChannel, CallCredentialsProvider.NO_CREDENTIALS, 3, retrier); ByteStreamBuildEventArtifactUploader artifactUploader = newArtifactUploader(uploader); ExecutionException e =
diff --git a/src/test/java/com/google/devtools/build/lib/remote/ByteStreamUploaderTest.java b/src/test/java/com/google/devtools/build/lib/remote/ByteStreamUploaderTest.java index 8bed33c..dcd9975 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/ByteStreamUploaderTest.java +++ b/src/test/java/com/google/devtools/build/lib/remote/ByteStreamUploaderTest.java
@@ -15,6 +15,7 @@ import static com.google.common.truth.Truth.assertThat; import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.fail; import build.bazel.remote.execution.v2.Digest; @@ -33,12 +34,14 @@ import com.google.common.util.concurrent.ListeningScheduledExecutorService; import com.google.common.util.concurrent.MoreExecutors; import com.google.devtools.build.lib.analysis.BlazeVersionInfo; +import com.google.devtools.build.lib.authandtls.CallCredentialsProvider; import com.google.devtools.build.lib.remote.util.DigestUtil; import com.google.devtools.build.lib.remote.util.TestUtils; import com.google.devtools.build.lib.remote.util.TracingMetadataUtils; import com.google.devtools.build.lib.vfs.DigestHashFunction; import com.google.protobuf.ByteString; import io.grpc.BindableService; +import io.grpc.CallCredentials; import io.grpc.Context; import io.grpc.ManagedChannel; import io.grpc.Metadata; @@ -72,6 +75,7 @@ import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import javax.annotation.Nullable; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -146,7 +150,7 @@ new ByteStreamUploader( INSTANCE_NAME, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -223,7 +227,11 @@ TestUtils.newRemoteRetrier(() -> mockBackoff, (e) -> true, retryService); ByteStreamUploader uploader = new ByteStreamUploader( - INSTANCE_NAME, new ReferenceCountedChannel(channel), null, 3, retrier); + INSTANCE_NAME, + new ReferenceCountedChannel(channel), + CallCredentialsProvider.NO_CREDENTIALS, + 3, + retrier); byte[] blob = new byte[CHUNK_SIZE * 2 + 1]; new Random().nextBytes(blob); @@ -339,7 +347,11 @@ TestUtils.newRemoteRetrier(() -> new FixedBackoff(1, 0), (e) -> true, retryService); ByteStreamUploader uploader = new ByteStreamUploader( - INSTANCE_NAME, new ReferenceCountedChannel(channel), null, 1, retrier); + INSTANCE_NAME, + new ReferenceCountedChannel(channel), + CallCredentialsProvider.NO_CREDENTIALS, + 1, + retrier); byte[] blob = new byte[CHUNK_SIZE * 2 + 1]; new Random().nextBytes(blob); @@ -397,7 +409,11 @@ TestUtils.newRemoteRetrier(() -> mockBackoff, (e) -> true, retryService); ByteStreamUploader uploader = new ByteStreamUploader( - INSTANCE_NAME, new ReferenceCountedChannel(channel), null, 3, retrier); + INSTANCE_NAME, + new ReferenceCountedChannel(channel), + CallCredentialsProvider.NO_CREDENTIALS, + 3, + retrier); byte[] blob = new byte[CHUNK_SIZE * 2 + 1]; new Random().nextBytes(blob); @@ -467,7 +483,11 @@ TestUtils.newRemoteRetrier(() -> mockBackoff, (e) -> true, retryService); ByteStreamUploader uploader = new ByteStreamUploader( - INSTANCE_NAME, new ReferenceCountedChannel(channel), null, 3, retrier); + INSTANCE_NAME, + new ReferenceCountedChannel(channel), + CallCredentialsProvider.NO_CREDENTIALS, + 3, + retrier); byte[] blob = new byte[CHUNK_SIZE * 2 + 1]; new Random().nextBytes(blob); @@ -504,7 +524,11 @@ TestUtils.newRemoteRetrier(() -> mockBackoff, (e) -> true, retryService); ByteStreamUploader uploader = new ByteStreamUploader( - INSTANCE_NAME, new ReferenceCountedChannel(channel), null, 3, retrier); + INSTANCE_NAME, + new ReferenceCountedChannel(channel), + CallCredentialsProvider.NO_CREDENTIALS, + 3, + retrier); byte[] blob = new byte[CHUNK_SIZE * 2 + 1]; new Random().nextBytes(blob); @@ -547,7 +571,7 @@ new ByteStreamUploader( INSTANCE_NAME, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -585,7 +609,7 @@ new ByteStreamUploader( INSTANCE_NAME, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -702,7 +726,7 @@ InProcessChannelBuilder.forName("Server for " + this.getClass()) .intercept(MetadataUtils.newAttachHeadersInterceptor(metadata)) .build()), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -765,7 +789,7 @@ new ByteStreamUploader( INSTANCE_NAME, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -833,7 +857,7 @@ new ByteStreamUploader( INSTANCE_NAME, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -868,7 +892,7 @@ new ByteStreamUploader( INSTANCE_NAME, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -935,7 +959,7 @@ new ByteStreamUploader( INSTANCE_NAME, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -975,7 +999,7 @@ new ByteStreamUploader( /* instanceName */ null, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -1022,7 +1046,7 @@ new ByteStreamUploader( /* instanceName */ null, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -1060,7 +1084,7 @@ new ByteStreamUploader( INSTANCE_NAME, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -1141,7 +1165,7 @@ new ByteStreamUploader( INSTANCE_NAME, new ReferenceCountedChannel(channel), - null, /* timeout seconds */ + CallCredentialsProvider.NO_CREDENTIALS, /* timeout seconds */ 60, retrier); @@ -1199,6 +1223,158 @@ withEmptyMetadata.detach(prevContext); } + @Test + public void unauthenticatedErrorShouldNotBeRetried() throws Exception { + Context prevContext = withEmptyMetadata.attach(); + RemoteRetrier retrier = + TestUtils.newRemoteRetrier( + () -> mockBackoff, RemoteRetrier.RETRIABLE_GRPC_ERRORS, retryService); + + AtomicInteger refreshTimes = new AtomicInteger(); + CallCredentialsProvider callCredentialsProvider = + new CallCredentialsProvider() { + @Nullable + @Override + public CallCredentials getCallCredentials() { + return null; + } + + @Override + public void refresh() throws IOException { + refreshTimes.incrementAndGet(); + } + }; + ByteStreamUploader uploader = + new ByteStreamUploader( + INSTANCE_NAME, + new ReferenceCountedChannel(channel), + callCredentialsProvider, /* timeout seconds */ + 60, + retrier); + + byte[] blob = new byte[CHUNK_SIZE * 2 + 1]; + new Random().nextBytes(blob); + + Chunker chunker = Chunker.builder().setInput(blob).setChunkSize(CHUNK_SIZE).build(); + HashCode hash = HashCode.fromString(DIGEST_UTIL.compute(blob).getHash()); + + AtomicInteger numUploads = new AtomicInteger(); + serviceRegistry.addService( + new ByteStreamImplBase() { + @Override + public StreamObserver<WriteRequest> write(StreamObserver<WriteResponse> streamObserver) { + numUploads.incrementAndGet(); + + streamObserver.onError(Status.UNAUTHENTICATED.asException()); + return new NoopStreamObserver(); + } + }); + + assertThrows( + IOException.class, + () -> { + uploader.uploadBlob(hash, chunker, true); + }); + + assertThat(refreshTimes.get()).isEqualTo(1); + assertThat(numUploads.get()).isEqualTo(2); + + // This test should not have triggered any retries. + Mockito.verifyZeroInteractions(mockBackoff); + + blockUntilInternalStateConsistent(uploader); + + withEmptyMetadata.detach(prevContext); + } + + @Test + public void shouldRefreshCredentialsOnAuthenticationError() throws Exception { + Context prevContext = withEmptyMetadata.attach(); + RemoteRetrier retrier = + TestUtils.newRemoteRetrier( + () -> mockBackoff, RemoteRetrier.RETRIABLE_GRPC_ERRORS, retryService); + + AtomicInteger refreshTimes = new AtomicInteger(); + CallCredentialsProvider callCredentialsProvider = + new CallCredentialsProvider() { + @Nullable + @Override + public CallCredentials getCallCredentials() { + return null; + } + + @Override + public void refresh() throws IOException { + refreshTimes.incrementAndGet(); + } + }; + ByteStreamUploader uploader = + new ByteStreamUploader( + INSTANCE_NAME, + new ReferenceCountedChannel(channel), + callCredentialsProvider, /* timeout seconds */ + 60, + retrier); + + byte[] blob = new byte[CHUNK_SIZE * 2 + 1]; + new Random().nextBytes(blob); + + Chunker chunker = Chunker.builder().setInput(blob).setChunkSize(CHUNK_SIZE).build(); + HashCode hash = HashCode.fromString(DIGEST_UTIL.compute(blob).getHash()); + + AtomicInteger numUploads = new AtomicInteger(); + serviceRegistry.addService( + new ByteStreamImplBase() { + @Override + public StreamObserver<WriteRequest> write(StreamObserver<WriteResponse> streamObserver) { + numUploads.incrementAndGet(); + + if (refreshTimes.get() == 0) { + streamObserver.onError(Status.UNAUTHENTICATED.asException()); + return new NoopStreamObserver(); + } + + return new StreamObserver<WriteRequest>() { + long nextOffset = 0; + + @Override + public void onNext(WriteRequest writeRequest) { + nextOffset += writeRequest.getData().size(); + boolean lastWrite = blob.length == nextOffset; + assertThat(writeRequest.getFinishWrite()).isEqualTo(lastWrite); + } + + @Override + public void onError(Throwable throwable) { + fail("onError should never be called."); + } + + @Override + public void onCompleted() { + assertThat(nextOffset).isEqualTo(blob.length); + + WriteResponse response = + WriteResponse.newBuilder().setCommittedSize(nextOffset).build(); + streamObserver.onNext(response); + streamObserver.onCompleted(); + } + }; + } + }); + + uploader.uploadBlob(hash, chunker, true); + + assertThat(refreshTimes.get()).isEqualTo(1); + assertThat(numUploads.get()).isEqualTo(2); + + // This test should not have triggered any retries. + Mockito.verifyZeroInteractions(mockBackoff); + + blockUntilInternalStateConsistent(uploader); + + withEmptyMetadata.detach(prevContext); + } + private static class NoopStreamObserver implements StreamObserver<WriteRequest> { @Override public void onNext(WriteRequest writeRequest) {
diff --git a/src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java b/src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java index 82bbf06..a75bcb0 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java +++ b/src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java
@@ -55,6 +55,7 @@ import com.google.devtools.build.lib.actions.ActionInputHelper; import com.google.devtools.build.lib.actions.cache.VirtualActionInput; import com.google.devtools.build.lib.authandtls.AuthAndTLSOptions; +import com.google.devtools.build.lib.authandtls.CallCredentialsProvider; import com.google.devtools.build.lib.authandtls.GoogleAuthUtils; import com.google.devtools.build.lib.clock.JavaClock; import com.google.devtools.build.lib.remote.RemoteRetrier.ExponentialBackoff; @@ -210,10 +211,12 @@ Scratch scratch = new Scratch(); scratch.file(authTlsOptions.googleCredentials, new JacksonFactory().toString(json)); - CallCredentials creds; + CallCredentialsProvider callCredentialsProvider; try (InputStream in = scratch.resolve(authTlsOptions.googleCredentials).getInputStream()) { - creds = GoogleAuthUtils.newCallCredentials(in, authTlsOptions.googleAuthScopes); + callCredentialsProvider = + GoogleAuthUtils.newCallCredentialsProvider(in, authTlsOptions.googleAuthScopes); } + CallCredentials creds = callCredentialsProvider.getCallCredentials(); RemoteRetrier retrier = TestUtils.newRemoteRetrier( @@ -229,7 +232,7 @@ new ByteStreamUploader( remoteOptions.remoteInstanceName, channel.retain(), - creds, + callCredentialsProvider, remoteOptions.remoteTimeout.getSeconds(), retrier); return new GrpcCacheClient(
diff --git a/src/test/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutionClientTest.java b/src/test/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutionClientTest.java index 823fe75..0d8fd7f 100644 --- a/src/test/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutionClientTest.java +++ b/src/test/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutionClientTest.java
@@ -61,6 +61,7 @@ import com.google.devtools.build.lib.actions.SpawnResult; import com.google.devtools.build.lib.analysis.BlazeVersionInfo; import com.google.devtools.build.lib.authandtls.AuthAndTLSOptions; +import com.google.devtools.build.lib.authandtls.CallCredentialsProvider; import com.google.devtools.build.lib.authandtls.GoogleAuthUtils; import com.google.devtools.build.lib.clock.JavaClock; import com.google.devtools.build.lib.collect.nestedset.NestedSetBuilder; @@ -259,13 +260,14 @@ .build()); GrpcRemoteExecutor executor = new GrpcRemoteExecutor(channel.retain(), null, retrier, remoteOptions); - CallCredentials creds = - GoogleAuthUtils.newCallCredentials(Options.getDefaults(AuthAndTLSOptions.class)); + CallCredentialsProvider callCredentialsProvider = + GoogleAuthUtils.newCallCredentialsProvider(Options.getDefaults(AuthAndTLSOptions.class)); + CallCredentials creds = callCredentialsProvider.getCallCredentials(); ByteStreamUploader uploader = new ByteStreamUploader( remoteOptions.remoteInstanceName, channel.retain(), - creds, + callCredentialsProvider, remoteOptions.remoteTimeout.getSeconds(), retrier); GrpcCacheClient cacheProtocol =