diff --git a/xds/src/main/java/io/grpc/xds/CdsLoadBalancer2.java b/xds/src/main/java/io/grpc/xds/CdsLoadBalancer2.java index d4ee46da8b4..70a9379d3bd 100644 --- a/xds/src/main/java/io/grpc/xds/CdsLoadBalancer2.java +++ b/xds/src/main/java/io/grpc/xds/CdsLoadBalancer2.java @@ -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; @@ -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; @@ -139,12 +141,13 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) { if (clusterConfig.getChildren() instanceof EndpointConfig) { addBackendServicePickDetailsLabel = true; - StatusOr edsUpdate = getEdsUpdate(xdsConfig, clusterName); + StatusOr endpointConfig = + getEndpointConfig(xdsConfig, clusterName); StatusOr statusOrResult = clusterState.edsUpdateToResult( clusterName, clusterConfig.getClusterResource(), clusterConfig.getClusterResource().lbPolicyConfig(), - edsUpdate); + endpointConfig); if (!statusOrResult.hasValue()) { Status status = Status.UNAVAILABLE .withDescription(statusOrResult.getStatus().getDescription()) @@ -286,7 +289,8 @@ private static long fixedPointMultiply(long a, long b) { return (a * b) >> FIXED_POINT_FRACTIONAL_BITS; } - private static StatusOr getEdsUpdate(XdsConfig xdsConfig, String cluster) { + private static StatusOr getEndpointConfig( + XdsConfig xdsConfig, String cluster) { StatusOr clusterConfig = xdsConfig.getClusters().get(cluster); if (clusterConfig == null) { return StatusOr.fromStatus(Status.INTERNAL @@ -299,9 +303,8 @@ private static StatusOr 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()); } /** @@ -332,7 +335,12 @@ StatusOr edsUpdateToResult( String clusterName, CdsUpdate discovery, Object lbConfig, - StatusOr updateOr) { + StatusOr endpointConfigOr) { + if (!endpointConfigOr.hasValue()) { + return StatusOr.fromStatus(endpointConfigOr.getStatus()); + } + XdsClusterConfig.EndpointConfig endpointConfig = endpointConfigOr.getValue(); + StatusOr updateOr = endpointConfig.getEndpoint(); if (!updateOr.hasValue()) { return StatusOr.fromStatus(updateOr.getStatus()); } @@ -369,6 +377,7 @@ StatusOr edsUpdateToResult( for (Locality locality : localityLbEndpoints.keySet()) { LocalityLbEndpoints localityLbInfo = localityLbEndpoints.get(locality); + List localityEndpoints = resolveEndpoints(localityLbInfo, endpointConfig); String priorityName = localityPriorityNames.get(locality); String localityName = localityName(locality); AddressFilter.PathChain pathChain = @@ -382,13 +391,13 @@ StatusOr 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; @@ -459,6 +468,34 @@ StatusOr 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 resolveEndpoints( + LocalityLbEndpoints localityLbInfo, XdsClusterConfig.EndpointConfig endpointConfig) { + if (localityLbInfo.endpointCollection() != null) { + return localityLbInfo.endpointCollection().endpoints(); + } + String collectionName = localityLbInfo.lbEndpointCollectionName(); + StatusOr 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 endpointMetadata, ImmutableMap localityMetadata) { diff --git a/xds/src/main/java/io/grpc/xds/Endpoints.java b/xds/src/main/java/io/grpc/xds/Endpoints.java index 558e3932ddc..f60afab169d 100644 --- a/xds/src/main/java/io/grpc/xds/Endpoints.java +++ b/xds/src/main/java/io/grpc/xds/Endpoints.java @@ -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; @@ -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 endpoints(); + + static LbEndpointCollection create(List 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 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(); @@ -45,11 +71,30 @@ abstract static class LocalityLbEndpoints { abstract ImmutableMap localityMetadata(); + /** Creates a locality whose endpoints are inlined in the EDS resource. */ static LocalityLbEndpoints create(List endpoints, int localityWeight, int priority, ImmutableMap 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 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 localityMetadata) { + return create( + null, checkNotNull(lbEndpointCollectionName, "lbEndpointCollectionName"), localityWeight, + priority, localityMetadata); } } diff --git a/xds/src/main/java/io/grpc/xds/XdsConfig.java b/xds/src/main/java/io/grpc/xds/XdsConfig.java index 9da5f970475..039f1f4e118 100644 --- a/xds/src/main/java/io/grpc/xds/XdsConfig.java +++ b/xds/src/main/java/io/grpc/xds/XdsConfig.java @@ -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; @@ -157,14 +158,23 @@ interface ClusterChild {} */ static final class EndpointConfig implements ClusterChild { private final StatusOr endpoint; + private final ImmutableMap> + lbEndpointCollectionResources; public EndpointConfig(StatusOr endpoint) { + this(endpoint, ImmutableMap.of()); + } + + public EndpointConfig(StatusOr endpoint, + Map> 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 @@ -172,20 +182,36 @@ 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 getEndpoint() { return endpoint; } + /** + * The {@code LbEndpointCollection} resources referenced by the localities of the EDS + * resource, keyed by resource name (gRFC A95). + */ + public ImmutableMap> + 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(); } } diff --git a/xds/src/main/java/io/grpc/xds/XdsDependencyManager.java b/xds/src/main/java/io/grpc/xds/XdsDependencyManager.java index a0af5974175..138c98c85c3 100644 --- a/xds/src/main/java/io/grpc/xds/XdsDependencyManager.java +++ b/xds/src/main/java/io/grpc/xds/XdsDependencyManager.java @@ -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; @@ -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 LDS_TYPE = @@ -76,6 +77,8 @@ private enum TrackedWatcherTypeEnum { new TrackedWatcherType<>(TrackedWatcherTypeEnum.CDS); private static final TrackedWatcherType EDS_TYPE = new TrackedWatcherType<>(TrackedWatcherTypeEnum.EDS); + private static final TrackedWatcherType LEDS_TYPE = + new TrackedWatcherType<>(TrackedWatcherTypeEnum.LEDS); private static final TrackedWatcherType> DNS_TYPE = new TrackedWatcherType<>(TrackedWatcherTypeEnum.DNS); @@ -361,7 +364,8 @@ private static void addConfigForCluster( TrackedWatcher edsWatcher = tracer.getWatcher(EDS_TYPE, cdsWatcher.getEdsServiceName()); if (edsWatcher != null) { - child = new EndpointConfig(edsWatcher.getData()); + StatusOr 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))); @@ -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> + getLbEndpointCollections( + StatusOr edsUpdateOr, WatcherTracer tracer) { + if (!edsUpdateOr.hasValue()) { + return ImmutableMap.of(); + } + Map> collections = new HashMap<>(); + for (LocalityLbEndpoints locality + : edsUpdateOr.getValue().localityLbEndpointsMap.values()) { + String collectionName = locality.lbEndpointCollectionName(); + if (collectionName == null || collections.containsKey(collectionName)) { + continue; + } + TrackedWatcher 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 dnsToEdsUpdate( StatusOr> dnsData, String dnsHostName) { if (!dnsData.hasValue()) { @@ -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; @@ -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 { + private LedsWatcher(String resourceName) { + super(XdsLbEndpointCollectionResource.getInstance(), + checkNotNull(resourceName, "resourceName")); + } + + @Override + public void subscribeToChildren(LbEndpointCollectionUpdate update) {} } private final class DnsWatcher implements TrackedWatcher> { diff --git a/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java b/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java index 9ad75595ea6..5aebffac658 100644 --- a/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java +++ b/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java @@ -30,6 +30,7 @@ import io.envoyproxy.envoy.config.core.v3.SocketAddress; import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; import io.envoyproxy.envoy.config.endpoint.v3.Endpoint; +import io.envoyproxy.envoy.config.endpoint.v3.LedsClusterLocalityConfig; import io.envoyproxy.envoy.type.v3.FractionalPercent; import io.grpc.EquivalentAddressGroup; import io.grpc.internal.GrpcUtil; @@ -37,6 +38,7 @@ import io.grpc.xds.Endpoints.LocalityLbEndpoints; import io.grpc.xds.MetadataRegistry.MetadataValueParser; import io.grpc.xds.XdsEndpointResource.EdsUpdate; +import io.grpc.xds.client.BootstrapperImpl; import io.grpc.xds.client.Locality; import io.grpc.xds.client.XdsClient.ResourceUpdate; import io.grpc.xds.client.XdsResourceType; @@ -112,6 +114,11 @@ private static boolean isEnabledXdsDualStack() { return GrpcUtil.getFlag(GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS, false); } + /** Whether gRFC A95 support (LEDS list collections) is enabled. */ + private static boolean isEnabledEndpointFallback() { + return BootstrapperImpl.enableEndpointFallback; + } + private static EdsUpdate processClusterLoadAssignment(ClusterLoadAssignment assignment) throws ResourceInvalidException { Map> priorities = new HashMap<>(); @@ -210,42 +217,76 @@ static StructOrError parseLocalityLbEndpoints( throw new ResourceInvalidException("Failed to parse Locality Endpoint metadata: " + e.getMessage(), e); } - List endpoints = new ArrayList<>(proto.getLbEndpointsCount()); - for (io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint endpoint : proto.getLbEndpointsList()) { - // The endpoint field of each lb_endpoints must be set. - // Inside of it: the address field must be set. - if (!endpoint.hasEndpoint() || !endpoint.getEndpoint().hasAddress()) { - return StructOrError.fromError("LbEndpoint with no endpoint/address"); + + // gRFC A95: if leds_cluster_locality_config is set, the lb_endpoints field is ignored and the + // endpoints are fetched separately as an LbEndpointCollection resource. + if (isEnabledEndpointFallback() && proto.hasLedsClusterLocalityConfig()) { + LedsClusterLocalityConfig ledsConfig = proto.getLedsClusterLocalityConfig(); + if (!ledsConfig.getLedsConfig().hasSelf()) { + return StructOrError.fromError( + "LedsClusterLocalityConfig with leds_config not set to self"); } - ImmutableMap endpointMetadata; - try { - endpointMetadata = registry.parseMetadata(endpoint.getMetadata()); - } catch (ResourceInvalidException e) { - throw new ResourceInvalidException("Failed to parse Endpoint metadata: " - + e.getMessage(), e); + String collectionName = ledsConfig.getLedsCollectionName(); + if (collectionName.endsWith("/*")) { + return StructOrError.fromError( + "LEDS glob collections are not supported: " + collectionName); } - List addresses = new ArrayList<>(); - addresses.add(getInetSocketAddress(endpoint.getEndpoint().getAddress())); - - if (isEnabledXdsDualStack()) { - for (Endpoint.AdditionalAddress additionalAddress - : endpoint.getEndpoint().getAdditionalAddressesList()) { - addresses.add(getInetSocketAddress(additionalAddress.getAddress())); - } + return StructOrError.fromStruct(Endpoints.LocalityLbEndpoints.createForCollectionName( + collectionName, proto.getLoadBalancingWeight().getValue(), proto.getPriority(), + localityMetadata)); + } + + List endpoints = new ArrayList<>(proto.getLbEndpointsCount()); + for (io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint endpoint : proto.getLbEndpointsList()) { + StructOrError endpointOrError = parseLbEndpoint(endpoint); + if (endpointOrError.getErrorDetail() != null) { + return StructOrError.fromError(endpointOrError.getErrorDetail()); } - boolean isHealthy = (endpoint.getHealthStatus() == HealthStatus.HEALTHY) - || (endpoint.getHealthStatus() == HealthStatus.UNKNOWN); - endpoints.add(Endpoints.LbEndpoint.create( - new EquivalentAddressGroup(addresses), - endpoint.getLoadBalancingWeight().getValue(), isHealthy, - endpoint.getEndpoint().getHostname(), - endpointMetadata)); + endpoints.add(endpointOrError.getStruct()); } return StructOrError.fromStruct(Endpoints.LocalityLbEndpoints.create( endpoints, proto.getLoadBalancingWeight().getValue(), proto.getPriority(), localityMetadata)); } + /** + * Parses a single {@code LbEndpoint} message. Used both for endpoints inlined in an EDS resource + * and for the entries of an {@code LbEndpointCollection} resource (gRFC A95), which share the + * same validation rules. + */ + static StructOrError parseLbEndpoint( + io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint endpoint) + throws ResourceInvalidException { + // The endpoint field of each lb_endpoints must be set. + // Inside of it: the address field must be set. + if (!endpoint.hasEndpoint() || !endpoint.getEndpoint().hasAddress()) { + return StructOrError.fromError("LbEndpoint with no endpoint/address"); + } + ImmutableMap endpointMetadata; + try { + endpointMetadata = MetadataRegistry.getInstance().parseMetadata(endpoint.getMetadata()); + } catch (ResourceInvalidException e) { + throw new ResourceInvalidException("Failed to parse Endpoint metadata: " + + e.getMessage(), e); + } + List addresses = new ArrayList<>(); + addresses.add(getInetSocketAddress(endpoint.getEndpoint().getAddress())); + + if (isEnabledXdsDualStack()) { + for (Endpoint.AdditionalAddress additionalAddress + : endpoint.getEndpoint().getAdditionalAddressesList()) { + addresses.add(getInetSocketAddress(additionalAddress.getAddress())); + } + } + boolean isHealthy = (endpoint.getHealthStatus() == HealthStatus.HEALTHY) + || (endpoint.getHealthStatus() == HealthStatus.UNKNOWN); + return StructOrError.fromStruct(Endpoints.LbEndpoint.create( + new EquivalentAddressGroup(addresses), + endpoint.getLoadBalancingWeight().getValue(), isHealthy, + endpoint.getEndpoint().getHostname(), + endpointMetadata)); + } + private static InetSocketAddress getInetSocketAddress(Address address) throws ResourceInvalidException { io.envoyproxy.envoy.config.core.v3.SocketAddress socketAddress = address.getSocketAddress(); diff --git a/xds/src/main/java/io/grpc/xds/XdsLbEndpointCollectionResource.java b/xds/src/main/java/io/grpc/xds/XdsLbEndpointCollectionResource.java new file mode 100644 index 00000000000..7a4a742e973 --- /dev/null +++ b/xds/src/main/java/io/grpc/xds/XdsLbEndpointCollectionResource.java @@ -0,0 +1,148 @@ +/* + * Copyright 2026 The gRPC Authors + * + * Licensed 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 + * + * http://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 io.grpc.xds; + +import static com.google.common.base.Preconditions.checkNotNull; + +import com.github.xds.core.v3.CollectionEntry; +import com.google.common.base.MoreObjects; +import com.google.protobuf.InvalidProtocolBufferException; +import com.google.protobuf.Message; +import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint; +import io.envoyproxy.envoy.config.endpoint.v3.LbEndpointCollection; +import io.grpc.xds.XdsLbEndpointCollectionResource.LbEndpointCollectionUpdate; +import io.grpc.xds.client.XdsClient.ResourceUpdate; +import io.grpc.xds.client.XdsResourceType; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; + +/** + * The {@code LbEndpointCollection} (LEDS) resource type, as described in gRFC A95. An + * {@code LbEndpointCollection} holds the list of endpoints for a single locality of an EDS + * resource, allowing that list to be fetched separately from the EDS resource itself. + */ +final class XdsLbEndpointCollectionResource extends XdsResourceType { + static final String ADS_TYPE_URL_LEDS = + "type.googleapis.com/envoy.config.endpoint.v3.LbEndpointCollection"; + + private static final String TYPE_URL_LB_ENDPOINT = + "type.googleapis.com/envoy.config.endpoint.v3.LbEndpoint"; + + private static final XdsLbEndpointCollectionResource instance = + new XdsLbEndpointCollectionResource(); + + static XdsLbEndpointCollectionResource getInstance() { + return instance; + } + + @Override + public String typeName() { + return "LEDS"; + } + + @Override + public String typeUrl() { + return ADS_TYPE_URL_LEDS; + } + + @Override + public boolean shouldRetrieveResourceKeysForArgs() { + return true; + } + + @Override + protected boolean isFullStateOfTheWorld() { + return false; + } + + @Override + protected Class unpackedClassName() { + return LbEndpointCollection.class; + } + + @Override + protected LbEndpointCollectionUpdate doParse(Args args, Message unpackedMessage) + throws ResourceInvalidException { + if (!(unpackedMessage instanceof LbEndpointCollection)) { + throw new ResourceInvalidException("Invalid message type: " + unpackedMessage.getClass()); + } + return processLbEndpointCollection((LbEndpointCollection) unpackedMessage); + } + + private static LbEndpointCollectionUpdate processLbEndpointCollection( + LbEndpointCollection collection) throws ResourceInvalidException { + // An empty entries field is valid; the locality is then considered unreachable. + List endpoints = new ArrayList<>(collection.getEntriesCount()); + for (CollectionEntry entry : collection.getEntriesList()) { + if (!entry.hasInlineEntry()) { + throw new ResourceInvalidException("CollectionEntry with no inline_entry"); + } + LbEndpoint lbEndpointProto; + try { + lbEndpointProto = unpackCompatibleType( + entry.getInlineEntry().getResource(), LbEndpoint.class, TYPE_URL_LB_ENDPOINT, null); + } catch (InvalidProtocolBufferException e) { + throw new ResourceInvalidException("Can't decode LbEndpoint: " + e.getMessage(), e); + } + StructOrError endpointOrError = + XdsEndpointResource.parseLbEndpoint(lbEndpointProto); + if (endpointOrError.getErrorDetail() != null) { + throw new ResourceInvalidException(endpointOrError.getErrorDetail()); + } + endpoints.add(endpointOrError.getStruct()); + } + return new LbEndpointCollectionUpdate(Endpoints.LbEndpointCollection.create(endpoints)); + } + + /** The parsed representation of an {@code LbEndpointCollection} resource. */ + static final class LbEndpointCollectionUpdate implements ResourceUpdate { + private final Endpoints.LbEndpointCollection endpointCollection; + + LbEndpointCollectionUpdate(Endpoints.LbEndpointCollection endpointCollection) { + this.endpointCollection = checkNotNull(endpointCollection, "endpointCollection"); + } + + Endpoints.LbEndpointCollection getEndpointCollection() { + return endpointCollection; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + LbEndpointCollectionUpdate that = (LbEndpointCollectionUpdate) o; + return Objects.equals(endpointCollection, that.endpointCollection); + } + + @Override + public int hashCode() { + return Objects.hash(endpointCollection); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("endpointCollection", endpointCollection) + .toString(); + } + } +} diff --git a/xds/src/main/java/io/grpc/xds/client/BootstrapperImpl.java b/xds/src/main/java/io/grpc/xds/client/BootstrapperImpl.java index 978966830be..add20a0d672 100644 --- a/xds/src/main/java/io/grpc/xds/client/BootstrapperImpl.java +++ b/xds/src/main/java/io/grpc/xds/client/BootstrapperImpl.java @@ -56,6 +56,9 @@ public abstract class BootstrapperImpl extends Bootstrapper { "envoy.lb.does_not_support_overprovisioning"; @VisibleForTesting public static final String CLIENT_FEATURE_RESOURCE_IN_SOTW = "xds.config.resource-in-sotw"; + @VisibleForTesting + public static final String CLIENT_FEATURE_LB_ENDPOINT_COLLECTION = + "xds.endpoint.supports_lb_endpoint_collection"; // Server features. private static final String SERVER_FEATURE_IGNORE_RESOURCE_DELETION = "ignore_resource_deletion"; @@ -176,6 +179,9 @@ protected BootstrapInfo.Builder bootstrapBuilder(Map rawData) nodeBuilder.setUserAgentVersion(buildVersion.getImplementationVersion()); nodeBuilder.addClientFeatures(CLIENT_FEATURE_DISABLE_OVERPROVISIONING); nodeBuilder.addClientFeatures(CLIENT_FEATURE_RESOURCE_IN_SOTW); + if (enableEndpointFallback) { + nodeBuilder.addClientFeatures(CLIENT_FEATURE_LB_ENDPOINT_COLLECTION); + } builder.node(nodeBuilder.build()); Map certProvidersBlob = JsonUtil.getObject(rawData, "certificate_providers"); diff --git a/xds/src/test/java/io/grpc/xds/CdsLoadBalancer2Test.java b/xds/src/test/java/io/grpc/xds/CdsLoadBalancer2Test.java index 515cae81521..4ca96ff5e07 100644 --- a/xds/src/test/java/io/grpc/xds/CdsLoadBalancer2Test.java +++ b/xds/src/test/java/io/grpc/xds/CdsLoadBalancer2Test.java @@ -22,6 +22,7 @@ import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_CDS; import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_EDS; import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_LDS; +import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_LEDS; import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_RDS; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; @@ -31,6 +32,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.github.xds.core.v3.CollectionEntry; import com.github.xds.type.v3.TypedStruct; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Iterables; @@ -54,12 +56,15 @@ import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; import io.envoyproxy.envoy.config.endpoint.v3.Endpoint; import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint; +import io.envoyproxy.envoy.config.endpoint.v3.LbEndpointCollection; +import io.envoyproxy.envoy.config.endpoint.v3.LedsClusterLocalityConfig; import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints; import io.envoyproxy.envoy.extensions.clusters.aggregate.v3.ClusterConfig; import io.envoyproxy.envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext; import io.grpc.Attributes; import io.grpc.ChannelLogger; import io.grpc.ConnectivityState; +import io.grpc.EquivalentAddressGroup; import io.grpc.LoadBalancer; import io.grpc.LoadBalancer.Helper; import io.grpc.LoadBalancer.PickDetailsConsumer; @@ -73,6 +78,7 @@ import io.grpc.NameResolverRegistry; import io.grpc.Status; import io.grpc.Status.Code; +import io.grpc.StatusOr; import io.grpc.SynchronizationContext; import io.grpc.inprocess.InProcessChannelBuilder; import io.grpc.inprocess.InProcessServerBuilder; @@ -81,8 +87,10 @@ import io.grpc.util.GracefulSwitchLoadBalancerAccessor; import io.grpc.xds.CdsLoadBalancerProvider.CdsConfig; import io.grpc.xds.ClusterImplLoadBalancerProvider.ClusterImplConfig; +import io.grpc.xds.client.BootstrapperImpl; import io.grpc.xds.client.XdsClient; import io.grpc.xds.internal.security.CommonTlsContextTestsUtil; +import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -114,6 +122,7 @@ public class CdsLoadBalancer2Test { private static final String CLUSTER = "cluster-foo.googleapis.com"; private static final String EDS_SERVICE_NAME = "backend-service-1.googleapis.com"; private static final String NODE_ID = "node-id"; + private static final String LEDS_NAME = "leds-collection-1"; private final io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext upstreamTlsContext = CommonTlsContextTestsUtil.buildUpstreamTlsContext("cert-instance-name", true); private static final Cluster EDS_CLUSTER = Cluster.newBuilder() @@ -145,6 +154,7 @@ public class CdsLoadBalancer2Test { private ArgumentCaptor pickerCaptor; private CdsLoadBalancer2 loadBalancer; private XdsConfig lastXdsConfig; + private boolean savedEnableEndpointFallback; @Before public void setUp() throws Exception { @@ -199,10 +209,13 @@ public void setUp() throws Exception { controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, ImmutableMap.of( EDS_SERVICE_NAME, ControlPlaneRule.buildClusterLoadAssignment( "127.0.0.1", "", 1234, EDS_SERVICE_NAME))); + + savedEnableEndpointFallback = BootstrapperImpl.enableEndpointFallback; } @After public void tearDown() { + BootstrapperImpl.enableEndpointFallback = savedEnableEndpointFallback; if (loadBalancer != null) { shutdownLoadBalancer(); } @@ -252,6 +265,129 @@ public void discoverTopLevelCluster() { assertThat(childBalancer.name).isEqualTo(PRIORITY_POLICY_NAME); } + @Test + public void edsCluster_ledsCollection_resolvesEndpoints() { + BootstrapperImpl.enableEndpointFallback = true; + controlPlaneService.setXdsConfig(ADS_TYPE_URL_CDS, ImmutableMap.of(CLUSTER, EDS_CLUSTER)); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, + ImmutableMap.of(EDS_SERVICE_NAME, edsWithLeds(LEDS_NAME))); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_LEDS, ImmutableMap.of( + LEDS_NAME, LbEndpointCollection.newBuilder() + .addEntries(inlineLbEndpoint("127.0.0.5", 1234)) + .build())); + + startXdsDepManager(); + + verify(helper, never()).updateBalancingState(eq(ConnectivityState.TRANSIENT_FAILURE), any()); + FakeLoadBalancer childBalancer = Iterables.getOnlyElement(childBalancers); + assertThat(childBalancer.addresses).hasSize(1); + assertThat(Iterables.getOnlyElement(childBalancer.addresses).getAddresses()) + .containsExactly(new InetSocketAddress("127.0.0.5", 1234)); + } + + @Test + public void edsCluster_ledsCollectionEmpty_localityUnreachable() { + BootstrapperImpl.enableEndpointFallback = true; + controlPlaneService.setXdsConfig(ADS_TYPE_URL_CDS, ImmutableMap.of(CLUSTER, EDS_CLUSTER)); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, + ImmutableMap.of(EDS_SERVICE_NAME, edsWithLeds(LEDS_NAME))); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_LEDS, + ImmutableMap.of(LEDS_NAME, LbEndpointCollection.getDefaultInstance())); + + startXdsDepManager(); + + verify(helper).updateBalancingState( + eq(ConnectivityState.TRANSIENT_FAILURE), pickerCaptor.capture()); + assertPickerStatus(pickerCaptor.getValue(), Status.UNAVAILABLE + .withDescription("No usable endpoint from cluster: " + CLUSTER)); + assertThat(childBalancers).isEmpty(); + } + + @Test + public void edsCluster_ledsCollectionNotFound_localityUnreachable() { + BootstrapperImpl.enableEndpointFallback = true; + controlPlaneService.setXdsConfig(ADS_TYPE_URL_CDS, ImmutableMap.of(CLUSTER, EDS_CLUSTER)); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, + ImmutableMap.of(EDS_SERVICE_NAME, edsWithLeds(LEDS_NAME))); + // The control plane never sends the referenced LbEndpointCollection resource. + + startXdsDepManager(); + + verify(helper).updateBalancingState( + eq(ConnectivityState.TRANSIENT_FAILURE), pickerCaptor.capture()); + assertPickerStatus(pickerCaptor.getValue(), Status.UNAVAILABLE + .withDescription("No usable endpoint from cluster: " + CLUSTER)); + assertThat(childBalancers).isEmpty(); + } + + /** + * XdsDependencyManager always puts an entry in the collection map for every collection a locality + * refers to, but the locality must still be treated as unreachable if that ever stops holding. + */ + @Test + public void edsCluster_ledsCollectionAbsentFromConfig_localityUnreachable() { + BootstrapperImpl.enableEndpointFallback = true; + controlPlaneService.setXdsConfig(ADS_TYPE_URL_CDS, ImmutableMap.of(CLUSTER, EDS_CLUSTER)); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, + ImmutableMap.of(EDS_SERVICE_NAME, edsWithLeds(LEDS_NAME))); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_LEDS, ImmutableMap.of( + LEDS_NAME, LbEndpointCollection.newBuilder() + .addEntries(inlineLbEndpoint("127.0.0.5", 1234)) + .build())); + startXdsDepManager(); + + // Re-deliver the published config with the collection dropped from the endpoint config. + XdsConfig.XdsClusterConfig clusterConfig = lastXdsConfig.getClusters().get(CLUSTER).getValue(); + XdsConfig.XdsClusterConfig.EndpointConfig endpointConfig = + (XdsConfig.XdsClusterConfig.EndpointConfig) clusterConfig.getChildren(); + XdsConfig strippedConfig = new XdsConfig.XdsConfigBuilder() + .setListener(lastXdsConfig.getListener()) + .setRoute(lastXdsConfig.getRoute()) + .setVirtualHost(lastXdsConfig.getVirtualHost()) + .addCluster(CLUSTER, StatusOr.fromValue(new XdsConfig.XdsClusterConfig( + CLUSTER, clusterConfig.getClusterResource(), + new XdsConfig.XdsClusterConfig.EndpointConfig(endpointConfig.getEndpoint())))) + .build(); + + Status status = loadBalancer.acceptResolvedAddresses(ResolvedAddresses.newBuilder() + .setAddresses(Collections.emptyList()) + .setAttributes(Attributes.newBuilder() + .set(XdsAttributes.XDS_CONFIG, strippedConfig) + .set(XdsAttributes.XDS_CLUSTER_SUBSCRIPT_REGISTRY, xdsDepManager) + .build()) + .setLoadBalancingPolicyConfig(new CdsConfig(CLUSTER)) + .build()); + + assertThat(status.getCode()).isEqualTo(Code.UNAVAILABLE); + assertThat(status.getDescription()).isEqualTo("No usable endpoint from cluster: " + CLUSTER); + } + + private static ClusterLoadAssignment edsWithLeds(String collectionName) { + return ClusterLoadAssignment.newBuilder() + .setClusterName(EDS_SERVICE_NAME) + .addEndpoints(LocalityLbEndpoints.newBuilder() + .setLoadBalancingWeight(UInt32Value.of(10)) + .setPriority(0) + .setLedsClusterLocalityConfig(LedsClusterLocalityConfig.newBuilder() + .setLedsConfig(ConfigSource.newBuilder() + .setSelf(SelfConfigSource.getDefaultInstance())) + .setLedsCollectionName(collectionName))) + .build(); + } + + private static CollectionEntry inlineLbEndpoint(String address, int port) { + return CollectionEntry.newBuilder() + .setInlineEntry(CollectionEntry.InlineEntry.newBuilder() + .setResource(Any.pack(LbEndpoint.newBuilder() + .setEndpoint(Endpoint.newBuilder() + .setAddress(Address.newBuilder() + .setSocketAddress(SocketAddress.newBuilder() + .setAddress(address) + .setPortValue(port)))) + .build()))) + .build(); + } + @Test public void nonAggregateCluster_resourceNotExist_returnErrorPicker() { startXdsDepManager(); @@ -839,6 +975,7 @@ private final class FakeLoadBalancer extends LoadBalancer { private final Helper helper; private Object config; private Attributes attributes; + private List addresses; private Status upstreamError; private boolean shutdown; @@ -851,6 +988,7 @@ private final class FakeLoadBalancer extends LoadBalancer { public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) { config = resolvedAddresses.getLoadBalancingPolicyConfig(); attributes = resolvedAddresses.getAttributes(); + addresses = resolvedAddresses.getAddresses(); return Status.OK; } diff --git a/xds/src/test/java/io/grpc/xds/GrpcBootstrapperImplTest.java b/xds/src/test/java/io/grpc/xds/GrpcBootstrapperImplTest.java index 221161b7957..4526df49375 100644 --- a/xds/src/test/java/io/grpc/xds/GrpcBootstrapperImplTest.java +++ b/xds/src/test/java/io/grpc/xds/GrpcBootstrapperImplTest.java @@ -1048,6 +1048,28 @@ public void authorityInfo_defaultsFallbackOnReachabilityOnlyToFalse() { assertThat(authorityInfo.fallbackOnReachabilityOnly()).isFalse(); } + @Test + public void lbEndpointCollectionClientFeature_advertisedWhenEnvVarEnabled() throws Exception { + CommonBootstrapperTestUtils.setEnableEndpointFallback(true); + bootstrapper.setFileReader( + createFileReader(BOOTSTRAP_FILE_PATH, buildAuthorityBootstrap(false))); + BootstrapInfo info = bootstrapper.bootstrap(); + assertThat(info.node()).isEqualTo(getNodeBuilder() + .addClientFeatures(GrpcBootstrapperImpl.CLIENT_FEATURE_LB_ENDPOINT_COLLECTION) + .build()); + } + + @Test + public void lbEndpointCollectionClientFeature_notAdvertisedWhenEnvVarDisabled() + throws Exception { + CommonBootstrapperTestUtils.setEnableEndpointFallback(false); + bootstrapper.setFileReader( + createFileReader(BOOTSTRAP_FILE_PATH, buildAuthorityBootstrap(false))); + BootstrapInfo info = bootstrapper.bootstrap(); + assertThat(info.node()).isEqualTo(getNodeBuilder().build()); + } + + private static String buildAuthorityBootstrap(boolean fallbackOnReachabilityOnly) { return "{\n" + " \"authorities\": {\n" diff --git a/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java b/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java index dc9590aed51..558e8e39b49 100644 --- a/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java +++ b/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java @@ -61,6 +61,7 @@ import io.envoyproxy.envoy.config.core.v3.TransportSocket; import io.envoyproxy.envoy.config.core.v3.TypedExtensionConfig; import io.envoyproxy.envoy.config.endpoint.v3.Endpoint; +import io.envoyproxy.envoy.config.endpoint.v3.LedsClusterLocalityConfig; import io.envoyproxy.envoy.config.listener.v3.Filter; import io.envoyproxy.envoy.config.listener.v3.FilterChain; import io.envoyproxy.envoy.config.listener.v3.FilterChainMatch; @@ -141,6 +142,7 @@ import io.grpc.xds.XdsClusterResource.CdsUpdate; import io.grpc.xds.client.BackendMetricPropagation; import io.grpc.xds.client.Bootstrapper.ServerInfo; +import io.grpc.xds.client.BootstrapperImpl; import io.grpc.xds.client.LoadStatsManager2; import io.grpc.xds.client.XdsClient; import io.grpc.xds.client.XdsResourceType; @@ -177,12 +179,14 @@ public class GrpcXdsClientImplDataTest { private boolean originalEnableRouteLookup; private boolean originalEnableLeastRequest; private boolean originalEnableUseSystemRootCerts; + private boolean originalEnableEndpointFallback; @Before public void setUp() { originalEnableRouteLookup = XdsRouteConfigureResource.enableRouteLookup; originalEnableLeastRequest = XdsClusterResource.enableLeastRequest; originalEnableUseSystemRootCerts = XdsClusterResource.enableSystemRootCerts; + originalEnableEndpointFallback = BootstrapperImpl.enableEndpointFallback; } @After @@ -190,6 +194,7 @@ public void tearDown() { XdsRouteConfigureResource.enableRouteLookup = originalEnableRouteLookup; XdsClusterResource.enableLeastRequest = originalEnableLeastRequest; XdsClusterResource.enableSystemRootCerts = originalEnableUseSystemRootCerts; + BootstrapperImpl.enableEndpointFallback = originalEnableEndpointFallback; } @Test @@ -1249,6 +1254,116 @@ public void parseLocalityLbEndpoints_invalidPriority() throws ResourceInvalidExc assertThat(struct.getErrorDetail()).isEqualTo("negative priority"); } + @Test + public void parseLocalityLbEndpoints_endpointWithoutAddress() throws ResourceInvalidException { + io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints proto = + io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints.newBuilder() + .setLocality(Locality.newBuilder() + .setRegion("region-foo").setZone("zone-foo").setSubZone("subZone-foo")) + .setLoadBalancingWeight(UInt32Value.newBuilder().setValue(100)) // locality weight + .setPriority(0) + .addLbEndpoints(io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint.getDefaultInstance()) + .build(); + StructOrError struct = XdsEndpointResource.parseLocalityLbEndpoints(proto); + assertThat(struct.getErrorDetail()).isEqualTo("LbEndpoint with no endpoint/address"); + } + + @Test + public void parseLocalityLbEndpoints_ledsClusterLocalityConfig() + throws ResourceInvalidException { + BootstrapperImpl.enableEndpointFallback = true; + io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints proto = + localityLbEndpointsWithLeds("collection-foo").build(); + StructOrError struct = XdsEndpointResource.parseLocalityLbEndpoints(proto); + assertThat(struct.getErrorDetail()).isNull(); + assertThat(struct.getStruct()).isEqualTo( + LocalityLbEndpoints.createForCollectionName( + "collection-foo", 100, 1, ImmutableMap.of())); + } + + @Test + public void parseLocalityLbEndpoints_ledsClusterLocalityConfig_ignoresLbEndpoints() + throws ResourceInvalidException { + BootstrapperImpl.enableEndpointFallback = true; + io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints proto = + localityLbEndpointsWithLeds("collection-foo") + .addLbEndpoints(io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint.newBuilder() + .setEndpoint(Endpoint.newBuilder() + .setAddress(Address.newBuilder() + .setSocketAddress( + SocketAddress.newBuilder() + .setAddress("172.14.14.5").setPortValue(8888))))) + .build(); + StructOrError struct = XdsEndpointResource.parseLocalityLbEndpoints(proto); + assertThat(struct.getErrorDetail()).isNull(); + assertThat(struct.getStruct().endpointCollection()).isNull(); + assertThat(struct.getStruct().lbEndpointCollectionName()).isEqualTo("collection-foo"); + } + + @Test + public void parseLocalityLbEndpoints_ledsConfigNotSelf() throws ResourceInvalidException { + BootstrapperImpl.enableEndpointFallback = true; + io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints proto = + io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints.newBuilder() + .setLocality(Locality.newBuilder() + .setRegion("region-foo").setZone("zone-foo").setSubZone("subZone-foo")) + .setLoadBalancingWeight(UInt32Value.newBuilder().setValue(100)) + .setPriority(1) + .setLedsClusterLocalityConfig(LedsClusterLocalityConfig.newBuilder() + .setLedsConfig(ConfigSource.getDefaultInstance()) + .setLedsCollectionName("collection-foo")) + .build(); + StructOrError struct = XdsEndpointResource.parseLocalityLbEndpoints(proto); + assertThat(struct.getErrorDetail()) + .isEqualTo("LedsClusterLocalityConfig with leds_config not set to self"); + } + + @Test + public void parseLocalityLbEndpoints_ledsGlobCollectionUnsupported() + throws ResourceInvalidException { + BootstrapperImpl.enableEndpointFallback = true; + io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints proto = + localityLbEndpointsWithLeds("xdstp://server/collection/*").build(); + StructOrError struct = XdsEndpointResource.parseLocalityLbEndpoints(proto); + assertThat(struct.getErrorDetail()) + .isEqualTo("LEDS glob collections are not supported: xdstp://server/collection/*"); + } + + @Test + public void parseLocalityLbEndpoints_ledsIgnoredWhenEnvVarDisabled() + throws ResourceInvalidException { + BootstrapperImpl.enableEndpointFallback = false; + io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints proto = + localityLbEndpointsWithLeds("collection-foo") + .addLbEndpoints(io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint.newBuilder() + .setEndpoint(Endpoint.newBuilder() + .setAddress(Address.newBuilder() + .setSocketAddress( + SocketAddress.newBuilder() + .setAddress("172.14.14.5").setPortValue(8888)))) + .setLoadBalancingWeight(UInt32Value.newBuilder().setValue(20))) + .build(); + StructOrError struct = XdsEndpointResource.parseLocalityLbEndpoints(proto); + assertThat(struct.getErrorDetail()).isNull(); + assertThat(struct.getStruct()).isEqualTo( + LocalityLbEndpoints.create( + Collections.singletonList(LbEndpoint.create("172.14.14.5", 8888, + 20, true, "", ImmutableMap.of())), + 100, 1, ImmutableMap.of())); + } + + private static io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints.Builder + localityLbEndpointsWithLeds(String collectionName) { + return io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints.newBuilder() + .setLocality(Locality.newBuilder() + .setRegion("region-foo").setZone("zone-foo").setSubZone("subZone-foo")) + .setLoadBalancingWeight(UInt32Value.newBuilder().setValue(100)) + .setPriority(1) + .setLedsClusterLocalityConfig(LedsClusterLocalityConfig.newBuilder() + .setLedsConfig(ConfigSource.newBuilder().setSelf(SelfConfigSource.getDefaultInstance())) + .setLedsCollectionName(collectionName)); + } + @Test public void parseHttpFilter_unsupportedButOptional() { HttpFilter httpFilter = HttpFilter.newBuilder() diff --git a/xds/src/test/java/io/grpc/xds/XdsDependencyManagerTest.java b/xds/src/test/java/io/grpc/xds/XdsDependencyManagerTest.java index 522eb29c001..c9ae196697f 100644 --- a/xds/src/test/java/io/grpc/xds/XdsDependencyManagerTest.java +++ b/xds/src/test/java/io/grpc/xds/XdsDependencyManagerTest.java @@ -22,8 +22,10 @@ import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_CDS; import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_EDS; import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_LDS; +import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_LEDS; import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_RDS; import static io.grpc.xds.XdsTestUtils.CLUSTER_NAME; +import static io.grpc.xds.XdsTestUtils.EDS_NAME; import static io.grpc.xds.XdsTestUtils.ENDPOINT_HOSTNAME; import static io.grpc.xds.XdsTestUtils.ENDPOINT_PORT; import static io.grpc.xds.XdsTestUtils.RDS_NAME; @@ -37,16 +39,23 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import com.github.xds.core.v3.CollectionEntry; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.google.common.util.concurrent.MoreExecutors; +import com.google.protobuf.Any; import com.google.protobuf.Message; +import com.google.protobuf.UInt32Value; import io.envoyproxy.envoy.config.cluster.v3.Cluster; import io.envoyproxy.envoy.config.core.v3.Address; +import io.envoyproxy.envoy.config.core.v3.ConfigSource; +import io.envoyproxy.envoy.config.core.v3.SelfConfigSource; import io.envoyproxy.envoy.config.core.v3.SocketAddress; import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; import io.envoyproxy.envoy.config.endpoint.v3.Endpoint; import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint; +import io.envoyproxy.envoy.config.endpoint.v3.LbEndpointCollection; +import io.envoyproxy.envoy.config.endpoint.v3.LedsClusterLocalityConfig; import io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints; import io.envoyproxy.envoy.config.listener.v3.Listener; import io.envoyproxy.envoy.config.route.v3.RouteConfiguration; @@ -68,6 +77,8 @@ import io.grpc.xds.XdsClusterResource.CdsUpdate; import io.grpc.xds.XdsConfig.XdsClusterConfig; import io.grpc.xds.XdsEndpointResource.EdsUpdate; +import io.grpc.xds.XdsLbEndpointCollectionResource.LbEndpointCollectionUpdate; +import io.grpc.xds.client.BootstrapperImpl; import io.grpc.xds.client.Locality; import io.grpc.xds.client.XdsClient; import io.grpc.xds.client.XdsClient.ResourceMetadata; @@ -122,6 +133,8 @@ public class XdsDependencyManagerTest { private TestWatcher xdsConfigWatcher; + private static final String LEDS_NAME = "leds-collection-0"; + private final String serverName = "the-service-name"; private final Queue loadReportCalls = new ArrayDeque<>(); private final AtomicBoolean adsEnded = new AtomicBoolean(true); @@ -153,6 +166,7 @@ public class XdsDependencyManagerTest { private XdsDependencyManager xdsDependencyManager = new XdsDependencyManager( xdsClient, syncContext, serverName, serverName, nameResolverArgs); private boolean savedEnableLogicalDns; + private boolean savedEnableEndpointFallback; @Before public void setUp() throws Exception { @@ -171,6 +185,7 @@ public void setUp() throws Exception { defaultXdsConfig = XdsTestUtils.getDefaultXdsConfig(serverName); savedEnableLogicalDns = XdsDependencyManager.enableLogicalDns; + savedEnableEndpointFallback = BootstrapperImpl.enableEndpointFallback; } @After @@ -185,6 +200,7 @@ public void tearDown() throws InterruptedException { assertThat(fakeClock.getPendingTasks()).isEmpty(); XdsDependencyManager.enableLogicalDns = savedEnableLogicalDns; + BootstrapperImpl.enableEndpointFallback = savedEnableEndpointFallback; } @Test @@ -260,6 +276,151 @@ private static StatusOr getEndpoint(StatusOr childC return endpoint; } + private static XdsClusterConfig.EndpointConfig getEndpointConfig( + StatusOr childConfigOr) { + XdsClusterConfig.ClusterChild clusterChild = childConfigOr.getValue().getChildren(); + assertThat(clusterChild).isInstanceOf(XdsClusterConfig.EndpointConfig.class); + return (XdsClusterConfig.EndpointConfig) clusterChild; + } + + @Test + public void verify_ledsCollection() { + BootstrapperImpl.enableEndpointFallback = true; + setEdsWithLeds(LEDS_NAME); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_LEDS, ImmutableMap.of( + LEDS_NAME, buildLbEndpointCollection("127.0.0.30"))); + + xdsDependencyManager.start(xdsConfigWatcher); + + verify(xdsConfigWatcher).onUpdate(xdsUpdateCaptor.capture()); + XdsConfig config = xdsUpdateCaptor.getValue().getValue(); + XdsClusterConfig.EndpointConfig endpointConfig = + getEndpointConfig(config.getClusters().get(CLUSTER_NAME)); + + Endpoints.LocalityLbEndpoints locality = endpointConfig.getEndpoint().getValue() + .localityLbEndpointsMap.values().iterator().next(); + assertThat(locality.endpointCollection()).isNull(); + assertThat(locality.lbEndpointCollectionName()).isEqualTo(LEDS_NAME); + + StatusOr collectionOr = + endpointConfig.getLbEndpointCollectionResources().get(LEDS_NAME); + assertThat(collectionOr).isNotNull(); + assertThat(collectionOr.hasValue()).isTrue(); + assertThat(collectionOr.getValue().getEndpointCollection().endpoints()).containsExactly( + Endpoints.LbEndpoint.create("127.0.0.30", ENDPOINT_PORT, 0, true, ENDPOINT_HOSTNAME, + ImmutableMap.of())); + } + + @Test + public void verify_ledsCollectionUpdate_republishesConfig() { + BootstrapperImpl.enableEndpointFallback = true; + setEdsWithLeds(LEDS_NAME); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_LEDS, ImmutableMap.of( + LEDS_NAME, buildLbEndpointCollection("127.0.0.30"))); + + InOrder inOrder = Mockito.inOrder(xdsConfigWatcher); + xdsDependencyManager.start(xdsConfigWatcher); + inOrder.verify(xdsConfigWatcher).onUpdate(xdsUpdateCaptor.capture()); + + // Only the LEDS resource changes; the EDS resource is untouched. + controlPlaneService.setXdsConfig(ADS_TYPE_URL_LEDS, ImmutableMap.of( + LEDS_NAME, buildLbEndpointCollection("127.0.0.31"))); + + inOrder.verify(xdsConfigWatcher).onUpdate(xdsUpdateCaptor.capture()); + StatusOr collectionOr = + getEndpointConfig(xdsUpdateCaptor.getValue().getValue().getClusters().get(CLUSTER_NAME)) + .getLbEndpointCollectionResources().get(LEDS_NAME); + assertThat(collectionOr.getValue().getEndpointCollection().endpoints()).containsExactly( + Endpoints.LbEndpoint.create("127.0.0.31", ENDPOINT_PORT, 0, true, ENDPOINT_HOSTNAME, + ImmutableMap.of())); + } + + @Test + public void verify_ledsCollection_unsubscribedWhenNoLongerReferenced() { + BootstrapperImpl.enableEndpointFallback = true; + setEdsWithLeds(LEDS_NAME); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_LEDS, ImmutableMap.of( + LEDS_NAME, buildLbEndpointCollection("127.0.0.30"))); + + xdsDependencyManager.start(xdsConfigWatcher); + verify(xdsConfigWatcher).onUpdate(xdsUpdateCaptor.capture()); + assertThat(getEndpointConfig(xdsUpdateCaptor.getValue().getValue().getClusters() + .get(CLUSTER_NAME)).getLbEndpointCollectionResources()).containsKey(LEDS_NAME); + + // Replace the EDS resource with one that inlines its endpoints. + XdsTestUtils.setAdsConfig(controlPlaneService, serverName); + + verify(xdsConfigWatcher, atLeastOnce()).onUpdate(xdsUpdateCaptor.capture()); + assertThat(getEndpointConfig(xdsUpdateCaptor.getValue().getValue().getClusters() + .get(CLUSTER_NAME)).getLbEndpointCollectionResources()).isEmpty(); + } + + @Test + public void verify_ledsCollection_sharedByTwoLocalities() throws Exception { + BootstrapperImpl.enableEndpointFallback = true; + // Two distinct localities referring to the same LbEndpointCollection resource. + controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, ImmutableMap.of(EDS_NAME, + ClusterLoadAssignment.newBuilder() + .setClusterName(EDS_NAME) + .addEndpoints(ledsLocality("region1", LEDS_NAME)) + .addEndpoints(ledsLocality("region2", LEDS_NAME)) + .build())); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_LEDS, ImmutableMap.of( + LEDS_NAME, buildLbEndpointCollection("127.0.0.30"))); + + xdsDependencyManager.start(xdsConfigWatcher); + + verify(xdsConfigWatcher).onUpdate(xdsUpdateCaptor.capture()); + XdsClusterConfig.EndpointConfig endpointConfig = + getEndpointConfig(xdsUpdateCaptor.getValue().getValue().getClusters().get(CLUSTER_NAME)); + assertThat(endpointConfig.getEndpoint().getValue().localityLbEndpointsMap).hasSize(2); + // The collection is stored once and subscribed to once. + assertThat(endpointConfig.getLbEndpointCollectionResources().keySet()) + .containsExactly(LEDS_NAME); + Map, Map> watches = + xdsClient.getSubscribedResourcesMetadataSnapshot().get(); + assertThat(watches.get(XdsLbEndpointCollectionResource.getInstance()).keySet()) + .containsExactly(LEDS_NAME); + } + + /** Replaces the default EDS resource with one whose only locality points at a LEDS resource. */ + private void setEdsWithLeds(String collectionName) { + ClusterLoadAssignment clusterLoadAssignment = ClusterLoadAssignment.newBuilder() + .setClusterName(EDS_NAME) + .addEndpoints(ledsLocality("", collectionName)) + .build(); + controlPlaneService.setXdsConfig(ADS_TYPE_URL_EDS, + ImmutableMap.of(EDS_NAME, clusterLoadAssignment)); + } + + /** Builds a locality in {@code region} whose endpoints come from a LEDS resource. */ + private static LocalityLbEndpoints.Builder ledsLocality(String region, String collectionName) { + return LocalityLbEndpoints.newBuilder() + .setLocality(io.envoyproxy.envoy.config.core.v3.Locality.newBuilder().setRegion(region)) + .setLoadBalancingWeight(UInt32Value.of(10)) + .setPriority(0) + .setLedsClusterLocalityConfig(LedsClusterLocalityConfig.newBuilder() + .setLedsConfig(ConfigSource.newBuilder() + .setSelf(SelfConfigSource.getDefaultInstance())) + .setLedsCollectionName(collectionName)); + } + + private static LbEndpointCollection buildLbEndpointCollection(String address) { + return LbEndpointCollection.newBuilder() + .addEntries(CollectionEntry.newBuilder() + .setInlineEntry(CollectionEntry.InlineEntry.newBuilder() + .setResource(Any.pack(LbEndpoint.newBuilder() + .setEndpoint(Endpoint.newBuilder() + .setHostname(ENDPOINT_HOSTNAME) + .setAddress(Address.newBuilder() + .setSocketAddress(SocketAddress.newBuilder() + .setAddress(address) + .setPortValue(ENDPOINT_PORT)))) + .build())))) + .build(); + } + + @Test public void testComplexRegisteredAggregate() throws IOException { InOrder inOrder = Mockito.inOrder(xdsConfigWatcher); @@ -680,7 +841,8 @@ public void testMultipleCdsReferToSameEds() { assertThat(edsForB.clusterName).isEqualTo(edsName); assertThat(edsForA).isEqualTo(edsForB); edsForA.localityLbEndpointsMap.values().forEach( - localityLbEndpoints -> assertThat(localityLbEndpoints.endpoints()).hasSize(1)); + localityLbEndpoints -> + assertThat(localityLbEndpoints.endpointCollection().endpoints()).hasSize(1)); } @Test diff --git a/xds/src/test/java/io/grpc/xds/XdsLbEndpointCollectionResourceTest.java b/xds/src/test/java/io/grpc/xds/XdsLbEndpointCollectionResourceTest.java new file mode 100644 index 00000000000..f4628199ea1 --- /dev/null +++ b/xds/src/test/java/io/grpc/xds/XdsLbEndpointCollectionResourceTest.java @@ -0,0 +1,173 @@ +/* + * Copyright 2026 The gRPC Authors + * + * Licensed 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 + * + * http://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 io.grpc.xds; + +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import com.github.xds.core.v3.CollectionEntry; +import com.github.xds.core.v3.ResourceLocator; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.testing.EqualsTester; +import com.google.protobuf.Any; +import com.google.protobuf.StringValue; +import com.google.protobuf.UInt32Value; +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.Endpoint; +import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint; +import io.envoyproxy.envoy.config.endpoint.v3.LbEndpointCollection; +import io.grpc.xds.XdsLbEndpointCollectionResource.LbEndpointCollectionUpdate; +import io.grpc.xds.client.XdsResourceType.ResourceInvalidException; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests for {@link XdsLbEndpointCollectionResource}. */ +@RunWith(JUnit4.class) +public class XdsLbEndpointCollectionResourceTest { + + private final XdsLbEndpointCollectionResource resource = + XdsLbEndpointCollectionResource.getInstance(); + + @Test + public void typeInfo() { + assertThat(resource.typeName()).isEqualTo("LEDS"); + assertThat(resource.typeUrl()) + .isEqualTo("type.googleapis.com/envoy.config.endpoint.v3.LbEndpointCollection"); + assertThat(resource.unpackedClassName()).isEqualTo(LbEndpointCollection.class); + assertThat(resource.isFullStateOfTheWorld()).isFalse(); + assertThat(resource.shouldRetrieveResourceKeysForArgs()).isTrue(); + } + + @Test + public void doParse_multipleEntries() throws ResourceInvalidException { + LbEndpointCollection collection = LbEndpointCollection.newBuilder() + .addEntries(inlineEntry(lbEndpoint("172.14.14.5", 8888, 20, HealthStatus.HEALTHY))) + .addEntries(inlineEntry(lbEndpoint("172.14.14.6", 8888, 30, HealthStatus.UNHEALTHY))) + .build(); + + LbEndpointCollectionUpdate update = resource.doParse(null, collection); + + assertThat(update.getEndpointCollection()).isEqualTo( + Endpoints.LbEndpointCollection.create(ImmutableList.of( + Endpoints.LbEndpoint.create("172.14.14.5", 8888, 20, true, "", ImmutableMap.of()), + Endpoints.LbEndpoint.create("172.14.14.6", 8888, 30, false, "", ImmutableMap.of())))); + } + + @Test + public void doParse_emptyEntriesIsValid() throws ResourceInvalidException { + LbEndpointCollectionUpdate update = + resource.doParse(null, LbEndpointCollection.getDefaultInstance()); + + assertThat(update.getEndpointCollection().endpoints()).isEmpty(); + } + + @Test + public void doParse_entryWithoutInlineEntry() { + LbEndpointCollection collection = LbEndpointCollection.newBuilder() + .addEntries(CollectionEntry.newBuilder() + .setLocator(ResourceLocator.getDefaultInstance())) + .build(); + + ResourceInvalidException ex = assertThrows(ResourceInvalidException.class, + () -> resource.doParse(null, collection)); + assertThat(ex).hasMessageThat().isEqualTo("CollectionEntry with no inline_entry"); + } + + @Test + public void doParse_inlineEntryWithWrongResourceType() { + LbEndpointCollection collection = LbEndpointCollection.newBuilder() + .addEntries(CollectionEntry.newBuilder() + .setInlineEntry(CollectionEntry.InlineEntry.newBuilder() + .setResource(Any.pack(StringValue.of("not-an-endpoint"))))) + .build(); + + ResourceInvalidException ex = assertThrows(ResourceInvalidException.class, + () -> resource.doParse(null, collection)); + assertThat(ex).hasMessageThat().contains("Can't decode LbEndpoint"); + } + + @Test + public void doParse_endpointWithoutAddress() { + LbEndpointCollection collection = LbEndpointCollection.newBuilder() + .addEntries(inlineEntry(LbEndpoint.getDefaultInstance())) + .build(); + + ResourceInvalidException ex = assertThrows(ResourceInvalidException.class, + () -> resource.doParse(null, collection)); + assertThat(ex).hasMessageThat().isEqualTo("LbEndpoint with no endpoint/address"); + } + + @Test + public void doParse_endpointWithNonIpAddress() { + LbEndpointCollection collection = LbEndpointCollection.newBuilder() + .addEntries(inlineEntry(lbEndpoint("example.com", 8888, 20, HealthStatus.HEALTHY))) + .build(); + + ResourceInvalidException ex = assertThrows(ResourceInvalidException.class, + () -> resource.doParse(null, collection)); + assertThat(ex).hasMessageThat().contains("Address is not an IP"); + } + + @Test + public void doParse_wrongMessageType() { + ResourceInvalidException ex = assertThrows(ResourceInvalidException.class, + () -> resource.doParse(null, LbEndpoint.getDefaultInstance())); + assertThat(ex).hasMessageThat().contains("Invalid message type"); + } + + @Test + public void lbEndpointCollectionUpdate_equalsAndHashCode() { + Endpoints.LbEndpointCollection collection = Endpoints.LbEndpointCollection.create( + ImmutableList.of( + Endpoints.LbEndpoint.create("172.14.14.5", 8888, 20, true, "", ImmutableMap.of()))); + Endpoints.LbEndpointCollection otherCollection = Endpoints.LbEndpointCollection.create( + ImmutableList.of( + Endpoints.LbEndpoint.create("172.14.14.6", 8888, 20, true, "", ImmutableMap.of()))); + + new EqualsTester() + .addEqualityGroup( + new LbEndpointCollectionUpdate(collection), + new LbEndpointCollectionUpdate(collection)) + .addEqualityGroup(new LbEndpointCollectionUpdate(otherCollection)) + .addEqualityGroup( + new LbEndpointCollectionUpdate( + Endpoints.LbEndpointCollection.create(ImmutableList.of()))) + .testEquals(); + } + + private static CollectionEntry inlineEntry(LbEndpoint endpoint) { + return CollectionEntry.newBuilder() + .setInlineEntry(CollectionEntry.InlineEntry.newBuilder().setResource(Any.pack(endpoint))) + .build(); + } + + private static LbEndpoint lbEndpoint( + String address, int port, int weight, HealthStatus healthStatus) { + return LbEndpoint.newBuilder() + .setEndpoint(Endpoint.newBuilder() + .setAddress(Address.newBuilder() + .setSocketAddress( + SocketAddress.newBuilder().setAddress(address).setPortValue(port)))) + .setHealthStatus(healthStatus) + .setLoadBalancingWeight(UInt32Value.newBuilder().setValue(weight)) + .build(); + } +} diff --git a/xds/src/test/java/io/grpc/xds/XdsTestControlPlaneService.java b/xds/src/test/java/io/grpc/xds/XdsTestControlPlaneService.java index a54893c9075..acd50f6f242 100644 --- a/xds/src/test/java/io/grpc/xds/XdsTestControlPlaneService.java +++ b/xds/src/test/java/io/grpc/xds/XdsTestControlPlaneService.java @@ -23,6 +23,7 @@ import io.envoyproxy.envoy.service.discovery.v3.AggregatedDiscoveryServiceGrpc; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; +import io.envoyproxy.envoy.service.discovery.v3.Resource; import io.grpc.SynchronizationContext; import io.grpc.stub.StreamObserver; import java.util.HashMap; @@ -74,6 +75,8 @@ public void uncaughtException(Thread t, Throwable e) { "type.googleapis.com/envoy.config.cluster.v3.Cluster"; static final String ADS_TYPE_URL_EDS = "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment"; + static final String ADS_TYPE_URL_LEDS = + "type.googleapis.com/envoy.config.endpoint.v3.LbEndpointCollection"; private final Map> xdsResources = new HashMap<>(); private ImmutableMap, Set>> subscribers @@ -81,19 +84,22 @@ public void uncaughtException(Thread t, Throwable e) { ADS_TYPE_URL_LDS, new ConcurrentHashMap, Set>(), ADS_TYPE_URL_RDS, new ConcurrentHashMap, Set>(), ADS_TYPE_URL_CDS, new ConcurrentHashMap, Set>(), - ADS_TYPE_URL_EDS, new ConcurrentHashMap, Set>()); + ADS_TYPE_URL_EDS, new ConcurrentHashMap, Set>(), + ADS_TYPE_URL_LEDS, new ConcurrentHashMap, Set>()); private final ImmutableMap xdsVersions = ImmutableMap.of( ADS_TYPE_URL_LDS, new AtomicInteger(1), ADS_TYPE_URL_RDS, new AtomicInteger(1), ADS_TYPE_URL_CDS, new AtomicInteger(1), - ADS_TYPE_URL_EDS, new AtomicInteger(1) + ADS_TYPE_URL_EDS, new AtomicInteger(1), + ADS_TYPE_URL_LEDS, new AtomicInteger(1) ); private final ImmutableMap, AtomicInteger>> xdsNonces = ImmutableMap.of( ADS_TYPE_URL_LDS, new ConcurrentHashMap, AtomicInteger>(), ADS_TYPE_URL_RDS, new ConcurrentHashMap, AtomicInteger>(), ADS_TYPE_URL_CDS, new ConcurrentHashMap, AtomicInteger>(), - ADS_TYPE_URL_EDS, new ConcurrentHashMap, AtomicInteger>() + ADS_TYPE_URL_EDS, new ConcurrentHashMap, AtomicInteger>(), + ADS_TYPE_URL_LEDS, new ConcurrentHashMap, AtomicInteger>() ); @@ -201,8 +207,18 @@ private DiscoveryResponse generateResponse(String resourceType, String version, for (String resourceName: resourceNames) { if (xdsResources.containsKey(resourceType) && xdsResources.get(resourceType).containsKey(resourceName)) { - responseBuilder.addResources(Any.pack(xdsResources.get(resourceType).get(resourceName), - resourceType)); + Message message = xdsResources.get(resourceType).get(resourceName); + if (ADS_TYPE_URL_LEDS.equals(resourceType)) { + // LbEndpointCollection has no name field, so it must be wrapped in a + // discovery.v3.Resource to convey the resource name. + responseBuilder.addResources(Any.pack( + Resource.newBuilder() + .setName(resourceName) + .setResource(Any.pack(message)) + .build())); + } else { + responseBuilder.addResources(Any.pack(message, resourceType)); + } } } return responseBuilder.build();