xds/xdsclient: add EDS resource endpoint address duplication check (#…

…5715)
grpc · Oct 12, 2022 · e163a90 · e163a90
1 parent 9eba574
commit e163a90
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 3 deletions.
diff --git a/xds/internal/xdsclient/xdsresource/unmarshal_eds.go b/xds/internal/xdsclient/xdsresource/unmarshal_eds.go
@@ -91,7 +91,7 @@ func parseDropPolicy(dropPolicy *v3endpointpb.ClusterLoadAssignment_Policy_DropO
 	}
 }
 
-func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint) ([]Endpoint, error) {
+func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs map[string]bool) ([]Endpoint, error) {
 	endpoints := make([]Endpoint, 0, len(lbEndpoints))
 	for _, lbEndpoint := range lbEndpoints {
 		// If the load_balancing_weight field is specified, it must be set to a
@@ -104,9 +104,14 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint) ([]Endpoint, error)
 			}
 			weight = w.GetValue()
 		}
+		addr := parseAddress(lbEndpoint.GetEndpoint().GetAddress().GetSocketAddress())
+		if uniqueEndpointAddrs[addr] {
+			return nil, fmt.Errorf("duplicate endpoint with the same address %s", addr)
+		}
+		uniqueEndpointAddrs[addr] = true
 		endpoints = append(endpoints, Endpoint{
 			HealthStatus: EndpointHealthStatus(lbEndpoint.GetHealthStatus()),
-			Address:      parseAddress(lbEndpoint.GetEndpoint().GetAddress().GetSocketAddress()),
+			Address:      addr,
 			Weight:       weight,
 		})
 	}
@@ -120,6 +125,7 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment, logger *grpclog.Pr
 	}
 	priorities := make(map[uint32]map[string]bool)
 	sumOfWeights := make(map[uint32]uint64)
+	uniqueEndpointAddrs := make(map[string]bool)
 	for _, locality := range m.Endpoints {
 		l := locality.GetLocality()
 		if l == nil {
@@ -150,7 +156,7 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment, logger *grpclog.Pr
 			return EndpointsUpdate{}, fmt.Errorf("duplicate locality %s with the same priority %v", lidStr, priority)
 		}
 		localitiesWithPriority[lidStr] = true
-		endpoints, err := parseEndpoints(locality.GetLbEndpoints())
+		endpoints, err := parseEndpoints(locality.GetLbEndpoints(), uniqueEndpointAddrs)
 		if err != nil {
 			return EndpointsUpdate{}, err
 		}

diff --git a/xds/internal/xdsclient/xdsresource/unmarshal_eds_test.go b/xds/internal/xdsclient/xdsresource/unmarshal_eds_test.go
@@ -111,6 +111,17 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 			want:    EndpointsUpdate{},
 			wantErr: true,
 		},
+		{
+			name: "duplicate endpoint address",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 1, []string{"addr:997"}, nil)
+				clab0.addLocality("locality-2", 1, 0, []string{"addr:997"}, nil)
+				return clab0.Build()
+			}(),
+			want:    EndpointsUpdate{},
+			wantErr: true,
+		},
 		{
 			name: "good",
 			m: func() *v3endpointpb.ClusterLoadAssignment {