xds: check that resources are received from the right servers

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,9 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.Any;
+import io.envoyproxy.envoy.config.listener.v3.Listener;
+import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
 import io.grpc.MetricRecorder;
 import io.grpc.Status;
 import io.grpc.StatusOr;
@@ -142,6 +145,51 @@ 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() {
+    trafficdirector.setLdsConfig(ControlPlaneRule.buildServerListener(),
+        ControlPlaneRule.buildClientListener("test-server"));
+
+    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());
+
+    // trafficdirector (top-level server) erroneously sends a response containing both an update
+    // for test-server and a listener owned by server-one.
+    Listener alienListener = ControlPlaneRule.buildClientListener(
+        "xdstp://server-one/envoy.config.listener.v3.Listener/test-server");
+    Listener updatedClientListener = ControlPlaneRule.buildClientListener("test-server",
+        "new-route-config.googleapis.com");
+    DiscoveryResponse rogueResponse = DiscoveryResponse.newBuilder()
+        .setTypeUrl(XdsTestControlPlaneService.ADS_TYPE_URL_LDS)
+        .setVersionInfo("rogue-version-1")
+        .setNonce("rogue-nonce-1")
+        .addResources(Any.pack(updatedClientListener))
+        .addResources(Any.pack(alienListener))
+        .build();
+    trafficdirector.getService().sendDiscoveryResponse(rogueResponse);
+
+    // Confirm that trafficdirector's second response was processed by waiting for mockWatcher
+    // to receive the updated test-server config.
+    verify(mockWatcher, timeout(10000).times(2)).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.

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

@@ -119,6 +119,21 @@ public void run() {
     });
   }
 
+  /**
+   * Directly sends the given DiscoveryResponse to all active subscribers of the response's type.
+   */
+  public void sendDiscoveryResponse(DiscoveryResponse response) {
+    syncContext.execute(new Runnable() {
+      @Override
+      public void run() {
+        for (StreamObserver<DiscoveryResponse> observer :
+            subscribers.get(response.getTypeUrl()).keySet()) {
+          observer.onNext(response);
+        }
+      }
+    });
+  }
+
   ImmutableMap<String, Message> getCurrentConfig(String type) {
     HashMap<String, Message> hashMap = xdsResources.get(type);
     return (hashMap != null) ? ImmutableMap.copyOf(hashMap) : ImmutableMap.of();