Skip to content

xds: Add LEDS list collection support (A95) #13055

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shivaspeaks wants to merge 3 commits into
base: master
Choose a base branch
from
Open

xds: Add LEDS list collection support (A95) #13055

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
256aacd
xds: Add LEDS list collection support (A95)
shivaspeaks Sep 13, 2026
8cce8d4
use 2026
shivaspeaks Sep 13, 2026
63fed73
add tests
shivaspeaks Sep 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 46 additions & 9 deletions xds/src/main/java/io/grpc/xds/CdsLoadBalancer2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import static io.grpc.xds.XdsLbPolicies.CDS_POLICY_NAME;
import static io.grpc.xds.XdsLbPolicies.PRIORITY_POLICY_NAME;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.primitives.UnsignedInts;
import com.google.errorprone.annotations.CheckReturnValue;
Expand Down Expand Up @@ -58,6 +59,7 @@
import io.grpc.xds.XdsConfig.XdsClusterConfig.AggregateConfig;
import io.grpc.xds.XdsConfig.XdsClusterConfig.EndpointConfig;
import io.grpc.xds.XdsEndpointResource.EdsUpdate;
import io.grpc.xds.XdsLbEndpointCollectionResource.LbEndpointCollectionUpdate;
import io.grpc.xds.client.Locality;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
Expand Down Expand Up @@ -139,12 +141,13 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {

if (clusterConfig.getChildren() instanceof EndpointConfig) {
addBackendServicePickDetailsLabel = true;
StatusOr<EdsUpdate> edsUpdate = getEdsUpdate(xdsConfig, clusterName);
StatusOr<XdsClusterConfig.EndpointConfig> endpointConfig =
getEndpointConfig(xdsConfig, clusterName);
StatusOr<ClusterResolutionResult> statusOrResult = clusterState.edsUpdateToResult(
clusterName,
clusterConfig.getClusterResource(),
clusterConfig.getClusterResource().lbPolicyConfig(),
edsUpdate);
endpointConfig);
if (!statusOrResult.hasValue()) {
Status status = Status.UNAVAILABLE
.withDescription(statusOrResult.getStatus().getDescription())
Expand Down Expand Up @@ -286,7 +289,8 @@ private static long fixedPointMultiply(long a, long b) {
return (a * b) >> FIXED_POINT_FRACTIONAL_BITS;
}

private static StatusOr<EdsUpdate> getEdsUpdate(XdsConfig xdsConfig, String cluster) {
private static StatusOr<XdsClusterConfig.EndpointConfig> getEndpointConfig(
XdsConfig xdsConfig, String cluster) {
StatusOr<XdsClusterConfig> clusterConfig = xdsConfig.getClusters().get(cluster);
if (clusterConfig == null) {
return StatusOr.fromStatus(Status.INTERNAL
Expand All @@ -299,9 +303,8 @@ private static StatusOr<EdsUpdate> getEdsUpdate(XdsConfig xdsConfig, String clus
return StatusOr.fromStatus(Status.INTERNAL
.withDescription("BUG: cluster resolver cluster with children of unknown type"));
}
XdsClusterConfig.EndpointConfig endpointConfig =
(XdsClusterConfig.EndpointConfig) clusterConfig.getValue().getChildren();
return endpointConfig.getEndpoint();
return StatusOr.fromValue(
(XdsClusterConfig.EndpointConfig) clusterConfig.getValue().getChildren());
}

/**
Expand Down Expand Up @@ -332,7 +335,12 @@ StatusOr<ClusterResolutionResult> edsUpdateToResult(
String clusterName,
CdsUpdate discovery,
Object lbConfig,
StatusOr<EdsUpdate> updateOr) {
StatusOr<XdsClusterConfig.EndpointConfig> endpointConfigOr) {
if (!endpointConfigOr.hasValue()) {
return StatusOr.fromStatus(endpointConfigOr.getStatus());
}
XdsClusterConfig.EndpointConfig endpointConfig = endpointConfigOr.getValue();
StatusOr<EdsUpdate> updateOr = endpointConfig.getEndpoint();
if (!updateOr.hasValue()) {
return StatusOr.fromStatus(updateOr.getStatus());
}
Expand Down Expand Up @@ -369,6 +377,7 @@ StatusOr<ClusterResolutionResult> edsUpdateToResult(

for (Locality locality : localityLbEndpoints.keySet()) {
LocalityLbEndpoints localityLbInfo = localityLbEndpoints.get(locality);
List<LbEndpoint> localityEndpoints = resolveEndpoints(localityLbInfo, endpointConfig);
String priorityName = localityPriorityNames.get(locality);
String localityName = localityName(locality);
AddressFilter.PathChain pathChain =
Expand All @@ -382,13 +391,13 @@ StatusOr<ClusterResolutionResult> edsUpdateToResult(
long endpointWeightSum = 0;
if (pickFirstWeightedShuffling) {
localityWeightSum = priorityLocalityWeightSums.get(priorityName);
for (LbEndpoint endpoint : localityLbInfo.endpoints()) {
for (LbEndpoint endpoint : localityEndpoints) {
if (endpoint.isHealthy()) {
endpointWeightSum += UnsignedInts.toLong(endpoint.loadBalancingWeight());
}
}
}
for (LbEndpoint endpoint : localityLbInfo.endpoints()) {
for (LbEndpoint endpoint : localityEndpoints) {
if (endpoint.isHealthy()) {
discard = false;
long weight;
Expand Down Expand Up @@ -459,6 +468,34 @@ StatusOr<ClusterResolutionResult> edsUpdateToResult(
sortedPriorityNames));
}

/**
* Returns the endpoints of a locality, which are either inlined in the EDS resource or fetched
* as a separate {@code LbEndpointCollection} resource (gRFC A95). If the referenced collection
* is missing or invalid, the locality is treated as unreachable.
*/
private List<LbEndpoint> resolveEndpoints(
LocalityLbEndpoints localityLbInfo, XdsClusterConfig.EndpointConfig endpointConfig) {
if (localityLbInfo.endpointCollection() != null) {
return localityLbInfo.endpointCollection().endpoints();
}
String collectionName = localityLbInfo.lbEndpointCollectionName();
StatusOr<LbEndpointCollectionUpdate> collectionOr =
endpointConfig.getLbEndpointCollectionResources().get(collectionName);
if (collectionOr == null) {
logger.log(XdsLogLevel.INFO,
"LbEndpointCollection {0} not found; treating locality as unreachable",
collectionName);
return ImmutableList.of();
}
if (!collectionOr.hasValue()) {
logger.log(XdsLogLevel.INFO,
"LbEndpointCollection {0} unavailable ({1}); treating locality as unreachable",
collectionName, collectionOr.getStatus());
return ImmutableList.of();
}
return collectionOr.getValue().getEndpointCollection().endpoints();
}

private SocketAddress rewriteAddress(SocketAddress addr,
ImmutableMap<String, Object> endpointMetadata,
ImmutableMap<String, Object> localityMetadata) {
Expand Down
51 changes: 48 additions & 3 deletions xds/src/main/java/io/grpc/xds/Endpoints.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package io.grpc.xds;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

import com.google.auto.value.AutoValue;
import com.google.common.annotations.VisibleForTesting;
Expand All @@ -26,16 +27,41 @@
import io.grpc.EquivalentAddressGroup;
import java.net.InetSocketAddress;
import java.util.List;
import javax.annotation.Nullable;

/** Locality and endpoint level load balancing configurations. */
final class Endpoints {
private Endpoints() {}

/**
* Represents a list of endpoints, either inlined in an EDS resource or fetched separately as an
* {@code LbEndpointCollection} resource (gRFC A95).
*/
@AutoValue
abstract static class LbEndpointCollection {
abstract ImmutableList<LbEndpoint> endpoints();

static LbEndpointCollection create(List<LbEndpoint> endpoints) {
return new AutoValue_Endpoints_LbEndpointCollection(ImmutableList.copyOf(endpoints));
}
}

/** Represents a group of endpoints belong to a single locality. */
@AutoValue
abstract static class LocalityLbEndpoints {
// Endpoints to be load balanced.
abstract ImmutableList<LbEndpoint> endpoints();
/**
* The endpoints for this locality, inlined in the EDS resource. Exactly one of this and {@link
* #lbEndpointCollectionName()} is non-null.
*/
@Nullable
abstract LbEndpointCollection endpointCollection();

/**
* The name of the {@code LbEndpointCollection} resource to fetch for this locality (gRFC A95).
* Exactly one of this and {@link #endpointCollection()} is non-null.
*/
@Nullable
abstract String lbEndpointCollectionName();

// Locality's weight for inter-locality load balancing. Guaranteed to be greater than 0.
abstract int localityWeight();
Expand All @@ -45,11 +71,30 @@ abstract static class LocalityLbEndpoints {

abstract ImmutableMap<String, Object> localityMetadata();

/** Creates a locality whose endpoints are inlined in the EDS resource. */
static LocalityLbEndpoints create(List<LbEndpoint> endpoints, int localityWeight,
int priority, ImmutableMap<String, Object> localityMetadata) {
return create(
LbEndpointCollection.create(endpoints), null, localityWeight, priority, localityMetadata);
}

private static LocalityLbEndpoints create(@Nullable LbEndpointCollection endpointCollection,
@Nullable String lbEndpointCollectionName, int localityWeight, int priority,
ImmutableMap<String, Object> localityMetadata) {
checkArgument(localityWeight > 0, "localityWeight must be greater than 0");
checkArgument((endpointCollection == null) != (lbEndpointCollectionName == null),
"exactly one of endpointCollection and lbEndpointCollectionName must be set");
return new AutoValue_Endpoints_LocalityLbEndpoints(
ImmutableList.copyOf(endpoints), localityWeight, priority, localityMetadata);
endpointCollection, lbEndpointCollectionName, localityWeight, priority,
localityMetadata);
}

/** Creates a locality whose endpoints are fetched via a separate LEDS resource. */
static LocalityLbEndpoints createForCollectionName(String lbEndpointCollectionName,
int localityWeight, int priority, ImmutableMap<String, Object> localityMetadata) {
return create(
null, checkNotNull(lbEndpointCollectionName, "lbEndpointCollectionName"), localityWeight,
priority, localityMetadata);
}
}

Expand Down
34 changes: 30 additions & 4 deletions xds/src/main/java/io/grpc/xds/XdsConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import io.grpc.StatusOr;
import io.grpc.xds.XdsClusterResource.CdsUpdate;
import io.grpc.xds.XdsEndpointResource.EdsUpdate;
import io.grpc.xds.XdsLbEndpointCollectionResource.LbEndpointCollectionUpdate;
import io.grpc.xds.XdsListenerResource.LdsUpdate;
import io.grpc.xds.XdsRouteConfigureResource.RdsUpdate;
import java.io.Closeable;
Expand Down Expand Up @@ -157,35 +158,60 @@ interface ClusterChild {}
*/
static final class EndpointConfig implements ClusterChild {
private final StatusOr<EdsUpdate> endpoint;
private final ImmutableMap<String, StatusOr<LbEndpointCollectionUpdate>>
lbEndpointCollectionResources;

public EndpointConfig(StatusOr<EdsUpdate> endpoint) {
this(endpoint, ImmutableMap.of());
}

public EndpointConfig(StatusOr<EdsUpdate> endpoint,
Map<String, StatusOr<LbEndpointCollectionUpdate>> lbEndpointCollectionResources) {
this.endpoint = checkNotNull(endpoint, "endpoint");
this.lbEndpointCollectionResources = ImmutableMap.copyOf(
checkNotNull(lbEndpointCollectionResources, "lbEndpointCollectionResources"));
}

@Override
public int hashCode() {
return endpoint.hashCode();
return Objects.hash(endpoint, lbEndpointCollectionResources);
}

@Override
public boolean equals(Object obj) {
if (!(obj instanceof EndpointConfig)) {
return false;
}
return Objects.equals(endpoint, ((EndpointConfig)obj).endpoint);
EndpointConfig that = (EndpointConfig) obj;
return Objects.equals(endpoint, that.endpoint)
&& Objects.equals(lbEndpointCollectionResources, that.lbEndpointCollectionResources);
}

public StatusOr<EdsUpdate> getEndpoint() {
return endpoint;
}

/**
* The {@code LbEndpointCollection} resources referenced by the localities of the EDS
* resource, keyed by resource name (gRFC A95).
*/
public ImmutableMap<String, StatusOr<LbEndpointCollectionUpdate>>
getLbEndpointCollectionResources() {
return lbEndpointCollectionResources;
}

@Override
public String toString() {
StringBuilder sb = new StringBuilder("EndpointConfig{");
if (endpoint.hasValue()) {
return "EndpointConfig{endpoint=" + endpoint.getValue() + "}";
sb.append("endpoint=").append(endpoint.getValue());
} else {
return "EndpointConfig{error=" + endpoint.getStatus() + "}";
sb.append("error=").append(endpoint.getStatus());
}
if (!lbEndpointCollectionResources.isEmpty()) {
sb.append(", lbEndpointCollectionResources=").append(lbEndpointCollectionResources);
}
return sb.append("}").toString();
}
}

Expand Down
65 changes: 62 additions & 3 deletions xds/src/main/java/io/grpc/xds/XdsDependencyManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import io.grpc.xds.XdsClusterResource.CdsUpdate.ClusterType;
import io.grpc.xds.XdsConfig.XdsClusterConfig.AggregateConfig;
import io.grpc.xds.XdsConfig.XdsClusterConfig.EndpointConfig;
import io.grpc.xds.XdsLbEndpointCollectionResource.LbEndpointCollectionUpdate;
import io.grpc.xds.XdsRouteConfigureResource.RdsUpdate;
import io.grpc.xds.client.Locality;
import io.grpc.xds.client.XdsClient;
Expand Down Expand Up @@ -65,7 +66,7 @@
*/
final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegistry {
private enum TrackedWatcherTypeEnum {
LDS, RDS, CDS, EDS, DNS
LDS, RDS, CDS, EDS, LEDS, DNS
}

private static final TrackedWatcherType<XdsListenerResource.LdsUpdate> LDS_TYPE =
Expand All @@ -76,6 +77,8 @@ private enum TrackedWatcherTypeEnum {
new TrackedWatcherType<>(TrackedWatcherTypeEnum.CDS);
private static final TrackedWatcherType<XdsEndpointResource.EdsUpdate> EDS_TYPE =
new TrackedWatcherType<>(TrackedWatcherTypeEnum.EDS);
private static final TrackedWatcherType<LbEndpointCollectionUpdate> LEDS_TYPE =
new TrackedWatcherType<>(TrackedWatcherTypeEnum.LEDS);
private static final TrackedWatcherType<List<EquivalentAddressGroup>> DNS_TYPE =
new TrackedWatcherType<>(TrackedWatcherTypeEnum.DNS);

Expand Down Expand Up @@ -361,7 +364,8 @@ private static void addConfigForCluster(
TrackedWatcher<XdsEndpointResource.EdsUpdate> edsWatcher =
tracer.getWatcher(EDS_TYPE, cdsWatcher.getEdsServiceName());
if (edsWatcher != null) {
child = new EndpointConfig(edsWatcher.getData());
StatusOr<XdsEndpointResource.EdsUpdate> edsUpdateOr = edsWatcher.getData();
child = new EndpointConfig(edsUpdateOr, getLbEndpointCollections(edsUpdateOr, tracer));
} else {
child = new EndpointConfig(StatusOr.fromStatus(Status.INTERNAL.withDescription(
"EDS resource not found for cluster " + clusterName)));
Expand Down Expand Up @@ -390,6 +394,36 @@ private static void addConfigForCluster(
new XdsConfig.XdsClusterConfig(clusterName, cdsUpdate, child)));
}

/**
* Collects the {@code LbEndpointCollection} resources referenced by the localities of an EDS
* resource (gRFC A95). Fetching them via the tracer also marks the watchers as used, so they are
* not garbage collected.
*/
private static ImmutableMap<String, StatusOr<LbEndpointCollectionUpdate>>
getLbEndpointCollections(
StatusOr<XdsEndpointResource.EdsUpdate> edsUpdateOr, WatcherTracer tracer) {
if (!edsUpdateOr.hasValue()) {
return ImmutableMap.of();
}
Map<String, StatusOr<LbEndpointCollectionUpdate>> collections = new HashMap<>();
for (LocalityLbEndpoints locality
: edsUpdateOr.getValue().localityLbEndpointsMap.values()) {
String collectionName = locality.lbEndpointCollectionName();
if (collectionName == null || collections.containsKey(collectionName)) {
continue;
}
TrackedWatcher<LbEndpointCollectionUpdate> watcher =
tracer.getWatcher(LEDS_TYPE, collectionName);
if (watcher == null) {
collections.put(collectionName, StatusOr.fromStatus(Status.INTERNAL.withDescription(
"LbEndpointCollection resource not found: " + collectionName)));
} else {
collections.put(collectionName, watcher.getData());
}
}
return ImmutableMap.copyOf(collections);
}

private static StatusOr<XdsEndpointResource.EdsUpdate> dnsToEdsUpdate(
StatusOr<List<EquivalentAddressGroup>> dnsData, String dnsHostName) {
if (!dnsData.hasValue()) {
Expand Down Expand Up @@ -427,6 +461,14 @@ private void addEdsWatcher(String edsServiceName) {
addWatcher(EDS_TYPE, new EdsWatcher(edsServiceName));
}

private void addLedsWatcher(String collectionName) {
if (getWatchers(LEDS_TYPE).containsKey(collectionName)) {
return;
}

addWatcher(LEDS_TYPE, new LedsWatcher(collectionName));
}

private void addClusterWatcher(String clusterName) {
if (getWatchers(CDS_TYPE).containsKey(clusterName)) {
return;
Expand Down Expand Up @@ -845,7 +887,24 @@ private EdsWatcher(String resourceName) {
}

@Override
public void subscribeToChildren(XdsEndpointResource.EdsUpdate update) {}
public void subscribeToChildren(XdsEndpointResource.EdsUpdate update) {
for (LocalityLbEndpoints localityLbEndpoints : update.localityLbEndpointsMap.values()) {
String collectionName = localityLbEndpoints.lbEndpointCollectionName();
if (collectionName != null) {
addLedsWatcher(collectionName);
}
}
}
}

private class LedsWatcher extends XdsWatcherBase<LbEndpointCollectionUpdate> {
private LedsWatcher(String resourceName) {
super(XdsLbEndpointCollectionResource.getInstance(),
checkNotNull(resourceName, "resourceName"));
}

@Override
public void subscribeToChildren(LbEndpointCollectionUpdate update) {}
}

private final class DnsWatcher implements TrackedWatcher<List<EquivalentAddressGroup>> {
Expand Down
Loading
Loading