Serve xDS resources that embed Armeria custom filter configs (e.g. Athenz)

Motivation:

The xDS control plane parses every stored Listener/Cluster/Route/Endpoint into
a proto with a curioswitch MessageMarshaller, which must resolve every
google.protobuf.Any it contains (such as an HttpFilter typed_config). The
marshaller only registered types under io.envoyproxy.envoy (from
java-control-plane) plus a couple of Central Dogma-specific ones. Armeria's xDS
filters (e.g. the Athenz access-token filters) are carried as typed_config Any
values whose types live in com.linecorp.armeria.xds.athenz and
jp.co.lycorp.ftd.athenz.v1, packaged in armeria-xds-api, which Central Dogma did
not depend on.

Modifications:
- Make armeria-xds-api the single provider of io.envoyproxy.envoy types in the
  :xds module.

Result:
- xDS resources that embed Armeria custom filter configs, such as the Athenz
  access-token filters, can now be created, stored and served by the control
  plane.

Author: minwoox

dependencies.toml

@@ -3,7 +3,7 @@
 #     If its classes are exposed in Javadoc, update offline links as well.
 #
 [versions]
-armeria = "1.40.0"
+armeria = "1.41.0"
 assertj = "3.27.7"
 awaitility = "4.3.0"
 bouncycastle = "1.84"

xds/build.gradle

@@ -5,15 +5,23 @@ dependencies {
     implementation libs.kubernetes.client.impl
 
     implementation libs.armeria.grpc
-    implementation libs.controlplane.cache
-    implementation libs.controlplane.server
+    // The Envoy proto types (io.envoyproxy.envoy.*) are provided by armeria-xds-api rather than
+    // java-control-plane's own 'api' artifact, so Central Dogma parses and serves resources against the
+    // same (Armeria-customized, evolving) schema that the Armeria xDS stack uses. java-control-plane's
+    // cache/server run against these classes; their transitive controlplane:api is excluded to keep a
+    // single io.envoyproxy.envoy provider on the classpath (both jars ship the same FQCNs).
+    implementation libs.armeria.xds.api
+    implementation(libs.controlplane.cache) {
+        exclude group: 'io.envoyproxy.controlplane', module: 'api'
+    }
+    implementation(libs.controlplane.server) {
+        exclude group: 'io.envoyproxy.controlplane', module: 'api'
+    }
 
     implementation libs.reflections
 
     testImplementation libs.armeria.junit5
-    testImplementation (libs.armeria.xds) {
-        exclude group: 'com.linecorp.armeria', module: 'armeria-xds-api'
-    }
+    testImplementation libs.armeria.xds
     testImplementation libs.kubernetes.server.mock
     testImplementation libs.kubernetes.junit.jupiter
     testImplementation libs.logback15

xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java

@@ -57,8 +57,6 @@
 import com.linecorp.centraldogma.xds.endpoint.v1.LocalityLbEndpoint;
 import com.linecorp.centraldogma.xds.k8s.v1.KubernetesEndpointAggregator;
 
-import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager;
-
 public final class XdsResourceManager {
 
     private static final Logger logger = LoggerFactory.getLogger(XdsResourceManager.class);
@@ -76,24 +74,40 @@ public final class XdsResourceManager {
 
     public static final MessageMarshaller JSON_MESSAGE_MARSHALLER;
 
+    // Packages of proto types that xDS resources may embed in a google.protobuf.Any (e.g. an HttpFilter
+    // typed_config) but that do not live under io.envoyproxy.envoy, so the Envoy scan does not pick them up.
+    // These are the Armeria xDS filter configs and their metadata messages, provided by armeria-xds-api.
+    private static final ImmutableList<String> CUSTOM_TYPE_PACKAGES = ImmutableList.of(
+            "com.linecorp.armeria.xds.athenz",
+            "jp.co.lycorp.ftd.athenz.v1");
+
     static {
         final MessageMarshaller.Builder builder =
                 MessageMarshaller.builder().omittingInsignificantWhitespace(true);
         // These Central Dogma-specific types are not in the io.envoyproxy.envoy package, so they
-        // must be registered explicitly; envoyExtension() does not pick them up.
+        // must be registered explicitly; the io.envoyproxy.envoy scan does not pick them up.
         builder.register(KubernetesEndpointAggregator.getDefaultInstance())
                .register(LocalityLbEndpoint.getDefaultInstance());
-        envoyExtension(builder);
+        // Envoy types use java_multiple_files, so every message is a top-level class.
+        registerPackage(builder, "io.envoyproxy.envoy", false);
+        // The custom (non-Envoy) protos do not set java_multiple_files, so their messages are nested.
+        for (String customTypePackage : CUSTOM_TYPE_PACKAGES) {
+            registerPackage(builder, customTypePackage, true);
+        }
         JSON_MESSAGE_MARSHALLER = builder.build();
     }
 
-    private static void envoyExtension(MessageMarshaller.Builder builder) {
+    private static void registerPackage(MessageMarshaller.Builder builder, String packageName,
+                                        boolean includeNested) {
         final Reflections reflections = new Reflections(
-                "io.envoyproxy.envoy", HttpConnectionManager.class.getClassLoader(),
+                packageName, XdsResourceManager.class.getClassLoader(),
                 new SubTypesScanner(true));
         reflections.getSubTypesOf(GeneratedMessageV3.class)
                    .stream()
-                   .filter(c -> !c.getName().contains("$")) // exclude inner classes
+                   // For java_multiple_files packages, a nested ('$') class is never a registrable message.
+                   // For custom packages the messages themselves are nested, so nested classes are kept and
+                   // the non-message ones (Builders, enums) are filtered out below by getDefaultInstance check.
+                   .filter(c -> includeNested || !c.getName().contains("$"))
                    .filter(XdsResourceManager::hasGetDefaultInstanceMethod)
                    .forEach(c -> {
                        // register() does not throw; build() does. A test build per class is needed

xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsAthenzResourceParseTest.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2026 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.centraldogma.xds.internal;
+
+import static com.linecorp.centraldogma.xds.internal.XdsResourceManager.JSON_MESSAGE_MARSHALLER;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+import com.google.protobuf.Any;
+
+import com.linecorp.armeria.xds.athenz.AthenzFilterConfig.AccessTokenConstraintConfig;
+import com.linecorp.centraldogma.internal.Yaml;
+
+import io.envoyproxy.envoy.config.listener.v3.ApiListener;
+import io.envoyproxy.envoy.config.listener.v3.Listener;
+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 jp.co.lycorp.ftd.athenz.v1.AthenzAccessToken.AccessTokenConstraint;
+
+class XdsAthenzResourceParseTest {
+
+    private static Listener athenzListener() {
+        // An inbound Athenz constraint filter config, packed as the http filter typed_config Any.
+        final AccessTokenConstraintConfig constraintConfig =
+                AccessTokenConstraintConfig.newBuilder()
+                                           .setZtsClusterName("zts-cluster")
+                                           .setAccessTokenConstraint(
+                                                   AccessTokenConstraint.newBuilder()
+                                                                        .setConstraintDomain("my.domain")
+                                                                        .setSyntaxVersion(1))
+                                           .build();
+        final HttpConnectionManager manager =
+                HttpConnectionManager.newBuilder()
+                                     .setStatPrefix("ingress_http")
+                                     .addHttpFilters(
+                                             HttpFilter.newBuilder()
+                                                       .setName("armeria.xds.athenz.access_token_constraint")
+                                                       .setTypedConfig(Any.pack(constraintConfig)))
+                                     .build();
+        return Listener.newBuilder()
+                       .setName("groups/foo/listeners/athenz")
+                       .setApiListener(ApiListener.newBuilder().setApiListener(Any.pack(manager)))
+                       .build();
+    }
+
+    @Test
+    void athenzTypedConfigRoundTrips() throws Exception {
+        final Listener listener = athenzListener();
+
+        // Serialize (as the serving/read path does) — the marshaller must resolve the Athenz Any type.
+        final String yaml = XdsResourceManager.toYamlBodyString(listener);
+        assertThat(yaml).contains("type.googleapis.com/armeria.xds.athenz.AccessTokenConstraintConfig");
+
+        // Parse it back (as the create/update and control-plane paths do).
+        final Listener.Builder builder = Listener.newBuilder();
+        JSON_MESSAGE_MARSHALLER.mergeValue(Yaml.readTree(yaml).traverse(), builder);
+        assertThat(builder.build()).isEqualTo(listener);
+    }
+
+    @Test
+    void athenzFilterConfigUnpacksFromParsedResource() throws Exception {
+        final Listener parsed = XdsResourceManager.parseYaml(
+                XdsResourceManager.toYamlBodyString(athenzListener()), Listener.newBuilder());
+        final HttpConnectionManager manager =
+                parsed.getApiListener().getApiListener().unpack(HttpConnectionManager.class);
+        final Any typedConfig = manager.getHttpFilters(0).getTypedConfig();
+        assertThat(typedConfig.getTypeUrl())
+                .isEqualTo("type.googleapis.com/armeria.xds.athenz.AccessTokenConstraintConfig");
+        final AccessTokenConstraintConfig constraintConfig =
+                typedConfig.unpack(AccessTokenConstraintConfig.class);
+        assertThat(constraintConfig.getZtsClusterName()).isEqualTo("zts-cluster");
+        assertThat(constraintConfig.getAccessTokenConstraint().getConstraintDomain()).isEqualTo("my.domain");
+    }
+}

xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsTestUtil.java

@@ -222,10 +222,15 @@ static Cluster staticCluster(String clusterName, ClusterLoadAssignment loadAssig
     }
 
     public static Listener exampleListener(String listenerName, String routeName, String statPrefix) {
-        final HttpConnectionManager manager = httpConnectionManager(routeName, rdsConfigSource());
+        // Apply statPrefix to the HttpConnectionManager (a field the Armeria xDS client marks as supported)
+        // rather than to Listener.stat_prefix, which the client does not mark supported and would reject
+        // during validation.
+        final HttpConnectionManager manager =
+                httpConnectionManager(routeName, rdsConfigSource()).toBuilder()
+                                                                   .setStatPrefix(statPrefix)
+                                                                   .build();
         return Listener.newBuilder()
                        .setName(listenerName)
-                       .setStatPrefix(statPrefix)
                        .setApiListener(ApiListener.newBuilder()
                                                   .setApiListener(Any.pack(manager)))
                        .build();

xds/src/test/java/com/linecorp/centraldogma/xds/listener/v1/XdsListenerServiceTest.java

@@ -41,15 +41,20 @@
 import com.linecorp.armeria.common.HttpMethod;
 import com.linecorp.armeria.common.HttpStatus;
 import com.linecorp.armeria.common.RequestHeaders;
+import com.linecorp.armeria.xds.athenz.AthenzFilterConfig.AccessTokenConstraintConfig;
 import com.linecorp.centraldogma.internal.Yaml;
 import com.linecorp.centraldogma.testing.junit.CentralDogmaExtension;
 
 import io.envoyproxy.controlplane.cache.Resources.V3;
+import io.envoyproxy.envoy.config.listener.v3.ApiListener;
 import io.envoyproxy.envoy.config.listener.v3.Listener;
+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.service.discovery.v3.DiscoveryRequest;
 import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
 import io.envoyproxy.envoy.service.listener.v3.ListenerDiscoveryServiceGrpc.ListenerDiscoveryServiceStub;
 import io.grpc.stub.StreamObserver;
+import jp.co.lycorp.ftd.athenz.v1.AthenzAccessToken.AccessTokenConstraint;
 
 class XdsListenerServiceTest {
 
@@ -87,6 +92,44 @@ void createListenerViaHttp() throws Exception {
         checkResourceViaDiscoveryRequest(actualListener, listenerName, true);
     }
 
+    @Test
+    void createAthenzListenerViaHttp() throws Exception {
+        // A listener whose HttpConnectionManager carries an Athenz inbound filter config in an HttpFilter
+        // typed_config Any. This exercises both parsing (the create API) and serving (LDS) of a resource that
+        // embeds a custom, non-io.envoyproxy.envoy proto type provided by armeria-xds-api.
+        final AccessTokenConstraintConfig constraintConfig =
+                AccessTokenConstraintConfig.newBuilder()
+                                           .setZtsClusterName("zts-cluster")
+                                           .setAccessTokenConstraint(
+                                                   AccessTokenConstraint.newBuilder()
+                                                                        .setConstraintDomain("my.domain")
+                                                                        .setSyntaxVersion(1))
+                                           .build();
+        final HttpConnectionManager manager =
+                HttpConnectionManager.newBuilder()
+                                     .setStatPrefix("ingress_http")
+                                     .addHttpFilters(
+                                             HttpFilter.newBuilder()
+                                                       .setName("armeria.xds.athenz.access_token_constraint")
+                                                       .setTypedConfig(Any.pack(constraintConfig)))
+                                     .build();
+        final Listener listener =
+                Listener.newBuilder()
+                        .setName("this_listener_name_will_be_ignored_and_replaced")
+                        .setApiListener(ApiListener.newBuilder().setApiListener(Any.pack(manager)))
+                        .build();
+
+        final AggregatedHttpResponse response =
+                createListener("groups/foo", "athenz-listener.1", listener, dogma.httpClient());
+        assertOk(response);
+        final Listener.Builder listenerBuilder = Listener.newBuilder();
+        JSON_MESSAGE_MARSHALLER.mergeValue(Yaml.readTree(response.contentUtf8()).traverse(), listenerBuilder);
+        final Listener actualListener = listenerBuilder.build();
+        final String listenerName = "groups/foo/listeners/athenz-listener.1";
+        assertThat(actualListener).isEqualTo(listener.toBuilder().setName(listenerName).build());
+        checkResourceViaDiscoveryRequest(actualListener, listenerName, true);
+    }
+
     private static void assertOk(AggregatedHttpResponse response) {
         assertThat(response.status()).isSameAs(HttpStatus.OK);
     }