flasn4nt
/
grpc-java
mirror of https://github.com/grpc/grpc-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

xds: remove ResourceType enum, use XdsResourceType instead (#9587) 

Now the xds resources are dynamically managed in resourceStore in xdsClient. The types is a xdsResourceType, singleton.
There is no longer hardcoded static list of known resource types, the subscription list is the source of truth.
AbstractXdsClient that manages AdsStream will only accept the xds resource types that has already has watchers subscribed to, same behaviour as before.
This commit is contained in:
yifeizhuang 2022-10-06 13:10:55 -07:00 committed by GitHub
parent cff8bd8d87
commit 68339250e4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 206 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
xds/src/main/java/io/grpc/xds/AbstractXdsClient.java
View File

@ -48,7 +48,6 @@ import io.grpc.stub.StreamObserver;
import io.grpc.xds.Bootstrapper.ServerInfo;
import io.grpc.xds.EnvoyProtoData.Node;
import io.grpc.xds.XdsClient.ResourceStore;
import io.grpc.xds.XdsClient.ResourceUpdate;
import io.grpc.xds.XdsClient.XdsResponseHandler;
import io.grpc.xds.XdsClientImpl.XdsChannelFactory;
import io.grpc.xds.XdsLogger.XdsLogLevel;
@ -82,7 +81,7 @@ final class AbstractXdsClient {
// Last successfully applied version_info for each resource type. Starts with empty string.
// A version_info is used to update management server with client's most recent knowledge of
// resources.
private final Map<ResourceType, String> versions = new HashMap<>();
private final Map<XdsResourceType<?>, String> versions = new HashMap<>();
private boolean shutdown;
@Nullable
@ -160,8 +159,7 @@ final class AbstractXdsClient {
if (adsStream == null) {
startRpcStream();
}
Collection<String> resources = resourceStore.getSubscribedResources(serverInfo,
resourceType.typeName());
Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, resourceType);
if (resources != null) {
adsStream.sendDiscoveryRequest(resourceType, resources);
}
@ -172,11 +170,10 @@ final class AbstractXdsClient {
* and sends an ACK request to the management server.
*/
// Must be synchronized.
void ackResponse(XdsResourceType<?> xdsResourceType, String versionInfo, String nonce) {
ResourceType type = xdsResourceType.typeName();
void ackResponse(XdsResourceType<?> type, String versionInfo, String nonce) {
versions.put(type, versionInfo);
logger.log(XdsLogLevel.INFO, "Sending ACK for {0} update, nonce: {1}, current version: {2}",
type, nonce, versionInfo);
type.typeName(), nonce, versionInfo);
Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);
if (resources == null) {
resources = Collections.emptyList();
@ -189,11 +186,10 @@ final class AbstractXdsClient {
* accepted version) to the management server.
*/
// Must be synchronized.
void nackResponse(XdsResourceType<?> xdsResourceType, String nonce, String errorDetail) {
ResourceType type = xdsResourceType.typeName();
void nackResponse(XdsResourceType<?> type, String nonce, String errorDetail) {
String versionInfo = versions.getOrDefault(type, "");
logger.log(XdsLogLevel.INFO, "Sending NACK for {0} update, nonce: {1}, current version: {2}",
type, nonce, versionInfo);
type.typeName(), nonce, versionInfo);
Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);
if (resources == null) {
resources = Collections.emptyList();
@ -239,77 +235,38 @@ final class AbstractXdsClient {
return;
}
startRpcStream();
for (ResourceType type : ResourceType.values()) {
if (type == ResourceType.UNKNOWN) {
continue;
}
for (XdsResourceType<?> type : resourceStore.getXdsResourceTypes()) {
Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, type);
if (resources != null) {
adsStream.sendDiscoveryRequest(resourceStore.getXdsResourceType(type), resources);
adsStream.sendDiscoveryRequest(type, resources);
}
}
xdsResponseHandler.handleStreamRestarted(serverInfo);
}
}
// TODO(zivy) : remove and replace with XdsResourceType
enum ResourceType {
UNKNOWN, LDS, RDS, CDS, EDS;
String typeUrl() {
switch (this) {
case LDS:
return ADS_TYPE_URL_LDS;
case RDS:
return ADS_TYPE_URL_RDS;
case CDS:
return ADS_TYPE_URL_CDS;
case EDS:
return ADS_TYPE_URL_EDS;
case UNKNOWN:
default:
throw new AssertionError("Unknown or missing case in enum switch: " + this);
}
}
String typeUrlV2() {
switch (this) {
case LDS:
return ADS_TYPE_URL_LDS_V2;
case RDS:
return ADS_TYPE_URL_RDS_V2;
case CDS:
return ADS_TYPE_URL_CDS_V2;
case EDS:
return ADS_TYPE_URL_EDS_V2;
case UNKNOWN:
default:
throw new AssertionError("Unknown or missing case in enum switch: " + this);
}
}
@VisibleForTesting
static ResourceType fromTypeUrl(String typeUrl) {
@Nullable
static XdsResourceType<?> fromTypeUrl(String typeUrl) {
switch (typeUrl) {
case ADS_TYPE_URL_LDS:
// fall trough
case ADS_TYPE_URL_LDS_V2:
return LDS;
return XdsListenerResource.getInstance();
case ADS_TYPE_URL_RDS:
// fall through
case ADS_TYPE_URL_RDS_V2:
return RDS;
return XdsRouteConfigureResource.getInstance();
case ADS_TYPE_URL_CDS:
// fall through
case ADS_TYPE_URL_CDS_V2:
return CDS;
return XdsClusterResource.getInstance();
case ADS_TYPE_URL_EDS:
// fall through
case ADS_TYPE_URL_EDS_V2:
return EDS;
return XdsEndpointResource.getInstance();
default:
return UNKNOWN;
}
return null;
}
}
@ -322,7 +279,7 @@ final class AbstractXdsClient {
// used for management server to identify which response the client is ACKing/NACking.
// To avoid confusion, client-initiated requests will always use the nonce in
// most recently received responses of each resource type.
private final Map<ResourceType, String> respNonces = new HashMap<>();
private final Map<XdsResourceType<?>, String> respNonces = new HashMap<>();
abstract void start();
@ -334,27 +291,27 @@ final class AbstractXdsClient {
* client-initiated discovery requests, use {@link
* #sendDiscoveryRequest(XdsResourceType, Collection)}.
*/
abstract void sendDiscoveryRequest(ResourceType type, String version,
abstract void sendDiscoveryRequest(XdsResourceType<?> type, String version,
Collection<String> resources, String nonce, @Nullable String errorDetail);
/**
* Sends a client-initiated discovery request.
*/
final void sendDiscoveryRequest(XdsResourceType<? extends ResourceUpdate> xdsResourceType,
Collection<String> resources) {
ResourceType type = xdsResourceType.typeName();
final void sendDiscoveryRequest(XdsResourceType<?> type, Collection<String> resources) {
logger.log(XdsLogLevel.INFO, "Sending {0} request for resources: {1}", type, resources);
sendDiscoveryRequest(type, versions.getOrDefault(type, ""), resources,
respNonces.getOrDefault(type, ""), null);
}
final void handleRpcResponse(
ResourceType type, String versionInfo, List<Any> resources, String nonce) {
final void handleRpcResponse(XdsResourceType<?> type, String versionInfo, List<Any> resources,
String nonce) {
if (closed) {
return;
}
responseReceived = true;
if (type != null) {
respNonces.put(type, nonce);
}
xdsResponseHandler.handleResourceResponse(type, serverInfo, versionInfo, resources, nonce);
}
@ -422,7 +379,7 @@ final class AbstractXdsClient {
syncContext.execute(new Runnable() {
@Override
public void run() {
ResourceType type = ResourceType.fromTypeUrl(response.getTypeUrl());
XdsResourceType<?> type = fromTypeUrl(response.getTypeUrl());
if (logger.isLoggable(XdsLogLevel.DEBUG)) {
logger.log(
XdsLogLevel.DEBUG, "Received {0} response:\n{1}", type,
@ -458,8 +415,9 @@ final class AbstractXdsClient {
}
@Override
void sendDiscoveryRequest(ResourceType type, String versionInfo, Collection<String> resources,
String nonce, @Nullable String errorDetail) {
void sendDiscoveryRequest(XdsResourceType<?> type, String versionInfo,
Collection<String> resources, String nonce,
@Nullable String errorDetail) {
checkState(requestWriter != null, "ADS stream has not been started");
io.envoyproxy.envoy.api.v2.DiscoveryRequest.Builder builder =
io.envoyproxy.envoy.api.v2.DiscoveryRequest.newBuilder()
@ -502,7 +460,7 @@ final class AbstractXdsClient {
syncContext.execute(new Runnable() {
@Override
public void run() {
ResourceType type = ResourceType.fromTypeUrl(response.getTypeUrl());
XdsResourceType<?> type = fromTypeUrl(response.getTypeUrl());
if (logger.isLoggable(XdsLogLevel.DEBUG)) {
logger.log(
XdsLogLevel.DEBUG, "Received {0} response:\n{1}", type,
@ -538,8 +496,9 @@ final class AbstractXdsClient {
}
@Override
void sendDiscoveryRequest(ResourceType type, String versionInfo, Collection<String> resources,
String nonce, @Nullable String errorDetail) {
void sendDiscoveryRequest(XdsResourceType<?> type, String versionInfo,
Collection<String> resources, String nonce,
@Nullable String errorDetail) {
checkState(requestWriter != null, "ADS stream has not been started");
DiscoveryRequest.Builder builder =
DiscoveryRequest.newBuilder()

12
xds/src/main/java/io/grpc/xds/CsdsService.java
View File

@ -33,7 +33,6 @@ import io.grpc.Status;
import io.grpc.StatusException;
import io.grpc.internal.ObjectPool;
import io.grpc.stub.StreamObserver;
import io.grpc.xds.AbstractXdsClient.ResourceType;
import io.grpc.xds.XdsClient.ResourceMetadata;
import io.grpc.xds.XdsClient.ResourceMetadata.ResourceMetadataStatus;
import io.grpc.xds.XdsClient.ResourceMetadata.UpdateFailureState;
@ -156,12 +155,12 @@ public final class CsdsService extends
ClientConfig.Builder builder = ClientConfig.newBuilder()
.setNode(xdsClient.getBootstrapInfo().node().toEnvoyProtoNode());
Map<ResourceType, Map<String, ResourceMetadata>> metadataByType =
Map<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByType =
awaitSubscribedResourcesMetadata(xdsClient.getSubscribedResourcesMetadataSnapshot());
for (Map.Entry<ResourceType, Map<String, ResourceMetadata>> metadataByTypeEntry
for (Map.Entry<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByTypeEntry
: metadataByType.entrySet()) {
ResourceType type = metadataByTypeEntry.getKey();
XdsResourceType<?> type = metadataByTypeEntry.getKey();
Map<String, ResourceMetadata> metadataByResourceName = metadataByTypeEntry.getValue();
for (Map.Entry<String, ResourceMetadata> metadataEntry : metadataByResourceName.entrySet()) {
String resourceName = metadataEntry.getKey();
@ -187,8 +186,9 @@ public final class CsdsService extends
return builder.build();
}
private static Map<ResourceType, Map<String, ResourceMetadata>> awaitSubscribedResourcesMetadata(
ListenableFuture<Map<ResourceType, Map<String, ResourceMetadata>>> future)
private static Map<XdsResourceType<?>, Map<String, ResourceMetadata>>
awaitSubscribedResourcesMetadata(
ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>> future)
throws InterruptedException {
try {
// Normally this shouldn't take long, but add some slack for cases like a cold JVM.

13
xds/src/main/java/io/grpc/xds/XdsClient.java
View File

@ -25,7 +25,6 @@ import com.google.common.net.UrlEscapers;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.protobuf.Any;
import io.grpc.Status;
import io.grpc.xds.AbstractXdsClient.ResourceType;
import io.grpc.xds.Bootstrapper.ServerInfo;
import io.grpc.xds.LoadStatsManager2.ClusterDropStats;
import io.grpc.xds.LoadStatsManager2.ClusterLocalityStats;
@ -296,7 +295,7 @@ abstract class XdsClient {
* a map ("resource name": "resource metadata").
*/
// Must be synchronized.
ListenableFuture<Map<ResourceType, Map<String, ResourceMetadata>>>
ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>
getSubscribedResourcesMetadataSnapshot() {
throw new UnsupportedOperationException();
}
@ -347,8 +346,8 @@ abstract class XdsClient {
interface XdsResponseHandler {
/** Called when a xds response is received. */
void handleResourceResponse(
ResourceType resourceType, ServerInfo serverInfo, String versionInfo, List<Any> resources,
String nonce);
XdsResourceType<?> resourceType, ServerInfo serverInfo, String versionInfo,
List<Any> resources, String nonce);
/** Called when the ADS stream is closed passively. */
// Must be synchronized.
@ -369,9 +368,9 @@ abstract class XdsClient {
*/
// Must be synchronized.
@Nullable
Collection<String> getSubscribedResources(ServerInfo serverInfo, ResourceType type);
Collection<String> getSubscribedResources(ServerInfo serverInfo,
XdsResourceType<? extends ResourceUpdate> type);
@Nullable
XdsResourceType<? extends ResourceUpdate> getXdsResourceType(ResourceType type);
Collection<XdsResourceType<? extends ResourceUpdate>> getXdsResourceTypes();
}
}

70
xds/src/main/java/io/grpc/xds/XdsClientImpl.java
View File

@ -18,10 +18,6 @@ package io.grpc.xds;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.xds.AbstractXdsClient.ResourceType.CDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.EDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.LDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.RDS;
import static io.grpc.xds.Bootstrapper.XDSTP_SCHEME;
import static io.grpc.xds.XdsResourceType.ParsedResource;
import static io.grpc.xds.XdsResourceType.ValidatedResourceUpdate;
@ -46,7 +42,6 @@ import io.grpc.SynchronizationContext;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.internal.BackoffPolicy;
import io.grpc.internal.TimeProvider;
import io.grpc.xds.AbstractXdsClient.ResourceType;
import io.grpc.xds.Bootstrapper.AuthorityInfo;
import io.grpc.xds.Bootstrapper.ServerInfo;
import io.grpc.xds.LoadStatsManager2.ClusterDropStats;
@ -96,12 +91,6 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
private final Map<XdsResourceType<? extends ResourceUpdate>,
Map<String, ResourceSubscriber<? extends ResourceUpdate>>>
resourceSubscribers = new HashMap<>();
private final Map<ResourceType, XdsResourceType<? extends ResourceUpdate>> xdsResourceTypeMap =
ImmutableMap.of(
LDS, XdsListenerResource.getInstance(),
RDS, XdsRouteConfigureResource.getInstance(),
CDS, XdsClusterResource.getInstance(),
EDS, XdsEndpointResource.getInstance());
private final LoadStatsManager2 loadStatsManager;
private final Map<ServerInfo, LoadReportClient> serverLrsClientMap = new HashMap<>();
private final XdsChannelFactory xdsChannelFactory;
@ -166,17 +155,16 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
@Override
public void handleResourceResponse(
ResourceType resourceType, ServerInfo serverInfo, String versionInfo, List<Any> resources,
String nonce) {
XdsResourceType<?> xdsResourceType, ServerInfo serverInfo, String versionInfo,
List<Any> resources, String nonce) {
syncContext.throwIfNotInThisSynchronizationContext();
XdsResourceType<? extends ResourceUpdate> xdsResourceType =
xdsResourceTypeMap.get(resourceType);
if (xdsResourceType == null) {
logger.log(XdsLogLevel.WARNING, "Ignore an unknown type of DiscoveryResponse");
return;
}
Set<String> toParseResourceNames = null;
if (!(resourceType == LDS || resourceType == RDS)
if (!(xdsResourceType == XdsListenerResource.getInstance()
|| xdsResourceType == XdsRouteConfigureResource.getInstance())
&& resourceSubscribers.containsKey(xdsResourceType)) {
toParseResourceNames = resourceSubscribers.get(xdsResourceType).keySet();
}
@ -239,23 +227,17 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
return isShutdown;
}
private Map<String, ResourceSubscriber<? extends ResourceUpdate>> getSubscribedResourcesMap(
ResourceType type) {
return resourceSubscribers.getOrDefault(xdsResourceTypeMap.get(type), Collections.emptyMap());
}
@Nullable
@Override
public XdsResourceType<? extends ResourceUpdate> getXdsResourceType(ResourceType type) {
return xdsResourceTypeMap.get(type);
public Collection<XdsResourceType<? extends ResourceUpdate>> getXdsResourceTypes() {
return resourceSubscribers.keySet();
}
@Nullable
@Override
public Collection<String> getSubscribedResources(ServerInfo serverInfo,
ResourceType type) {
XdsResourceType<? extends ResourceUpdate> type) {
Map<String, ResourceSubscriber<? extends ResourceUpdate>> resources =
getSubscribedResourcesMap(type);
resourceSubscribers.getOrDefault(type, Collections.emptyMap());
ImmutableSet.Builder<String> builder = ImmutableSet.builder();
for (String key : resources.keySet()) {
if (resources.get(key).serverInfo.equals(serverInfo)) {
@ -266,26 +248,26 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
return retVal.isEmpty() ? null : retVal;
}
// As XdsClient APIs becomes resource agnostic, subscribed resource types are dynamic.
// ResourceTypes that do not have subscribers does not show up in the snapshot keys.
@Override
ListenableFuture<Map<ResourceType, Map<String, ResourceMetadata>>>
ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>
getSubscribedResourcesMetadataSnapshot() {
final SettableFuture<Map<ResourceType, Map<String, ResourceMetadata>>> future =
final SettableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>> future =
SettableFuture.create();
syncContext.execute(new Runnable() {
@Override
public void run() {
// A map from a "resource type" to a map ("resource name": "resource metadata")
ImmutableMap.Builder<ResourceType, Map<String, ResourceMetadata>> metadataSnapshot =
ImmutableMap.Builder<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataSnapshot =
ImmutableMap.builder();
for (XdsResourceType<? extends ResourceUpdate> resourceType: xdsResourceTypeMap.values()) {
for (XdsResourceType<?> resourceType: resourceSubscribers.keySet()) {
ImmutableMap.Builder<String, ResourceMetadata> metadataMap = ImmutableMap.builder();
Map<String, ResourceSubscriber<? extends ResourceUpdate>> resourceSubscriberMap =
resourceSubscribers.getOrDefault(resourceType, Collections.emptyMap());
for (Map.Entry<String, ResourceSubscriber<? extends ResourceUpdate>> resourceEntry
: resourceSubscriberMap.entrySet()) {
: resourceSubscribers.get(resourceType).entrySet()) {
metadataMap.put(resourceEntry.getKey(), resourceEntry.getValue().metadata);
}
metadataSnapshot.put(resourceType.typeName(), metadataMap.buildOrThrow());
metadataSnapshot.put(resourceType, metadataMap.buildOrThrow());
}
future.set(metadataSnapshot.buildOrThrow());
}
@ -312,7 +294,7 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
(ResourceSubscriber<T>) resourceSubscribers.get(type).get(resourceName);;
if (subscriber == null) {
logger.log(XdsLogLevel.INFO, "Subscribe {0} resource {1}", type, resourceName);
subscriber = new ResourceSubscriber<>(type.typeName(), resourceName);
subscriber = new ResourceSubscriber<>(type, resourceName);
resourceSubscribers.get(type).put(resourceName, subscriber);
if (subscriber.xdsChannel != null) {
subscriber.xdsChannel.adjustResourceSubscription(type);
@ -427,8 +409,9 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
}
long updateTime = timeProvider.currentTimeNanos();
for (Map.Entry<String, ResourceSubscriber<?>> entry :
getSubscribedResourcesMap(xdsResourceType.typeName()).entrySet()) {
Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscribedResources =
resourceSubscribers.getOrDefault(xdsResourceType, Collections.emptyMap());
for (Map.Entry<String, ResourceSubscriber<?>> entry : subscribedResources.entrySet()) {
String resourceName = entry.getKey();
ResourceSubscriber<T> subscriber = (ResourceSubscriber<T>) entry.getValue();
@ -473,7 +456,7 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
// LDS/CDS responses represents the state of the world, RDS/EDS resources not referenced in
// LDS/CDS resources should be deleted.
if (xdsResourceType.dependentResource() != null) {
XdsResourceType<?> dependency = xdsResourceTypeMap.get(xdsResourceType.dependentResource());
XdsResourceType<?> dependency = xdsResourceType.dependentResource();
Map<String, ResourceSubscriber<? extends ResourceUpdate>> dependentSubscribers =
resourceSubscribers.get(dependency);
if (dependentSubscribers == null) {
@ -493,13 +476,13 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
return;
}
String resourceName = null;
if (subscriber.type == LDS) {
if (subscriber.type == XdsListenerResource.getInstance()) {
LdsUpdate ldsUpdate = (LdsUpdate) subscriber.data;
io.grpc.xds.HttpConnectionManager hcm = ldsUpdate.httpConnectionManager();
if (hcm != null) {
resourceName = hcm.rdsName();
}
} else if (subscriber.type == CDS) {
} else if (subscriber.type == XdsClusterResource.getInstance()) {
CdsUpdate cdsUpdate = (CdsUpdate) subscriber.data;
resourceName = cdsUpdate.edsServiceName();
}
@ -515,7 +498,7 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
private final class ResourceSubscriber<T extends ResourceUpdate> {
@Nullable private final ServerInfo serverInfo;
@Nullable private final AbstractXdsClient xdsChannel;
private final ResourceType type;
private final XdsResourceType<T> type;
private final String resource;
private final Set<ResourceWatcher<T>> watchers = new HashSet<>();
@Nullable private T data;
@ -527,7 +510,7 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
@Nullable private ResourceMetadata metadata;
@Nullable private String errorDescription;
ResourceSubscriber(ResourceType type, String resource) {
ResourceSubscriber(XdsResourceType<T> type, String resource) {
syncContext.throwIfNotInThisSynchronizationContext();
this.type = type;
this.resource = resource;
@ -669,7 +652,8 @@ final class XdsClientImpl extends XdsClient implements XdsResponseHandler, Resou
// and the resource is reusable.
boolean ignoreResourceDeletionEnabled =
serverInfo != null && serverInfo.ignoreResourceDeletion();
boolean isStateOfTheWorld = (type == LDS || type == CDS);
boolean isStateOfTheWorld = (type == XdsListenerResource.getInstance()
|| type == XdsClusterResource.getInstance());
if (ignoreResourceDeletionEnabled && isStateOfTheWorld && data != null) {
if (!resourceDeletionIgnored) {
logger.log(XdsLogLevel.FORCE_WARNING,

11
xds/src/main/java/io/grpc/xds/XdsClusterResource.java
View File

@ -17,9 +17,6 @@
package io.grpc.xds;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.xds.AbstractXdsClient.ResourceType;
import static io.grpc.xds.AbstractXdsClient.ResourceType.CDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.EDS;
import static io.grpc.xds.Bootstrapper.ServerInfo;
import com.google.auto.value.AutoValue;
@ -77,8 +74,8 @@ class XdsClusterResource extends XdsResourceType<CdsUpdate> {
}
@Override
ResourceType typeName() {
return CDS;
String typeName() {
return "CDS";
}
@Override
@ -93,8 +90,8 @@ class XdsClusterResource extends XdsResourceType<CdsUpdate> {
@Nullable
@Override
ResourceType dependentResource() {
return EDS;
XdsResourceType<?> dependentResource() {
return XdsEndpointResource.getInstance();
}
@Override

8
xds/src/main/java/io/grpc/xds/XdsEndpointResource.java
View File

@ -17,8 +17,6 @@
package io.grpc.xds;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.xds.AbstractXdsClient.ResourceType;
import static io.grpc.xds.AbstractXdsClient.ResourceType.EDS;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
@ -66,8 +64,8 @@ class XdsEndpointResource extends XdsResourceType<EdsUpdate> {
}
@Override
ResourceType typeName() {
return EDS;
String typeName() {
return "EDS";
}
@Override
@ -82,7 +80,7 @@ class XdsEndpointResource extends XdsResourceType<EdsUpdate> {
@Nullable
@Override
ResourceType dependentResource() {
XdsResourceType<?> dependentResource() {
return null;
}

11
xds/src/main/java/io/grpc/xds/XdsListenerResource.java
View File

@ -17,9 +17,6 @@
package io.grpc.xds;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.xds.AbstractXdsClient.ResourceType;
import static io.grpc.xds.AbstractXdsClient.ResourceType.LDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.RDS;
import static io.grpc.xds.XdsClient.ResourceUpdate;
import static io.grpc.xds.XdsClientImpl.ResourceInvalidException;
import static io.grpc.xds.XdsClusterResource.validateCommonTlsContext;
@ -81,8 +78,8 @@ class XdsListenerResource extends XdsResourceType<LdsUpdate> {
}
@Override
ResourceType typeName() {
return LDS;
String typeName() {
return "LDS";
}
@Override
@ -102,8 +99,8 @@ class XdsListenerResource extends XdsResourceType<LdsUpdate> {
@Nullable
@Override
ResourceType dependentResource() {
return RDS;
XdsResourceType<?> dependentResource() {
return XdsRouteConfigureResource.getInstance();
}
@Override

6
xds/src/main/java/io/grpc/xds/XdsNameResolver.java
View File

@ -47,7 +47,6 @@ import io.grpc.Status.Code;
import io.grpc.SynchronizationContext;
import io.grpc.internal.GrpcUtil;
import io.grpc.internal.ObjectPool;
import io.grpc.xds.AbstractXdsClient.ResourceType;
import io.grpc.xds.Bootstrapper.AuthorityInfo;
import io.grpc.xds.Bootstrapper.BootstrapInfo;
import io.grpc.xds.ClusterSpecifierPlugin.PluginConfig;
@ -202,8 +201,9 @@ final class XdsNameResolver extends NameResolver {
replacement = XdsClient.percentEncodePath(replacement);
}
String ldsResourceName = expandPercentS(listenerNameTemplate, replacement);
if (!XdsClient.isResourceNameValid(ldsResourceName, ResourceType.LDS.typeUrl())
&& !XdsClient.isResourceNameValid(ldsResourceName, ResourceType.LDS.typeUrlV2())) {
if (!XdsClient.isResourceNameValid(ldsResourceName, XdsListenerResource.getInstance().typeUrl())
&& !XdsClient.isResourceNameValid(ldsResourceName,
XdsListenerResource.getInstance().typeUrlV2())) {
listener.onError(Status.INVALID_ARGUMENT.withDescription(
"invalid listener resource URI for service authority: " + serviceAuthority));
return;

5
xds/src/main/java/io/grpc/xds/XdsResourceType.java
View File

@ -17,7 +17,6 @@
package io.grpc.xds;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.xds.AbstractXdsClient.ResourceType;
import static io.grpc.xds.Bootstrapper.ServerInfo;
import static io.grpc.xds.XdsClient.ResourceUpdate;
import static io.grpc.xds.XdsClient.canonifyResourceName;
@ -80,7 +79,7 @@ abstract class XdsResourceType<T extends ResourceUpdate> {
abstract Class<? extends com.google.protobuf.Message> unpackedClassName();
abstract ResourceType typeName();
abstract String typeName();
abstract String typeUrl();
@ -88,7 +87,7 @@ abstract class XdsResourceType<T extends ResourceUpdate> {
// Non-null for State of the World resources.
@Nullable
abstract ResourceType dependentResource();
abstract XdsResourceType<?> dependentResource();
static class Args {
final ServerInfo serverInfo;

8
xds/src/main/java/io/grpc/xds/XdsRouteConfigureResource.java
View File

@ -17,8 +17,6 @@
package io.grpc.xds;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.xds.AbstractXdsClient.ResourceType;
import static io.grpc.xds.AbstractXdsClient.ResourceType.RDS;
import static io.grpc.xds.XdsRouteConfigureResource.RdsUpdate;
import com.github.udpa.udpa.type.v1.TypedStruct;
@ -95,8 +93,8 @@ class XdsRouteConfigureResource extends XdsResourceType<RdsUpdate> {
}
@Override
ResourceType typeName() {
return RDS;
String typeName() {
return "RDS";
}
@Override
@ -111,7 +109,7 @@ class XdsRouteConfigureResource extends XdsResourceType<RdsUpdate> {
@Nullable
@Override
ResourceType dependentResource() {
XdsResourceType<?> dependentResource() {
return null;
}

5
xds/src/test/java/io/grpc/xds/CdsLoadBalancer2Test.java
View File

@ -17,7 +17,6 @@
package io.grpc.xds;
import static com.google.common.truth.Truth.assertThat;
import static io.grpc.xds.AbstractXdsClient.ResourceType.CDS;
import static io.grpc.xds.XdsLbPolicies.CLUSTER_RESOLVER_POLICY_NAME;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
@ -657,7 +656,7 @@ public class CdsLoadBalancer2Test {
@SuppressWarnings("unchecked")
<T extends ResourceUpdate> void watchXdsResource(XdsResourceType<T> type, String resourceName,
ResourceWatcher<T> watcher) {
assertThat(type.typeName()).isEqualTo(CDS);
assertThat(type.typeName()).isEqualTo("CDS");
assertThat(watchers).doesNotContainKey(resourceName);
watchers.put(resourceName, (ResourceWatcher<CdsUpdate>)watcher);
}
@ -667,7 +666,7 @@ public class CdsLoadBalancer2Test {
<T extends ResourceUpdate> void cancelXdsResourceWatch(XdsResourceType<T> type,
String resourceName,
ResourceWatcher<T> watcher) {
assertThat(type.typeName()).isEqualTo(CDS);
assertThat(type.typeName()).isEqualTo("CDS");
assertThat(watchers).containsKey(resourceName);
watchers.remove(resourceName);
}

5
xds/src/test/java/io/grpc/xds/ClusterResolverLoadBalancerTest.java
View File

@ -17,7 +17,6 @@
package io.grpc.xds;
import static com.google.common.truth.Truth.assertThat;
import static io.grpc.xds.AbstractXdsClient.ResourceType.EDS;
import static io.grpc.xds.XdsLbPolicies.CLUSTER_IMPL_POLICY_NAME;
import static io.grpc.xds.XdsLbPolicies.PRIORITY_POLICY_NAME;
import static io.grpc.xds.XdsLbPolicies.WEIGHTED_TARGET_POLICY_NAME;
@ -1176,7 +1175,7 @@ public class ClusterResolverLoadBalancerTest {
@SuppressWarnings("unchecked")
<T extends ResourceUpdate> void watchXdsResource(XdsResourceType<T> type, String resourceName,
ResourceWatcher<T> watcher) {
assertThat(type.typeName()).isEqualTo(EDS);
assertThat(type.typeName()).isEqualTo("EDS");
assertThat(watchers).doesNotContainKey(resourceName);
watchers.put(resourceName, (ResourceWatcher<EdsUpdate>) watcher);
}
@ -1186,7 +1185,7 @@ public class ClusterResolverLoadBalancerTest {
<T extends ResourceUpdate> void cancelXdsResourceWatch(XdsResourceType<T> type,
String resourceName,
ResourceWatcher<T> watcher) {
assertThat(type.typeName()).isEqualTo(EDS);
assertThat(type.typeName()).isEqualTo("EDS");
assertThat(watchers).containsKey(resourceName);
watchers.remove(resourceName);
}

40
xds/src/test/java/io/grpc/xds/CsdsServiceTest.java
View File

@ -17,10 +17,6 @@
package io.grpc.xds;
import static com.google.common.truth.Truth.assertThat;
import static io.grpc.xds.AbstractXdsClient.ResourceType.CDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.EDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.LDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.RDS;
import static org.junit.Assert.fail;
import com.google.common.collect.ImmutableList;
@ -49,13 +45,12 @@ import io.grpc.internal.ObjectPool;
import io.grpc.internal.testing.StreamRecorder;
import io.grpc.stub.StreamObserver;
import io.grpc.testing.GrpcServerRule;
import io.grpc.xds.AbstractXdsClient.ResourceType;
import io.grpc.xds.Bootstrapper.BootstrapInfo;
import io.grpc.xds.Bootstrapper.ServerInfo;
import io.grpc.xds.XdsClient.ResourceMetadata;
import io.grpc.xds.XdsClient.ResourceMetadata.ResourceMetadataStatus;
import io.grpc.xds.XdsNameResolverProvider.XdsClientPoolFactory;
import java.util.EnumMap;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
@ -81,6 +76,10 @@ public class CsdsServiceTest {
.node(BOOTSTRAP_NODE)
.build();
private static final XdsClient XDS_CLIENT_NO_RESOURCES = new FakeXdsClient();
private static final XdsResourceType<?> LDS = XdsListenerResource.getInstance();
private static final XdsResourceType<?> CDS = XdsClusterResource.getInstance();
private static final XdsResourceType<?> RDS = XdsRouteConfigureResource.getInstance();
private static final XdsResourceType<?> EDS = XdsEndpointResource.getInstance();
@RunWith(JUnit4.class)
public static class ServiceTests {
@ -126,7 +125,7 @@ public class CsdsServiceTest {
public void fetchClientConfig_unexpectedException() {
XdsClient throwingXdsClient = new FakeXdsClient() {
@Override
ListenableFuture<Map<ResourceType, Map<String, ResourceMetadata>>>
ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>
getSubscribedResourcesMetadataSnapshot() {
return Futures.immediateFailedFuture(
new IllegalArgumentException("IllegalArgumentException"));
@ -150,12 +149,12 @@ public class CsdsServiceTest {
public void fetchClientConfig_interruptedException() {
XdsClient throwingXdsClient = new FakeXdsClient() {
@Override
ListenableFuture<Map<ResourceType, Map<String, ResourceMetadata>>>
ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>
getSubscribedResourcesMetadataSnapshot() {
return Futures.submit(
new Callable<Map<ResourceType, Map<String, ResourceMetadata>>>() {
new Callable<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>() {
@Override
public Map<ResourceType, Map<String, ResourceMetadata>> call() {
public Map<XdsResourceType<?>, Map<String, ResourceMetadata>> call() {
Thread.currentThread().interrupt();
return null;
}
@ -323,9 +322,9 @@ public class CsdsServiceTest {
throws InterruptedException {
ClientConfig clientConfig = CsdsService.getClientConfigForXdsClient(new FakeXdsClient() {
@Override
protected Map<ResourceType, Map<String, ResourceMetadata>>
protected Map<XdsResourceType<?>, Map<String, ResourceMetadata>>
getSubscribedResourcesMetadata() {
return new ImmutableMap.Builder<ResourceType, Map<String, ResourceMetadata>>()
return new ImmutableMap.Builder<XdsResourceType<?>, Map<String, ResourceMetadata>>()
.put(LDS, ImmutableMap.of("subscribedResourceName.LDS", METADATA_ACKED_LDS))
.put(RDS, ImmutableMap.of("subscribedResourceName.RDS", METADATA_ACKED_RDS))
.put(CDS, ImmutableMap.of("subscribedResourceName.CDS", METADATA_ACKED_CDS))
@ -340,7 +339,7 @@ public class CsdsServiceTest {
// is propagated to the correct resource types.
int xdsConfigCount = clientConfig.getGenericXdsConfigsCount();
assertThat(xdsConfigCount).isEqualTo(4);
EnumMap<ResourceType, GenericXdsConfig> configDumps = mapConfigDumps(clientConfig);
Map<XdsResourceType<?>, GenericXdsConfig> configDumps = mapConfigDumps(clientConfig);
assertThat(configDumps.keySet()).containsExactly(LDS, RDS, CDS, EDS);
// LDS.
@ -384,7 +383,7 @@ public class CsdsServiceTest {
private static void verifyClientConfigNoResources(ClientConfig clientConfig) {
int xdsConfigCount = clientConfig.getGenericXdsConfigsCount();
assertThat(xdsConfigCount).isEqualTo(0);
EnumMap<ResourceType, GenericXdsConfig> configDumps = mapConfigDumps(clientConfig);
Map<XdsResourceType<?>, GenericXdsConfig> configDumps = mapConfigDumps(clientConfig);
assertThat(configDumps).isEmpty();
}
@ -398,12 +397,12 @@ public class CsdsServiceTest {
assertThat(node).isEqualTo(BOOTSTRAP_NODE.toEnvoyProtoNode());
}
private static EnumMap<ResourceType, GenericXdsConfig> mapConfigDumps(ClientConfig config) {
EnumMap<ResourceType, GenericXdsConfig> xdsConfigMap = new EnumMap<>(ResourceType.class);
private static Map<XdsResourceType<?>, GenericXdsConfig> mapConfigDumps(ClientConfig config) {
Map<XdsResourceType<?>, GenericXdsConfig> xdsConfigMap = new HashMap<>();
List<GenericXdsConfig> xdsConfigList = config.getGenericXdsConfigsList();
for (GenericXdsConfig genericXdsConfig : xdsConfigList) {
ResourceType type = ResourceType.fromTypeUrl(genericXdsConfig.getTypeUrl());
assertThat(type).isNotEqualTo(ResourceType.UNKNOWN);
XdsResourceType<?> type = AbstractXdsClient.fromTypeUrl(genericXdsConfig.getTypeUrl());
assertThat(type).isNotNull();
assertThat(xdsConfigMap).doesNotContainKey(type);
xdsConfigMap.put(type, genericXdsConfig);
}
@ -411,12 +410,13 @@ public class CsdsServiceTest {
}
private static class FakeXdsClient extends XdsClient {
protected Map<ResourceType, Map<String, ResourceMetadata>> getSubscribedResourcesMetadata() {
protected Map<XdsResourceType<?>, Map<String, ResourceMetadata>>
getSubscribedResourcesMetadata() {
return ImmutableMap.of();
}
@Override
ListenableFuture<Map<ResourceType, Map<String, ResourceMetadata>>>
ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>
getSubscribedResourcesMetadataSnapshot() {
return Futures.immediateFuture(getSubscribedResourcesMetadata());
}

22
xds/src/test/java/io/grpc/xds/XdsClientImplDataTest.java
View File

@ -115,7 +115,6 @@ import io.grpc.lookup.v1.GrpcKeyBuilder.Name;
import io.grpc.lookup.v1.NameMatcher;
import io.grpc.lookup.v1.RouteLookupClusterSpecifier;
import io.grpc.lookup.v1.RouteLookupConfig;
import io.grpc.xds.AbstractXdsClient.ResourceType;
import io.grpc.xds.Bootstrapper.ServerInfo;
import io.grpc.xds.ClusterSpecifierPlugin.NamedPluginConfig;
import io.grpc.xds.ClusterSpecifierPlugin.PluginConfig;
@ -2699,21 +2698,28 @@ public class XdsClientImplDataTest {
@Test
public void validateResourceName() {
String traditionalResource = "cluster1.google.com";
assertThat(XdsClient.isResourceNameValid(traditionalResource, ResourceType.CDS.typeUrl()))
assertThat(XdsClient.isResourceNameValid(traditionalResource,
XdsClusterResource.getInstance().typeUrl()))
.isTrue();
assertThat(XdsClient.isResourceNameValid(traditionalResource, ResourceType.RDS.typeUrlV2()))
assertThat(XdsClient.isResourceNameValid(traditionalResource,
XdsRouteConfigureResource.getInstance().typeUrlV2()))
.isTrue();
String invalidPath = "xdstp:/abc/efg";
assertThat(XdsClient.isResourceNameValid(invalidPath, ResourceType.CDS.typeUrl())).isFalse();
assertThat(XdsClient.isResourceNameValid(invalidPath,
XdsClusterResource.getInstance().typeUrl())).isFalse();
String invalidPath2 = "xdstp:///envoy.config.route.v3.RouteConfiguration";
assertThat(XdsClient.isResourceNameValid(invalidPath2, ResourceType.RDS.typeUrl())).isFalse();
assertThat(XdsClient.isResourceNameValid(invalidPath2,
XdsRouteConfigureResource.getInstance().typeUrl())).isFalse();
String typeMatch = "xdstp:///envoy.config.route.v3.RouteConfiguration/foo/route1";
assertThat(XdsClient.isResourceNameValid(typeMatch, ResourceType.LDS.typeUrl())).isFalse();
assertThat(XdsClient.isResourceNameValid(typeMatch, ResourceType.RDS.typeUrl())).isTrue();
assertThat(XdsClient.isResourceNameValid(typeMatch, ResourceType.RDS.typeUrlV2())).isFalse();
assertThat(XdsClient.isResourceNameValid(typeMatch,
XdsListenerResource.getInstance().typeUrl())).isFalse();
assertThat(XdsClient.isResourceNameValid(typeMatch,
XdsRouteConfigureResource.getInstance().typeUrl())).isTrue();
assertThat(XdsClient.isResourceNameValid(typeMatch,
XdsRouteConfigureResource.getInstance().typeUrlV2())).isFalse();
}
@Test

90
xds/src/test/java/io/grpc/xds/XdsClientImplTestBase.java
View File

@ -18,10 +18,6 @@ package io.grpc.xds;
import static com.google.common.truth.Truth.assertThat;
import static com.google.common.truth.Truth.assertWithMessage;
import static io.grpc.xds.AbstractXdsClient.ResourceType.CDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.EDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.LDS;
import static io.grpc.xds.AbstractXdsClient.ResourceType.RDS;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.mock;
@ -66,7 +62,6 @@ import io.grpc.internal.ServiceConfigUtil;
import io.grpc.internal.ServiceConfigUtil.LbConfig;
import io.grpc.internal.TimeProvider;
import io.grpc.testing.GrpcCleanupRule;
import io.grpc.xds.AbstractXdsClient.ResourceType;
import io.grpc.xds.Bootstrapper.AuthorityInfo;
import io.grpc.xds.Bootstrapper.CertificateProviderInfo;
import io.grpc.xds.Bootstrapper.ServerInfo;
@ -137,6 +132,10 @@ public abstract class XdsClientImplTestBase {
private static final Node NODE = Node.newBuilder().setId(NODE_ID).build();
private static final Any FAILING_ANY = MessageFactory.FAILING_ANY;
private static final ChannelCredentials CHANNEL_CREDENTIALS = InsecureChannelCredentials.create();
private static final XdsResourceType<?> LDS = XdsListenerResource.getInstance();
private static final XdsResourceType<?> CDS = XdsClusterResource.getInstance();
private static final XdsResourceType<?> RDS = XdsRouteConfigureResource.getInstance();
private static final XdsResourceType<?> EDS = XdsEndpointResource.getInstance();
// xDS control plane server info.
private ServerInfo xdsServerInfo;
@ -399,15 +398,27 @@ public abstract class XdsClientImplTestBase {
private void verifySubscribedResourcesMetadataSizes(
int ldsSize, int cdsSize, int rdsSize, int edsSize) {
Map<ResourceType, Map<String, ResourceMetadata>> subscribedResourcesMetadata =
Map<XdsResourceType<?>, Map<String, ResourceMetadata>> subscribedResourcesMetadata =
awaitSubscribedResourcesMetadata();
assertThat(subscribedResourcesMetadata.get(LDS)).hasSize(ldsSize);
assertThat(subscribedResourcesMetadata.get(CDS)).hasSize(cdsSize);
assertThat(subscribedResourcesMetadata.get(RDS)).hasSize(rdsSize);
assertThat(subscribedResourcesMetadata.get(EDS)).hasSize(edsSize);
verifyResourceCount(subscribedResourcesMetadata, LDS, ldsSize);
verifyResourceCount(subscribedResourcesMetadata, CDS, cdsSize);
verifyResourceCount(subscribedResourcesMetadata, RDS, rdsSize);
verifyResourceCount(subscribedResourcesMetadata, EDS, edsSize);
}
private Map<ResourceType, Map<String, ResourceMetadata>> awaitSubscribedResourcesMetadata() {
private void verifyResourceCount(
Map<XdsResourceType<?>, Map<String, ResourceMetadata>> subscribedResourcesMetadata,
XdsResourceType<?> type,
int size) {
if (size == 0) {
assertThat(subscribedResourcesMetadata.containsKey(type)).isFalse();
} else {
assertThat(subscribedResourcesMetadata.get(type)).hasSize(size);
}
}
private Map<XdsResourceType<?>, Map<String, ResourceMetadata>>
awaitSubscribedResourcesMetadata() {
try {
return xdsClient.getSubscribedResourcesMetadataSnapshot().get(20, TimeUnit.SECONDS);
} catch (Exception e) {
@ -419,20 +430,20 @@ public abstract class XdsClientImplTestBase {
}
/** Verify the resource requested, but not updated. */
private void verifyResourceMetadataRequested(ResourceType type, String resourceName) {
private void verifyResourceMetadataRequested(XdsResourceType<?> type, String resourceName) {
verifyResourceMetadata(
type, resourceName, null, ResourceMetadataStatus.REQUESTED, "", 0, false);
}
/** Verify that the requested resource does not exist. */
private void verifyResourceMetadataDoesNotExist(ResourceType type, String resourceName) {
private void verifyResourceMetadataDoesNotExist(XdsResourceType<?> type, String resourceName) {
verifyResourceMetadata(
type, resourceName, null, ResourceMetadataStatus.DOES_NOT_EXIST, "", 0, false);
}
/** Verify the resource to be acked. */
private void verifyResourceMetadataAcked(
ResourceType type, String resourceName, Any rawResource, String versionInfo,
XdsResourceType<?> type, String resourceName, Any rawResource, String versionInfo,
long updateTimeNanos) {
verifyResourceMetadata(type, resourceName, rawResource, ResourceMetadataStatus.ACKED,
versionInfo, updateTimeNanos, false);
@ -443,7 +454,7 @@ public abstract class XdsClientImplTestBase {
* corresponding i-th element of {@code List<String> failedDetails}.
*/
private void verifyResourceMetadataNacked(
ResourceType type, String resourceName, Any rawResource, String versionInfo,
XdsResourceType<?> type, String resourceName, Any rawResource, String versionInfo,
long updateTime, String failedVersion, long failedUpdateTimeNanos,
List<String> failedDetails) {
ResourceMetadata resourceMetadata =
@ -465,7 +476,7 @@ public abstract class XdsClientImplTestBase {
}
private ResourceMetadata verifyResourceMetadata(
ResourceType type, String resourceName, Any rawResource, ResourceMetadataStatus status,
XdsResourceType<?> type, String resourceName, Any rawResource, ResourceMetadataStatus status,
String versionInfo, long updateTimeNanos, boolean hasErrorState) {
ResourceMetadata metadata = awaitSubscribedResourcesMetadata().get(type).get(resourceName);
assertThat(metadata).isNotNull();
@ -1297,7 +1308,7 @@ public abstract class XdsClientImplTestBase {
RDS_RESOURCE, rdsResourceWatcher);
Any routeConfig = Any.pack(mf.buildRouteConfiguration("route-bar.googleapis.com",
mf.buildOpaqueVirtualHosts(2)));
call.sendResponse(ResourceType.RDS, routeConfig, VERSION_1, "0000");
call.sendResponse(RDS, routeConfig, VERSION_1, "0000");
// Client sends an ACK RDS request.
call.verifyRequest(RDS, RDS_RESOURCE, VERSION_1, "0000", NODE);
@ -1934,7 +1945,7 @@ public abstract class XdsClientImplTestBase {
mf.buildEdsCluster(CDS_RESOURCE, null, "least_request_experimental", null,
leastRequestConfig, false, null, "envoy.transport_sockets.tls", null, null
));
call.sendResponse(ResourceType.CDS, clusterRingHash, VERSION_1, "0000");
call.sendResponse(CDS, clusterRingHash, VERSION_1, "0000");
// Client sent an ACK CDS request.
call.verifyRequest(CDS, CDS_RESOURCE, VERSION_1, "0000", NODE);
@ -1966,7 +1977,7 @@ public abstract class XdsClientImplTestBase {
mf.buildEdsCluster(CDS_RESOURCE, null, "ring_hash_experimental", ringHashConfig, null,
false, null, "envoy.transport_sockets.tls", null, null
));
call.sendResponse(ResourceType.CDS, clusterRingHash, VERSION_1, "0000");
call.sendResponse(CDS, clusterRingHash, VERSION_1, "0000");
// Client sent an ACK CDS request.
call.verifyRequest(CDS, CDS_RESOURCE, VERSION_1, "0000", NODE);
@ -3353,10 +3364,9 @@ public abstract class XdsClientImplTestBase {
Message listener =
mf.buildListenerWithFilterChain(LISTENER_RESOURCE, 7000, "0.0.0.0", filterChain);
List<Any> listeners = ImmutableList.of(Any.pack(listener));
call.sendResponse(ResourceType.LDS, listeners, "0", "0000");
call.sendResponse(LDS, listeners, "0", "0000");
// Client sends an ACK LDS request.
call.verifyRequest(
ResourceType.LDS, Collections.singletonList(LISTENER_RESOURCE), "0", "0000", NODE);
call.verifyRequest(LDS, Collections.singletonList(LISTENER_RESOURCE), "0", "0000", NODE);
verify(ldsResourceWatcher).onChanged(ldsUpdateCaptor.capture());
EnvoyServerProtoData.Listener parsedListener = ldsUpdateCaptor.getValue().listener();
assertThat(parsedListener.name()).isEqualTo(LISTENER_RESOURCE);
@ -3390,10 +3400,9 @@ public abstract class XdsClientImplTestBase {
Message listener = mf.buildListenerWithFilterChain(
"grpc/server?xds.resource.listening_address=0.0.0.0:8000", 7000, "0.0.0.0", filterChain);
List<Any> listeners = ImmutableList.of(Any.pack(listener));
call.sendResponse(ResourceType.LDS, listeners, "0", "0000");
call.sendResponse(LDS, listeners, "0", "0000");
// Client sends an ACK LDS request.
call.verifyRequest(
ResourceType.LDS, Collections.singletonList(LISTENER_RESOURCE), "0", "0000", NODE);
call.verifyRequest(LDS, Collections.singletonList(LISTENER_RESOURCE), "0", "0000", NODE);
verifyNoInteractions(ldsResourceWatcher);
fakeClock.forwardTime(XdsClientImpl.INITIAL_RESOURCE_FETCH_TIMEOUT_SEC, TimeUnit.SECONDS);
@ -3418,7 +3427,7 @@ public abstract class XdsClientImplTestBase {
Message listener =
mf.buildListenerWithFilterChain(LISTENER_RESOURCE, 7000, "0.0.0.0", filterChain);
List<Any> listeners = ImmutableList.of(Any.pack(listener));
call.sendResponse(ResourceType.LDS, listeners, "0", "0000");
call.sendResponse(LDS, listeners, "0", "0000");
// The response NACKed with errors indicating indices of the failed resources.
String errorMsg = "LDS response Listener \'grpc/server?xds.resource.listening_address="
+ "0.0.0.0:7000\' validation error: "
@ -3445,7 +3454,7 @@ public abstract class XdsClientImplTestBase {
Message listener =
mf.buildListenerWithFilterChain(LISTENER_RESOURCE, 7000, "0.0.0.0", filterChain);
List<Any> listeners = ImmutableList.of(Any.pack(listener));
call.sendResponse(ResourceType.LDS, listeners, "0", "0000");
call.sendResponse(LDS, listeners, "0", "0000");
// The response NACKed with errors indicating indices of the failed resources.
String errorMsg = "LDS response Listener \'grpc/server?xds.resource.listening_address="
+ "0.0.0.0:7000\' validation error: "
@ -3460,28 +3469,27 @@ public abstract class XdsClientImplTestBase {
XdsResourceType<T> type, String name, ResourceWatcher<T> watcher) {
FakeClock.TaskFilter timeoutTaskFilter;
switch (type.typeName()) {
case LDS:
case "LDS":
timeoutTaskFilter = LDS_RESOURCE_FETCH_TIMEOUT_TASK_FILTER;
xdsClient.watchXdsResource(type, name, watcher);
break;
case RDS:
case "RDS":
timeoutTaskFilter = RDS_RESOURCE_FETCH_TIMEOUT_TASK_FILTER;
xdsClient.watchXdsResource(type, name, watcher);
break;
case CDS:
case "CDS":
timeoutTaskFilter = CDS_RESOURCE_FETCH_TIMEOUT_TASK_FILTER;
xdsClient.watchXdsResource(type, name, watcher);
break;
case EDS:
case "EDS":
timeoutTaskFilter = EDS_RESOURCE_FETCH_TIMEOUT_TASK_FILTER;
xdsClient.watchXdsResource(type, name, watcher);
break;
case UNKNOWN:
default:
throw new AssertionError("should never be here");
}
DiscoveryRpcCall call = resourceDiscoveryCalls.poll();
call.verifyRequest(type.typeName(), Collections.singletonList(name), "", "", NODE);
call.verifyRequest(type, Collections.singletonList(name), "", "", NODE);
ScheduledTask timeoutTask =
Iterables.getOnlyElement(fakeClock.getPendingTasks(timeoutTaskFilter));
assertThat(timeoutTask.getDelay(TimeUnit.SECONDS))
@ -3492,19 +3500,20 @@ public abstract class XdsClientImplTestBase {
protected abstract static class DiscoveryRpcCall {
protected abstract void verifyRequest(
ResourceType type, List<String> resources, String versionInfo, String nonce, Node node);
XdsResourceType<?> type, List<String> resources, String versionInfo, String nonce,
Node node);
protected void verifyRequest(
ResourceType type, String resource, String versionInfo, String nonce, Node node) {
XdsResourceType<?> type, String resource, String versionInfo, String nonce, Node node) {
verifyRequest(type, ImmutableList.of(resource), versionInfo, nonce, node);
}
protected abstract void verifyRequestNack(
ResourceType type, List<String> resources, String versionInfo, String nonce, Node node,
List<String> errorMessages);
XdsResourceType<?> type, List<String> resources, String versionInfo, String nonce,
Node node, List<String> errorMessages);
protected void verifyRequestNack(
ResourceType type, String resource, String versionInfo, String nonce, Node node,
XdsResourceType<?> type, String resource, String versionInfo, String nonce, Node node,
List<String> errorMessages) {
verifyRequestNack(type, ImmutableList.of(resource), versionInfo, nonce, node, errorMessages);
}
@ -3512,9 +3521,10 @@ public abstract class XdsClientImplTestBase {
protected abstract void verifyNoMoreRequest();
protected abstract void sendResponse(
ResourceType type, List<Any> resources, String versionInfo, String nonce);
XdsResourceType<?> type, List<Any> resources, String versionInfo, String nonce);
protected void sendResponse(ResourceType type, Any resource, String versionInfo, String nonce) {
protected void sendResponse(XdsResourceType<?> type, Any resource, String versionInfo,
String nonce) {
sendResponse(type, ImmutableList.of(resource), versionInfo, nonce);
}

7
xds/src/test/java/io/grpc/xds/XdsClientImplV2Test.java
View File

@ -92,7 +92,6 @@ import io.grpc.Context;
import io.grpc.Context.CancellationListener;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import io.grpc.xds.AbstractXdsClient.ResourceType;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@ -197,7 +196,7 @@ public class XdsClientImplV2Test extends XdsClientImplTestBase {
@Override
protected void verifyRequest(
ResourceType type, List<String> resources, String versionInfo, String nonce,
XdsResourceType<?> type, List<String> resources, String versionInfo, String nonce,
EnvoyProtoData.Node node) {
verify(requestObserver).onNext(argThat(new DiscoveryRequestMatcher(
node.toEnvoyProtoNodeV2(), versionInfo, resources, type.typeUrlV2(), nonce, null, null)));
@ -205,7 +204,7 @@ public class XdsClientImplV2Test extends XdsClientImplTestBase {
@Override
protected void verifyRequestNack(
ResourceType type, List<String> resources, String versionInfo, String nonce,
XdsResourceType<?> type, List<String> resources, String versionInfo, String nonce,
EnvoyProtoData.Node node, List<String> errorMessages) {
verify(requestObserver).onNext(argThat(new DiscoveryRequestMatcher(
node.toEnvoyProtoNodeV2(), versionInfo, resources, type.typeUrlV2(), nonce,
@ -219,7 +218,7 @@ public class XdsClientImplV2Test extends XdsClientImplTestBase {
@Override
protected void sendResponse(
ResourceType type, List<Any> resources, String versionInfo, String nonce) {
XdsResourceType<?> type, List<Any> resources, String versionInfo, String nonce) {
DiscoveryResponse response =
DiscoveryResponse.newBuilder()
.setVersionInfo(versionInfo)

7
xds/src/test/java/io/grpc/xds/XdsClientImplV3Test.java
View File

@ -99,7 +99,6 @@ import io.grpc.Context;
import io.grpc.Context.CancellationListener;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import io.grpc.xds.AbstractXdsClient.ResourceType;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@ -205,7 +204,7 @@ public class XdsClientImplV3Test extends XdsClientImplTestBase {
@Override
protected void verifyRequest(
ResourceType type, List<String> resources, String versionInfo, String nonce,
XdsResourceType<?> type, List<String> resources, String versionInfo, String nonce,
EnvoyProtoData.Node node) {
verify(requestObserver).onNext(argThat(new DiscoveryRequestMatcher(
node.toEnvoyProtoNode(), versionInfo, resources, type.typeUrl(), nonce, null, null)));
@ -213,7 +212,7 @@ public class XdsClientImplV3Test extends XdsClientImplTestBase {
@Override
protected void verifyRequestNack(
ResourceType type, List<String> resources, String versionInfo, String nonce,
XdsResourceType<?> type, List<String> resources, String versionInfo, String nonce,
EnvoyProtoData.Node node, List<String> errorMessages) {
verify(requestObserver).onNext(argThat(new DiscoveryRequestMatcher(
node.toEnvoyProtoNode(), versionInfo, resources, type.typeUrl(), nonce,
@ -227,7 +226,7 @@ public class XdsClientImplV3Test extends XdsClientImplTestBase {
@Override
protected void sendResponse(
ResourceType type, List<Any> resources, String versionInfo, String nonce) {
XdsResourceType<?> type, List<Any> resources, String versionInfo, String nonce) {
DiscoveryResponse response =
DiscoveryResponse.newBuilder()
.setVersionInfo(versionInfo)

8
xds/src/test/java/io/grpc/xds/XdsNameResolverTest.java
View File

@ -2089,14 +2089,14 @@ public class XdsNameResolverTest {
ResourceWatcher<T> watcher) {
switch (resourceType.typeName()) {
case LDS:
case "LDS":
assertThat(ldsResource).isNull();
assertThat(ldsWatcher).isNull();
assertThat(resourceName).isEqualTo(expectedLdsResourceName);
ldsResource = resourceName;
ldsWatcher = (ResourceWatcher<LdsUpdate>) watcher;
break;
case RDS:
case "RDS":
assertThat(rdsResource).isNull();
assertThat(rdsWatcher).isNull();
rdsResource = resourceName;
@ -2111,14 +2111,14 @@ public class XdsNameResolverTest {
String resourceName,
ResourceWatcher<T> watcher) {
switch (type.typeName()) {
case LDS:
case "LDS":
assertThat(ldsResource).isNotNull();
assertThat(ldsWatcher).isNotNull();
assertThat(resourceName).isEqualTo(expectedLdsResourceName);
ldsResource = null;
ldsWatcher = null;
break;
case RDS:
case "RDS":
assertThat(rdsResource).isNotNull();
assertThat(rdsWatcher).isNotNull();
rdsResource = null;

8
xds/src/test/java/io/grpc/xds/XdsServerTestHelper.java
View File

@ -184,12 +184,12 @@ public class XdsServerTestHelper {
String resourceName,
ResourceWatcher<T> watcher) {
switch (resourceType.typeName()) {
case LDS:
case "LDS":
assertThat(ldsWatcher).isNull();
ldsWatcher = (ResourceWatcher<LdsUpdate>) watcher;
ldsResource.set(resourceName);
break;
case RDS:
case "RDS":
//re-register is not allowed.
assertThat(rdsWatchers.put(resourceName, (ResourceWatcher<RdsUpdate>)watcher)).isNull();
rdsCount.countDown();
@ -203,12 +203,12 @@ public class XdsServerTestHelper {
String resourceName,
ResourceWatcher<T> watcher) {
switch (type.typeName()) {
case LDS:
case "LDS":
assertThat(ldsWatcher).isNotNull();
ldsResource = null;
ldsWatcher = null;
break;
case RDS:
case "RDS":
rdsWatchers.remove(resourceName);
break;
default: