xdsclient: NACK cluster resource if config_source_specifier in lrs_server is not self (#5613)

xds/internal/xdsclient/xdsresource/unmarshal_cds.go

@@ -155,7 +155,14 @@ func validateClusterAndConstructClusterUpdate(cluster *v3clusterpb.Cluster) (Clu
 	// xdsclient bootstrap information now (can be added if necessary). The
 	// ServerConfig will be read and populated by the CDS balancer when
 	// processing this field.
-	if cluster.GetLrsServer().GetSelf() != nil {
+	// According to A27:
+	// If the `lrs_server` field is set, it must have its `self` field set, in
+	// which case the client should use LRS for load reporting. Otherwise
+	// (the `lrs_server` field is not set), LRS load reporting will be disabled.
+	if lrs := cluster.GetLrsServer(); lrs != nil {
+		if lrs.GetSelf() == nil {
+			return ClusterUpdate{}, fmt.Errorf("unsupported config_source_specifier %T in lrs_server field", lrs.ConfigSourceSpecifier)
+		}
 		ret.LRSServerConfig = ClusterLRSServerSelf
 	}
 

xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go

@@ -1503,6 +1503,41 @@ func (s) TestUnmarshalCluster(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name: "cluster resource with non-self lrs_server field",
+			resources: []*anypb.Any{
+				testutils.MarshalAny(&v3clusterpb.Cluster{
+					Name:                 "test",
+					ClusterDiscoveryType: &v3clusterpb.Cluster_Type{Type: v3clusterpb.Cluster_EDS},
+					EdsClusterConfig: &v3clusterpb.Cluster_EdsClusterConfig{
+						EdsConfig: &v3corepb.ConfigSource{
+							ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{
+								Ads: &v3corepb.AggregatedConfigSource{},
+							},
+						},
+						ServiceName: v3Service,
+					},
+					LbPolicy: v3clusterpb.Cluster_ROUND_ROBIN,
+					LrsServer: &v3corepb.ConfigSource{
+						ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{
+							Ads: &v3corepb.AggregatedConfigSource{},
+						},
+					},
+				}),
+			},
+			wantUpdate: map[string]ClusterUpdateErrTuple{
+				"test": {Err: cmpopts.AnyError},
+			},
+			wantMD: UpdateMetadata{
+				Status:  ServiceStatusNACKed,
+				Version: testVersion,
+				ErrState: &UpdateErrorMetadata{
+					Version: testVersion,
+					Err:     cmpopts.AnyError,
+				},
+			},
+			wantErr: true,
+		},
 		{
 			name:      "v2 cluster",
 			resources: []*anypb.Any{v2ClusterAny},