diff --git a/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java b/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java index 7e9f712db17..0bc16f6864c 100644 --- a/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java +++ b/xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java @@ -606,8 +606,13 @@ private void handleResourceUpdate( Map> subscribedResources = resourceSubscribers.getOrDefault(xdsResourceType, Collections.emptyMap()); for (Map.Entry> entry : subscribedResources.entrySet()) { - String resourceName = entry.getKey(); ResourceSubscriber subscriber = (ResourceSubscriber) 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 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); } } diff --git a/xds/src/test/java/io/grpc/xds/XdsClientFederationTest.java b/xds/src/test/java/io/grpc/xds/XdsClientFederationTest.java index da310871c25..70fb09aed29 100644 --- a/xds/src/test/java/io/grpc/xds/XdsClientFederationTest.java +++ b/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. diff --git a/xds/src/test/java/io/grpc/xds/XdsTestControlPlaneService.java b/xds/src/test/java/io/grpc/xds/XdsTestControlPlaneService.java index a54893c9075..7ec446cdeef 100644 --- a/xds/src/test/java/io/grpc/xds/XdsTestControlPlaneService.java +++ b/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 observer : + subscribers.get(response.getTypeUrl()).keySet()) { + observer.onNext(response); + } + } + }); + } + ImmutableMap getCurrentConfig(String type) { HashMap hashMap = xdsResources.get(type); return (hashMap != null) ? ImmutableMap.copyOf(hashMap) : ImmutableMap.of();