xds/internal/xdsclient: Process string metadata in CDS for com.google.csm.telemetry_labels (#7085)

Author: zasweq

xds/internal/xdsclient/xdsresource/type_cds.go

@@ -85,4 +85,8 @@ type ClusterUpdate struct {
 
 	// Raw is the resource from the xds response.
 	Raw *anypb.Any
+	// TelemetryLabels are the string valued metadata of filter_metadata type
+	// "com.google.csm.telemetry_labels" with keys "service_name" or
+	// "service_namespace".
+	TelemetryLabels map[string]string
 }

xds/internal/xdsclient/xdsresource/unmarshal_cds.go

@@ -39,6 +39,7 @@ import (
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 // ValidateClusterAndConstructClusterUpdateForTesting exports the
@@ -81,6 +82,24 @@ const (
 )
 
 func validateClusterAndConstructClusterUpdate(cluster *v3clusterpb.Cluster) (ClusterUpdate, error) {
+	telemetryLabels := make(map[string]string)
+	if fmd := cluster.GetMetadata().GetFilterMetadata(); fmd != nil {
+		if val, ok := fmd["com.google.csm.telemetry_labels"]; ok {
+			if fields := val.GetFields(); fields != nil {
+				if val, ok := fields["service_name"]; ok {
+					if _, ok := val.GetKind().(*structpb.Value_StringValue); ok {
+						telemetryLabels["service_name"] = val.GetStringValue()
+					}
+				}
+				if val, ok := fields["service_namespace"]; ok {
+					if _, ok := val.GetKind().(*structpb.Value_StringValue); ok {
+						telemetryLabels["service_namespace"] = val.GetStringValue()
+					}
+				}
+			}
+		}
+	}
+
 	var lbPolicy json.RawMessage
 	var err error
 	switch cluster.GetLbPolicy() {
@@ -160,6 +179,7 @@ func validateClusterAndConstructClusterUpdate(cluster *v3clusterpb.Cluster) (Clu
 		MaxRequests:      circuitBreakersFromCluster(cluster),
 		LBPolicy:         lbPolicy,
 		OutlierDetection: od,
+		TelemetryLabels:  telemetryLabels,
 	}
 
 	// Note that this is different from the gRFC (gRFC A47 says to include the

xds/internal/xdsclient/xdsresource/unmarshal_cds_test.go

@@ -41,6 +41,7 @@ import (
 	v3matcherpb "github.com/envoyproxy/go-control-plane/envoy/type/matcher/v3"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/structpb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
@@ -1216,6 +1217,71 @@ func (s) TestUnmarshalCluster(t *testing.T) {
 				},
 			},
 		})
+
+		v3ClusterAnyWithTelemetryLabels = testutils.MarshalAny(t, &v3clusterpb.Cluster{
+			Name:                 v3ClusterName,
+			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_Self{
+					Self: &v3corepb.SelfConfigSource{},
+				},
+			},
+			Metadata: &v3corepb.Metadata{
+				FilterMetadata: map[string]*structpb.Struct{
+					"com.google.csm.telemetry_labels": {
+						Fields: map[string]*structpb.Value{
+							"service_name":      structpb.NewStringValue("grpc-service"),
+							"service_namespace": structpb.NewStringValue("grpc-service-namespace"),
+						},
+					},
+				},
+			},
+		})
+		v3ClusterAnyWithTelemetryLabelsIgnoreSome = testutils.MarshalAny(t, &v3clusterpb.Cluster{
+			Name:                 v3ClusterName,
+			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_Self{
+					Self: &v3corepb.SelfConfigSource{},
+				},
+			},
+			Metadata: &v3corepb.Metadata{
+				FilterMetadata: map[string]*structpb.Struct{
+					"com.google.csm.telemetry_labels": {
+						Fields: map[string]*structpb.Value{
+							"string-value-should-ignore": structpb.NewStringValue("string-val"),
+							"float-value-ignore":         structpb.NewNumberValue(3),
+							"bool-value-ignore":          structpb.NewBoolValue(false),
+							"service_name":               structpb.NewStringValue("grpc-service"), // shouldn't ignore
+							"service_namespace":          structpb.NewNullValue(),                 // should ignore - wrong type
+						},
+					},
+					"ignore-this-metadata": { // should ignore this filter_metadata
+						Fields: map[string]*structpb.Value{
+							"service_namespace": structpb.NewStringValue("string-val-should-ignore"),
+						},
+					},
+				},
+			},
+		})
 	)
 
 	tests := []struct {
@@ -1300,6 +1366,35 @@ func (s) TestUnmarshalCluster(t *testing.T) {
 				Raw: v3ClusterAnyWithEDSConfigSourceSelf,
 			},
 		},
+		{
+			name:     "v3 cluster with telemetry case",
+			resource: v3ClusterAnyWithTelemetryLabels,
+			wantName: v3ClusterName,
+			wantUpdate: ClusterUpdate{
+				ClusterName:     v3ClusterName,
+				EDSServiceName:  v3Service,
+				LRSServerConfig: ClusterLRSServerSelf,
+				Raw:             v3ClusterAnyWithTelemetryLabels,
+				TelemetryLabels: map[string]string{
+					"service_name":      "grpc-service",
+					"service_namespace": "grpc-service-namespace",
+				},
+			},
+		},
+		{
+			name:     "v3 metadata ignore other types not string and not com.google.csm.telemetry_labels",
+			resource: v3ClusterAnyWithTelemetryLabelsIgnoreSome,
+			wantName: v3ClusterName,
+			wantUpdate: ClusterUpdate{
+				ClusterName:     v3ClusterName,
+				EDSServiceName:  v3Service,
+				LRSServerConfig: ClusterLRSServerSelf,
+				Raw:             v3ClusterAnyWithTelemetryLabelsIgnoreSome,
+				TelemetryLabels: map[string]string{
+					"service_name": "grpc-service",
+				},
+			},
+		},
 		{
 			name: "xdstp cluster resource with unset EDS service name",
 			resource: testutils.MarshalAny(t, &v3clusterpb.Cluster{