Introduce XdsClusterManager to cache Clusters

xds/src/main/java/com/linecorp/armeria/xds/AbstractResourceNode.java

@@ -1,7 +1,7 @@
 /*
- * Copyright 2023 LINE Corporation
+ * Copyright 2025 LY Corporation
  *
- * LINE Corporation licenses this file to you under the Apache License,
+ * LY Corporation licenses this file to you 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:
  *
@@ -16,119 +16,166 @@
 
 package com.linecorp.armeria.xds;
 
-import java.util.ArrayDeque;
-import java.util.Deque;
+import static com.google.common.base.Preconditions.checkArgument;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.MoreObjects;
 
 import com.linecorp.armeria.common.annotation.Nullable;
 
 import io.envoyproxy.envoy.config.core.v3.ConfigSource;
 import io.grpc.Status;
 
-abstract class AbstractResourceNode<T extends XdsResource> implements ResourceNode<T> {
+abstract class AbstractResourceNode<T extends XdsResource, S extends Snapshot<T>> implements ResourceNode<T> {
 
-    private final Deque<ResourceNode<?>> children = new ArrayDeque<>();
+    private static final Logger logger = LoggerFactory.getLogger(AbstractResourceNode.class);
 
-    private final XdsBootstrapImpl xdsBootstrap;
+    private final SubscriptionContext context;
     @Nullable
     private final ConfigSource configSource;
     private final XdsType type;
     private final String resourceName;
-    private final SnapshotWatcher<? extends Snapshot<T>> parentWatcher;
+    private final Set<SnapshotWatcher<S>> watchers = new HashSet<>();
     private final ResourceNodeType resourceNodeType;
     @Nullable
-    private T current;
+    private S snapshot;
 
-    AbstractResourceNode(XdsBootstrapImpl xdsBootstrap, @Nullable ConfigSource configSource,
-                         XdsType type, String resourceName,
-                         SnapshotWatcher<? extends Snapshot<T>> parentWatcher,
+    AbstractResourceNode(SubscriptionContext context, @Nullable ConfigSource configSource,
+                         XdsType type, String resourceName, SnapshotWatcher<S> parentWatcher,
                          ResourceNodeType resourceNodeType) {
-        this.xdsBootstrap = xdsBootstrap;
+        if (resourceNodeType == ResourceNodeType.DYNAMIC) {
+            checkArgument(configSource != null, "Dynamic node <%s.%s> received a null config source",
+                          type, resourceName);
+        } else if (resourceNodeType == ResourceNodeType.STATIC) {
+            checkArgument(configSource == null, "Static node <%s.%s> received a config source <%s>",
+                          type, resourceName, configSource);
+        }
+        this.context = context;
+        this.configSource = configSource;
+        this.type = type;
+        this.resourceName = resourceName;
+        this.resourceNodeType = resourceNodeType;
+        watchers.add(parentWatcher);
+    }
+
+    AbstractResourceNode(SubscriptionContext context, @Nullable ConfigSource configSource,
+                         XdsType type, String resourceName, ResourceNodeType resourceNodeType) {
+        this.context = context;
         this.configSource = configSource;
         this.type = type;
         this.resourceName = resourceName;
-        this.parentWatcher = parentWatcher;
         this.resourceNodeType = resourceNodeType;
     }
 
-    XdsBootstrapImpl xdsBootstrap() {
-        return xdsBootstrap;
+    final SubscriptionContext context() {
+        return context;
     }
 
     @Nullable
     @Override
-    public ConfigSource configSource() {
+    public final ConfigSource configSource() {
         return configSource;
     }
 
-    private void setCurrent(@Nullable T current) {
-        this.current = current;
+    final void addWatcher(SnapshotWatcher<S> watcher) {
+        watchers.add(watcher);
+        if (snapshot != null) {
+            watcher.snapshotUpdated(snapshot);
+        }
     }
 
-    @Nullable
-    @Override
-    public T currentResource() {
-        return current;
+    final void removeWatcher(SnapshotWatcher<S> watcher) {
+        watchers.remove(watcher);
     }
 
-    @Override
-    public void onError(XdsType type, Status error) {
-        parentWatcher.onError(type, error);
+    final boolean hasWatchers() {
+        return !watchers.isEmpty();
     }
 
     @Override
-    public void onResourceDoesNotExist(XdsType type, String resourceName) {
-        setCurrent(null);
+    public final void onError(XdsType type, Status error) {
+        notifyOnError(type, error);
+    }
 
-        for (ResourceNode<?> child: children) {
-            child.close();
+    final void notifyOnError(XdsType type, Status error) {
+        for (SnapshotWatcher<S> watcher : watchers) {
+            try {
+                watcher.onError(type, error);
+            } catch (Exception e) {
+                logger.warn("Unexpected exception notifying <{}> for 'onError' <{},{}> for error <{}> e: ",
+                            watcher, resourceName, type, error, e);
+            }
         }
-        children.clear();
-        parentWatcher.onMissing(type, resourceName);
     }
 
     @Override
-    public void onChanged(T update) {
-        assert update.type() == type();
-        setCurrent(update);
+    public final void onResourceDoesNotExist(XdsType type, String resourceName) {
+        notifyOnMissing(type, resourceName);
+    }
 
-        final Deque<ResourceNode<?>> prevChildren = new ArrayDeque<>(children);
-        children.clear();
+    final void notifyOnMissing(XdsType type, String resourceName) {
+        for (SnapshotWatcher<S> watcher : watchers) {
+            try {
+                watcher.onMissing(type, resourceName);
+            } catch (Exception e) {
+                logger.warn("Unexpected exception notifying <{}> for 'onMissing' <{},{}> e: ",
+                            watcher, resourceName, type, e);
+            }
+        }
+    }
 
+    @Override
+    public final void onChanged(T update) {
+        assert update.type() == type();
         doOnChanged(update);
-
-        for (ResourceNode<?> child: prevChildren) {
-            child.close();
-        }
     }
 
     abstract void doOnChanged(T update);
 
+    final void notifyOnChanged(S snapshot) {
+        this.snapshot = snapshot;
+        for (SnapshotWatcher<S> watcher : watchers) {
+            try {
+                watcher.snapshotUpdated(snapshot);
+            } catch (Exception e) {
+                logger.warn("Unexpected exception notifying <{}> for 'snapshotUpdated' <{}> e: ",
+                            watcher, snapshot, e);
+            }
+        }
+    }
+
     @Override
     public void close() {
-        for (ResourceNode<?> child: children) {
-            child.close();
-        }
-        children.clear();
         if (resourceNodeType == ResourceNodeType.DYNAMIC) {
-            xdsBootstrap.unsubscribe(this);
+            context.unsubscribe(this);
         }
     }
 
-    Deque<ResourceNode<?>> children() {
-        return children;
-    }
-
     @Override
-    public XdsType type() {
+    public final XdsType type() {
         return type;
     }
 
     @Override
-    public String name() {
+    public final String name() {
         return resourceName;
     }
 
-    ConfigSourceMapper configSourceMapper() {
-        return xdsBootstrap.configSourceMapper().withParentConfigSource(configSource);
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                          .add("context", context)
+                          .add("configSource", configSource)
+                          .add("type", type)
+                          .add("resourceName", resourceName)
+                          .add("watchers", watchers)
+                          .add("resourceNodeType", resourceNodeType)
+                          .add("snapshot", snapshot)
+                          .toString();
     }
 }

xds/src/main/java/com/linecorp/armeria/xds/AbstractRoot.java

@@ -1,7 +1,7 @@
 /*
- * Copyright 2023 LINE Corporation
+ * Copyright 2025 LY Corporation
  *
- * LINE Corporation licenses this file to you under the Apache License,
+ * LY Corporation licenses this file to you 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:
  *
@@ -80,23 +80,6 @@ public void addSnapshotWatcher(SnapshotWatcher<? super T> watcher) {
         }
     }
 
-    /**
-     * Removes a watcher which waits for a snapshot update.
-     */
-    public void removeSnapshotWatcher(SnapshotWatcher<? super T> watcher) {
-        requireNonNull(watcher, "watcher");
-        checkState(!closed, "Watcher %s can't be removed since %s is already closed.",
-                   watcher, getClass().getSimpleName());
-        if (!eventLoop.inEventLoop()) {
-            eventLoop.execute(() -> removeSnapshotWatcher(watcher));
-            return;
-        }
-        if (closed) {
-            return;
-        }
-        snapshotWatchers.remove(watcher);
-    }
-
     @Override
     public void snapshotUpdated(T newSnapshot) {
         if (closed) {

xds/src/main/java/com/linecorp/armeria/xds/BootstrapClusters.java

@@ -1,7 +1,7 @@
 /*
- * Copyright 2024 LINE Corporation
+ * Copyright 2025 LY Corporation
  *
- * LINE Corporation licenses this file to you under the Apache License,
+ * LY Corporation licenses this file to you 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:
  *
@@ -22,43 +22,54 @@
 import com.linecorp.armeria.common.annotation.Nullable;
 
 import io.envoyproxy.envoy.config.bootstrap.v3.Bootstrap;
-import io.envoyproxy.envoy.config.bootstrap.v3.Bootstrap.StaticResources;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster;
 import io.grpc.Status;
+import io.netty.util.concurrent.EventExecutor;
 
 final class BootstrapClusters implements SnapshotWatcher<ClusterSnapshot> {
 
     private final Map<String, ClusterSnapshot> clusterSnapshots = new HashMap<>();
-    private final Map<String, Cluster> clusters = new HashMap<>();
+    private final Bootstrap bootstrap;
+    private final EventExecutor eventLoop;
+    private final XdsClusterManager clusterManager;
 
-    BootstrapClusters(Bootstrap bootstrap, XdsBootstrapImpl xdsBootstrap) {
-        if (bootstrap.hasStaticResources()) {
-            final StaticResources staticResources = bootstrap.getStaticResources();
-            for (Cluster cluster : staticResources.getClustersList()) {
-                if (cluster.hasLoadAssignment()) {
-                    // no need to clean this cluster up since it is fully static
-                    StaticResourceUtils.staticCluster(xdsBootstrap, cluster.getName(), this, cluster);
-                }
-                clusters.put(cluster.getName(), cluster);
+    BootstrapClusters(Bootstrap bootstrap, EventExecutor eventLoop, XdsClusterManager clusterManager) {
+        this.bootstrap = bootstrap;
+        this.eventLoop = eventLoop;
+        this.clusterManager = clusterManager;
+        initializePrimary(bootstrap);
+    }
+
+    private void initializePrimary(Bootstrap bootstrap) {
+        final StaticSubscriptionContext context = new StaticSubscriptionContext(eventLoop);
+        for (Cluster cluster: bootstrap.getStaticResources().getClustersList()) {
+            if (!cluster.hasLoadAssignment()) {
+                continue;
             }
+            clusterManager.register(cluster, context, this);
+        }
+    }
+
+    void initializeSecondary(SubscriptionContext context) {
+        for (Cluster cluster: bootstrap.getStaticResources().getClustersList()) {
+            if (!cluster.hasEdsClusterConfig()) {
+                continue;
+            }
+            clusterManager.register(cluster, context, this);
         }
     }
 
     @Override
     public void snapshotUpdated(ClusterSnapshot newSnapshot) {
-        clusterSnapshots.put(newSnapshot.xdsResource().name(), newSnapshot);
+        final String name = newSnapshot.xdsResource().name();
+        clusterSnapshots.put(name, newSnapshot);
     }
 
     @Nullable
     ClusterSnapshot clusterSnapshot(String clusterName) {
         return clusterSnapshots.get(clusterName);
     }
 
-    @Nullable
-    Cluster cluster(String clusterName) {
-        return clusters.get(clusterName);
-    }
-
     @Override
     public void onMissing(XdsType type, String resourceName) {
         throw new IllegalArgumentException("Bootstrap cluster not found for type: '" +