Introduce an extension point for specifying custom ConfigSource

it/xds-client/src/main/java/com/linecorp/armeria/xds/it/XdsResourceReader.java

@@ -16,67 +16,28 @@
 
 package com.linecorp.armeria.xds.it;
 
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
 import com.google.common.escape.Escaper;
 import com.google.common.escape.Escapers;
-import com.google.protobuf.GeneratedMessage;
-import com.google.protobuf.util.JsonFormat;
-import com.google.protobuf.util.JsonFormat.Parser;
-import com.google.protobuf.util.JsonFormat.TypeRegistry;
+import com.google.protobuf.GeneratedMessageV3;
 
 import io.envoyproxy.envoy.config.bootstrap.v3.Bootstrap;
-import io.envoyproxy.envoy.extensions.filters.http.router.v3.Router;
-import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager;
-import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext;
 
+/**
+ * Delegates to {@link com.linecorp.armeria.xds.XdsResourceReader} for YAML/JSON parsing.
+ * Keeps test-only utilities like {@link #escapeMultiLine(String)}.
+ */
 public final class XdsResourceReader {
 
-    private static final ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
-    private static final ObjectMapper jsonMapper = new ObjectMapper();
-    private static final Parser parser =
-            JsonFormat.parser().usingTypeRegistry(TypeRegistry.newBuilder()
-                                                              .add(HttpConnectionManager.getDescriptor())
-                                                              .add(Router.getDescriptor())
-                                                              .add(UpstreamTlsContext.getDescriptor())
-                                                              .build());
-
     public static Bootstrap fromYaml(String yaml) {
-        final Bootstrap.Builder bootstrapBuilder = Bootstrap.newBuilder();
-        try {
-            final JsonNode jsonNode = mapper.reader().readTree(yaml);
-            parser.merge(jsonNode.toString(), bootstrapBuilder);
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-        return bootstrapBuilder.build();
+        return com.linecorp.armeria.xds.XdsResourceReader.from(yaml, Bootstrap.class);
     }
 
-    @SuppressWarnings("unchecked")
-    public static <T extends GeneratedMessage> T fromYaml(String yaml, Class<T> clazz) {
-        final GeneratedMessage.Builder<?> builder;
-        try {
-            builder = (GeneratedMessage.Builder<?>) clazz.getMethod("newBuilder").invoke(null);
-            final JsonNode jsonNode = mapper.reader().readTree(yaml);
-            parser.merge(jsonNode.toString(), builder);
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-        return (T) builder.build();
+    public static <T extends GeneratedMessageV3> T fromYaml(String yaml, Class<T> clazz) {
+        return com.linecorp.armeria.xds.XdsResourceReader.from(yaml, clazz);
     }
 
-    @SuppressWarnings("unchecked")
-    public static <T extends GeneratedMessage> T fromJson(String json, Class<T> clazz) {
-        final GeneratedMessage.Builder<?> builder;
-        try {
-            builder = (GeneratedMessage.Builder<?>) clazz.getMethod("newBuilder").invoke(null);
-            final JsonNode jsonNode = jsonMapper.reader().readTree(json);
-            parser.merge(jsonNode.toString(), builder);
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-        return (T) builder.build();
+    public static <T extends GeneratedMessageV3> T fromJson(String json, Class<T> clazz) {
+        return com.linecorp.armeria.xds.XdsResourceReader.from(json, clazz);
     }
 
     private static final Escaper multiLineEscaper = Escapers.builder()

it/xds-client/src/test/java/com/linecorp/armeria/xds/it/PathConfigSourceTest.java

@@ -0,0 +1,153 @@
+/*
+ * 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.armeria.xds.it;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import com.linecorp.armeria.xds.ClusterSnapshot;
+import com.linecorp.armeria.xds.SnapshotWatcher;
+import com.linecorp.armeria.xds.XdsBootstrap;
+
+import io.envoyproxy.envoy.config.bootstrap.v3.Bootstrap;
+
+class PathConfigSourceTest {
+
+    @TempDir
+    Path tempDir;
+
+    @Test
+    void clusterFromJsonPathConfigSource() throws Exception {
+        //language=JSON
+        final String json = """
+                {
+                  "typeUrl": "type.googleapis.com/envoy.config.cluster.v3.Cluster",
+                  "resources": [
+                    {
+                      "@type": "type.googleapis.com/envoy.config.cluster.v3.Cluster",
+                      "name": "path-cluster",
+                      "type": "STATIC",
+                      "loadAssignment": {
+                        "clusterName": "path-cluster",
+                        "endpoints": [
+                          {
+                            "lbEndpoints": [
+                              {
+                                "endpoint": {
+                                  "address": {
+                                    "socketAddress": {
+                                      "address": "127.0.0.1",
+                                      "portValue": 8080
+                                    }
+                                  }
+                                }
+                              }
+                            ]
+                          }
+                        ]
+                      }
+                    }
+                  ],
+                  "versionInfo": "1"
+                }
+                """;
+
+        final Path targetFile = tempDir.resolve("clusters.json");
+        Files.writeString(targetFile, json);
+
+        //language=YAML
+        final String bootstrapYaml =
+                """
+                dynamic_resources:
+                  cds_config:
+                    path_config_source:
+                      path: %s
+                """.formatted(targetFile);
+
+        verifyClusterFromPathConfigSource(bootstrapYaml);
+    }
+
+    @Test
+    void clusterFromYamlPathConfigSource() throws Exception {
+        //language=YAML
+        final String yaml = """
+                typeUrl: "type.googleapis.com/envoy.config.cluster.v3.Cluster"
+                resources:
+                  - "@type": "type.googleapis.com/envoy.config.cluster.v3.Cluster"
+                    name: path-cluster
+                    type: STATIC
+                    loadAssignment:
+                      clusterName: path-cluster
+                      endpoints:
+                        - lbEndpoints:
+                            - endpoint:
+                                address:
+                                  socketAddress:
+                                    address: 127.0.0.1
+                                    portValue: 8080
+                versionInfo: "1"
+                """;
+
+        final Path targetFile = tempDir.resolve("clusters.yaml");
+        Files.writeString(targetFile, yaml);
+
+        //language=YAML
+        final String bootstrapYaml =
+                """
+                dynamic_resources:
+                  cds_config:
+                    path_config_source:
+                      path: %s
+                """.formatted(targetFile);
+
+        verifyClusterFromPathConfigSource(bootstrapYaml);
+    }
+
+    private static void verifyClusterFromPathConfigSource(String bootstrapYaml) {
+        final Bootstrap bootstrap = XdsResourceReader.fromYaml(bootstrapYaml, Bootstrap.class);
+        final AtomicReference<ClusterSnapshot> snapshotRef = new AtomicReference<>();
+        final AtomicReference<Throwable> errorRef = new AtomicReference<>();
+        final SnapshotWatcher<Object> watcher = (snapshot, t) -> {
+            if (t != null) {
+                errorRef.set(t);
+                return;
+            }
+            if (snapshot instanceof ClusterSnapshot) {
+                snapshotRef.set((ClusterSnapshot) snapshot);
+            }
+        };
+
+        try (XdsBootstrap xdsBootstrap = XdsBootstrap.builder(bootstrap)
+                                                     .defaultSnapshotWatcher(watcher)
+                                                     .build()) {
+            xdsBootstrap.clusterRoot("path-cluster");
+            await().untilAsserted(() -> {
+                assertThat(errorRef.get()).isNull();
+                assertThat(snapshotRef.get()).isNotNull();
+                assertThat(snapshotRef.get().xdsResource().resource().getName())
+                        .isEqualTo("path-cluster");
+            });
+        }
+    }
+}

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

@@ -168,6 +168,7 @@ message RateLimitSettings {
 
 // Local filesystem path configuration source.
 message PathConfigSource {
+  option (armeria.xds.supported.field) = 1;
   // Path on the filesystem to source and watch for configuration updates.
   // When sourcing configuration for a :ref:`secret <envoy_v3_api_msg_extensions.transport_sockets.tls.v3.Secret>`,
   // the certificate and key files are also watched for updates.
@@ -186,6 +187,7 @@ message PathConfigSource {
   //   this path. This is required in certain deployment scenarios. See below for more information.
   string path = 1 [(validate.rules).string = {min_len: 1}];
 
+  option (armeria.xds.supported.field) = 2;
   // If configured, this directory will be watched for *moves*. When an entry in this directory is
   // moved to, the ``path`` will be reloaded. This is required in certain deployment scenarios.
   //
@@ -225,6 +227,7 @@ message ConfigSource {
     string path = 1 [deprecated = true, (envoy.annotations.deprecated_at_minor_version) = "3.0"];
 
     // Local filesystem path configuration source.
+    option (armeria.xds.supported.oneof_field) = 8;
     PathConfigSource path_config_source = 8;
 
     // API configuration source.
@@ -249,6 +252,14 @@ message ConfigSource {
     // is provided via ADS and the specified data can also be obtained via ADS.]
     option (armeria.xds.supported.oneof_field) = 5;
     SelfConfigSource self = 5;
+
+    // Armeria extension: a custom config source implementation resolved via
+    // the extension registry by the Any's type URL. This allows plugging in
+    // non-standard config sources (e.g. CentralDogma, etcd, Consul) without
+    // modifying the upstream proto definition.
+    // The large field number avoids conflicts with future upstream additions.
+    option (armeria.xds.supported.oneof_field) = 1000;
+    google.protobuf.Any custom_config_source = 1000;
   }
 
   // When this timeout is specified, Envoy will wait no longer than the specified time for first

xds/build.gradle

@@ -10,6 +10,7 @@ dependencies {
     api project(':xds-api')
     api project(':xds-validator')
 
+    implementation libs.jackson.dataformat.yaml
     implementation libs.re2j
 
     testImplementation libs.controlplane.server

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

@@ -47,6 +47,10 @@ interface ActualStreamFactory {
     private final ConfigSourceLifecycleObserver lifecycleObserver;
     private final Set<XdsType> targetTypes;
 
+    StateCoordinator stateCoordinator() {
+        return stateCoordinator;
+    }
+
     private int connBackoffAttempts = 1;
     private boolean stopped;
     @Nullable