Introduce virtual host snapshots (#6105)

Motivation:

We are planning on deprecating `XdsEndpointGroup` and introducing
`XdsPreprocessor` at #6096.

Previously, `XdsEndpointGroup` created sub-`EndpointGroup`s which
consume resources. (e.g. `HealthCheckedEndpointGroup`,
`DnsEndpointGroup`, etc..) assuming that all xDS clients will share a
common `EndpointGroup`.

However, `XdsPreprocessor` cannot be shared among all client types.
For instance, thrift clients will need to use `RpcPreprocessor` whereas
web clients will need to use normal `Preprocessor`s.

I propose that `XdsBootstrap` contains a single `ClusterManager` which
caches all `EndpointGroup`s derived from `Cluster`s. This
`ClusterManager` is aligned with the xDS API
[cluster_manager](https://www.envoyproxy.io/docs/envoy/latest/api-v3/config/bootstrap/v3/bootstrap.proto#envoy-v3-api-msg-config-bootstrap-v3-clustermanager)
and is aligned with the [upstream
implementation](https://github.com/envoyproxy/envoy/blob/e3a97f1a81a65b168c5cf822d4a0861958f94162/source/common/upstream/cluster_manager_impl.h#L243)
as well.

One prerequisite for the introduction of `ClusterManager` is the ability
to cache `Cluster`s. This is difficult since the current implementation
of `ClusterSnapshot` also contains a reference to the containing
`Route`.


https://github.com/line/armeria/blob/533f798efdf827326c3a8d70ce21d13f20245c20/xds/src/main/java/com/linecorp/armeria/xds/ClusterSnapshot.java#L82

I propose that `VirtualHostSnapshot` and `RouteEntry` is introduced, and
snapshot mappings are unidirectional. This direction aligns with a
possible future implementation of
[VHDS](https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_conn_man/vhds)
as well.

As this is a refactor of the previous implementation, there is no
expected change in behavior except for subset load balancing (which
actually matches upstream now)

Modifications:

- Introduced `VirtualHostSnapshot` which is equivalent to `VirtualHost`,
and `RouteEntry` which is equivalent to `Route`.
- Refactored `ClusterManager` and `SubsetLoadBalancer` to not use
`ClusterSnapshot#route`
- `SubsetLoadBalancer` now is aligned with the upstream implementation
([ref](https://github.com/envoyproxy/envoy/blob/e3a97f1a81a65b168c5cf822d4a0861958f94162/source/extensions/load_balancing_policies/subset/subset_lb.cc#L478))
- The previous test case for skipping failure on empty metadata is
removed as it does not match the upstream implementation.
- (Misc) Cleaned up the `ConfigSourceMapper` implementation by removing
unused fields and the corresponding logic.

Result:

- The `ClusterSnapshot` API is prepared for the introduction of a
central `ClusterManager` maintained by `XdsBootstrapImpl`

<!--
Visit this URL to learn more about how to write a pull request
description:

https://armeria.dev/community/developer-guide#how-to-write-pull-request-description
-->

Author: jrhee17

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

@@ -18,7 +18,6 @@
 
 import static com.google.common.base.Strings.isNullOrEmpty;
 import static com.linecorp.armeria.xds.XdsType.CLUSTER;
-import static java.util.Objects.requireNonNull;
 
 import java.util.Objects;
 
@@ -28,16 +27,10 @@
 import io.envoyproxy.envoy.config.cluster.v3.Cluster.EdsClusterConfig;
 import io.envoyproxy.envoy.config.core.v3.ConfigSource;
 import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
-import io.envoyproxy.envoy.config.route.v3.Route;
-import io.envoyproxy.envoy.config.route.v3.VirtualHost;
 import io.grpc.Status;
 
 final class ClusterResourceNode extends AbstractResourceNodeWithPrimer<ClusterXdsResource> {
 
-    @Nullable
-    private final VirtualHost virtualHost;
-    @Nullable
-    private final Route route;
     private final int index;
     private final EndpointSnapshotWatcher snapshotWatcher = new EndpointSnapshotWatcher();
     private final SnapshotWatcher<ClusterSnapshot> parentWatcher;
@@ -49,19 +42,15 @@ final class ClusterResourceNode extends AbstractResourceNodeWithPrimer<ClusterXd
                         ResourceNodeType resourceNodeType) {
         super(xdsBootstrap, configSource, CLUSTER, resourceName, primer, parentWatcher, resourceNodeType);
         this.parentWatcher = parentWatcher;
-        virtualHost = null;
-        route = null;
         index = -1;
     }
 
     ClusterResourceNode(@Nullable ConfigSource configSource,
                         String resourceName, XdsBootstrapImpl xdsBootstrap,
-                        @Nullable RouteXdsResource primer, SnapshotWatcher<ClusterSnapshot> parentWatcher,
-                        VirtualHost virtualHost, Route route, int index, ResourceNodeType resourceNodeType) {
+                        @Nullable VirtualHostXdsResource primer, SnapshotWatcher<ClusterSnapshot> parentWatcher,
+                        int index, ResourceNodeType resourceNodeType) {
         super(xdsBootstrap, configSource, CLUSTER, resourceName, primer, parentWatcher, resourceNodeType);
         this.parentWatcher = parentWatcher;
-        this.virtualHost = requireNonNull(virtualHost, "virtualHost");
-        this.route = requireNonNull(route, "route");
         this.index = index;
     }
 
@@ -86,7 +75,8 @@ public void doOnChanged(ClusterXdsResource resource) {
             children().add(node);
             xdsBootstrap().subscribe(node);
         } else {
-            parentWatcher.snapshotUpdated(new ClusterSnapshot(resource));
+            final ClusterSnapshot clusterSnapshot = new ClusterSnapshot(resource);
+            parentWatcher.snapshotUpdated(clusterSnapshot);
         }
     }
 
@@ -100,8 +90,8 @@ public void snapshotUpdated(EndpointSnapshot newSnapshot) {
             if (!Objects.equals(newSnapshot.xdsResource().primer(), current)) {
                 return;
             }
-            parentWatcher.snapshotUpdated(
-                    new ClusterSnapshot(current, newSnapshot, virtualHost, route, index));
+            final ClusterSnapshot clusterSnapshot = new ClusterSnapshot(current, newSnapshot, index);
+            parentWatcher.snapshotUpdated(clusterSnapshot);
         }
 
         @Override

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

@@ -39,7 +39,7 @@ public final class ClusterRoot extends AbstractRoot<ClusterSnapshot> {
         if (cluster != null) {
             node = staticCluster(xdsBootstrap, resourceName, this, cluster);
         } else {
-            final ConfigSource configSource = configSourceMapper.cdsConfigSource(null, resourceName);
+            final ConfigSource configSource = configSourceMapper.cdsConfigSource(resourceName);
             node = new ClusterResourceNode(configSource, resourceName, xdsBootstrap,
                                            null, this, ResourceNodeType.DYNAMIC);
             xdsBootstrap.subscribe(node);

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

@@ -23,8 +23,6 @@
 import com.linecorp.armeria.common.annotation.UnstableApi;
 
 import io.envoyproxy.envoy.config.cluster.v3.Cluster;
-import io.envoyproxy.envoy.config.route.v3.Route;
-import io.envoyproxy.envoy.config.route.v3.VirtualHost;
 
 /**
  * A snapshot of a {@link Cluster} resource.
@@ -34,27 +32,17 @@ public final class ClusterSnapshot implements Snapshot<ClusterXdsResource> {
     private final ClusterXdsResource clusterXdsResource;
     @Nullable
     private final EndpointSnapshot endpointSnapshot;
-    @Nullable
-    private final VirtualHost virtualHost;
-    @Nullable
-    private final Route route;
     private final int index;
 
-    ClusterSnapshot(ClusterXdsResource clusterXdsResource, EndpointSnapshot endpointSnapshot,
-                    @Nullable VirtualHost virtualHost, @Nullable Route route, int index) {
+    ClusterSnapshot(ClusterXdsResource clusterXdsResource,
+                    @Nullable EndpointSnapshot endpointSnapshot, int index) {
         this.clusterXdsResource = clusterXdsResource;
         this.endpointSnapshot = endpointSnapshot;
-        this.virtualHost = virtualHost;
-        this.route = route;
         this.index = index;
     }
 
     ClusterSnapshot(ClusterXdsResource clusterXdsResource) {
-        this.clusterXdsResource = clusterXdsResource;
-        endpointSnapshot = null;
-        virtualHost = null;
-        route = null;
-        index = -1;
+        this(clusterXdsResource, null, -1);
     }
 
     @Override
@@ -70,22 +58,6 @@ public EndpointSnapshot endpointSnapshot() {
         return endpointSnapshot;
     }
 
-    /**
-     * The {@link VirtualHost} this {@link Cluster} belongs to.
-     */
-    @Nullable
-    public VirtualHost virtualHost() {
-        return virtualHost;
-    }
-
-    /**
-     * The {@link Route} this {@link Cluster} belongs to.
-     */
-    @Nullable
-    public Route route() {
-        return route;
-    }
-
     int index() {
         return index;
     }
@@ -99,15 +71,13 @@ public boolean equals(Object object) {
             return false;
         }
         final ClusterSnapshot that = (ClusterSnapshot) object;
-        return index == that.index && Objects.equal(clusterXdsResource, that.clusterXdsResource) &&
-               Objects.equal(endpointSnapshot, that.endpointSnapshot) &&
-               Objects.equal(virtualHost, that.virtualHost) &&
-               Objects.equal(route, that.route);
+        return Objects.equal(clusterXdsResource, that.clusterXdsResource) &&
+               Objects.equal(endpointSnapshot, that.endpointSnapshot);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hashCode(clusterXdsResource, endpointSnapshot, virtualHost, route, index);
+        return Objects.hashCode(clusterXdsResource, endpointSnapshot);
     }
 
     @Override
@@ -116,9 +86,6 @@ public String toString() {
                           .omitNullValues()
                           .add("clusterXdsResource", clusterXdsResource)
                           .add("endpointSnapshot", endpointSnapshot)
-                          .add("virtualHost", virtualHost)
-                          .add("route", route)
-                          .add("index", index)
                           .toString();
     }
 }

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

@@ -34,7 +34,7 @@ final class CompositeXdsStream implements XdsStream {
     CompositeXdsStream(GrpcClientBuilder clientBuilder, Node node, Backoff backoff,
                        EventExecutor eventLoop, XdsResponseHandler handler,
                        SubscriberStorage subscriberStorage) {
-        for (XdsType type: XdsType.values()) {
+        for (XdsType type: XdsType.discoverableTypes()) {
             final SotwXdsStream stream = new SotwXdsStream(
                     SotwDiscoveryStub.basic(type, clientBuilder), node, backoff, eventLoop,
                     handler, subscriberStorage, EnumSet.of(type));

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

@@ -32,7 +32,7 @@ final class ConfigSourceMapper {
     @Nullable
     private final ConfigSource bootstrapAdsConfig;
     @Nullable
-    private ConfigSource parentConfigSource;
+    private final ConfigSource parentConfigSource;
 
     ConfigSourceMapper(Bootstrap bootstrap) {
         this(bootstrap, null);
@@ -78,18 +78,7 @@ ConfigSource edsConfigSource(@Nullable ConfigSource configSource, String resourc
         throw new IllegalArgumentException("Cannot find an EDS config source for " + resourceName);
     }
 
-    ConfigSource cdsConfigSource(@Nullable ConfigSource configSource, String resourceName) {
-        if (configSource != null) {
-            if (configSource.hasApiConfigSource()) {
-                return configSource;
-            }
-            if (configSource.hasSelf() && parentConfigSource != null) {
-                return parentConfigSource;
-            }
-            if (configSource.hasAds() && bootstrapAdsConfig != null) {
-                return bootstrapAdsConfig;
-            }
-        }
+    ConfigSource cdsConfigSource(String resourceName) {
         if (bootstrapCdsConfig != null && bootstrapCdsConfig.hasApiConfigSource()) {
             return bootstrapCdsConfig;
         }
@@ -114,18 +103,7 @@ ConfigSource rdsConfigSource(@Nullable ConfigSource configSource, String resourc
         throw new IllegalArgumentException("Cannot find an RDS config source for route: " + resourceName);
     }
 
-    ConfigSource ldsConfigSource(@Nullable ConfigSource configSource, String resourceName) {
-        if (configSource != null) {
-            if (configSource.hasApiConfigSource()) {
-                return configSource;
-            }
-            if (configSource.hasSelf() && parentConfigSource != null) {
-                return parentConfigSource;
-            }
-            if (configSource.hasAds() && bootstrapAdsConfig != null) {
-                return bootstrapAdsConfig;
-            }
-        }
+    ConfigSource ldsConfigSource(String resourceName) {
         if (bootstrapLdsConfig != null && bootstrapLdsConfig.hasApiConfigSource()) {
             return bootstrapLdsConfig;
         }

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

@@ -39,7 +39,7 @@ public final class ListenerRoot extends AbstractRoot<ListenerSnapshot> {
             node = new ListenerResourceNode(null, resourceName, xdsBootstrap, this, ResourceNodeType.STATIC);
             node.onChanged(listenerXdsResource);
         } else {
-            final ConfigSource configSource = configSourceMapper.ldsConfigSource(null, resourceName);
+            final ConfigSource configSource = configSourceMapper.ldsConfigSource(resourceName);
             node = new ListenerResourceNode(configSource, resourceName, xdsBootstrap,
                                             this, ResourceNodeType.DYNAMIC);
             xdsBootstrap.subscribe(node);

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

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2025 LY Corporation
+ *
+ * LINE 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:
+ *
+ *   https://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 com.linecorp.armeria.xds;
+
+import java.util.Objects;
+
+import com.google.common.base.MoreObjects;
+
+import com.linecorp.armeria.common.annotation.Nullable;
+
+import io.envoyproxy.envoy.config.route.v3.Route;
+import io.envoyproxy.envoy.config.route.v3.RouteAction;
+
+/**
+ * Represents a {@link Route}.
+ */
+public final class RouteEntry {
+
+    private final Route route;
+    @Nullable
+    private final ClusterSnapshot clusterSnapshot;
+
+    RouteEntry(Route route, @Nullable ClusterSnapshot clusterSnapshot) {
+        this.route = route;
+        this.clusterSnapshot = clusterSnapshot;
+    }
+
+    /**
+     * The {@link Route}.
+     */
+    public Route route() {
+        return route;
+    }
+
+    /**
+     * The {@link ClusterSnapshot} that is represented by {@link RouteAction#getCluster()}.
+     * If the {@link RouteAction} does not reference a cluster, the returned value may be {@code null}.
+     */
+    @Nullable
+    public ClusterSnapshot clusterSnapshot() {
+        return clusterSnapshot;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final RouteEntry that = (RouteEntry) o;
+        return Objects.equals(route, that.route) &&
+               Objects.equals(clusterSnapshot, that.clusterSnapshot);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(route, clusterSnapshot);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                          .add("route", route)
+                          .add("clusterSnapshot", clusterSnapshot)
+                          .toString();
+    }
+}