xds: use a standalone Context for xDS control plane RPCs (#8153)

Control plane RPCs are independent of application RPCs, they can stand for completely different lifetime. So the context for making application RPCs should not be propagated to control plane RPCs. This change makes control plane RPCs use the ROOT Context.

Author: voidzcy

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

@@ -28,6 +28,7 @@
 import io.envoyproxy.envoy.service.discovery.v3.AggregatedDiscoveryServiceGrpc;
 import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
 import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
+import io.grpc.Context;
 import io.grpc.InternalLogId;
 import io.grpc.ManagedChannel;
 import io.grpc.Status;
@@ -81,6 +82,7 @@ public void uncaughtException(Thread t, Throwable e) {
   private final InternalLogId logId;
   private final XdsLogger logger;
   private final ManagedChannel channel;
+  private final Context context;
   private final ScheduledExecutorService timeService;
   private final BackoffPolicy.Provider backoffPolicyProvider;
   private final Stopwatch stopwatch;
@@ -103,10 +105,11 @@ public void uncaughtException(Thread t, Throwable e) {
   private ScheduledHandle rpcRetryTimer;
 
   AbstractXdsClient(ManagedChannel channel, Bootstrapper.BootstrapInfo bootstrapInfo,
-      ScheduledExecutorService timeService, BackoffPolicy.Provider backoffPolicyProvider,
-      Supplier<Stopwatch> stopwatchSupplier) {
+      Context context, ScheduledExecutorService timeService,
+      BackoffPolicy.Provider backoffPolicyProvider, Supplier<Stopwatch> stopwatchSupplier) {
     this.channel = checkNotNull(channel, "channel");
     this.bootstrapInfo = checkNotNull(bootstrapInfo, "bootstrapInfo");
+    this.context = checkNotNull(context, "context");
     this.timeService = checkNotNull(timeService, "timeService");
     this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");
     stopwatch = checkNotNull(stopwatchSupplier, "stopwatchSupplier").get();
@@ -305,7 +308,12 @@ private void startRpcStream() {
     } else {
       adsStream = new AdsStreamV2();
     }
-    adsStream.start();
+    Context prevContext = context.attach();
+    try {
+      adsStream.start();
+    } finally {
+      context.detach(prevContext);
+    }
     logger.log(XdsLogLevel.INFO, "ADS stream started");
     stopwatch.reset().start();
   }

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

@@ -48,6 +48,7 @@
 import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.Rds;
 import io.envoyproxy.envoy.type.v3.FractionalPercent;
 import io.envoyproxy.envoy.type.v3.FractionalPercent.DenominatorType;
+import io.grpc.Context;
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.ManagedChannel;
 import io.grpc.Status;
@@ -132,13 +133,13 @@ final class ClientXdsClient extends AbstractXdsClient {
   private boolean reportingLoad;
 
   ClientXdsClient(
-      ManagedChannel channel, Bootstrapper.BootstrapInfo bootstrapInfo,
+      ManagedChannel channel, Bootstrapper.BootstrapInfo bootstrapInfo, Context context,
       ScheduledExecutorService timeService, BackoffPolicy.Provider backoffPolicyProvider,
       Supplier<Stopwatch> stopwatchSupplier, TimeProvider timeProvider) {
-    super(channel, bootstrapInfo, timeService, backoffPolicyProvider, stopwatchSupplier);
+    super(channel, bootstrapInfo, context, timeService, backoffPolicyProvider, stopwatchSupplier);
     loadStatsManager = new LoadStatsManager2(stopwatchSupplier);
     this.timeProvider = timeProvider;
-    lrsClient = new LoadReportClient(loadStatsManager, channel,
+    lrsClient = new LoadReportClient(loadStatsManager, channel, context,
         bootstrapInfo.getServers().get(0).isUseProtocolV3(), bootstrapInfo.getNode(),
         getSyncContext(), timeService, backoffPolicyProvider, stopwatchSupplier);
   }

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

@@ -28,6 +28,7 @@
 import io.envoyproxy.envoy.service.load_stats.v3.LoadReportingServiceGrpc.LoadReportingServiceStub;
 import io.envoyproxy.envoy.service.load_stats.v3.LoadStatsRequest;
 import io.envoyproxy.envoy.service.load_stats.v3.LoadStatsResponse;
+import io.grpc.Context;
 import io.grpc.InternalLogId;
 import io.grpc.ManagedChannel;
 import io.grpc.Status;
@@ -55,6 +56,7 @@ final class LoadReportClient {
   private final InternalLogId logId;
   private final XdsLogger logger;
   private final ManagedChannel channel;
+  private final Context context;
   private final boolean useProtocolV3;
   private final Node node;
   private final SynchronizationContext syncContext;
@@ -74,6 +76,7 @@ final class LoadReportClient {
   LoadReportClient(
       LoadStatsManager2 loadStatsManager,
       ManagedChannel channel,
+      Context context,
       boolean useProtocolV3,
       Node node,
       SynchronizationContext syncContext,
@@ -82,6 +85,7 @@ final class LoadReportClient {
       Supplier<Stopwatch> stopwatchSupplier) {
     this.loadStatsManager = checkNotNull(loadStatsManager, "loadStatsManager");
     this.channel = checkNotNull(channel, "xdsChannel");
+    this.context = checkNotNull(context, "context");
     this.useProtocolV3 = useProtocolV3;
     this.syncContext = checkNotNull(syncContext, "syncContext");
     this.timerService = checkNotNull(scheduledExecutorService, "timeService");
@@ -163,7 +167,12 @@ private void startLrsRpc() {
       lrsStream = new LrsStreamV2();
     }
     retryStopwatch.reset().start();
-    lrsStream.start();
+    Context prevContext = context.attach();
+    try {
+      lrsStream.start();
+    } finally {
+      context.detach(prevContext);
+    }
   }
 
   private abstract class LrsStream {

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

@@ -20,6 +20,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import io.grpc.ChannelCredentials;
+import io.grpc.Context;
 import io.grpc.Grpc;
 import io.grpc.ManagedChannel;
 import io.grpc.internal.ExponentialBackoffPolicy;
@@ -105,6 +106,7 @@ private static class SharedXdsClientPoolProviderHolder {
   @ThreadSafe
   @VisibleForTesting
   static class RefCountedXdsClientObjectPool implements ObjectPool<XdsClient> {
+    private final Context context = Context.ROOT;
     private final BootstrapInfo bootstrapInfo;
     private final Object lock = new Object();
     @GuardedBy("lock")
@@ -132,7 +134,7 @@ public XdsClient getObject() {
               .keepAliveTime(5, TimeUnit.MINUTES)
               .build();
           scheduler = SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE);
-          xdsClient = new ClientXdsClient(channel, bootstrapInfo, scheduler,
+          xdsClient = new ClientXdsClient(channel, bootstrapInfo, context, scheduler,
               new ExponentialBackoffPolicy.Provider(), GrpcUtil.STOPWATCH_SUPPLIER,
               TimeProvider.SYSTEM_TIME_PROVIDER);
         }

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

@@ -22,7 +22,9 @@
 import static io.grpc.xds.AbstractXdsClient.ResourceType.EDS;
 import static io.grpc.xds.AbstractXdsClient.ResourceType.LDS;
 import static io.grpc.xds.AbstractXdsClient.ResourceType.RDS;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
@@ -40,6 +42,8 @@
 import io.envoyproxy.envoy.config.route.v3.FilterConfig;
 import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.SdsSecretConfig;
 import io.grpc.BindableService;
+import io.grpc.Context;
+import io.grpc.Context.CancellableContext;
 import io.grpc.InsecureChannelCredentials;
 import io.grpc.ManagedChannel;
 import io.grpc.Status;
@@ -270,6 +274,7 @@ public void setUp() throws IOException {
         new ClientXdsClient(
             channel,
             bootstrapInfo,
+            Context.ROOT,
             fakeClock.getScheduledExecutorService(),
             backoffPolicyProvider,
             fakeClock.getStopwatchSupplier(),
@@ -1776,6 +1781,26 @@ public void multipleEdsWatchers() {
     verifySubscribedResourcesMetadataSizes(0, 0, 0, 2);
   }
 
+  @Test
+  public void useIndependentRpcContext() {
+    // Simulates making RPCs within the context of an inbound RPC.
+    CancellableContext cancellableContext = Context.current().withCancellation();
+    Context prevContext = cancellableContext.attach();
+    try {
+      DiscoveryRpcCall call = startResourceWatcher(LDS, LDS_RESOURCE, ldsResourceWatcher);
+
+      // The inbound RPC finishes and closes its context. The outbound RPC's control plane RPC
+      // should not be impacted.
+      cancellableContext.close();
+      verify(ldsResourceWatcher, never()).onError(any(Status.class));
+
+      call.sendResponse(LDS, testListenerRds, VERSION_1, "0000");
+      verify(ldsResourceWatcher).onChanged(any(LdsUpdate.class));
+    } finally {
+      cancellableContext.detach(prevContext);
+    }
+  }
+
   @Test
   public void streamClosedAndRetryWithBackoff() {
     InOrder inOrder = Mockito.inOrder(backoffPolicyProvider, backoffPolicy1, backoffPolicy2);

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

@@ -172,8 +172,8 @@ public void cancelled(Context context) {
     when(backoffPolicy2.nextBackoffNanos())
         .thenReturn(TimeUnit.SECONDS.toNanos(2L), TimeUnit.SECONDS.toNanos(20L));
     addFakeStatsData();
-    lrsClient = new LoadReportClient(loadStatsManager, channel, false, NODE, syncContext,
-        fakeClock.getScheduledExecutorService(), backoffPolicyProvider,
+    lrsClient = new LoadReportClient(loadStatsManager, channel, Context.ROOT, false, NODE,
+        syncContext, fakeClock.getScheduledExecutorService(), backoffPolicyProvider,
         fakeClock.getStopwatchSupplier());
     syncContext.execute(new Runnable() {
       @Override