diff --git a/it/xds-client/build.gradle b/it/xds-client/build.gradle index 16490854965..16d80203048 100644 --- a/it/xds-client/build.gradle +++ b/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 } diff --git a/it/xds-client/src/test/java/com/linecorp/armeria/xds/it/KubernetesClusterTypeIntegrationTest.java b/it/xds-client/src/test/java/com/linecorp/armeria/xds/it/KubernetesClusterTypeIntegrationTest.java new file mode 100644 index 00000000000..e68dfc22773 --- /dev/null +++ b/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 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); + } +} diff --git a/kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaHttpClient.java b/kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaHttpClient.java index 29a7e6a88d1..2808b8430cc 100644 --- a/kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaHttpClient.java +++ b/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(); } diff --git a/kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaWebSocketClient.java b/kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaWebSocketClient.java index 11a60e4bd10..bf1454aec7e 100644 --- a/kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaWebSocketClient.java +++ b/kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/ArmeriaWebSocketClient.java @@ -131,7 +131,7 @@ CompletableFuture execute(StandardWebSocketBuilder webSocketR public void close() { final WebSocketClient webSocketClient = this.webSocketClient; if (webSocketClient != null) { - webSocketClient.options().factory().close(); + webSocketClient.options().factory().closeAsync(); } } diff --git a/settings.gradle b/settings.gradle index c7c2db93ca9..fcc6831664e 100644 --- a/settings.gradle +++ b/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' diff --git a/xds-api/src/main/proto/armeria/xds/kubernetes/kubernetes_cluster_config.proto b/xds-api/src/main/proto/armeria/xds/kubernetes/kubernetes_cluster_config.proto new file mode 100644 index 00000000000..095b90d6ac7 --- /dev/null +++ b/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; +} diff --git a/xds-kubernetes/build.gradle b/xds-kubernetes/build.gradle new file mode 100644 index 00000000000..555c16fbda8 --- /dev/null +++ b/xds-kubernetes/build.gradle @@ -0,0 +1,4 @@ +dependencies { + api project(':xds') + api project(':kubernetes') +} diff --git a/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/DefaultKubernetesEndpointMapper.java b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/DefaultKubernetesEndpointMapper.java new file mode 100644 index 00000000000..04ce5eb859c --- /dev/null +++ b/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 endpoints) { + final LocalityLbEndpoints.Builder localityBuilder = LocalityLbEndpoints.newBuilder(); + final Set 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(); + } +} diff --git a/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesClusterTypeFactory.java b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesClusterTypeFactory.java new file mode 100644 index 00000000000..f2d76726a1f --- /dev/null +++ b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesClusterTypeFactory.java @@ -0,0 +1,183 @@ +/* + * 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 static java.util.Objects.requireNonNull; + +import java.util.List; + +import com.google.common.collect.ImmutableList; +import com.google.protobuf.Any; + +import com.linecorp.armeria.client.kubernetes.endpoints.KubernetesEndpointGroup; +import com.linecorp.armeria.client.kubernetes.endpoints.KubernetesEndpointGroupBuilder; +import com.linecorp.armeria.client.kubernetes.endpoints.KubernetesEndpointMode; +import com.linecorp.armeria.common.annotation.UnstableApi; +import com.linecorp.armeria.xds.ClusterXdsResource; +import com.linecorp.armeria.xds.EndpointSnapshot; +import com.linecorp.armeria.xds.client.endpoint.ClusterTypeFactory; +import com.linecorp.armeria.xds.filter.FactoryContext; +import com.linecorp.armeria.xds.stream.SnapshotStream; + +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; +import io.fabric8.kubernetes.client.Config; +import io.fabric8.kubernetes.client.ConfigBuilder; + +/** + * A {@link ClusterTypeFactory} that resolves endpoints using a Kubernetes + * {@link KubernetesEndpointGroup}. + * + *

The cluster's {@code typed_config} must be a + * {@link com.linecorp.armeria.xds.kubernetes.KubernetesClusterConfig KubernetesClusterConfig} + * protobuf message specifying the Kubernetes service to watch. + * + *

If the config includes a {@code credential} field referencing an SDS secret, the factory + * uses it as a bearer token for Kubernetes API authentication. When the secret rotates, the + * {@link KubernetesEndpointGroup} is recreated with the new token. + * + *

Example xDS cluster configuration: + * 

{@code
+ * cluster_type:
+ *   name: armeria.cluster.kubernetes
+ *   typed_config:
+ *     "@type": type.googleapis.com/armeria.xds.kubernetes.KubernetesClusterConfig
+ *     service_name: my-service
+ *     namespace: production
+ *     port_name: http
+ *     mode: POD
+ * }
+ */ +@UnstableApi +public final class KubernetesClusterTypeFactory implements ClusterTypeFactory { + + private static final String NAME = "armeria.cluster.kubernetes"; + private static final String TYPE_URL = + "type.googleapis.com/armeria.xds.kubernetes.KubernetesClusterConfig"; + private static final List TYPE_URLS = ImmutableList.of(TYPE_URL); + + /** + * Returns a new factory with the default Kubernetes client configuration and the default + * {@link KubernetesEndpointMapper}. + */ + public static KubernetesClusterTypeFactory of() { + return of(new ConfigBuilder().build(), KubernetesEndpointMapper.ofDefault()); + } + + /** + * Returns a new factory with the specified {@link KubernetesEndpointMapper}. + */ + public static KubernetesClusterTypeFactory of(KubernetesEndpointMapper mapper) { + return of(new ConfigBuilder().build(), requireNonNull(mapper, "mapper")); + } + + /** + * Returns a new factory with the specified base {@link Config} and the default + * {@link KubernetesEndpointMapper}. The base config provides TLS trust settings, + * authentication defaults, and other Kubernetes client options that can be overridden + * by the proto config fields ({@code api_server_url}, {@code credential}). + */ + public static KubernetesClusterTypeFactory of(Config baseConfig) { + return of(requireNonNull(baseConfig, "baseConfig"), KubernetesEndpointMapper.ofDefault()); + } + + /** + * Returns a new factory with the specified base {@link Config} and + * {@link KubernetesEndpointMapper}. + */ + public static KubernetesClusterTypeFactory of(Config baseConfig, KubernetesEndpointMapper mapper) { + return new KubernetesClusterTypeFactory(requireNonNull(baseConfig, "baseConfig"), + requireNonNull(mapper, "mapper")); + } + + private final Config baseConfig; + private final KubernetesEndpointMapper mapper; + + private KubernetesClusterTypeFactory(Config baseConfig, KubernetesEndpointMapper mapper) { + this.baseConfig = baseConfig; + this.mapper = mapper; + } + + @Override + public String name() { + return NAME; + } + + @Override + public List typeUrls() { + return TYPE_URLS; + } + + @Override + public SnapshotStream createEndpointStream( + ClusterXdsResource clusterXdsResource, FactoryContext context) { + final Any typedConfig = clusterXdsResource.resource().getClusterType().getTypedConfig(); + final KubernetesClusterConfig config = context.validator().unpack( + typedConfig, KubernetesClusterConfig.class); + final String clusterName = clusterXdsResource.name(); + + if (config.hasCredential()) { + return context.genericSecretStream(config.getCredential()) + .switchMapEager(secretSnapshot -> { + final ConfigBuilder configBuilder = newConfigBuilder(config); + if (secretSnapshot.credential() != null) { + configBuilder.withOauthToken(secretSnapshot.credential()); + } + return createSnapshot(configBuilder, config, clusterName); + }); + } + return createSnapshot(newConfigBuilder(config), config, clusterName); + } + + private ConfigBuilder newConfigBuilder(KubernetesClusterConfig config) { + final ConfigBuilder configBuilder = new ConfigBuilder(baseConfig); + if (!config.getApiServerUrl().isEmpty()) { + configBuilder.withMasterUrl(config.getApiServerUrl()); + } + return configBuilder; + } + + private SnapshotStream createSnapshot( + ConfigBuilder configBuilder, KubernetesClusterConfig config, String clusterName) { + final KubernetesEndpointGroupBuilder builder = + KubernetesEndpointGroup.builder(configBuilder.build()) + .serviceName(config.getServiceName()); + if (!config.getNamespace().isEmpty()) { + builder.namespace(config.getNamespace()); + } + if (!config.getPortName().isEmpty()) { + builder.portName(config.getPortName()); + } + if (config.getMode() == + com.linecorp.armeria.xds.kubernetes.KubernetesEndpointMode.NODE_PORT) { + builder.mode(KubernetesEndpointMode.NODE_PORT); + } else { + builder.mode(KubernetesEndpointMode.POD); + } + return endpointGroupToSnapshot(builder.build(), clusterName); + } + + private SnapshotStream endpointGroupToSnapshot( + KubernetesEndpointGroup endpointGroup, String clusterName) { + return watcher -> { + endpointGroup.addListener(endpoints -> { + final ClusterLoadAssignment cla = mapper.map(clusterName, endpoints); + watcher.onUpdate(EndpointSnapshot.of(cla), null); + }, true); + return endpointGroup::closeAsync; + }; + } +} diff --git a/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesClusterTypeFactoryProvider.java b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesClusterTypeFactoryProvider.java new file mode 100644 index 00000000000..2167acf2c20 --- /dev/null +++ b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesClusterTypeFactoryProvider.java @@ -0,0 +1,33 @@ +/* + * 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 com.linecorp.armeria.common.annotation.UnstableApi; +import com.linecorp.armeria.xds.XdsExtensionFactory; +import com.linecorp.armeria.xds.XdsExtensionFactoryProvider; + +/** + * Provides the {@link KubernetesClusterTypeFactory} for xDS extension discovery. + */ +@UnstableApi +public final class KubernetesClusterTypeFactoryProvider implements XdsExtensionFactoryProvider { + + @Override + public XdsExtensionFactory newFactory() { + return KubernetesClusterTypeFactory.of(); + } +} diff --git a/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesEndpointMapper.java b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesEndpointMapper.java new file mode 100644 index 00000000000..f1a73c8202c --- /dev/null +++ b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesEndpointMapper.java @@ -0,0 +1,61 @@ +/* + * 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 com.linecorp.armeria.client.Endpoint; +import com.linecorp.armeria.common.annotation.UnstableApi; + +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; + +/** + * Maps a list of Armeria {@link Endpoint}s from a + * {@link com.linecorp.armeria.client.kubernetes.endpoints.KubernetesEndpointGroup KubernetesEndpointGroup} + * into a {@link ClusterLoadAssignment} protobuf for xDS endpoint resolution. + * + *

Users who need locality-aware load balancing, priority tiers, or custom health mapping + * can provide their own implementation via + * {@link KubernetesClusterTypeFactory#of(KubernetesEndpointMapper)}. + * + *

The default mapping places all endpoints in a single + * {@link io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints LocalityLbEndpoints} + * with equal weight, priority 0, and HEALTHY status. + */ +@UnstableApi +@FunctionalInterface +public interface KubernetesEndpointMapper { + + /** + * Returns the default {@link KubernetesEndpointMapper} that places all endpoints in a single + * {@link io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints LocalityLbEndpoints} + * with equal weight, priority 0, and + * {@link io.envoyproxy.envoy.config.core.v3.HealthStatus#HEALTHY HEALTHY} status. + */ + static KubernetesEndpointMapper ofDefault() { + return DefaultKubernetesEndpointMapper.get(); + } + + /** + * Converts the given list of {@link Endpoint}s into a {@link ClusterLoadAssignment}. + * + * @param clusterName the xDS cluster name + * @param endpoints the endpoints discovered from Kubernetes + * @return the constructed {@link ClusterLoadAssignment} + */ + ClusterLoadAssignment map(String clusterName, List endpoints); +} diff --git a/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesTypeRegistryPackageProvider.java b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesTypeRegistryPackageProvider.java new file mode 100644 index 00000000000..a8d4ff9c500 --- /dev/null +++ b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/KubernetesTypeRegistryPackageProvider.java @@ -0,0 +1,34 @@ +/* + * 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 com.google.common.collect.ImmutableList; + +import com.linecorp.armeria.common.annotation.UnstableApi; +import com.linecorp.armeria.xds.XdsTypeRegistryPackageProvider; + +/** + * Provides the Kubernetes cluster config protobuf package for xDS type registry discovery. + */ +@UnstableApi +public final class KubernetesTypeRegistryPackageProvider implements XdsTypeRegistryPackageProvider { + + @Override + public Iterable packages() { + return ImmutableList.of("com.linecorp.armeria.xds.kubernetes"); + } +} diff --git a/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/package-info.java b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/package-info.java new file mode 100644 index 00000000000..16447cc7f2b --- /dev/null +++ b/xds-kubernetes/src/main/java/com/linecorp/armeria/xds/kubernetes/package-info.java @@ -0,0 +1,25 @@ +/* + * 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. + */ + +/** + * Kubernetes cluster type integration for xDS. + */ +@NonNullByDefault +@UnstableApi +package com.linecorp.armeria.xds.kubernetes; + +import com.linecorp.armeria.common.annotation.NonNullByDefault; +import com.linecorp.armeria.common.annotation.UnstableApi; diff --git a/xds-kubernetes/src/main/resources/META-INF/services/com.linecorp.armeria.xds.XdsExtensionFactoryProvider b/xds-kubernetes/src/main/resources/META-INF/services/com.linecorp.armeria.xds.XdsExtensionFactoryProvider new file mode 100644 index 00000000000..5f5cc9950b1 --- /dev/null +++ b/xds-kubernetes/src/main/resources/META-INF/services/com.linecorp.armeria.xds.XdsExtensionFactoryProvider @@ -0,0 +1 @@ +com.linecorp.armeria.xds.kubernetes.KubernetesClusterTypeFactoryProvider diff --git a/xds-kubernetes/src/main/resources/META-INF/services/com.linecorp.armeria.xds.XdsTypeRegistryPackageProvider b/xds-kubernetes/src/main/resources/META-INF/services/com.linecorp.armeria.xds.XdsTypeRegistryPackageProvider new file mode 100644 index 00000000000..345d8d365d7 --- /dev/null +++ b/xds-kubernetes/src/main/resources/META-INF/services/com.linecorp.armeria.xds.XdsTypeRegistryPackageProvider @@ -0,0 +1 @@ +com.linecorp.armeria.xds.kubernetes.KubernetesTypeRegistryPackageProvider diff --git a/xds-kubernetes/src/test/java/com/linecorp/armeria/xds/kubernetes/DefaultKubernetesEndpointMapperTest.java b/xds-kubernetes/src/test/java/com/linecorp/armeria/xds/kubernetes/DefaultKubernetesEndpointMapperTest.java new file mode 100644 index 00000000000..f5f6ddee170 --- /dev/null +++ b/xds-kubernetes/src/test/java/com/linecorp/armeria/xds/kubernetes/DefaultKubernetesEndpointMapperTest.java @@ -0,0 +1,108 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThat; + +import java.util.List; + +import org.junit.jupiter.api.Test; + +import com.google.common.collect.ImmutableList; + +import com.linecorp.armeria.client.Endpoint; + +import io.envoyproxy.envoy.config.core.v3.HealthStatus; +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; +import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint; +import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints; + +class DefaultKubernetesEndpointMapperTest { + + @Test + void mapsEndpointsToSingleLocality() { + final List endpoints = ImmutableList.of( + Endpoint.of("10.0.0.1", 8080), + Endpoint.of("10.0.0.2", 8080), + Endpoint.of("10.0.0.3", 9090)); + + final ClusterLoadAssignment cla = + DefaultKubernetesEndpointMapper.get().map("test-cluster", endpoints); + + assertThat(cla.getClusterName()).isEqualTo("test-cluster"); + assertThat(cla.getEndpointsList()).hasSize(1); + + final LocalityLbEndpoints locality = cla.getEndpoints(0); + assertThat(locality.getLbEndpointsList()).hasSize(3); + + final LbEndpoint ep0 = locality.getLbEndpoints(0); + assertThat(ep0.getHealthStatus()).isEqualTo(HealthStatus.HEALTHY); + assertThat(ep0.getEndpoint().getAddress().getSocketAddress().getAddress()) + .isEqualTo("10.0.0.1"); + assertThat(ep0.getEndpoint().getAddress().getSocketAddress().getPortValue()) + .isEqualTo(8080); + + final LbEndpoint ep2 = locality.getLbEndpoints(2); + assertThat(ep2.getEndpoint().getAddress().getSocketAddress().getAddress()) + .isEqualTo("10.0.0.3"); + assertThat(ep2.getEndpoint().getAddress().getSocketAddress().getPortValue()) + .isEqualTo(9090); + } + + @Test + void emptyEndpointsProducesEmptyLocality() { + final ClusterLoadAssignment cla = + DefaultKubernetesEndpointMapper.get().map("empty-cluster", ImmutableList.of()); + + assertThat(cla.getClusterName()).isEqualTo("empty-cluster"); + assertThat(cla.getEndpointsList()).hasSize(1); + assertThat(cla.getEndpoints(0).getLbEndpointsList()).isEmpty(); + } + + @Test + void deduplicatesByHostPort() { + // Simulates NODE_PORT mode where multiple pods on the same node produce + // duplicate endpoints with the same nodeIP:nodePort. + final List endpoints = ImmutableList.of( + Endpoint.of("192.168.1.1", 30000), + Endpoint.of("192.168.1.1", 30000), + Endpoint.of("192.168.1.2", 30000)); + + final ClusterLoadAssignment cla = + DefaultKubernetesEndpointMapper.get().map("dedup-cluster", endpoints); + + final LocalityLbEndpoints locality = cla.getEndpoints(0); + assertThat(locality.getLbEndpointsList()).hasSize(2); + assertThat(locality.getLbEndpoints(0).getEndpoint().getAddress() + .getSocketAddress().getAddress()).isEqualTo("192.168.1.1"); + assertThat(locality.getLbEndpoints(1).getEndpoint().getAddress() + .getSocketAddress().getAddress()).isEqualTo("192.168.1.2"); + } + + @Test + void endpointWithoutPortOmitsPortValue() { + final List endpoints = ImmutableList.of(Endpoint.of("10.0.0.1")); + + final ClusterLoadAssignment cla = + DefaultKubernetesEndpointMapper.get().map("no-port-cluster", endpoints); + + final LbEndpoint ep = cla.getEndpoints(0).getLbEndpoints(0); + assertThat(ep.getEndpoint().getAddress().getSocketAddress().getAddress()) + .isEqualTo("10.0.0.1"); + assertThat(ep.getEndpoint().getAddress().getSocketAddress().hasPortValue()).isFalse(); + } +}