xds: store server config for LRS server in xdsresource.ClusterUpdate (#…

…7191)

* xds: support LRS server config

* switch to the new bootstrap package in internal/xds

xds/internal/balancer/cdsbalancer/cdsbalancer.go

@@ -609,21 +609,7 @@ func (b *cdsBalancer) generateDMsForCluster(name string, depth int, dms []cluste
  Cluster: cluster.ClusterName,
  EDSServiceName: cluster.EDSServiceName,
  MaxConcurrentRequests: cluster.MaxRequests,
- }
- if cluster.LRSServerConfig == xdsresource.ClusterLRSServerSelf {
- bootstrapConfig := b.xdsClient.BootstrapConfig()
- parsedName := xdsresource.ParseName(cluster.ClusterName)
- if parsedName.Scheme == xdsresource.FederationScheme {
- // Is a federation resource name, find the corresponding
- // authority server config.
- if cfg, ok := bootstrapConfig.Authorities[parsedName.Authority]; ok {
- dm.LoadReportingServer = cfg.XDSServer
- }
- } else {
- // Not a federation resource name, use the default
- // authority.
- dm.LoadReportingServer = bootstrapConfig.XDSServer
- }
+ LoadReportingServer: cluster.LRSServerConfig,
  }
  case xdsresource.ClusterTypeLogicalDNS:
  dm = clusterresolver.DiscoveryMechanism{

xds/internal/xdsclient/authority.go

@@ -94,12 +94,6 @@ type authorityArgs struct {
  // (although the former is part of the latter) is because authorities in the
  // bootstrap config might contain an empty server config, and in this case,
  // the top-level server config is to be used.
- //
- // There are two code paths from where a new authority struct might be
- // created. One is when a watch is registered for a resource, and one is
- // when load reporting needs to be started. We have the authority name in
- // the first case, but do in the second. We only have the server config in
- // the second case.
  serverCfg *bootstrap.ServerConfig
  bootstrapCfg *bootstrap.Config
  serializer *grpcsync.CallbackSerializer
@@ -156,7 +150,10 @@ func (a *authority) handleResourceUpdate(resourceUpdate transport.ResourceUpdate
  return xdsresource.NewErrorf(xdsresource.ErrorTypeResourceTypeUnsupported, "Resource URL %v unknown in response from server", resourceUpdate.URL)
  }
 
- opts := &xdsresource.DecodeOptions{BootstrapConfig: a.bootstrapCfg}
+ opts := &xdsresource.DecodeOptions{
+ BootstrapConfig: a.bootstrapCfg,
+ ServerConfig: a.serverCfg,
+ }
  updates, md, err := decodeAllResources(opts, rType, resourceUpdate)
  a.updateResourceStateAndScheduleCallbacks(rType, updates, md)
  return err

xds/internal/xdsclient/clientimpl_authority.go

@@ -36,7 +36,7 @@ import (
 // authority, without holding c.authorityMu.
 //
 // Caller must not hold c.authorityMu.
-func (c *clientImpl) findAuthority(n *xdsresource.Name) (_ *authority, unref func(), _ error) {
+func (c *clientImpl) findAuthority(n *xdsresource.Name) (*authority, func(), error) {
  scheme, authority := n.Scheme, n.Authority
 
  c.authorityMu.Lock()

xds/internal/xdsclient/clientimpl_watchers.go

@@ -48,15 +48,6 @@ func (c *clientImpl) WatchResource(rType xdsresource.Type, resourceName string,
  return func() {}
  }
 
- // TODO: replace this with the code does the following when we have
- // implemented generic watch API on the authority:
- // - Parse the resource name and extract the authority.
- // - Locate the corresponding authority object and acquire a reference to
- // it. If the authority is not found, error out.
- // - Call the watchResource() method on the authority.
- // - Return a cancel function to cancel the watch on the authority and to
- // release the reference.
-
  // TODO: Make ParseName return an error if parsing fails, and
  // schedule the OnError callback in that case.
  n := xdsresource.ParseName(resourceName)

xds/internal/xdsclient/tests/resource_update_test.go

@@ -669,9 +669,8 @@ func (s) TestHandleClusterResponseFromManagementServer(t *testing.T) {
  Resources: []*anypb.Any{testutils.MarshalAny(t, resource1)},
  },
  wantUpdate: xdsresource.ClusterUpdate{
- ClusterName: "resource-name-1",
- EDSServiceName: "eds-service-name",
- LRSServerConfig: xdsresource.ClusterLRSServerSelf,
+ ClusterName: "resource-name-1",
+ EDSServiceName: "eds-service-name",
  },
  wantUpdateMetadata: map[string]xdsresource.UpdateWithMD{
  "resource-name-1": {
@@ -689,9 +688,8 @@ func (s) TestHandleClusterResponseFromManagementServer(t *testing.T) {
  Resources: []*anypb.Any{testutils.MarshalAny(t, resource1), testutils.MarshalAny(t, resource2)},
  },
  wantUpdate: xdsresource.ClusterUpdate{
- ClusterName: "resource-name-1",
- EDSServiceName: "eds-service-name",
- LRSServerConfig: xdsresource.ClusterLRSServerSelf,
+ ClusterName: "resource-name-1",
+ EDSServiceName: "eds-service-name",
  },
  wantUpdateMetadata: map[string]xdsresource.UpdateWithMD{
  "resource-name-1": {
@@ -763,6 +761,16 @@ func (s) TestHandleClusterResponseFromManagementServer(t *testing.T) {
  if gotErr != nil && !strings.Contains(gotErr.Error(), test.wantErr) {
  t.Fatalf("Got error from handling update: %v, want %v", gotErr, test.wantErr)
  }
+
+ // For tests expected to succeed, we expect an LRS server config in
+ // the update from the xDS client, because the LRS bit is turned on
+ // in the cluster resource. We *cannot* set the LRS server config in
+ // the test table because we do not have the address of the xDS
+ // server at that point, hence we do it here before verifying the
+ // received update.
+ if test.wantErr == "" {
+ test.wantUpdate.LRSServerConfig = xdstestutils.ServerConfigForAddress(t, mgmtServer.Address)
+ }
  cmpOpts := []cmp.Option{
  cmpopts.EquateEmpty(),
  cmpopts.IgnoreFields(xdsresource.ClusterUpdate{}, "Raw", "LBPolicy"),

xds/internal/xdsclient/xdsresource/cluster_resource_type.go

@@ -55,7 +55,7 @@ type clusterResourceType struct {
 // Decode deserializes and validates an xDS resource serialized inside the
 // provided `Any` proto, as received from the xDS management server.
 func (clusterResourceType) Decode(opts *DecodeOptions, resource *anypb.Any) (*DecodeResult, error) {
- name, cluster, err := unmarshalClusterResource(resource)
+ name, cluster, err := unmarshalClusterResource(resource, opts.ServerConfig)
  switch {
  case name == "":
  // Name is unset only when protobuf deserialization fails.

xds/internal/xdsclient/xdsresource/resource_type.go

@@ -133,9 +133,13 @@ type ResourceData interface {
 // DecodeOptions wraps the options required by ResourceType implementation for
 // decoding configuration received from the xDS management server.
 type DecodeOptions struct {
- // BootstrapConfig contains the bootstrap configuration passed to the
- // top-level xdsClient. This contains useful data for resource validation.
+ // BootstrapConfig contains the complete bootstrap configuration passed to
+ // the xDS client. This contains useful data for resource validation.
  BootstrapConfig *bootstrap.Config
+ // ServerConfig contains the server config (from the above bootstrap
+ // configuration) of the xDS server from which the current resource, for
+ // which Decode() is being invoked, was received.
+ ServerConfig *bootstrap.ServerConfig
 }
 
 // DecodeResult is the result of a decode operation.

xds/internal/xdsclient/xdsresource/tests/unmarshal_cds_test.go

@@ -26,18 +26,20 @@ import (
  "github.com/google/go-cmp/cmp"
  "github.com/google/go-cmp/cmp/cmpopts"
  "google.golang.org/grpc/balancer/leastrequest"
- _ "google.golang.org/grpc/balancer/roundrobin" // To register round_robin load balancer.
  "google.golang.org/grpc/internal/balancer/stub"
  "google.golang.org/grpc/internal/envconfig"
  "google.golang.org/grpc/internal/grpctest"
  iserviceconfig "google.golang.org/grpc/internal/serviceconfig"
  "google.golang.org/grpc/internal/testutils"
  "google.golang.org/grpc/internal/testutils/xds/e2e"
+ "google.golang.org/grpc/internal/xds/bootstrap"
  "google.golang.org/grpc/serviceconfig"
- _ "google.golang.org/grpc/xds" // Register the xDS LB Registry Converters.
  "google.golang.org/grpc/xds/internal/balancer/ringhash"
  "google.golang.org/grpc/xds/internal/balancer/wrrlocality"
  "google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
+ "google.golang.org/protobuf/proto"
+ "google.golang.org/protobuf/types/known/anypb"
+ "google.golang.org/protobuf/types/known/structpb"
  "google.golang.org/protobuf/types/known/wrapperspb"
 
  v3xdsxdstypepb "github.com/cncf/xds/go/xds/type/v3"
@@ -48,9 +50,9 @@ import (
  v3ringhashpb "github.com/envoyproxy/go-control-plane/envoy/extensions/load_balancing_policies/ring_hash/v3"
  v3roundrobinpb "github.com/envoyproxy/go-control-plane/envoy/extensions/load_balancing_policies/round_robin/v3"
  v3wrrlocalitypb "github.com/envoyproxy/go-control-plane/envoy/extensions/load_balancing_policies/wrr_locality/v3"
- "google.golang.org/protobuf/proto"
- "google.golang.org/protobuf/types/known/anypb"
- "google.golang.org/protobuf/types/known/structpb"
+
+ _ "google.golang.org/grpc/balancer/roundrobin" // To register round_robin load balancer.
+ _ "google.golang.org/grpc/xds" // Register the xDS LB Registry Converters.
 )
 
 type s struct {
@@ -66,8 +68,6 @@ const (
  serviceName = "service"
 )
 
-var emptyUpdate = xdsresource.ClusterUpdate{ClusterName: clusterName, LRSServerConfig: xdsresource.ClusterLRSOff}
-
 func wrrLocality(t *testing.T, m proto.Message) *v3wrrlocalitypb.WrrLocality {
  return &v3wrrlocalitypb.WrrLocality{
  EndpointPickingPolicy: &v3clusterpb.LoadBalancingPolicy{
@@ -105,6 +105,7 @@ func (s) TestValidateCluster_Success(t *testing.T) {
  tests := []struct {
  name string
  cluster *v3clusterpb.Cluster
+ serverCfg *bootstrap.ServerConfig
  wantUpdate xdsresource.ClusterUpdate
  wantLBConfig *iserviceconfig.BalancerConfig
  }{
@@ -164,7 +165,8 @@ func (s) TestValidateCluster_Success(t *testing.T) {
  LbPolicy: v3clusterpb.Cluster_ROUND_ROBIN,
  },
  wantUpdate: xdsresource.ClusterUpdate{
- ClusterName: clusterName, LRSServerConfig: xdsresource.ClusterLRSOff, ClusterType: xdsresource.ClusterTypeAggregate,
+ ClusterName: clusterName,
+ ClusterType: xdsresource.ClusterTypeAggregate,
  PrioritizedClusterNames: []string{"a", "b", "c"},
  },
  wantLBConfig: &iserviceconfig.BalancerConfig{
@@ -179,7 +181,7 @@ func (s) TestValidateCluster_Success(t *testing.T) {
  {
  name: "happy-case-no-service-name-no-lrs",
  cluster: e2e.DefaultCluster(clusterName, "", e2e.SecurityLevelNone),
- wantUpdate: emptyUpdate,
+ wantUpdate: xdsresource.ClusterUpdate{ClusterName: clusterName},
  wantLBConfig: &iserviceconfig.BalancerConfig{
  Name: wrrlocality.Name,
  Config: &wrrlocality.LBConfig{
@@ -206,16 +208,17 @@ func (s) TestValidateCluster_Success(t *testing.T) {
  },
  },
  {
- name: "happiest-case",
+ name: "happiest-case-with-lrs",
  cluster: e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
  ClusterName: clusterName,
  ServiceName: serviceName,
  EnableLRS: true,
  }),
+ serverCfg: &bootstrap.ServerConfig{ServerURI: "test-server-uri"},
  wantUpdate: xdsresource.ClusterUpdate{
  ClusterName: clusterName,
  EDSServiceName: serviceName,
- LRSServerConfig: xdsresource.ClusterLRSServerSelf,
+ LRSServerConfig: &bootstrap.ServerConfig{ServerURI: "test-server-uri"},
  },
  wantLBConfig: &iserviceconfig.BalancerConfig{
  Name: wrrlocality.Name,
@@ -248,10 +251,11 @@ func (s) TestValidateCluster_Success(t *testing.T) {
  }
  return c
  }(),
+ serverCfg: &bootstrap.ServerConfig{ServerURI: "test-server-uri"},
  wantUpdate: xdsresource.ClusterUpdate{
  ClusterName: clusterName,
  EDSServiceName: serviceName,
- LRSServerConfig: xdsresource.ClusterLRSServerSelf,
+ LRSServerConfig: &bootstrap.ServerConfig{ServerURI: "test-server-uri"},
  MaxRequests: func() *uint32 { i := uint32(512); return &i }(),
  },
  wantLBConfig: &iserviceconfig.BalancerConfig{
@@ -298,7 +302,8 @@ func (s) TestValidateCluster_Success(t *testing.T) {
  LbPolicy: v3clusterpb.Cluster_LEAST_REQUEST,
  },
  wantUpdate: xdsresource.ClusterUpdate{
- ClusterName: clusterName, EDSServiceName: serviceName,
+ ClusterName: clusterName,
+ EDSServiceName: serviceName,
  },
  wantLBConfig: &iserviceconfig.BalancerConfig{
  Name: "least_request_experimental",
@@ -353,7 +358,8 @@ func (s) TestValidateCluster_Success(t *testing.T) {
  },
  },
  wantUpdate: xdsresource.ClusterUpdate{
- ClusterName: clusterName, EDSServiceName: serviceName,
+ ClusterName: clusterName,
+ EDSServiceName: serviceName,
  },
  wantLBConfig: &iserviceconfig.BalancerConfig{
  Name: "least_request_experimental",
@@ -527,7 +533,7 @@ func (s) TestValidateCluster_Success(t *testing.T) {
 
  for _, test := range tests {
  t.Run(test.name, func(t *testing.T) {
- update, err := xdsresource.ValidateClusterAndConstructClusterUpdateForTesting(test.cluster)
+ update, err := xdsresource.ValidateClusterAndConstructClusterUpdateForTesting(test.cluster, test.serverCfg)
  if err != nil {
  t.Errorf("validateClusterAndConstructClusterUpdate(%+v) failed: %v", test.cluster, err)
  }

xds/internal/xdsclient/xdsresource/type_cds.go

@@ -20,6 +20,7 @@ package xdsresource
 import (
  "encoding/json"
 
+ "google.golang.org/grpc/internal/xds/bootstrap"
  "google.golang.org/protobuf/types/known/anypb"
 )
 
@@ -39,18 +40,6 @@ const (
  ClusterTypeAggregate
 )
 
-// ClusterLRSServerConfigType is the type of LRS server config.
-type ClusterLRSServerConfigType int
-
-const (
- // ClusterLRSOff indicates LRS is off (loads are not reported for this
- // cluster).
- ClusterLRSOff ClusterLRSServerConfigType = iota
- // ClusterLRSServerSelf indicates loads should be reported to the same
- // server (the authority) where the CDS resp is received from.
- ClusterLRSServerSelf
-)
-
 // ClusterUpdate contains information from a received CDS response, which is of
 // interest to the registered CDS watcher.
 type ClusterUpdate struct {
@@ -60,10 +49,10 @@ type ClusterUpdate struct {
  // EDSServiceName is an optional name for EDS. If it's not set, the balancer
  // should watch ClusterName for the EDS resources.
  EDSServiceName string
- // LRSServerConfig contains the server where the load reports should be sent
- // to. This can be change to an interface, to support other types, e.g. a
- // ServerConfig with ServerURI, creds.
- LRSServerConfig ClusterLRSServerConfigType
+ // LRSServerConfig contains configuration about the xDS server that sent
+ // this cluster resource. This is also the server where load reports are to
+ // be sent, for this cluster.
+ LRSServerConfig *bootstrap.ServerConfig
  // SecurityCfg contains security configuration sent by the control plane.
  SecurityCfg *SecurityConfig
  // MaxRequests for circuit breaking, if any (otherwise nil).

xds/internal/xdsclient/xdsresource/unmarshal_cds.go

@@ -34,6 +34,7 @@ import (
  "google.golang.org/grpc/internal/envconfig"
  "google.golang.org/grpc/internal/pretty"
  iserviceconfig "google.golang.org/grpc/internal/serviceconfig"
+ "google.golang.org/grpc/internal/xds/bootstrap"
  "google.golang.org/grpc/internal/xds/matcher"
  "google.golang.org/grpc/xds/internal/xdsclient/xdslbregistry"
  "google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"
@@ -50,7 +51,7 @@ var ValidateClusterAndConstructClusterUpdateForTesting = validateClusterAndConst
 // to this value by the management server.
 const transportSocketName = "envoy.transport_sockets.tls"
 
-func unmarshalClusterResource(r *anypb.Any) (string, ClusterUpdate, error) {
+func unmarshalClusterResource(r *anypb.Any, serverCfg *bootstrap.ServerConfig) (string, ClusterUpdate, error) {
  r, err := UnwrapResource(r)
  if err != nil {
  return "", ClusterUpdate{}, fmt.Errorf("failed to unwrap resource: %v", err)
@@ -64,7 +65,7 @@ func unmarshalClusterResource(r *anypb.Any) (string, ClusterUpdate, error) {
  if err := proto.Unmarshal(r.GetValue(), cluster); err != nil {
  return "", ClusterUpdate{}, fmt.Errorf("failed to unmarshal resource: %v", err)
  }
- cu, err := validateClusterAndConstructClusterUpdate(cluster)
+ cu, err := validateClusterAndConstructClusterUpdate(cluster, serverCfg)
  if err != nil {
  return cluster.GetName(), ClusterUpdate{}, err
  }
@@ -81,7 +82,7 @@ const (
  defaultLeastRequestChoiceCount = 2
 )
 
-func validateClusterAndConstructClusterUpdate(cluster *v3clusterpb.Cluster) (ClusterUpdate, error) {
+func validateClusterAndConstructClusterUpdate(cluster *v3clusterpb.Cluster, serverCfg *bootstrap.ServerConfig) (ClusterUpdate, error) {
  telemetryLabels := make(map[string]string)
  if fmd := cluster.GetMetadata().GetFilterMetadata(); fmd != nil {
  if val, ok := fmd["com.google.csm.telemetry_labels"]; ok {
@@ -182,21 +183,11 @@ func validateClusterAndConstructClusterUpdate(cluster *v3clusterpb.Cluster) (Clu
  TelemetryLabels: telemetryLabels,
  }
 
- // Note that this is different from the gRFC (gRFC A47 says to include the
- // full ServerConfig{URL,creds,server feature} here). This information is
- // not available here, because this function doesn't have access to the
- // xdsclient bootstrap information now (can be added if necessary). The
- // ServerConfig will be read and populated by the CDS balancer when
- // processing this field.
- // 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
+ ret.LRSServerConfig = serverCfg
  }
 
  // Validate and set cluster type from the response.