Repository: grpc/grpc-go

Issue:
xds/e2e: Use the `Metadata` field from `BackendOptions` only when it is non-nil
In the `e2e` package, we define a `BackendOptions` struct with a `Metadata` field here: https://github.com/grpc/grpc-go/blob/3074bcdd8f3c52e4b5cc9527592d475780b062a8/internal/testutils/xds/e2e/clientresources.go#L655

This `Metadata` is supposed to appear as endpoint level metadata in the `ClusterLoadAssignment` resource. The problem though is that when we build the resource in `EndpointResourceWithOptions`, we unconditionally use the `Metadata` field without checking if it is non-nil. See: https://github.com/grpc/grpc-go/blob/3074bcdd8f3c52e4b5cc9527592d475780b062a8/internal/testutils/xds/e2e/clientresources.go#L715. This results in the `ClusterLoadAssignment` resource having an empty JSON metadata.

See here for where this came up: https://github.com/grpc/grpc-go/pull/8537#discussion_r2298844760

Ideally:
- We should only use the `Metadata` field from `BackendOptions` when it is non-nil
- Once we do the above, we should be able to fix existing tests from not having to expect an empty JSON metadata at the endpoint level

Discussion:
@hugehoo: @easwars hello, i would like to take this issue. 

When `b.Metadata` is nil, `&v3endpointpb.LbEndpoint` struct's Metadata field would remain nil/unset, and it seems ok because it's configured as omitempty field. 
``` go
type LbEndpoint struct {
	state         protoimpl.MessageState
	sizeCache     protoimpl.SizeCache
	unknownFields protoimpl.UnknownFields

	...

	// The endpoint metadata specifies values that may be used by the load
	// balancer to select endpoints in a cluster for a given request. The filter
	// name should be specified as “envoy.lb“. An example boolean key-value pair
	// is “canary“, providing the optional canary status of the upstream host.
	// This may be matched against in a route's
	// :ref:`RouteAction <envoy_v3_api_msg_config.route.v3.RouteAction>` metadata_match field
	// to subset the endpoints considered in cluster load balancing.
	Metadata *v3.Metadata `protobuf:"bytes,3,opt,name=metadata,proto3" json:"metadata,omitempty"`
...
}
```

Solution patch:
diff --git a/internal/xds/resolver/serviceconfig.go b/internal/xds/resolver/serviceconfig.go
index 2b4e5a05..f2ceabe7 100644
--- a/internal/xds/resolver/serviceconfig.go
+++ b/internal/xds/resolver/serviceconfig.go
@@ -151,8 +151,11 @@ func (cs *erroringConfigSelector) SelectConfig(iresolver.RPCInfo) (*iresolver.RP
 func (cs *erroringConfigSelector) stop() {}
 
 type configSelector struct {
-	r                *xdsResolver
-	xdsNodeID        string
+	channelID            uint64 // Static hash when hash policy is HashPolicyTypeChannelID
+	xdsNodeID            string // xDS node ID, for annotating errors.
+	sendNewServiceConfig func() // Function to send a new service config to gRPC.
+
+	// Configuration received from the xDS management server.
 	virtualHost      virtualHost
 	routes           []route
 	clusters         map[string]*clusterInfo
@@ -214,9 +217,7 @@ func (cs *configSelector) SelectConfig(rpcInfo iresolver.RPCInfo) (*iresolver.RP
 			if v := atomic.AddInt32(ref, -1); v == 0 {
 				// This entry will be removed from activeClusters when
 				// producing the service config for the empty update.
-				cs.r.serializer.TrySchedule(func(context.Context) {
-					cs.r.onClusterRefDownToZero()
-				})
+				cs.sendNewServiceConfig()
 			}
 		},
 		Interceptor: interceptor,
@@ -281,7 +282,7 @@ func (cs *configSelector) generateHash(rpcInfo iresolver.RPCInfo, hashPolicies [
 			generatedPolicyHash = true
 		case xdsresource.HashPolicyTypeChannelID:
 			// Use the static channel ID as the hash for this policy.
-			policyHash = cs.r.channelID
+			policyHash = cs.channelID
 			generatedHash = true
 			generatedPolicyHash = true
 		}
@@ -358,9 +359,7 @@ func (cs *configSelector) stop() {
 	// selector; we need another update to delete clusters from the config (if
 	// we don't have another update pending already).
 	if needUpdate {
-		cs.r.serializer.TrySchedule(func(context.Context) {
-			cs.r.onClusterRefDownToZero()
-		})
+		cs.sendNewServiceConfig()
 	}
 }
 
diff --git a/internal/xds/resolver/xds_resolver.go b/internal/xds/resolver/xds_resolver.go
index 70291f8c..4ed400c3 100644
--- a/internal/xds/resolver/xds_resolver.go
+++ b/internal/xds/resolver/xds_resolver.go
@@ -330,8 +330,13 @@ func (r *xdsResolver) sendNewServiceConfig(cs stoppableConfigSelector) bool {
 // Only executed in the context of a serializer callback.
 func (r *xdsResolver) newConfigSelector() *configSelector {
 	cs := &configSelector{
-		r:         r,
+		channelID: r.channelID,
 		xdsNodeID: r.xdsClient.BootstrapConfig().Node().GetId(),
+		sendNewServiceConfig: func() {
+			r.serializer.TrySchedule(func(context.Context) {
+				r.sendNewServiceConfig(r.curConfigSelector)
+			})
+		},
 		virtualHost: virtualHost{
 			httpFilterConfigOverride: r.currentVirtualHost.HTTPFilterConfigOverride,
 			retryConfig:              r.currentVirtualHost.RetryConfig,
@@ -607,8 +612,3 @@ func (r *xdsResolver) onRouteConfigResourceError(name string, err error) {
 	}
 	r.onResourceError(err)
 }
-
-// Only executed in the context of a serializer callback.
-func (r *xdsResolver) onClusterRefDownToZero() {
-	r.sendNewServiceConfig(r.curConfigSelector)
-}



Test patch:
diff --git a/internal/testutils/xds/e2e/clientresources.go b/internal/testutils/xds/e2e/clientresources.go
index b8ebf8ba..be375d40 100644
--- a/internal/testutils/xds/e2e/clientresources.go
+++ b/internal/testutils/xds/e2e/clientresources.go
@@ -712,11 +712,7 @@ func EndpointResourceWithOptions(opts EndpointOptions) *v3endpointpb.ClusterLoad
 					},
 				}
 			}
-			metadata, err := structpb.NewStruct(b.Metadata)
-			if err != nil {
-				panic(err)
-			}
-			lbEndpoints = append(lbEndpoints, &v3endpointpb.LbEndpoint{
+			lbEndpoint := &v3endpointpb.LbEndpoint{
 				HostIdentifier: &v3endpointpb.LbEndpoint_Endpoint{Endpoint: &v3endpointpb.Endpoint{
 					Address: &v3corepb.Address{Address: &v3corepb.Address_SocketAddress{
 						SocketAddress: &v3corepb.SocketAddress{
@@ -729,12 +725,20 @@ func EndpointResourceWithOptions(opts EndpointOptions) *v3endpointpb.ClusterLoad
 				}},
 				HealthStatus:        b.HealthStatus,
 				LoadBalancingWeight: &wrapperspb.UInt32Value{Value: b.Weight},
-				Metadata: &v3corepb.Metadata{
+			}
+
+			if b.Metadata != nil {
+				metadata, err := structpb.NewStruct(b.Metadata)
+				if err != nil {
+					panic(fmt.Sprintf("failed to marshal metadata: %v", err))
+				}
+				lbEndpoint.Metadata = &v3corepb.Metadata{
 					FilterMetadata: map[string]*structpb.Struct{
 						"envoy.lb": metadata,
 					},
-				},
-			})
+				}
+			}
+			lbEndpoints = append(lbEndpoints, lbEndpoint)
 		}
 
 		l := locality.Locality
diff --git a/internal/xds/resolver/serviceconfig_test.go b/internal/xds/resolver/serviceconfig_test.go
index 3570ff2e..f05a7d91 100644
--- a/internal/xds/resolver/serviceconfig_test.go
+++ b/internal/xds/resolver/serviceconfig_test.go
@@ -29,7 +29,6 @@ import (
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/grpcutil"
 	iresolver "google.golang.org/grpc/internal/resolver"
-	"google.golang.org/grpc/internal/testutils"
 	_ "google.golang.org/grpc/internal/xds/balancer/cdsbalancer" // To parse LB config
 	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
 	"google.golang.org/grpc/metadata"
@@ -64,12 +63,8 @@ func (s) TestPruneActiveClusters(t *testing.T) {
 
 func (s) TestGenerateRequestHash(t *testing.T) {
 	const channelID = 12378921
-	cs := &configSelector{
-		r: &xdsResolver{
-			cc:        &testutils.ResolverClientConn{Logger: t},
-			channelID: channelID,
-		},
-	}
+	cs := &configSelector{channelID: channelID}
+
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	tests := []struct {