Add xds-kubernetes module for Kubernetes-based xDS endpoint discovery

it/xds-client/build.gradle

@@ -12,6 +12,10 @@ dependencies {
     testImplementation project(':athenz')
     testImplementation project(':thrift0.18')
     testImplementation project(':xds-athenz')
+    testImplementation project(':xds-kubernetes')
+    testImplementation(variantOf(libs.kubernetes.client.api) { classifier("tests") })
+    testImplementation libs.kubernetes.server.mock
+    testImplementation libs.kubernetes.junit.jupiter
     testImplementation libs.athenz.zms.client
     testImplementation libs.testcontainers.junit.jupiter
 }

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

@@ -0,0 +1,224 @@
+/*
+ * 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 java.util.Map;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import com.google.common.collect.ImmutableMap;
+
+import com.linecorp.armeria.client.BlockingWebClient;
+import com.linecorp.armeria.client.WebClient;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.server.ServerBuilder;
+import com.linecorp.armeria.testing.junit5.server.ServerExtension;
+import com.linecorp.armeria.xds.XdsBootstrap;
+import com.linecorp.armeria.xds.client.endpoint.XdsHttpPreprocessor;
+import com.linecorp.armeria.xds.kubernetes.KubernetesClusterTypeFactory;
+
+import io.envoyproxy.envoy.config.bootstrap.v3.Bootstrap;
+import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
+import io.fabric8.kubernetes.api.model.LabelSelectorBuilder;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.api.model.PodSpec;
+import io.fabric8.kubernetes.api.model.PodSpecBuilder;
+import io.fabric8.kubernetes.api.model.PodStatusBuilder;
+import io.fabric8.kubernetes.api.model.PodTemplateSpec;
+import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder;
+import io.fabric8.kubernetes.api.model.Service;
+import io.fabric8.kubernetes.api.model.ServiceBuilder;
+import io.fabric8.kubernetes.api.model.ServicePortBuilder;
+import io.fabric8.kubernetes.api.model.ServiceSpecBuilder;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;
+import io.fabric8.kubernetes.api.model.apps.DeploymentSpecBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+
+@EnableKubernetesMockClient(crud = true)
+class KubernetesClusterTypeIntegrationTest {
+
+    private static final Map<String, String> LABELS = ImmutableMap.of("app", "test-app");
+
+    KubernetesClient client;
+
+    @RegisterExtension
+    static final ServerExtension backendServer = new ServerExtension() {
+        @Override
+        protected void configure(ServerBuilder sb) {
+            sb.service("/hello", (ctx, req) -> HttpResponse.of("world"));
+        }
+    };
+
+    @BeforeEach
+    void setUp() {
+        createK8sResources();
+    }
+
+    @Test
+    void basicEndpointDiscovery() {
+        final Bootstrap bootstrap = bootstrapYaml(client.getMasterUrl().toString());
+        final KubernetesClusterTypeFactory factory = KubernetesClusterTypeFactory.of(
+                client.getConfiguration(),
+                (clusterName, endpoints) -> backendCla(clusterName));
+
+        try (XdsBootstrap xdsBootstrap = XdsBootstrap.builder(bootstrap)
+                                                     .extensionFactories(factory)
+                                                     .build();
+             XdsHttpPreprocessor preprocessor =
+                     XdsHttpPreprocessor.ofListener("listener1", xdsBootstrap)) {
+            final BlockingWebClient webClient = WebClient.of(preprocessor).blocking();
+            assertThat(webClient.get("/hello").contentUtf8()).isEqualTo("world");
+        }
+    }
+
+    private void createK8sResources() {
+        final Deployment deployment = newDeployment();
+        final Service service = newService();
+        client.apps().deployments().resource(deployment).create();
+        client.services().resource(service).create();
+
+        final PodTemplateSpec template = deployment.getSpec().getTemplate();
+        client.pods().resource(newPodWithIp(template, "pod-0", "10.0.0.1")).create();
+    }
+
+    private static Deployment newDeployment() {
+        final ObjectMeta metadata = new ObjectMetaBuilder()
+                .withName("test-deployment")
+                .build();
+        return new DeploymentBuilder()
+                .withMetadata(metadata)
+                .withSpec(new DeploymentSpecBuilder()
+                                  .withSelector(new LabelSelectorBuilder().withMatchLabels(LABELS).build())
+                                  .withTemplate(newPodTemplate())
+                                  .build())
+                .build();
+    }
+
+    private static PodTemplateSpec newPodTemplate() {
+        final ObjectMeta metadata = new ObjectMetaBuilder()
+                .withLabels(LABELS)
+                .build();
+        final Container container = new ContainerBuilder()
+                .withName("app")
+                .withImage("app:latest")
+                .withPorts(new ContainerPortBuilder()
+                                   .withContainerPort(8080)
+                                   .build())
+                .build();
+        final PodSpec spec = new PodSpecBuilder()
+                .withContainers(container)
+                .build();
+        return new PodTemplateSpecBuilder()
+                .withMetadata(metadata)
+                .withSpec(spec)
+                .build();
+    }
+
+    private static Pod newPodWithIp(PodTemplateSpec template, String podName, String podIp) {
+        final PodSpec spec = template.getSpec()
+                                     .toBuilder()
+                                     .withNodeName("dummy-node")
+                                     .build();
+        final ObjectMeta metadata = template.getMetadata()
+                                            .toBuilder()
+                                            .withName(podName)
+                                            .build();
+        return new PodBuilder()
+                .withMetadata(metadata)
+                .withSpec(spec)
+                .withStatus(new PodStatusBuilder().withPodIP(podIp).build())
+                .build();
+    }
+
+    private static Service newService() {
+        final ObjectMeta metadata = new ObjectMetaBuilder().withName("test-service")
+                                                           .build();
+        return new ServiceBuilder()
+                .withMetadata(metadata)
+                .withSpec(new ServiceSpecBuilder().withPorts(new ServicePortBuilder().withPort(8080).build())
+                                                  .withSelector(LABELS)
+                                                  .withType("ClusterIP")
+                                                  .build())
+                .build();
+    }
+
+    private static ClusterLoadAssignment backendCla(String clusterName) {
+        //language=YAML
+        final String yaml = """
+                cluster_name: %s
+                endpoints:
+                - lb_endpoints:
+                  - endpoint:
+                      address:
+                        socket_address:
+                          address: 127.0.0.1
+                          port_value: %s
+                """.formatted(clusterName, backendServer.httpPort());
+        return XdsResourceReader.fromYaml(yaml, ClusterLoadAssignment.class);
+    }
+
+    private static Bootstrap bootstrapYaml(String apiServerUrl) {
+        //language=YAML
+        final String yaml = """
+                static_resources:
+                  listeners:
+                  - name: listener1
+                    api_listener:
+                      api_listener:
+                        "@type": type.googleapis.com/envoy.extensions.filters.network\
+                .http_connection_manager.v3.HttpConnectionManager
+                        stat_prefix: http
+                        route_config:
+                          name: route1
+                          virtual_hosts:
+                          - name: local_service1
+                            domains: [ "*" ]
+                            routes:
+                            - match:
+                                prefix: /
+                              route:
+                                cluster: cluster1
+                        http_filters:
+                        - name: envoy.filters.http.router
+                          typed_config:
+                            "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
+                  clusters:
+                  - name: cluster1
+                    cluster_type:
+                      name: armeria.cluster.kubernetes
+                      typed_config:
+                        "@type": type.googleapis.com/armeria.xds.kubernetes.KubernetesClusterConfig
+                        service_name: test-service
+                        namespace: test
+                        mode: POD
+                        api_server_url: "%s"
+                """.formatted(apiServerUrl);
+        return XdsResourceReader.fromYaml(yaml, Bootstrap.class);
+    }
+}

kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaHttpClient.java

@@ -60,7 +60,7 @@ final class ArmeriaHttpClient
 
     @Override
     public void doClose() {
-        webClient.options().factory().close();
+        webClient.options().factory().closeAsync();
         webSocketClient.close();
     }
 

kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaWebSocketClient.java

@@ -131,7 +131,7 @@ CompletableFuture<WebSocketResponse> execute(StandardWebSocketBuilder webSocketR
     public void close() {
         final WebSocketClient webSocketClient = this.webSocketClient;
         if (webSocketClient != null) {
-            webSocketClient.options().factory().close();
+            webSocketClient.options().factory().closeAsync();
         }
     }
 

settings.gradle

@@ -221,6 +221,7 @@ includeWithFlags ':tomcat9',                             'java', 'publish', 'rel
 includeWithFlags ':tomcat10',                            'java11', 'publish', 'relocate'
 includeWithFlags ':xds',                                 'java', 'publish', 'relocate'
 includeWithFlags ':xds-athenz',                          'java11', 'publish', 'relocate'
+includeWithFlags ':xds-kubernetes',                      'java11', 'publish', 'relocate'
 includeWithFlags ':xds-api',                             'java', 'publish', 'relocate', 'javapgv', 'no_aggregation'
 includeWithFlags ':xds-validator',                       'java', 'publish', 'relocate', 'no_aggregation'
 includeWithFlags ':xds-pgv-shaded',                      'java', 'publish', 'relocate', 'no_aggregation'

xds-api/src/main/proto/armeria/xds/kubernetes/kubernetes_cluster_config.proto

@@ -0,0 +1,47 @@
+syntax = "proto3";
+
+package armeria.xds.kubernetes;
+
+option java_package = "com.linecorp.armeria.xds.kubernetes";
+option java_multiple_files = true;
+
+import "validate/validate.proto";
+import "armeria/xds/supported.proto";
+import "envoy/extensions/transport_sockets/tls/v3/secret.proto";
+
+message KubernetesClusterConfig {
+  // The Kubernetes Service name to watch for endpoints. (Required)
+  option (armeria.xds.supported.field) = 1;
+  string service_name = 1 [(validate.rules).string = {min_len: 1}];
+
+  // The Kubernetes namespace. If empty, uses the client's default namespace.
+  option (armeria.xds.supported.field) = 2;
+  string namespace = 2;
+
+  // The port name to select. If empty, uses the first port.
+  option (armeria.xds.supported.field) = 3;
+  string port_name = 3;
+
+  // Endpoint discovery mode. Default: POD.
+  option (armeria.xds.supported.field) = 4;
+  KubernetesEndpointMode mode = 4;
+
+  // The Kubernetes API server URL (e.g. "https://kubernetes.default.svc").
+  // If empty, uses the default from kubeconfig or in-cluster config.
+  option (armeria.xds.supported.field) = 5;
+  string api_server_url = 5;
+
+  // SDS secret config for the bearer token to authenticate with the K8s API.
+  // The secret must be a generic_secret whose value is the raw bearer token.
+  // When the secret rotates, the KubernetesEndpointGroup is recreated.
+  // If unset, uses the default KubernetesClient auth (kubeconfig/in-cluster SA token).
+  option (armeria.xds.supported.field) = 6;
+  envoy.extensions.transport_sockets.tls.v3.SdsSecretConfig credential = 6;
+}
+
+enum KubernetesEndpointMode {
+  // Direct pod IP connections (default). True client-side load balancing.
+  POD = 0;
+  // NodeIP:NodePort connections via Kubernetes NodePort/LoadBalancer services.
+  NODE_PORT = 1;
+}

xds-kubernetes/build.gradle

@@ -0,0 +1,4 @@
+dependencies {
+    api project(':xds')
+    api project(':kubernetes')
+}

xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/DefaultKubernetesEndpointMapper.java

@@ -0,0 +1,67 @@
+/*
+ * 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.kubernetes;
+
+import java.util.List;
+import java.util.Set;
+
+import com.google.common.collect.ImmutableSet;
+
+import com.linecorp.armeria.client.Endpoint;
+
+import io.envoyproxy.envoy.config.core.v3.Address;
+import io.envoyproxy.envoy.config.core.v3.HealthStatus;
+import io.envoyproxy.envoy.config.core.v3.SocketAddress;
+import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
+import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint;
+import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints;
+
+final class DefaultKubernetesEndpointMapper implements KubernetesEndpointMapper {
+
+    private static final DefaultKubernetesEndpointMapper INSTANCE = new DefaultKubernetesEndpointMapper();
+
+    static DefaultKubernetesEndpointMapper get() {
+        return INSTANCE;
+    }
+
+    private DefaultKubernetesEndpointMapper() {}
+
+    @Override
+    public ClusterLoadAssignment map(String clusterName, List<Endpoint> endpoints) {
+        final LocalityLbEndpoints.Builder localityBuilder = LocalityLbEndpoints.newBuilder();
+        final Set<Endpoint> deduped = ImmutableSet.copyOf(endpoints);
+        for (Endpoint endpoint : deduped) {
+            final SocketAddress.Builder sa = SocketAddress.newBuilder()
+                                                          .setAddress(endpoint.host());
+            if (endpoint.hasPort()) {
+                sa.setPortValue(endpoint.port());
+            }
+            localityBuilder.addLbEndpoints(
+                    LbEndpoint.newBuilder()
+                              .setHealthStatus(HealthStatus.HEALTHY)
+                              .setEndpoint(
+                                      io.envoyproxy.envoy.config.endpoint.v3.Endpoint.newBuilder()
+                                              .setAddress(Address.newBuilder()
+                                                                 .setSocketAddress(sa))));
+        }
+
+        return ClusterLoadAssignment.newBuilder()
+                                    .setClusterName(clusterName)
+                                    .addEndpoints(localityBuilder)
+                                    .build();
+    }
+}