Skip to content

Commit eea921b

Browse files
ikhoonikhoon
authored
Support distinct endpoints and Pod/Node metadata mapping for xDS K8s endpoints (#1326)
Motivation: The xDS control plane converts Kubernetes service endpoints into Envoy ClusterLoadAssignment resources, but had two gaps: in NODE_PORT mode multiple Pods on the same node produce duplicate LbEndpoints, and LbEndpoint.metadata was never populated, so users could not drive Envoy subset load balancing from Kubernetes Pod/Node labels or annotations. Modifications: - Add `distinct_endpoint` to `ServiceEndpointWatcher`; when true, endpoints that share the same host:port are collapsed into a single LbEndpoint. - Add `metadata_mapping` (repeated `MetadataMapping`) to `ServiceEndpointWatcher`. Each rule copies a Pod/Node label or annotation into `LbEndpoint.metadata`, selecting an exact `source_key` or a `source_key_prefix` (original keys kept), under a configurable `metadata_namespace` (default `envoy.lb`) and `metadata_key`. - Add `KubernetesEndpointConverter` centralizing endpoint-to-LbEndpoint conversion (dedup + metadata) via Armeria `KubernetesResourceAccess`. - Route both build sites (background fetching service and preview) through the converter, removing duplicated conversion code. - Validate metadata mappings on create/update, rejecting malformed rules with INVALID_ARGUMENT. - Add tests for exact/prefix metadata copy, custom namespace, annotations, endpoint dedup, and validation errors. Result: Users can deduplicate Kubernetes endpoints and carry Pod/Node labels and annotations as Envoy endpoint metadata by configuring the watcher. Both fields are optional; existing configurations are unaffected.
1 parent e1d6e2a commit eea921b

5 files changed

Lines changed: 542 additions & 39 deletions

File tree

‎xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/KubernetesEndpointConverter.java‎

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
/*
2+
* Copyright 2026 LY Corporation
3+
*
4+
* LY Corporation licenses this file to you under the Apache License,
5+
* version 2.0 (the "License"); you may not use this file except in compliance
6+
* with the License. You may obtain a copy of the License at:
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations
14+
* under the License.
15+
*/
16+
package com.linecorp.centraldogma.xds.k8s.v1;
17+
18+
import java.util.HashSet;
19+
import java.util.LinkedHashMap;
20+
import java.util.List;
21+
import java.util.Map;
22+
import java.util.Set;
23+
24+
import org.jspecify.annotations.Nullable;
25+
26+
import com.google.protobuf.Struct;
27+
import com.google.protobuf.Value;
28+
29+
import com.linecorp.armeria.client.Endpoint;
30+
import com.linecorp.armeria.client.kubernetes.endpoints.KubernetesResourceAccess;
31+
32+
import io.envoyproxy.envoy.config.core.v3.Address;
33+
import io.envoyproxy.envoy.config.core.v3.Metadata;
34+
import io.envoyproxy.envoy.config.core.v3.SocketAddress;
35+
import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint;
36+
import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints;
37+
import io.fabric8.kubernetes.api.model.Node;
38+
import io.fabric8.kubernetes.api.model.ObjectMeta;
39+
import io.fabric8.kubernetes.api.model.Pod;
40+
41+
/**
42+
* Converts Kubernetes {@link Endpoint}s into Envoy {@link LbEndpoint}s, applying the
43+
* {@link ServiceEndpointWatcher#getDistinctEndpoint() distinct_endpoint} and
44+
* {@link ServiceEndpointWatcher#getMetadataMappingList() metadata_mapping} options.
45+
*/
46+
final class KubernetesEndpointConverter {
47+
48+
private static final String DEFAULT_METADATA_NAMESPACE = "envoy.lb";
49+
50+
/**
51+
* Appends an {@link LbEndpoint} to the specified {@code builder} for each {@link Endpoint}, collapsing
52+
* endpoints that share the same host and port when {@code distinct_endpoint} is enabled and copying the
53+
* Pod/Node label/annotation values described by {@code metadata_mapping} into the endpoint metadata.
54+
*/
55+
static void addLbEndpoints(LocalityLbEndpoints.Builder builder, Iterable<Endpoint> endpoints,
56+
ServiceEndpointWatcher watcher) {
57+
final boolean distinct = watcher.getDistinctEndpoint();
58+
final List<MetadataMapping> mappings = watcher.getMetadataMappingList();
59+
final Set<String> seen = distinct ? new HashSet<>() : null;
60+
for (Endpoint endpoint : endpoints) {
61+
if (!endpoint.hasPort()) {
62+
continue;
63+
}
64+
if (seen != null && !seen.add(endpoint.host() + ':' + endpoint.port())) {
65+
continue;
66+
}
67+
final SocketAddress socketAddress = SocketAddress.newBuilder()
68+
.setAddress(endpoint.host())
69+
.setPortValue(endpoint.port())
70+
.build();
71+
final LbEndpoint.Builder lbEndpointBuilder =
72+
LbEndpoint.newBuilder()
73+
.setEndpoint(io.envoyproxy.envoy.config.endpoint.v3.Endpoint.newBuilder()
74+
.setAddress(Address.newBuilder()
75+
.setSocketAddress(socketAddress)
76+
.build())
77+
.build());
78+
final Metadata metadata = buildMetadata(endpoint, mappings);
79+
if (metadata != null) {
80+
lbEndpointBuilder.setMetadata(metadata);
81+
}
82+
builder.addLbEndpoints(lbEndpointBuilder.build());
83+
}
84+
}
85+
86+
@Nullable
87+
private static Metadata buildMetadata(Endpoint endpoint, List<MetadataMapping> mappings) {
88+
if (mappings.isEmpty()) {
89+
return null;
90+
}
91+
final Map<String, Struct.Builder> structsByNamespace = new LinkedHashMap<>();
92+
for (MetadataMapping mapping : mappings) {
93+
final ObjectMeta objectMeta = objectMeta(endpoint, mapping.getResourceType());
94+
if (objectMeta == null) {
95+
continue;
96+
}
97+
final Map<String, String> source =
98+
mapping.getEntryType() == MetadataMapping.EntryType.ANNOTATION ? objectMeta.getAnnotations()
99+
: objectMeta.getLabels();
100+
if (source == null || source.isEmpty()) {
101+
continue;
102+
}
103+
final String namespace = mapping.getMetadataNamespace().isEmpty() ? DEFAULT_METADATA_NAMESPACE
104+
: mapping.getMetadataNamespace();
105+
switch (mapping.getSourceCase()) {
106+
case SOURCE_KEY:
107+
// Exact match: the destination key is metadata_key or, if empty, the source key.
108+
final String value = source.get(mapping.getSourceKey());
109+
if (value != null) {
110+
final String key = mapping.getMetadataKey().isEmpty() ? mapping.getSourceKey()
111+
: mapping.getMetadataKey();
112+
putField(structsByNamespace, namespace, key, value);
113+
}
114+
break;
115+
case SOURCE_KEY_PREFIX:
116+
// Prefix match: copy every matching entry, preserving the original key.
117+
final String prefix = mapping.getSourceKeyPrefix();
118+
for (Map.Entry<String, String> entry : source.entrySet()) {
119+
if (entry.getKey().startsWith(prefix)) {
120+
putField(structsByNamespace, namespace, entry.getKey(), entry.getValue());
121+
}
122+
}
123+
break;
124+
default:
125+
// SOURCE_NOT_SET is rejected by validation.
126+
break;
127+
}
128+
}
129+
if (structsByNamespace.isEmpty()) {
130+
return null;
131+
}
132+
final Metadata.Builder metadataBuilder = Metadata.newBuilder();
133+
structsByNamespace.forEach((namespace, struct) ->
134+
metadataBuilder.putFilterMetadata(namespace, struct.build()));
135+
return metadataBuilder.build();
136+
}
137+
138+
private static void putField(Map<String, Struct.Builder> structsByNamespace, String namespace,
139+
String key, String value) {
140+
structsByNamespace.computeIfAbsent(namespace, unused -> Struct.newBuilder())
141+
.putFields(key, Value.newBuilder().setStringValue(value).build());
142+
}
143+
144+
@Nullable
145+
private static ObjectMeta objectMeta(Endpoint endpoint, MetadataMapping.ResourceType resourceType) {
146+
switch (resourceType) {
147+
case POD:
148+
final Pod pod = KubernetesResourceAccess.pod(endpoint);
149+
return pod != null ? pod.getMetadata() : null;
150+
case NODE:
151+
final Node node = KubernetesResourceAccess.node(endpoint);
152+
return node != null ? node.getMetadata() : null;
153+
default:
154+
return null;
155+
}
156+
}
157+
158+
private KubernetesEndpointConverter() {}
159+
}

‎xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesEndpointFetchingService.java‎

Lines changed: 3 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -61,11 +61,7 @@
6161
import com.linecorp.centraldogma.server.storage.project.Project;
6262
import com.linecorp.centraldogma.xds.internal.XdsResourceWatchingService;
6363

64-
import io.envoyproxy.envoy.config.core.v3.Address;
65-
import io.envoyproxy.envoy.config.core.v3.SocketAddress;
6664
import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
67-
import io.envoyproxy.envoy.config.endpoint.v3.Endpoint;
68-
import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint;
6965
import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints;
7066
import io.micrometer.core.instrument.MeterRegistry;
7167
import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
@@ -362,20 +358,9 @@ private static void addLocalityLbEndpoints(
362358
kubernetesLocalityLbEndpoints.getLoadBalancingWeight());
363359
}
364360
localityLbEndpointsBuilder.setPriority(kubernetesLocalityLbEndpoints.getPriority());
365-
for (com.linecorp.armeria.client.Endpoint endpoint : kubernetesEndpointGroup.endpoints()) {
366-
assert endpoint.hasPort();
367-
final SocketAddress socketAddress = SocketAddress.newBuilder()
368-
.setAddress(endpoint.host())
369-
.setPortValue(endpoint.port())
370-
.build();
371-
localityLbEndpointsBuilder.addLbEndpoints(
372-
LbEndpoint.newBuilder()
373-
.setEndpoint(Endpoint.newBuilder()
374-
.setAddress(Address.newBuilder()
375-
.setSocketAddress(socketAddress)
376-
.build()).build())
377-
.build());
378-
}
361+
KubernetesEndpointConverter.addLbEndpoints(localityLbEndpointsBuilder,
362+
kubernetesEndpointGroup.endpoints(),
363+
kubernetesLocalityLbEndpoints.getWatcher());
379364
clusterLoadAssignmentBuilder.addEndpoints(localityLbEndpointsBuilder.build());
380365
}
381366

‎xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesService.java‎

Lines changed: 40 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -55,10 +55,7 @@
5555
import com.linecorp.centraldogma.xds.internal.XdsResourceManager;
5656
import com.linecorp.centraldogma.xds.k8s.v1.XdsKubernetesServiceGrpc.XdsKubernetesServiceImplBase;
5757

58-
import io.envoyproxy.envoy.config.core.v3.Address;
59-
import io.envoyproxy.envoy.config.core.v3.SocketAddress;
6058
import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
61-
import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint;
6259
import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints;
6360
import io.fabric8.kubernetes.client.Config;
6461
import io.fabric8.kubernetes.client.ConfigBuilder;
@@ -154,6 +151,9 @@ private void validateKubernetesEndpointAndPush(
154151
StreamObserver<KubernetesEndpointAggregator> responseObserver,
155152
List<KubernetesLocalityLbEndpoints> kubernetesLocalityLbEndpointsList,
156153
String group, String fileName, Runnable onSuccess) {
154+
for (KubernetesLocalityLbEndpoints kubernetesLocalityLbEndpoints : kubernetesLocalityLbEndpointsList) {
155+
validateMetadataMappings(kubernetesLocalityLbEndpoints.getWatcher());
156+
}
157157
// Create a KubernetesEndpointGroup to check if the watcher is valid.
158158
// We use KubernetesEndpointGroup for simplicity, but we will implement a custom implementation
159159
// for better debugging and error handling in the future.
@@ -228,6 +228,41 @@ private void validateKubernetesEndpointAndPush(
228228
});
229229
}
230230

231+
private static void validateMetadataMappings(ServiceEndpointWatcher watcher) {
232+
for (MetadataMapping mapping : watcher.getMetadataMappingList()) {
233+
if (mapping.getResourceType() == MetadataMapping.ResourceType.RESOURCE_TYPE_UNSPECIFIED) {
234+
throw Status.INVALID_ARGUMENT.withDescription(
235+
"resource_type must be specified in metadata_mapping: " + mapping)
236+
.asRuntimeException();
237+
}
238+
if (mapping.getEntryType() == MetadataMapping.EntryType.ENTRY_TYPE_UNSPECIFIED) {
239+
throw Status.INVALID_ARGUMENT.withDescription(
240+
"entry_type must be specified in metadata_mapping: " + mapping)
241+
.asRuntimeException();
242+
}
243+
switch (mapping.getSourceCase()) {
244+
case SOURCE_KEY:
245+
if (mapping.getSourceKey().isEmpty()) {
246+
throw Status.INVALID_ARGUMENT.withDescription(
247+
"source_key must not be empty in metadata_mapping: " + mapping)
248+
.asRuntimeException();
249+
}
250+
break;
251+
case SOURCE_KEY_PREFIX:
252+
if (mapping.getSourceKeyPrefix().isEmpty()) {
253+
throw Status.INVALID_ARGUMENT.withDescription(
254+
"source_key_prefix must not be empty in metadata_mapping: " + mapping)
255+
.asRuntimeException();
256+
}
257+
break;
258+
default:
259+
throw Status.INVALID_ARGUMENT.withDescription(
260+
"either source_key or source_key_prefix must be set in metadata_mapping: " +
261+
mapping).asRuntimeException();
262+
}
263+
}
264+
}
265+
231266
/**
232267
* Creates a {@link KubernetesEndpointGroup} from the specified {@link ServiceEndpointWatcher}.
233268
*/
@@ -472,24 +507,8 @@ private static LocalityLbEndpoints toLocalityLbEndpoints(
472507
builder.setLoadBalancingWeight(localityLbEndpoints.getLoadBalancingWeight());
473508
}
474509
builder.setPriority(localityLbEndpoints.getPriority());
475-
for (Endpoint endpoint : endpointGroup.endpoints()) {
476-
if (!endpoint.hasPort()) {
477-
continue;
478-
}
479-
final SocketAddress socketAddress = SocketAddress.newBuilder()
480-
.setAddress(endpoint.host())
481-
.setPortValue(endpoint.port())
482-
.build();
483-
builder.addLbEndpoints(
484-
LbEndpoint.newBuilder()
485-
.setEndpoint(
486-
io.envoyproxy.envoy.config.endpoint.v3.Endpoint.newBuilder()
487-
.setAddress(Address.newBuilder()
488-
.setSocketAddress(socketAddress)
489-
.build())
490-
.build())
491-
.build());
492-
}
510+
KubernetesEndpointConverter.addLbEndpoints(builder, endpointGroup.endpoints(),
511+
localityLbEndpoints.getWatcher());
493512
return builder.build();
494513
}
495514
}

‎xds/src/main/proto/centraldogma/xds/k8s/v1/xds_kubernetes.proto‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,51 @@ message ServiceEndpointWatcher {
9292
string port_name = 2 [(google.api.field_behavior) = OPTIONAL];
9393
Kubeconfig kubeconfig = 3 [(google.api.field_behavior) = REQUIRED];
9494
map<string, string> additional_properties = 4 [(google.api.field_behavior) = OPTIONAL];
95+
// When true, endpoints that share the same host and port are collapsed into a
96+
// single LbEndpoint. This is useful in NODE_PORT mode where multiple Pods on
97+
// the same node resolve to the same nodeIP:nodePort. The first occurrence
98+
// supplies the endpoint metadata.
99+
bool distinct_endpoint = 5 [(google.api.field_behavior) = OPTIONAL];
100+
// Rules that copy Kubernetes Pod or Node label/annotation values into the
101+
// metadata of the generated LbEndpoint.
102+
repeated MetadataMapping metadata_mapping = 6 [(google.api.field_behavior) = OPTIONAL];
103+
}
104+
105+
// A rule that copies a Kubernetes Pod or Node label/annotation value into the
106+
// metadata of the generated LbEndpoint.
107+
message MetadataMapping {
108+
// The Kubernetes resource to read the value from.
109+
ResourceType resource_type = 1 [(google.api.field_behavior) = REQUIRED];
110+
// Whether the value is read from a label or an annotation.
111+
EntryType entry_type = 2 [(google.api.field_behavior) = REQUIRED];
112+
// Exactly one selector must be set.
113+
oneof source {
114+
// Copies the value of this single label/annotation key,
115+
// e.g. "topology.kubernetes.io/zone".
116+
string source_key = 3;
117+
// Copies the value of every label/annotation key that starts with this
118+
// prefix, e.g. "topology.kubernetes.io/". The original keys are preserved
119+
// as the metadata keys.
120+
string source_key_prefix = 4;
121+
}
122+
// The Envoy filter_metadata namespace to store the value under. Defaults to
123+
// "envoy.lb" when empty.
124+
string metadata_namespace = 5 [(google.api.field_behavior) = OPTIONAL];
125+
// The metadata key to store the value under. Defaults to source_key when
126+
// empty. Ignored in source_key_prefix mode where the original keys are kept.
127+
string metadata_key = 6 [(google.api.field_behavior) = OPTIONAL];
128+
129+
enum ResourceType {
130+
RESOURCE_TYPE_UNSPECIFIED = 0;
131+
POD = 1;
132+
NODE = 2;
133+
}
134+
135+
enum EntryType {
136+
ENTRY_TYPE_UNSPECIFIED = 0;
137+
LABEL = 1;
138+
ANNOTATION = 2;
139+
}
95140
}
96141

97142
message Kubeconfig {

0 commit comments

Comments
 (0)