Allow XdsFilter to handle unknown Message config (#6710)

Subset of #6700

Motivation

`HttpFilterFactory<T extends Message>` forced each factory
implementation to declare a specific protobuf `Message` type for its
config, and required the framework to own config parsing and
`FilterConfig` envelope unwrapping. This made it difficult to handle
unknown or dynamically-typed xDS filter configs (e.g. Istio-specific
filters), and leaked framework concerns into the factory API.

Additionally, `ParsedFilterConfig` wrapped per-route filter configs and
required the framework to merge and propagate them through the snapshot
hierarchy (`RouteSnapshot` → `VirtualHostSnapshot` → `RouteEntry`),
coupling snapshot construction to filter logic.

Modifications

- Replace `HttpFilterFactory<T extends Message>` with an untyped
`HttpFilterFactory` interface; `create(HttpFilter, Any)` now receives
the raw `Any` config directly, and factories are responsible for all
parsing including `FilterConfig` envelope unwrapping. Returning `null`
skips the filter.
- Introduce `XdsHttpFilter` as the return type of `create()`, with
default no-op `httpPreprocessor()`, `rpcPreprocessor()`,
`httpDecorator()`, and `rpcDecorator()` methods.
- Delete `ParsedFilterConfig`; per-route filter configs are now stored
and merged as `Map<String, Any>` directly.
- Move 3-level `typed_per_filter_config` merging (route-config → vhost →
route) into `RouteEntry`'s constructor, eliminating
`withFilterConfigs()` from `VirtualHostSnapshot` and `withRouter()` from
`RouteSnapshot`.
- Thread `@Nullable ListenerXdsResource` through `RouteStream` inner
classes so `RouteEntry` can access upstream HTTP filters from the Router
directly.
- Update `RouterFilterFactory` to implement the new untyped interface.
- Simplify `FilterUtil`: remove `toParsedFilterConfigs()`, update
`mergeFilterConfigs` to operate on `Map<String, Any>`,
`buildUpstreamFilter` takes `@Nullable RetryPolicy` directly.

Result

- `HttpFilterFactory` implementations no longer need to declare a
protobuf type parameter, and can handle any xDS filter config including
unknown or Istio-specific ones.
- Per-route filter config merging is self-contained in `RouteEntry`
rather than spread across the snapshot hierarchy.
- `RouteEntry.filterConfig(String)` now returns `@Nullable Any` instead
of `@Nullable ParsedFilterConfig`.

Breaking

- `HttpFilterFactory` implementations must now implement `XdsHttpFilter
create(HttpFilter httpFilter, Any config)` for custom filter
implementations

Author: jrhee17

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

@@ -20,48 +20,33 @@
 
 import java.util.List;
 import java.util.Map;
-import java.util.Map.Entry;
-import java.util.function.Function;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.protobuf.Any;
-import com.google.protobuf.Message;
 
 import com.linecorp.armeria.client.ClientDecoration;
 import com.linecorp.armeria.client.ClientDecorationBuilder;
 import com.linecorp.armeria.client.ClientPreprocessors;
 import com.linecorp.armeria.client.ClientPreprocessorsBuilder;
-import com.linecorp.armeria.client.DecoratingHttpClientFunction;
-import com.linecorp.armeria.client.DecoratingRpcClientFunction;
-import com.linecorp.armeria.client.HttpClient;
-import com.linecorp.armeria.client.HttpPreprocessor;
-import com.linecorp.armeria.client.RpcPreprocessor;
-import com.linecorp.armeria.client.retry.RetryingClient;
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.xds.filter.HttpFilterFactory;
 import com.linecorp.armeria.xds.filter.HttpFilterFactoryRegistry;
+import com.linecorp.armeria.xds.filter.XdsHttpFilter;
 
+import io.envoyproxy.envoy.config.route.v3.RetryPolicy;
 import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager;
 import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpFilter;
 import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpFilter.ConfigTypeCase;
 
 final class FilterUtil {
 
-    static Map<String, ParsedFilterConfig> mergeFilterConfigs(
-            Map<String, ParsedFilterConfig> filterConfigs1,
-            Map<String, ParsedFilterConfig> filterConfigs2) {
-        final ImmutableMap.Builder<String, ParsedFilterConfig> builder = ImmutableMap.builder();
-        builder.putAll(filterConfigs1);
-        builder.putAll(filterConfigs2);
-        return builder.buildKeepingLast();
-    }
-
-    static Map<String, ParsedFilterConfig> toParsedFilterConfigs(Map<String, Any> filterConfigMap) {
-        final ImmutableMap.Builder<String, ParsedFilterConfig> filterConfigsBuilder = ImmutableMap.builder();
-        for (Entry<String, Any> e: filterConfigMap.entrySet()) {
-            filterConfigsBuilder.put(e.getKey(), ParsedFilterConfig.of(e.getKey(), e.getValue()));
-        }
-        return filterConfigsBuilder.buildKeepingLast();
+    static Map<String, Any> mergeFilterConfigs(
+            Map<String, Any> filterConfigs1,
+            Map<String, Any> filterConfigs2) {
+        return ImmutableMap.<String, Any>builder()
+                           .putAll(filterConfigs1)
+                           .putAll(filterConfigs2)
+                           .buildKeepingLast();
     }
 
     static ClientPreprocessors buildDownstreamFilter(
@@ -73,98 +58,58 @@ static ClientPreprocessors buildDownstreamFilter(
         final ClientPreprocessorsBuilder builder = ClientPreprocessors.builder();
         for (int i = httpFilters.size() - 1; i >= 0; i--) {
             final HttpFilter httpFilter = httpFilters.get(i);
-            final XdsFilter<?> xdsFilter = xdsHttpFilter(httpFilter, null);
-            if (xdsFilter == null) {
+            final XdsHttpFilter instance = resolveInstance(httpFilter, null);
+            if (instance == null) {
                 continue;
             }
-            if (xdsFilter.filterConfig().disabled()) {
-                continue;
-            }
-            builder.add(xdsFilter.httpPreprocessor());
-            builder.addRpc(xdsFilter.rpcPreprocessor());
+            builder.add(instance.httpPreprocessor());
+            builder.addRpc(instance.rpcPreprocessor());
         }
         return builder.build();
     }
 
     static ClientDecoration buildUpstreamFilter(
-            List<HttpFilter> httpFilters, Map<String, ParsedFilterConfig> filterConfigs,
-            @Nullable Function<? super HttpClient, RetryingClient> retryingDecorator) {
+            List<HttpFilter> httpFilters, Map<String, Any> filterConfigs,
+            @Nullable RetryPolicy retryPolicy) {
         final ClientDecorationBuilder builder = ClientDecoration.builder();
         for (int i = httpFilters.size() - 1; i >= 0; i--) {
             final HttpFilter httpFilter = httpFilters.get(i);
-            final ParsedFilterConfig parsedFilterConfig = filterConfigs.get(httpFilter.getName());
-            final XdsFilter<?> xdsFilter = xdsHttpFilter(httpFilter, parsedFilterConfig);
-            if (xdsFilter == null) {
-                continue;
-            }
-            if (xdsFilter.filterConfig().disabled()) {
+            final Any perRouteConfig = filterConfigs.get(httpFilter.getName());
+            final XdsHttpFilter instance = resolveInstance(httpFilter, perRouteConfig);
+            if (instance == null) {
                 continue;
             }
-            builder.add(xdsFilter.httpDecorator());
-            builder.addRpc(xdsFilter.rpcDecorator());
+            builder.add(instance.httpDecorator());
+            builder.addRpc(instance.rpcDecorator());
         }
-        if (retryingDecorator != null) {
+        if (retryPolicy != null) {
             // add the retrying decorator as the first (outermost) decorator if exists
-            builder.add(retryingDecorator);
+            builder.add(new RetryStateFactory(retryPolicy).retryingDecorator());
         }
         return builder.build();
     }
 
     @Nullable
-    private static XdsFilter<?> xdsHttpFilter(HttpFilter httpFilter,
-                                              @Nullable ParsedFilterConfig parsedFilterConfig) {
-        final HttpFilterFactory<?> filterFactory =
+    private static XdsHttpFilter resolveInstance(
+            HttpFilter httpFilter, @Nullable Any perRouteConfig) {
+        final HttpFilterFactory filterFactory =
                 HttpFilterFactoryRegistry.filterFactory(httpFilter.getName());
         if (filterFactory == null) {
-            if (httpFilter.getIsOptional()) {
-                return null;
+            if (!httpFilter.getIsOptional()) {
+                throw new IllegalArgumentException(
+                        "Unknown HTTP filter '" + httpFilter.getName() +
+                        "': no HttpFilterFactory registered. Register an SPI " +
+                        "HttpFilterFactory implementation to handle this filter.");
             }
-            throw new IllegalArgumentException("Couldn't find filter factory: " + httpFilter.getName());
+            return null;
         }
         checkArgument(httpFilter.getConfigTypeCase() == ConfigTypeCase.TYPED_CONFIG ||
                       httpFilter.getConfigTypeCase() == ConfigTypeCase.CONFIGTYPE_NOT_SET,
                       "Only 'typed_config' is supported, but '%s' was supplied",
                       httpFilter.getConfigTypeCase());
-        return new XdsFilter<>(filterFactory, httpFilter, parsedFilterConfig);
-    }
-
-    private static class XdsFilter<T extends Message> {
-
-        private final HttpFilterFactory<T> filterFactory;
-        private final T config;
-        private final ParsedFilterConfig filterConfig;
-
-        XdsFilter(HttpFilterFactory<T> filterFactory, HttpFilter httpFilter,
-                  @Nullable ParsedFilterConfig filterConfig) {
-            this.filterFactory = filterFactory;
-            if (filterConfig != null) {
-                this.filterConfig = filterConfig;
-            } else {
-                this.filterConfig = ParsedFilterConfig.of(httpFilter.getName(), httpFilter.getTypedConfig(),
-                                                          httpFilter.getIsOptional(), httpFilter.getDisabled());
-            }
-            config = this.filterConfig.parsedConfig(filterFactory.defaultConfig());
-        }
-
-        public ParsedFilterConfig filterConfig() {
-            return filterConfig;
-        }
-
-        public HttpPreprocessor httpPreprocessor() {
-            return filterFactory.httpPreprocessor(config);
-        }
-
-        public RpcPreprocessor rpcPreprocessor() {
-            return filterFactory.rpcPreprocessor(config);
-        }
-
-        public DecoratingHttpClientFunction httpDecorator() {
-            return filterFactory.httpDecorator(config);
-        }
-
-        public DecoratingRpcClientFunction rpcDecorator() {
-            return filterFactory.rpcDecorator(config);
-        }
+        final Any effectiveConfig =
+                perRouteConfig != null ? perRouteConfig : httpFilter.getTypedConfig();
+        return filterFactory.create(httpFilter, effectiveConfig);
     }
 
     private FilterUtil() {}

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

@@ -42,11 +42,7 @@ public final class ListenerSnapshot implements Snapshot<ListenerXdsResource> {
 
     ListenerSnapshot(ListenerXdsResource listenerXdsResource, @Nullable RouteSnapshot routeSnapshot) {
         this.listenerXdsResource = listenerXdsResource;
-        if (listenerXdsResource.router() != null && routeSnapshot != null) {
-            this.routeSnapshot = routeSnapshot.withRouter(listenerXdsResource.router());
-        } else {
-            this.routeSnapshot = routeSnapshot;
-        }
+        this.routeSnapshot = routeSnapshot;
         downstreamFilter = FilterUtil.buildDownstreamFilter(listenerXdsResource.connectionManager());
     }
 

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

@@ -62,14 +62,14 @@ protected Subscription onStart(SnapshotWatcher<ListenerSnapshot> watcher) {
                 .subscribe(watcher);
     }
 
-    private SnapshotStream<ListenerSnapshot> resource2snapshot(ListenerXdsResource resource,
-                                                               @Nullable ConfigSource parentConfigSource) {
+    private SnapshotStream<ListenerSnapshot> resource2snapshot(
+            ListenerXdsResource resource, @Nullable ConfigSource parentConfigSource) {
         SnapshotStream<ListenerSnapshot> node = null;
         final HttpConnectionManager connectionManager = resource.connectionManager();
         if (connectionManager != null) {
             if (connectionManager.hasRouteConfig()) {
                 final RouteConfiguration routeConfig = connectionManager.getRouteConfig();
-                node = new RouteStream(context, routeConfig)
+                node = new RouteStream(context, routeConfig, resource)
                         .map(routeSnapshot -> new ListenerSnapshot(resource, routeSnapshot));
             } else if (connectionManager.hasRds()) {
                 final Rds rds = connectionManager.getRds();
@@ -81,7 +81,7 @@ private SnapshotStream<ListenerSnapshot> resource2snapshot(ListenerXdsResource r
                     return SnapshotStream.error(new XdsResourceException(LISTENER, resourceName,
                                                                          "config source not found"));
                 }
-                node = new RouteStream(configSource, routeName, context)
+                node = new RouteStream(configSource, routeName, context, resource)
                         .map(routeSnapshot -> new ListenerSnapshot(resource, routeSnapshot));
             }
         }