xds: deletion only to watchers of same control plane (#9896)

When XdsClient learns that a control plane no longer tracks a resource,
it should only notify watchers associated with that control plane.

This matters in control plane federation cases when more than one
control plane is in use.

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

@@ -477,10 +477,13 @@ final class XdsClientImpl extends XdsClient
       }
 
       // For State of the World services, notify watchers when their watched resource is missing
-      // from the ADS update.
+      // 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 (subscriber.serverInfo.equals(args.serverInfo)) {
         subscriber.onAbsent();
       }
     }
+  }
 
   /**
    * Tracks a single subscribed resource.

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

@@ -76,15 +76,24 @@ public class ControlPlaneRule extends TestWatcher {
   private static final String EDS_NAME = "eds-service-0";
   private static final String SERVER_LISTENER_TEMPLATE_NO_REPLACEMENT =
       "grpc/server?udpa.resource.listening_address=";
-  private static final String SERVER_HOST_NAME = "test-server";
   private static final String HTTP_CONNECTION_MANAGER_TYPE_URL =
       "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3"
           + ".HttpConnectionManager";
 
+  private String serverHostName;
   private Server server;
   private XdsTestControlPlaneService controlPlaneService;
   private XdsNameResolverProvider nameResolverProvider;
 
+  public ControlPlaneRule() {
+    serverHostName = "test-server";
+  }
+
+  public ControlPlaneRule setServerHostName(String serverHostName) {
+    this.serverHostName = serverHostName;
+    return this;
+  }
+
   /**
    * Returns the test control plane service interface.
    */
@@ -155,7 +164,7 @@ public class ControlPlaneRule extends TestWatcher {
   void setLdsConfig(Listener serverListener, Listener clientListener) {
     getService().setXdsConfig(ADS_TYPE_URL_LDS,
         ImmutableMap.of(SERVER_LISTENER_TEMPLATE_NO_REPLACEMENT, serverListener,
-                        SERVER_HOST_NAME, clientListener));
+                        serverHostName, clientListener));
   }
 
   void setRdsConfig(RouteConfiguration routeConfiguration) {

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

@@ -0,0 +1,150 @@
+/*
+ * Copyright 2023 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds;
+
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.grpc.internal.ObjectPool;
+import io.grpc.xds.Filter.NamedFilterConfig;
+import io.grpc.xds.XdsClient.ResourceWatcher;
+import io.grpc.xds.XdsListenerResource.LdsUpdate;
+import java.util.Collections;
+import java.util.Map;
+import java.util.UUID;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+/**
+ * Tests for xDS control plane federation scenarios.
+ */
+@RunWith(JUnit4.class)
+public class XdsClientFederationTest {
+
+  @Mock
+  private ResourceWatcher<LdsUpdate> mockDirectPathWatcher;
+
+  @Mock
+  private ResourceWatcher<LdsUpdate> mockWatcher;
+
+  @Rule
+  public ControlPlaneRule trafficdirector = new ControlPlaneRule().setServerHostName("test-server");
+
+  @Rule
+  public ControlPlaneRule directpathPa = new ControlPlaneRule().setServerHostName(
+      "xdstp://server-one/envoy.config.listener.v3.Listener/test-server");
+
+  private ObjectPool<XdsClient> xdsClientPool;
+  private XdsClient xdsClient;
+  private boolean originalFederationStatus;
+
+  @Before
+  public void setUp() throws XdsInitializationException {
+    MockitoAnnotations.initMocks(this);
+
+    originalFederationStatus = BootstrapperImpl.enableFederation;
+    BootstrapperImpl.enableFederation = true;
+
+    SharedXdsClientPoolProvider clientPoolProvider = new SharedXdsClientPoolProvider();
+    clientPoolProvider.setBootstrapOverride(defaultBootstrapOverride());
+    xdsClientPool = clientPoolProvider.getOrCreate();
+    xdsClient = xdsClientPool.getObject();
+  }
+
+  @After
+  public void cleanUp() throws InterruptedException {
+    BootstrapperImpl.enableFederation = originalFederationStatus;
+    xdsClientPool.returnObject(xdsClient);
+  }
+
+  /**
+   * Assures that resource deletions happening in one control plane do not trigger deletion events
+   * in watchers of resources on other control planes.
+   */
+  @Test
+  public void isolatedResourceDeletions() throws InterruptedException {
+    trafficdirector.setLdsConfig(ControlPlaneRule.buildServerListener(),
+        ControlPlaneRule.buildClientListener("test-server"));
+    directpathPa.setLdsConfig(ControlPlaneRule.buildServerListener(),
+        ControlPlaneRule.buildClientListener(
+            "xdstp://server-one/envoy.config.listener.v3.Listener/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(2000)).onChanged(
+        LdsUpdate.forApiListener(
+            HttpConnectionManager.forRdsName(0, "route-config.googleapis.com", ImmutableList.of(
+                new NamedFilterConfig("terminal-filter", RouterFilter.ROUTER_CONFIG)))));
+    verify(mockDirectPathWatcher, timeout(2000)).onChanged(
+        LdsUpdate.forApiListener(
+            HttpConnectionManager.forRdsName(0, "route-config.googleapis.com", ImmutableList.of(
+                new NamedFilterConfig("terminal-filter", RouterFilter.ROUTER_CONFIG)))));
+
+    // By setting the LDS config with a new server name we effectively make the old server to go
+    // away as it is not in the configuration anymore. This change in one control plane (here the
+    // "normal TrafficDirector" one) should not trigger an onResourceDoesNotExist() call on a
+    // watcher of another control plane (here the DirectPath one).
+    trafficdirector.setLdsConfig(ControlPlaneRule.buildServerListener(),
+        ControlPlaneRule.buildClientListener("new-server"));
+    verify(mockWatcher, timeout(20000)).onResourceDoesNotExist("test-server");
+    verify(mockDirectPathWatcher, times(0)).onResourceDoesNotExist(
+            "xdstp://server-one/envoy.config.listener.v3.Listener/test-server");
+  }
+
+  private Map<String, ?> defaultBootstrapOverride() {
+    return ImmutableMap.of(
+        "node", ImmutableMap.of(
+            "id", UUID.randomUUID().toString(),
+            "cluster", "cluster0"),
+        "xds_servers", ImmutableList.of(
+            ImmutableMap.of(
+                "server_uri", "localhost:" + trafficdirector.getServer().getPort(),
+                "channel_creds", Collections.singletonList(
+                    ImmutableMap.of("type", "insecure")
+                ),
+                "server_features", Collections.singletonList("xds_v3")
+            )
+        ),
+        "authorities", ImmutableMap.of(
+            "", ImmutableMap.of(),
+            "server-one", ImmutableMap.of(
+                "xds_servers", ImmutableList.of(
+                    ImmutableMap.of(
+                        "server_uri", "localhost:" + directpathPa.getServer().getPort(),
+                        "channel_creds", Collections.singletonList(
+                            ImmutableMap.of("type", "insecure")
+                        ),
+                        "server_features", Collections.singletonList("xds_v3")
+                    )
+                )
+            ),
+            "server-two", ImmutableMap.of()
+        )
+    );
+  }
+}