xds: apply ext_authz check deadline per-RPC

xds/src/main/java/io/grpc/xds/ExtAuthzFilter.java

@@ -42,7 +42,6 @@
 import io.grpc.xds.internal.headermutations.HeaderMutationFilter;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.Immutable;
 import javax.annotation.concurrent.ThreadSafe;
@@ -218,11 +217,6 @@ public ClientInterceptor buildClientInterceptor(FilterConfig config,
       stub = stub.withCallCredentials(
           extAuthzConfig.grpcService().googleGrpc().callCredentials().get());
     }
-    if (extAuthzConfig.grpcService().timeout().isPresent()) {
-      stub = stub.withDeadlineAfter(
-          extAuthzConfig.grpcService().timeout().get().toMillis(),
-          TimeUnit.MILLISECONDS);
-    }
     return new ExtAuthzClientInterceptor(extAuthzConfig, stub, random,
         new CheckRequestBuilder(extAuthzConfig),
         new CheckResponseHandler(

xds/src/main/java/io/grpc/xds/internal/extauthz/ExtAuthzClientCall.java

@@ -31,6 +31,7 @@
 import io.grpc.internal.DelayedClientCall;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 
 /**
@@ -64,7 +65,9 @@ public ExtAuthzClientCall(
     this.callOptions = callOptions;
     this.next = next;
     this.method = method;
-    this.authzStub = authzStub;
+    this.authzStub = config.grpcService().timeout()
+        .map(t -> authzStub.withDeadlineAfter(t.toMillis(), TimeUnit.MILLISECONDS))
+        .orElse(authzStub);
     this.checkRequestBuilder = checkRequestBuilder;
     this.responseHandler = responseHandler;
     this.config = config;

xds/src/test/java/io/grpc/xds/ExtAuthzFilterTest.java

@@ -174,7 +174,7 @@ public void buildClientInterceptor_success() {
   }
 
   @Test
-  public void buildClientInterceptor_withTimeout_appliesDeadline() {
+  public void buildClientInterceptor_withTimeout_doesNotBakeDeadlineIntoSharedStub() {
     GrpcServiceConfig.GoogleGrpcConfig googleGrpc = GrpcServiceConfig.GoogleGrpcConfig.builder()
         .target("test-cluster")
         .configuredChannelCredentials(io.grpc.xds.client.ConfiguredChannelCredentials.create(
@@ -204,7 +204,10 @@ public void buildClientInterceptor_withTimeout_appliesDeadline() {
     assertThat(created).isInstanceOf(ExtAuthzFilter.ExtAuthzClientInterceptor.class);
     ExtAuthzFilter.ExtAuthzClientInterceptor interceptor =
         (ExtAuthzFilter.ExtAuthzClientInterceptor) created;
-    assertThat(interceptor.getAuthzStubForTest().getCallOptions().getDeadline()).isNotNull();
+    // The stub is shared by every RPC this interceptor handles. Resolving the timeout into an
+    // absolute Deadline here would expire it for all later calls, so it is applied per-RPC by
+    // ExtAuthzClientCall instead.
+    assertThat(interceptor.getAuthzStubForTest().getCallOptions().getDeadline()).isNull();
   }
 
   @Test

xds/src/test/java/io/grpc/xds/internal/extauthz/ExtAuthzClientCallTest.java

@@ -34,6 +34,7 @@
 import io.grpc.ClientInterceptor;
 import io.grpc.ClientInterceptors;
 import io.grpc.Context;
+import io.grpc.Deadline;
 import io.grpc.ManagedChannel;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
@@ -56,6 +57,7 @@
 import io.grpc.xds.internal.grpcservice.HeaderValue;
 import io.grpc.xds.internal.headermutations.HeaderMutations;
 import io.grpc.xds.internal.headermutations.HeaderValueOption;
+import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -627,6 +629,28 @@ public void drain_executesOnCallExecutor() throws Exception {
     verify(mockExecutor, org.mockito.Mockito.atLeastOnce()).execute(any(Runnable.class));
   }
 
+  @Test
+  public void start_withTimeout_appliesDeadlineToCheckRpc() throws Exception {
+    ExtAuthzConfig configWithTimeout = buildConfigWithTimeout(Duration.ofSeconds(10));
+    AtomicReference<Deadline> observedDeadline = new AtomicReference<>();
+    CountDownLatch checkReceived = new CountDownLatch(1);
+    doAnswer(invocation -> {
+      observedDeadline.set(Context.current().getDeadline());
+      checkReceived.countDown();
+      return null;
+    }).when(authzService).check(any(CheckRequest.class), ArgumentMatchers.any());
+
+    createCall(com.google.common.util.concurrent.MoreExecutors.directExecutor(), channel,
+        configWithTimeout).start(new CapturingListener<>(), new Metadata());
+
+    assertThat(checkReceived.await(5, TimeUnit.SECONDS)).isTrue();
+    Deadline deadline = observedDeadline.get();
+    // ExtAuthzFilterTest asserts the shared stub carries no deadline, so one observed here can
+    // only have been derived for this individual call.
+    assertThat(deadline).isNotNull();
+    assertThat(deadline.timeRemaining(TimeUnit.MILLISECONDS)).isAtMost(10_000L);
+  }
+
   private ExtAuthzClientCall<SimpleRequest, SimpleResponse>
       createCall() {
     return createCall(channel, config);
@@ -675,4 +699,32 @@ private static ExtAuthzConfig buildConfig() {
         .statusOnError(Status.PERMISSION_DENIED)
         .build();
   }
+
+  private static ExtAuthzConfig buildConfigWithTimeout(Duration timeout) {
+    GrpcServiceConfig.GoogleGrpcConfig googleGrpc =
+        GrpcServiceConfig.GoogleGrpcConfig.builder()
+            .target("test-cluster")
+            .configuredChannelCredentials(
+                ConfiguredChannelCredentials.create(
+                    mock(ChannelCredentials.class),
+                    mock(ConfiguredChannelCredentials
+                        .ChannelCredsConfig.class)))
+            .build();
+    GrpcServiceConfig grpcServiceConfig =
+        GrpcServiceConfig.builder()
+            .googleGrpc(googleGrpc)
+            .initialMetadata(ImmutableList.of())
+            .timeout(timeout)
+            .build();
+    return ExtAuthzConfig.builder()
+        .grpcService(grpcServiceConfig)
+        .failureModeAllow(false)
+        .failureModeAllowHeaderAdd(false)
+        .includePeerCertificate(false)
+        .denyAtDisable(false)
+        .filterEnabled(
+            Matchers.FractionMatcher.create(100, 100))
+        .statusOnError(Status.PERMISSION_DENIED)
+        .build();
+  }
 }