xds: fix difference between user target name and resource name (#2845)

balancer/xds/lrs/lrs.go

@@ -64,7 +64,6 @@ func newRPCCountData() *rpcCountData {
 // lrsStore collects loads from xds balancer, and periodically sends load to the
 // server.
 type lrsStore struct {
-	serviceName  string
 	node         *basepb.Node
 	backoff      backoff.Strategy
 	lastReported time.Time
@@ -76,7 +75,6 @@ type lrsStore struct {
 // NewStore creates a store for load reports.
 func NewStore(serviceName string) Store {
 	return &lrsStore{
-		serviceName: serviceName,
 		node: &basepb.Node{
 			Metadata: &structpb.Struct{
 				Fields: map[string]*structpb.Value{
@@ -130,7 +128,7 @@ func (ls *lrsStore) CallFinished(l internal.Locality, err error) {
 	}
 }
 
-func (ls *lrsStore) buildStats() []*loadreportpb.ClusterStats {
+func (ls *lrsStore) buildStats(clusterName string) []*loadreportpb.ClusterStats {
 	var (
 		totalDropped  uint64
 		droppedReqs   []*loadreportpb.ClusterStats_DroppedRequests
@@ -179,7 +177,7 @@ func (ls *lrsStore) buildStats() []*loadreportpb.ClusterStats {
 
 	var ret []*loadreportpb.ClusterStats
 	ret = append(ret, &loadreportpb.ClusterStats{
-		ClusterName:           ls.serviceName,
+		ClusterName:           clusterName,
 		UpstreamLocalityStats: localityStats,
 
 		TotalDroppedRequests: totalDropped,
@@ -239,8 +237,8 @@ func (ls *lrsStore) ReportTo(ctx context.Context, cc *grpc.ClientConn) {
 			grpclog.Infof("lrs: failed to convert report interval: %v", err)
 			continue
 		}
-		if len(first.Clusters) != 1 || first.Clusters[0] != ls.serviceName {
+		if len(first.Clusters) != 1 {
-			grpclog.Infof("lrs: received clusters %v, expect one cluster %q", first.Clusters, ls.serviceName)
+			grpclog.Infof("lrs: received multiple clusters %v, expect one cluster", first.Clusters)
 			continue
 		}
 		if first.ReportEndpointGranularity {
@@ -267,7 +265,7 @@ func (ls *lrsStore) sendLoads(ctx context.Context, stream lrsgrpc.LoadReportingS
 		}
 		if err := stream.Send(&lrspb.LoadStatsRequest{
 			Node:         ls.node,
-			ClusterStats: ls.buildStats(),
+			ClusterStats: ls.buildStats(clusterName),
 		}); err != nil {
 			grpclog.Infof("lrs: failed to send report: %v", err)
 			return

balancer/xds/lrs/lrs_test.go

@@ -158,7 +158,7 @@ func Test_lrsStore_buildStats_drops(t *testing.T) {
 				}
 				wg.Wait()
 
-				if got := ls.buildStats(); !equalClusterStats(got, want) {
+				if got := ls.buildStats(testService); !equalClusterStats(got, want) {
 					t.Errorf("lrsStore.buildStats() = %v, want %v", got, want)
 					t.Errorf("%s", cmp.Diff(got, want))
 				}
@@ -286,7 +286,7 @@ func Test_lrsStore_buildStats_rpcCounts(t *testing.T) {
 				}
 				wg.Wait()
 
-				if got := ls.buildStats(); !equalClusterStats(got, want) {
+				if got := ls.buildStats(testService); !equalClusterStats(got, want) {
 					t.Errorf("lrsStore.buildStats() = %v, want %v", got, want)
 					t.Errorf("%s", cmp.Diff(got, want))
 				}

balancer/xds/xds.go

@@ -423,9 +423,8 @@ func (x *xdsBalancer) newADSResponse(ctx context.Context, resp proto.Message) er
 	var update interface{}
 	switch u := resp.(type) {
 	case *cdspb.Cluster:
-		if u.GetName() != x.buildOpts.Target.Endpoint {
+		// TODO: EDS requests should use CDS response's Name. Store
-			return fmt.Errorf("unmatched service name, got %s, want %s", u.GetName(), x.buildOpts.Target.Endpoint)
+		// `u.GetName()` in `x.clusterName` and use it in xds_client.
-		}
 		if u.GetType() != cdspb.Cluster_EDS {
 			return fmt.Errorf("unexpected service discovery type, got %v, want %v", u.GetType(), cdspb.Cluster_EDS)
 		}

balancer/xds/xds_client.go

@@ -149,14 +149,26 @@ func (c *client) newEDSRequest() *discoverypb.DiscoveryRequest {
 		Node: &basepb.Node{
 			Metadata: &structpb.Struct{
 				Fields: map[string]*structpb.Value{
+					internal.GrpcHostname: {
+						Kind: &structpb.Value_StringValue{StringValue: c.serviceName},
+					},
 					endpointRequired: {
 						Kind: &structpb.Value_BoolValue{BoolValue: c.enableCDS},
 					},
 				},
 			},
 		},
-		ResourceNames: []string{c.serviceName},
+		// TODO: the expected ResourceName could be in a different format from
-		TypeUrl:       edsType,
+		// dial target. (test_service.test_namespace.traffic_director.com vs
+		// test_namespace:test_service).
+		//
+		// The solution today is to always include GrpcHostname in metadata,
+		// with the value set to dial target.
+		//
+		// A future solution could be: always do CDS, get cluster name from CDS
+		// response, and use it here.
+		// `ResourceNames: []string{c.clusterName},`
+		TypeUrl: edsType,
 	}
 	return edsReq
 }

balancer/xds/xds_client_test.go

@@ -65,27 +65,31 @@ var (
 		Node: &basepb.Node{
 			Metadata: &structpb.Struct{
 				Fields: map[string]*structpb.Value{
+					internal.GrpcHostname: {
+						Kind: &structpb.Value_StringValue{StringValue: testServiceName},
+					},
 					endpointRequired: {
 						Kind: &structpb.Value_BoolValue{BoolValue: true},
 					},
 				},
 			},
 		},
-		ResourceNames: []string{testServiceName},
+		TypeUrl: edsType,
-		TypeUrl:       edsType,
 	}
 	testEDSReqWithoutEndpoints = &discoverypb.DiscoveryRequest{
 		Node: &basepb.Node{
 			Metadata: &structpb.Struct{
 				Fields: map[string]*structpb.Value{
+					internal.GrpcHostname: {
+						Kind: &structpb.Value_StringValue{StringValue: testServiceName},
+					},
 					endpointRequired: {
 						Kind: &structpb.Value_BoolValue{BoolValue: false},
 					},
 				},
 			},
 		},
-		ResourceNames: []string{testServiceName},
+		TypeUrl: edsType,
-		TypeUrl:       edsType,
 	}
 	testCluster = &cdspb.Cluster{
 		Name:                 testServiceName,