xds: enforce control plane isolation in handleResourceUpdate

Ensure that handleResourceUpdate only applies resource updates, errors,
or SotW absent notices to subscribers whose active control plane client
matches the controlPlaneClient that received the ADS response.

TAG=agy
CONV=e27e4459-da3e-4c14-b2a1-2055f0b9b5d1

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java

@@ -606,8 +606,13 @@ private <T extends ResourceUpdate> void handleResourceUpdate(
     Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscribedResources =
         resourceSubscribers.getOrDefault(xdsResourceType, Collections.emptyMap());
     for (Map.Entry<String, ResourceSubscriber<?>> entry : subscribedResources.entrySet()) {
-      String resourceName = entry.getKey();
       ResourceSubscriber<T> subscriber = (ResourceSubscriber<T>) entry.getValue();
+      // Only process resources subscribed to the control plane that sent this update.
+      if (getActiveCpc(subscriber.authority) != controlPlaneClient) {
+        continue;
+      }
+
+      String resourceName = entry.getKey();
       if (parsedResources.containsKey(resourceName)) {
         // Happy path: the resource updated successfully. Notify the watchers of the update.
         subscriber.onData(parsedResources.get(resourceName), args.versionInfo, updateTime,
@@ -641,9 +646,7 @@ private <T extends ResourceUpdate> void handleResourceUpdate(
       // For State of the World services, notify watchers when their watched resource is missing
       // from the ADS update. Note that we can only do this if the resource update is coming from
       // the same xDS server that the ResourceSubscriber is subscribed to.
-      if (getActiveCpc(subscriber.authority) == controlPlaneClient) {
-        subscriber.onAbsent(processingTracker, args.serverInfo);
-      }
+      subscriber.onAbsent(processingTracker, args.serverInfo);
     }
   }
 

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

@@ -26,6 +26,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import io.envoyproxy.envoy.config.listener.v3.Listener;
 import io.grpc.MetricRecorder;
 import io.grpc.Status;
 import io.grpc.StatusOr;
@@ -142,6 +143,39 @@ public void isolatedResourceDeletions() {
     verify(mockDirectPathWatcher, never()).onAmbientError(any());
   }
 
+  /**
+   * Assures that resource updates happening in one control plane do not trigger update events
+   * in watchers of resources owned by other control planes.
+   */
+  @Test
+  public void isolatedResourceUpdates() {
+    Listener serverListener = ControlPlaneRule.buildServerListener();
+    Listener clientListener1 = ControlPlaneRule.buildClientListener("test-server");
+    Listener clientListener2 = ControlPlaneRule.buildClientListener(
+        "xdstp://server-one/envoy.config.listener.v3.Listener/test-server");
+    // trafficdirector (top-level server) erroneously sends a listener owned by server-one.
+    trafficdirector.getService().setXdsConfig(XdsTestControlPlaneService.ADS_TYPE_URL_LDS,
+        ImmutableMap.of(
+            serverListener.getName(), serverListener,
+            clientListener1.getName(), clientListener1,
+            clientListener2.getName(), clientListener2));
+
+    xdsClient.watchXdsResource(XdsListenerResource.getInstance(), "test-server", mockWatcher);
+    xdsClient.watchXdsResource(XdsListenerResource.getInstance(),
+        "xdstp://server-one/envoy.config.listener.v3.Listener/test-server", mockDirectPathWatcher);
+
+    verify(mockWatcher, timeout(10000)).onResourceChanged(any());
+    // mockDirectPathWatcher should not receive the update sent by trafficdirector.
+    verify(mockDirectPathWatcher, never()).onResourceChanged(any());
+
+    // When server-one sends its LDS config, mockDirectPathWatcher should then receive the update.
+    directpathPa.setLdsConfig(ControlPlaneRule.buildServerListener(),
+        ControlPlaneRule.buildClientListener(
+            "xdstp://server-one/envoy.config.listener.v3.Listener/test-server"));
+
+    verify(mockDirectPathWatcher, timeout(10000)).onResourceChanged(any());
+  }
+
   /**
    * Assures that when an {@link XdsClient} is asked to add cluster locality stats it appropriately
    * starts {@link LoadReportClient}s to do that.