commit 240f731e00f659b6dd271bfb9941e4ba587e154f
author Eric Anderson <ejona@google.com> Sat May 31 07:44:55 2025 -0700
committer Eric Anderson <ejona@google.com> Fri Jun 13 15:18:14 2025 +0000
tree 2fa5f3e31ab01aed9c26a41c9e7aa9e93c571a77
parent 26bd0eee4705e4ddac0e351fded5a8bd4e49f863

xds: Avoid changing cache when watching children in XdsDepManager

The watchers can be completely regular, so the base class can do the
cache management while the subclasses are only concerned with
subscribing to children.

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

diff --git a/xds/src/main/java/io/grpc/xds/XdsDependencyManager.java b/xds/src/main/java/io/grpc/xds/XdsDependencyManager.java
index 0428852..c7333b9 100644
--- a/xds/src/main/java/io/grpc/xds/XdsDependencyManager.java
+++ b/xds/src/main/java/io/grpc/xds/XdsDependencyManager.java
@@ -333,7 +333,8 @@
             Status.INTERNAL.withDescription("Logical DNS in dependency manager unsupported")));
         break;
       default:
-        throw new IllegalStateException("Unexpected value: " + cdsUpdate.clusterType());
+        child = new EndpointConfig(StatusOr.fromStatus(Status.UNAVAILABLE.withDescription(
+              "Unknown type in cluster " + clusterName + " " + cdsUpdate.clusterType())));
     }
     if (clusters.containsKey(clusterName)) {
       // If a cycle is detected, we'll have detected it while recursing, so now there will be a key
@@ -520,7 +521,7 @@
       }
       // Don't update configuration on error, if we've already received configuration
       if (!hasDataValue()) {
-        setDataAsStatus(Status.UNAVAILABLE.withDescription(
+        this.data = StatusOr.fromStatus(Status.UNAVAILABLE.withDescription(
             String.format("Error retrieving %s: %s: %s",
               toContextString(), error.getCode(), error.getDescription())));
         maybePublishConfig();
@@ -534,11 +535,25 @@
       }
 
       checkArgument(this.resourceName.equals(resourceName), "Resource name does not match");
-      setDataAsStatus(Status.UNAVAILABLE.withDescription(
+      this.data = StatusOr.fromStatus(Status.UNAVAILABLE.withDescription(
           toContextString() + " does not exist" + nodeInfo()));
       maybePublishConfig();
     }
 
+    @Override
+    public void onChanged(T update) {
+      checkNotNull(update, "update");
+      if (cancelled) {
+        return;
+      }
+
+      this.data = StatusOr.fromValue(update);
+      subscribeToChildren(update);
+      maybePublishConfig();
+    }
+
+    protected abstract void subscribeToChildren(T update);
+
     public void close() {
       cancelled = true;
       xdsClient.cancelXdsResourceWatch(type, resourceName, this);
@@ -557,20 +572,6 @@
       return data != null && data.hasValue();
     }
 
-    String resourceName() {
-      return resourceName;
-    }
-
-    protected void setData(T data) {
-      checkNotNull(data, "data");
-      this.data = StatusOr.fromValue(data);
-    }
-
-    protected void setDataAsStatus(Status status) {
-      checkNotNull(status, "status");
-      this.data = StatusOr.fromStatus(status);
-    }
-
     public String toContextString() {
       return toContextStr(type.typeName(), resourceName);
     }
@@ -588,12 +589,7 @@
     }
 
     @Override
-    public void onChanged(XdsListenerResource.LdsUpdate update) {
-      checkNotNull(update, "update");
-      if (cancelled) {
-        return;
-      }
-
+    public void subscribeToChildren(XdsListenerResource.LdsUpdate update) {
       HttpConnectionManager httpConnectionManager = update.httpConnectionManager();
       List<VirtualHost> virtualHosts;
       if (httpConnectionManager == null) {
@@ -610,9 +606,6 @@
       if (rdsName != null) {
         addRdsWatcher(rdsName);
       }
-
-      setData(update);
-      maybePublishConfig();
     }
 
     private String getRdsName(XdsListenerResource.LdsUpdate update) {
@@ -680,14 +673,8 @@
     }
 
     @Override
-    public void onChanged(RdsUpdate update) {
-      checkNotNull(update, "update");
-      if (cancelled) {
-        return;
-      }
-      setData(update);
+    public void subscribeToChildren(RdsUpdate update) {
       updateRoutes(update.virtualHosts);
-      maybePublishConfig();
     }
 
     @Override
@@ -705,31 +692,20 @@
     }
 
     @Override
-    public void onChanged(XdsClusterResource.CdsUpdate update) {
-      checkNotNull(update, "update");
-      if (cancelled) {
-        return;
-      }
+    public void subscribeToChildren(XdsClusterResource.CdsUpdate update) {
       switch (update.clusterType()) {
         case EDS:
-          setData(update);
           addEdsWatcher(getEdsServiceName());
           break;
         case LOGICAL_DNS:
-          setData(update);
           // no eds needed
           break;
         case AGGREGATE:
-          setData(update);
           update.prioritizedClusterNames()
               .forEach(name -> addClusterWatcher(name));
           break;
         default:
-          Status error = Status.UNAVAILABLE.withDescription(
-              "unknown cluster type in " + resourceName() + " " + update.clusterType());
-          setDataAsStatus(error);
       }
-      maybePublishConfig();
     }
 
     public String getEdsServiceName() {
@@ -749,12 +725,6 @@
     }
 
     @Override
-    public void onChanged(XdsEndpointResource.EdsUpdate update) {
-      if (cancelled) {
-        return;
-      }
-      setData(checkNotNull(update, "update"));
-      maybePublishConfig();
-    }
+    public void subscribeToChildren(XdsEndpointResource.EdsUpdate update) {}
   }
 }