diff --git a/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java b/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java index 9ad75595ea6..2a44b5f297b 100644 --- a/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java +++ b/xds/src/main/java/io/grpc/xds/XdsEndpointResource.java @@ -116,6 +116,7 @@ private static EdsUpdate processClusterLoadAssignment(ClusterLoadAssignment assi throws ResourceInvalidException { Map> priorities = new HashMap<>(); Map localityLbEndpointsMap = new LinkedHashMap<>(); + Set endpointAddresses = new HashSet<>(); List dropOverloads = new ArrayList<>(); int maxPriority = -1; for (io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints localityLbEndpointsProto @@ -144,6 +145,14 @@ private static EdsUpdate processClusterLoadAssignment(ClusterLoadAssignment assi throw new ResourceInvalidException("ClusterLoadAssignment has duplicate locality:" + locality + " for priority:" + priority); } + for (Endpoints.LbEndpoint endpoint : localityLbEndpoints.endpoints()) { + for (java.net.SocketAddress address : endpoint.eag().getAddresses()) { + if (!endpointAddresses.add(address)) { + throw new ResourceInvalidException( + "ClusterLoadAssignment has duplicate endpoint address: " + address); + } + } + } } if (priorities.size() != maxPriority + 1) { throw new ResourceInvalidException("ClusterLoadAssignment has sparse priorities"); diff --git a/xds/src/test/java/io/grpc/xds/ClusterResolverLoadBalancerTest.java b/xds/src/test/java/io/grpc/xds/ClusterResolverLoadBalancerTest.java index f5fb6beb175..3c34b3e5557 100644 --- a/xds/src/test/java/io/grpc/xds/ClusterResolverLoadBalancerTest.java +++ b/xds/src/test/java/io/grpc/xds/ClusterResolverLoadBalancerTest.java @@ -682,11 +682,15 @@ public void edsUpdatePriorityName_mergeTwoPriorities() { } private LocalityLbEndpoints createEndpoints(Locality locality, int priority) { + // Localities must have distinct endpoint addresses even when their priorities are merged. + String address = ImmutableMap.of( + LOCALITY1, "127.0.0.1", LOCALITY2, "127.0.0.2", LOCALITY3, "127.0.0.3") + .get(locality); return LocalityLbEndpoints.newBuilder() .setLoadBalancingWeight(UInt32Value.of(70)) .setLocality(locality) .setPriority(priority) - .addLbEndpoints(newSocketLbEndpoint("127.0." + priority + ".1", 8080)) + .addLbEndpoints(newSocketLbEndpoint(address, 8080)) .build(); } diff --git a/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java b/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java index 60bb9ab8da2..45cfe1ebbd7 100644 --- a/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java +++ b/xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java @@ -19,6 +19,7 @@ import static com.google.common.truth.Truth.assertThat; import static com.google.common.truth.Truth.assertWithMessage; import static io.grpc.StatusMatcher.statusHasCode; +import static io.grpc.xds.XdsEndpointResource.GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; @@ -47,6 +48,9 @@ import com.google.protobuf.UInt32Value; import com.google.protobuf.util.Durations; import io.envoyproxy.envoy.config.cluster.v3.OutlierDetection; +import io.envoyproxy.envoy.config.core.v3.Address; +import io.envoyproxy.envoy.config.core.v3.SocketAddress; +import io.envoyproxy.envoy.config.endpoint.v3.Endpoint; import io.envoyproxy.envoy.config.route.v3.FilterConfig; import io.envoyproxy.envoy.config.route.v3.WeightedCluster; import io.envoyproxy.envoy.extensions.filters.http.router.v3.Router; @@ -4068,6 +4072,215 @@ public void edsResourceUpdated() { verifySubscribedResourcesMetadataSizes(0, 0, 0, 1); } + @Test + public void edsDuplicateEndpointAcrossLocalities_retainsLastGoodResource() { + DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE, + edsResourceWatcher); + call.sendResponse(EDS, testClusterLoadAssignment, VERSION_1, "0000"); + call.verifyRequest(EDS, EDS_RESOURCE, VERSION_1, "0000", NODE); + verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture()); + EdsUpdate lastGoodUpdate = edsUpdateCaptor.getValue().getValue(); + + Any invalidAssignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE, + ImmutableList.of( + mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1", + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0), + mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2", + mf.buildLbEndpoint("172.44.2.2", 8000, "unhealthy", 2, "host2"), 1, 0)), + ImmutableList.of())); + List errors = ImmutableList.of("EDS response ClusterLoadAssignment '" + EDS_RESOURCE + + "' validation error: ClusterLoadAssignment has duplicate endpoint address: " + + "/172.44.2.2:8000"); + call.sendResponse(EDS, invalidAssignment, VERSION_2, "0001"); + call.verifyRequestNack(EDS, EDS_RESOURCE, VERSION_1, "0001", NODE, errors); + verifyResourceMetadataNacked(EDS, EDS_RESOURCE, testClusterLoadAssignment, VERSION_1, + TIME_INCREMENT, VERSION_2, TIME_INCREMENT * 2, errors, true); + ArgumentCaptor statusCaptor = ArgumentCaptor.forClass(Status.class); + verify(edsResourceWatcher).onAmbientError(statusCaptor.capture()); + assertThat(statusCaptor.getValue().getCode()).isEqualTo(Code.UNAVAILABLE); + assertThat(statusCaptor.getValue().getDescription()).contains(errors.get(0)); + verifyNoMoreInteractions(edsResourceWatcher); + + // A new watcher still receives the last accepted resource. + @SuppressWarnings("unchecked") + ResourceWatcher watcher = mock(ResourceWatcher.class); + xdsClient.watchXdsResource(XdsEndpointResource.getInstance(), EDS_RESOURCE, watcher); + verify(watcher).onResourceChanged(StatusOr.fromValue(lastGoodUpdate)); + verify(watcher).onAmbientError(statusCaptor.getValue()); + + call.sendResponse(EDS, invalidAssignment, VERSION_3, "0002"); + call.verifyRequestNack(EDS, EDS_RESOURCE, VERSION_1, "0002", NODE, errors); + verifyResourceMetadataNacked(EDS, EDS_RESOURCE, testClusterLoadAssignment, VERSION_1, + TIME_INCREMENT, VERSION_3, TIME_INCREMENT * 3, errors, true); + verify(edsResourceWatcher, times(2)).onAmbientError(any()); + verify(watcher, times(2)).onAmbientError(any()); + verifyNoMoreInteractions(edsResourceWatcher, watcher); + + Any validAssignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE, + ImmutableList.of(mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2", + mf.buildLbEndpoint("172.44.2.2", 8001, "healthy", 1, "host2"), 1, 0)), + ImmutableList.of())); + call.sendResponse(EDS, validAssignment, "45", "0003"); + call.verifyRequest(EDS, EDS_RESOURCE, "45", "0003", NODE); + verifyResourceMetadataAcked(EDS, EDS_RESOURCE, validAssignment, "45", TIME_INCREMENT * 4); + verify(edsResourceWatcher, times(2)).onResourceChanged(edsUpdateCaptor.capture()); + assertThat(edsUpdateCaptor.getValue().getValue()).isNotEqualTo(lastGoodUpdate); + verify(watcher).onResourceChanged(edsUpdateCaptor.getValue()); + } + + @Test + public void edsDuplicateEndpointOnColdStart() { + DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE, + edsResourceWatcher); + Any invalidAssignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE, + ImmutableList.of( + mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1", + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0), + mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2", + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host2"), 1, 0)), + ImmutableList.of())); + call.sendResponse(EDS, invalidAssignment, VERSION_1, "0000"); + call.verifyRequestNack(EDS, EDS_RESOURCE, "", "0000", NODE, + ImmutableList.of("EDS response ClusterLoadAssignment '" + EDS_RESOURCE + + "' validation error: ClusterLoadAssignment has duplicate endpoint address:")); + verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture()); + assertThat(edsUpdateCaptor.getValue().hasValue()).isFalse(); + + call.sendResponse(EDS, testClusterLoadAssignment, VERSION_2, "0001"); + call.verifyRequest(EDS, EDS_RESOURCE, VERSION_2, "0001", NODE); + verify(edsResourceWatcher, times(2)).onResourceChanged(edsUpdateCaptor.capture()); + assertThat(edsUpdateCaptor.getValue().hasValue()).isTrue(); + validateGoldenClusterLoadAssignment(edsUpdateCaptor.getValue().getValue()); + } + + @Test + public void edsDuplicateEndpointWithinLocality() { + verifyEdsDuplicateEndpoint(ImmutableList.of( + mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1", + ImmutableList.of( + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host2")), 1, 0))); + } + + @Test + public void edsDuplicateEndpointAcrossPriorities() { + verifyEdsDuplicateEndpoint(ImmutableList.of( + mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1", + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0), + mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2", + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host2"), 1, 1))); + } + + @Test + public void edsDuplicateEndpointInOverwrittenLocality() { + // The second entry overwrites the first in the locality-keyed output map. + verifyEdsDuplicateEndpoint(ImmutableList.of( + mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1", + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0), + mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1", + mf.buildLbEndpoint("172.44.2.3", 8000, "healthy", 1, "host2"), 1, 1), + mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2", + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host3"), 1, 0))); + } + + @Test + public void edsDuplicateEndpointWithEquivalentIpv6Addresses() { + verifyEdsDuplicateEndpoint(ImmutableList.of( + mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1", + mf.buildLbEndpoint("2001:db8::1", 8000, "healthy", 1, "host1"), 1, 0), + mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2", + mf.buildLbEndpoint("2001:0db8:0:0:0:0:0:1", 8000, "healthy", 1, "host2"), 1, 0))); + } + + @Test + public void edsDuplicateEndpointInAdditionalAddresses() { + String originalDualStackProp = + System.setProperty(GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS, "false"); + try { + io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint endpoint = + (io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint) + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"); + Message dualStackEndpoint = endpoint.toBuilder() + .setEndpoint(endpoint.getEndpoint().toBuilder() + .addAdditionalAddresses(Endpoint.AdditionalAddress.newBuilder() + .setAddress(Address.newBuilder().setSocketAddress(SocketAddress.newBuilder() + .setAddress("2001:db8::1").setPortValue(8000))))) + .build(); + Any assignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE, + ImmutableList.of( + mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1", + dualStackEndpoint, 1, 0), + mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2", + mf.buildLbEndpoint("2001:db8::1", 8000, "healthy", 1, "host2"), 1, 0)), + ImmutableList.of())); + DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE, + edsResourceWatcher); + call.sendResponse(EDS, assignment, VERSION_1, "0000"); + call.verifyRequest(EDS, EDS_RESOURCE, VERSION_1, "0000", NODE); + + System.setProperty(GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS, "true"); + call.sendResponse(EDS, assignment, VERSION_2, "0001"); + List errors = ImmutableList.of("EDS response ClusterLoadAssignment '" + EDS_RESOURCE + + "' validation error: ClusterLoadAssignment has duplicate endpoint address:"); + call.verifyRequestNack(EDS, EDS_RESOURCE, VERSION_1, "0001", NODE, errors); + verifyResourceMetadataNacked(EDS, EDS_RESOURCE, assignment, VERSION_1, TIME_INCREMENT, + VERSION_2, TIME_INCREMENT * 2, errors, true); + verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture()); + assertThat(edsUpdateCaptor.getValue().hasValue()).isTrue(); + } finally { + if (originalDualStackProp == null) { + System.clearProperty(GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS); + } else { + System.setProperty(GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS, originalDualStackProp); + } + } + } + + private void verifyEdsDuplicateEndpoint(List localities) { + DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE, + edsResourceWatcher); + Any assignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE, localities, + ImmutableList.of())); + call.sendResponse(EDS, assignment, VERSION_1, "0000"); + call.verifyRequestNack(EDS, EDS_RESOURCE, "", "0000", NODE, + ImmutableList.of("EDS response ClusterLoadAssignment '" + EDS_RESOURCE + + "' validation error: ClusterLoadAssignment has duplicate endpoint address:")); + verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture()); + assertThat(edsUpdateCaptor.getValue().hasValue()).isFalse(); + } + + @Test + public void edsEndpointsWithSameIpAndDifferentPorts() { + DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE, + edsResourceWatcher); + Any assignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE, + ImmutableList.of( + mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1", + mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0), + mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2", + mf.buildLbEndpoint("172.44.2.2", 8001, "healthy", 1, "host2"), 1, 0)), + ImmutableList.of())); + call.sendResponse(EDS, assignment, VERSION_1, "0000"); + call.verifyRequest(EDS, EDS_RESOURCE, VERSION_1, "0000", NODE); + verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture()); + assertThat(edsUpdateCaptor.getValue().getValue().localityLbEndpointsMap).hasSize(2); + } + + @Test + public void edsDuplicateEndpointInZeroWeightLocalityIgnored() { + DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE, + edsResourceWatcher); + Any assignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE, + ImmutableList.of(lbEndpointHealthy, + mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2", + mf.buildLbEndpoint("192.168.0.1", 8080, "healthy", 1, "host2"), 0, 0)), + ImmutableList.of())); + call.sendResponse(EDS, assignment, VERSION_1, "0000"); + call.verifyRequest(EDS, EDS_RESOURCE, VERSION_1, "0000", NODE); + verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture()); + assertThat(edsUpdateCaptor.getValue().getValue().localityLbEndpointsMap).hasSize(1); + } + @Test public void edsDuplicateLocalityInTheSamePriority() { DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE,