Add support for credential_injector

it/xds-istio/src/test/java/com/linecorp/armeria/it/xds/filter/IstioFilterFactories.java

@@ -21,7 +21,7 @@
 
 import com.linecorp.armeria.client.HttpPreprocessor;
 import com.linecorp.armeria.common.annotation.Nullable;
-import com.linecorp.armeria.xds.XdsResourceValidator;
+import com.linecorp.armeria.xds.filter.FactoryContext;
 import com.linecorp.armeria.xds.filter.HttpFilterFactory;
 import com.linecorp.armeria.xds.filter.XdsHttpFilter;
 import com.linecorp.armeria.xds.internal.XdsCommonUtil;
@@ -40,7 +40,7 @@ public final class IstioFilterFactories {
     public abstract static class Base implements HttpFilterFactory {
         @Override
         @Nullable
-        public XdsHttpFilter create(HttpFilter httpFilter, Any config, XdsResourceValidator validator) {
+        public XdsHttpFilter create(HttpFilter httpFilter, Any config, FactoryContext context) {
             return null;
         }
     }
@@ -62,7 +62,7 @@ public String name() {
         }
 
         @Override
-        public XdsHttpFilter create(HttpFilter httpFilter, Any config, XdsResourceValidator validator) {
+        public XdsHttpFilter create(HttpFilter httpFilter, Any config, FactoryContext context) {
             // we need to also generate istio proto configs to actually dynamically decide alpn
             // based on filter config, but for testing purposes we just assume h2 is always used
             return ALPN_FILTER;

xds-api/src/main/proto/envoy/config/core/v3/extension.proto

@@ -4,6 +4,7 @@ package envoy.config.core.v3;
 
 import "google/protobuf/any.proto";
 
+import "armeria/xds/supported.proto";
 import "udpa/annotations/status.proto";
 import "validate/validate.proto";
 
@@ -20,6 +21,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 message TypedExtensionConfig {
   // The name of an extension. This is not used to select the extension, instead
   // it serves the role of an opaque identifier.
+  option (armeria.xds.supported.field) = 1;
   string name = 1 [(validate.rules).string = {min_len: 1}];
 
   // The typed config for the extension. The type URL will be used to identify
@@ -28,5 +30,6 @@ message TypedExtensionConfig {
   // URL of ``TypedStruct`` will be utilized. See the
   // :ref:`extension configuration overview
   // <config_overview_extension_configuration>` for further details.
+  option (armeria.xds.supported.field) = 2;
   google.protobuf.Any typed_config = 2 [(validate.rules).any = {required: true}];
 }

xds-api/src/main/proto/envoy/extensions/filters/http/credential_injector/v3/credential_injector.proto

@@ -4,6 +4,7 @@ package envoy.extensions.filters.http.credential_injector.v3;
 
 import "envoy/config/core/v3/extension.proto";
 
+import "armeria/xds/supported.proto";
 import "udpa/annotations/status.proto";
 import "validate/validate.proto";
 
@@ -71,6 +72,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 message CredentialInjector {
   // Whether to overwrite the value or not if the injected headers already exist.
   // Value defaults to false.
+  option (armeria.xds.supported.field) = 1;
   bool overwrite = 1;
 
   // Whether to send the request to upstream if the credential is not present or if the credential injection
@@ -79,9 +81,11 @@ message CredentialInjector {
   // By default, a request will fail with ``401 Unauthorized`` if the
   // credential is not present or the injection of the credential to the request fails.
   // If set to true, the request will be sent to upstream without the credential.
+  option (armeria.xds.supported.field) = 2;
   bool allow_request_without_credential = 2;
 
   // The credential to inject into the proxied requests
   // [#extension-category: envoy.http.injected_credentials]
+  option (armeria.xds.supported.field) = 3;
   config.core.v3.TypedExtensionConfig credential = 3 [(validate.rules).message = {required: true}];
 }

xds-api/src/main/proto/envoy/extensions/http/injected_credentials/generic/v3/generic.proto

@@ -4,6 +4,7 @@ package envoy.extensions.http.injected_credentials.generic.v3;
 
 import "envoy/extensions/transport_sockets/tls/v3/secret.proto";
 
+import "armeria/xds/supported.proto";
 import "udpa/annotations/status.proto";
 import "validate/validate.proto";
 
@@ -24,11 +25,13 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 message Generic {
   // The SDS configuration for the credential that will be injected to the specified HTTP request header.
   // It must be a generic secret.
+  option (armeria.xds.supported.field) = 1;
   transport_sockets.tls.v3.SdsSecretConfig credential = 1
       [(validate.rules).message = {required: true}];
 
   // The header that will be injected to the HTTP request with the provided credential.
   // If not set, filter will default to: ``Authorization``
+  option (armeria.xds.supported.field) = 2;
   string header = 2
       [(validate.rules).string = {well_known_regex: HTTP_HEADER_NAME ignore_empty: true}];
 

xds-api/src/main/proto/envoy/extensions/transport_sockets/tls/v3/secret.proto

@@ -26,6 +26,7 @@ message GenericSecret {
 
   // Secret of generic type and is available to filters. It is expected
   // that only only one of secret and secrets is set.
+  option (armeria.xds.supported.field) = 1;
   config.core.v3.DataSource secret = 1 [(udpa.annotations.sensitive) = true];
 
   // For cases where multiple associated secrets need to be distributed together. It is expected
@@ -63,6 +64,7 @@ message Secret {
     option (armeria.xds.supported.oneof_field) = 4;
     CertificateValidationContext validation_context = 4;
 
+    option (armeria.xds.supported.oneof_field) = 5;
     GenericSecret generic_secret = 5;
   }
 }

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

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Map;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.protobuf.Any;
 
@@ -29,8 +30,10 @@
 import com.linecorp.armeria.client.ClientPreprocessors;
 import com.linecorp.armeria.client.ClientPreprocessorsBuilder;
 import com.linecorp.armeria.common.annotation.Nullable;
+import com.linecorp.armeria.xds.filter.FactoryContext;
 import com.linecorp.armeria.xds.filter.HttpFilterFactory;
 import com.linecorp.armeria.xds.filter.XdsHttpFilter;
+import com.linecorp.armeria.xds.stream.SnapshotStream;
 
 import io.envoyproxy.envoy.config.route.v3.RetryPolicy;
 import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpFilter;
@@ -47,51 +50,77 @@ static Map<String, Any> mergeFilterConfigs(
                            .buildKeepingLast();
     }
 
-    static ClientPreprocessors buildDownstreamFilter(
-            XdsExtensionRegistry extensionRegistry,
+    static SnapshotStream<ClientPreprocessors> buildDownstreamFilter(
+            XdsExtensionRegistry extensionRegistry, FactoryContext factoryContext,
             List<HttpFilter> httpFilters, Map<String, Any> filterConfigs) {
         if (httpFilters.isEmpty()) {
-            return ClientPreprocessors.of();
+            return SnapshotStream.just(ClientPreprocessors.of());
         }
-        final ClientPreprocessorsBuilder builder = ClientPreprocessors.builder();
+        final ImmutableList.Builder<SnapshotStream<XdsHttpFilter>> streams = ImmutableList.builder();
         for (int i = httpFilters.size() - 1; i >= 0; i--) {
             final HttpFilter httpFilter = httpFilters.get(i);
             final Any perRouteConfig = filterConfigs.get(httpFilter.getName());
-            final XdsHttpFilter instance = resolveInstance(extensionRegistry, httpFilter, perRouteConfig);
-            if (instance == null) {
-                continue;
+            final SnapshotStream<XdsHttpFilter> stream =
+                    resolveInstance(extensionRegistry, factoryContext, httpFilter, perRouteConfig);
+            if (stream != null) {
+                streams.add(stream);
             }
-            builder.add(instance.httpPreprocessor());
-            builder.addRpc(instance.rpcPreprocessor());
         }
-        return builder.build();
+        final ImmutableList<SnapshotStream<XdsHttpFilter>> streamList = streams.build();
+        if (streamList.isEmpty()) {
+            return SnapshotStream.just(ClientPreprocessors.of());
+        }
+        return SnapshotStream.combineNLatest(streamList).map(filters -> {
+            final ClientPreprocessorsBuilder builder = ClientPreprocessors.builder();
+            for (XdsHttpFilter f : filters) {
+                builder.add(f.httpPreprocessor());
+                builder.addRpc(f.rpcPreprocessor());
+            }
+            return builder.build();
+        });
     }
 
-    static ClientDecoration buildUpstreamFilter(
-            XdsExtensionRegistry extensionRegistry,
+    static SnapshotStream<ClientDecoration> buildUpstreamFilter(
+            XdsExtensionRegistry extensionRegistry, FactoryContext factoryContext,
             List<HttpFilter> httpFilters, Map<String, Any> filterConfigs,
             @Nullable RetryPolicy retryPolicy) {
-        final ClientDecorationBuilder builder = ClientDecoration.builder();
+        final ImmutableList.Builder<SnapshotStream<XdsHttpFilter>> streams = ImmutableList.builder();
         for (int i = httpFilters.size() - 1; i >= 0; i--) {
             final HttpFilter httpFilter = httpFilters.get(i);
             final Any perRouteConfig = filterConfigs.get(httpFilter.getName());
-            final XdsHttpFilter instance = resolveInstance(extensionRegistry, httpFilter, perRouteConfig);
-            if (instance == null) {
-                continue;
+            final SnapshotStream<XdsHttpFilter> stream =
+                    resolveInstance(extensionRegistry, factoryContext, httpFilter, perRouteConfig);
+            if (stream != null) {
+                streams.add(stream);
+            }
+        }
+        final ImmutableList<SnapshotStream<XdsHttpFilter>> streamList = streams.build();
+        if (streamList.isEmpty()) {
+            return SnapshotStream.just(buildDecoration(null, retryPolicy));
+        }
+        return SnapshotStream.combineNLatest(streamList).map(filters -> {
+            return buildDecoration(filters, retryPolicy);
+        });
+    }
+
+    private static ClientDecoration buildDecoration(@Nullable List<XdsHttpFilter> filters,
+                                                    @Nullable RetryPolicy retryPolicy) {
+        final ClientDecorationBuilder builder = ClientDecoration.builder();
+        if (filters != null) {
+            for (XdsHttpFilter f : filters) {
+                builder.add(f.httpDecorator());
+                builder.addRpc(f.rpcDecorator());
             }
-            builder.add(instance.httpDecorator());
-            builder.addRpc(instance.rpcDecorator());
         }
         if (retryPolicy != null) {
-            // add the retrying decorator as the first (outermost) decorator if exists
             builder.add(new RetryStateFactory(retryPolicy).retryingDecorator());
         }
         return builder.build();
     }
 
     @Nullable
-    static XdsHttpFilter resolveInstance(
-            XdsExtensionRegistry extensionRegistry,
+    static SnapshotStream<XdsHttpFilter> resolveInstance(
+            XdsExtensionRegistry extensionRegistry, FactoryContext factoryContext,
             HttpFilter httpFilter, @Nullable Any perRouteConfig) {
         final Any defaultConfig = httpFilter.getTypedConfig();
         final Any filterConfig = perRouteConfig != null ? perRouteConfig : defaultConfig;
@@ -110,7 +139,8 @@ static XdsHttpFilter resolveInstance(
                       httpFilter.getConfigTypeCase() == ConfigTypeCase.CONFIGTYPE_NOT_SET,
                       "Only 'typed_config' is supported, but '%s' was supplied",
                       httpFilter.getConfigTypeCase());
-        return factory.create(httpFilter, filterConfig, extensionRegistry.validator());
+        return factory.createStream(httpFilter, filterConfig, factoryContext)
+                      .rescheduleEventsOn(factoryContext.eventLoop());
     }
 
     private FilterUtil() {}

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

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2025 LY Corporation
+ *
+ * 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:
+ *
+ *   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 com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
+
+import com.linecorp.armeria.common.annotation.Nullable;
+import com.linecorp.armeria.common.annotation.UnstableApi;
+
+import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.GenericSecret;
+
+/**
+ * A snapshot of a {@link GenericSecret} resource with its resolved credential value.
+ * This snapshot is created when resolving generic secret configuration from xDS resources,
+ * analogous to {@link TlsCertificateSnapshot} for TLS certificates.
+ */
+@UnstableApi
+public final class GenericSecretSnapshot implements Snapshot<GenericSecret> {
+
+    private final GenericSecret resource;
+    @Nullable
+    private final String credential;
+
+    GenericSecretSnapshot(GenericSecret resource, @Nullable String credential) {
+        this.resource = resource;
+        this.credential = credential;
+    }
+
+    /**
+     * Returns the resolved credential value, or {@code null} if the credential
+     * is not available or empty.
+     */
+    @Nullable
+    public String credential() {
+        return credential;
+    }
+
+    @Override
+    public GenericSecret xdsResource() {
+        return resource;
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        if (this == object) {
+            return true;
+        }
+        if (object == null || getClass() != object.getClass()) {
+            return false;
+        }
+        final GenericSecretSnapshot that = (GenericSecretSnapshot) object;
+        return Objects.equal(resource, that.resource);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(resource);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                          .toString();
+    }
+
+    @Override
+    public String toDebugString() {
+        return MoreObjects.toStringHelper(this)
+                          .add("genericSecret", resource)
+                          .toString();
+    }
+}