Cherry-pick #8411, #8419, #8422, #8445 and #8451 to v1.74.x (#8454)

* xdsclient: preserve original bytes for decoding when the resource is wrapped (#8411)

* xds: Avoid error logs when setting fallback bootstrap config (#8419)

* xdsclient: relay marshalled bytes of complete resource proto to decoders (#8422)

* xds: give up pool lock before closing xdsclient channel (#8445)

* transport: release mutex before returning on expired deadlines in server streams (#8451)

---------

Co-authored-by: Easwar Swaminathan <[email protected]>
Co-authored-by: Arjan Singh Bal <[email protected]>
Co-authored-by: Purnesh Dixit <[email protected]>
Co-authored-by: Doug Fawley <[email protected]>

Author: 5 people

internal/transport/http2_server.go

@@ -602,6 +602,7 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 	}
 
 	if s.ctx.Err() != nil {
+		t.mu.Unlock()
 		// Early abort in case the timeout was zero or so low it already fired.
 		t.controlBuf.put(&earlyAbortStream{
 			httpStatus:     http.StatusOK,

internal/xds/bootstrap/bootstrap.go

@@ -576,6 +576,9 @@ func (c *Config) UnmarshalJSON(data []byte) error {
 // the presence of the errors) and may return a Config object with certain
 // fields left unspecified, in which case the caller should use some sane
 // defaults.
+//
+// This function returns an error if it's unable to parse the contents of the
+// bootstrap config. It returns (nil, nil) if none of the env vars are set.
 func GetConfiguration() (*Config, error) {
 	fName := envconfig.XDSBootstrapFileName
 	fContent := envconfig.XDSBootstrapFileContent
@@ -598,7 +601,7 @@ func GetConfiguration() (*Config, error) {
 		return NewConfigFromContents([]byte(fContent))
 	}
 
-	return nil, fmt.Errorf("bootstrap environment variables (%q or %q) not defined", envconfig.XDSBootstrapFileNameEnv, envconfig.XDSBootstrapFileContentEnv)
+	return nil, nil
 }
 
 // NewConfigFromContents creates a new bootstrap configuration from the provided

internal/xds/bootstrap/bootstrap_test.go

@@ -383,12 +383,19 @@ func (s) TestGetConfiguration_Failure(t *testing.T) {
 	cancel := setupBootstrapOverride(bootstrapFileMap)
 	defer cancel()
 
-	for _, name := range []string{"nonExistentBootstrapFile", "empty", "badJSON", "noBalancerName", "emptyXdsServer"} {
+	for _, name := range []string{"nonExistentBootstrapFile", "badJSON", "noBalancerName", "emptyXdsServer"} {
 		t.Run(name, func(t *testing.T) {
 			testGetConfigurationWithFileNameEnv(t, name, true, nil)
 			testGetConfigurationWithFileContentEnv(t, name, true, nil)
 		})
 	}
+	const name = "empty"
+	t.Run(name, func(t *testing.T) {
+		testGetConfigurationWithFileNameEnv(t, name, true, nil)
+		// If both the env vars are empty, a nil config with a nil error must be
+		// returned.
+		testGetConfigurationWithFileContentEnv(t, name, false, nil)
+	})
 }
 
 // Tests the functionality in GetConfiguration with different bootstrap file
@@ -462,9 +469,9 @@ func (s) TestGetConfiguration_BootstrapEnvPriority(t *testing.T) {
 	envconfig.XDSBootstrapFileContent = ""
 	defer func() { envconfig.XDSBootstrapFileContent = origBootstrapContent }()
 
-	// When both env variables are empty, GetConfiguration should fail.
-	if _, err := GetConfiguration(); err == nil {
-		t.Errorf("GetConfiguration() returned nil error, expected to fail")
+	// When both env variables are empty, GetConfiguration should return nil.
+	if cfg, err := GetConfiguration(); err != nil || cfg != nil {
+		t.Errorf("GetConfiguration() returned (%v, %v), want (<nil>, <nil>)", cfg, err)
 	}
 
 	// When one of them is set, it should be used.

xds/internal/clients/xdsclient/channel.go

@@ -253,16 +253,10 @@ func decodeResponse(opts *DecodeOptions, rType *ResourceType, resp response) (ma
 	perResourceErrors := make(map[string]error) // Tracks resource validation errors, where we have a resource name.
 	ret := make(map[string]dataAndErrTuple)     // Return result, a map from resource name to either resource data or error.
 	for _, r := range resp.resources {
-		r, err := xdsresource.UnwrapResource(r)
-		if err != nil {
-			topLevelErrors = append(topLevelErrors, err)
-			continue
-		}
-		if _, ok := opts.Config.ResourceTypes[r.TypeUrl]; !ok || r.TypeUrl != resp.typeURL {
-			topLevelErrors = append(topLevelErrors, xdsresource.NewErrorf(xdsresource.ErrorTypeResourceTypeUnsupported, "unexpected resource type: %q ", r.GetTypeUrl()))
-			continue
-		}
-		result, err := rType.Decoder.Decode(r.GetValue(), *opts)
+		result, err := rType.Decoder.Decode(AnyProto{
+			TypeURL: r.GetTypeUrl(),
+			Value:   r.GetValue(),
+		}, *opts)
 
 		// Name field of the result is left unpopulated only when resource
 		// deserialization fails.

xds/internal/clients/xdsclient/helpers_test.go

@@ -30,6 +30,7 @@ import (
 	"google.golang.org/grpc/xds/internal/clients/internal/pretty"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
 
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	v3httppb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
@@ -62,17 +63,24 @@ var (
 	}
 )
 
-func unmarshalListenerResource(r []byte) (string, listenerUpdate, error) {
+func unmarshalListenerResource(rProto *anypb.Any) (string, listenerUpdate, error) {
+	rProto, err := xdsresource.UnwrapResource(rProto)
+	if err != nil {
+		return "", listenerUpdate{}, fmt.Errorf("failed to unwrap resource: %v", err)
+	}
+	if !xdsresource.IsListenerResource(rProto.GetTypeUrl()) {
+		return "", listenerUpdate{}, fmt.Errorf("unexpected listener resource type: %q", rProto.GetTypeUrl())
+	}
 	lis := &v3listenerpb.Listener{}
-	if err := proto.Unmarshal(r, lis); err != nil {
+	if err := proto.Unmarshal(rProto.GetValue(), lis); err != nil {
 		return "", listenerUpdate{}, fmt.Errorf("failed to unmarshal resource: %v", err)
 	}
 
 	lu, err := processListener(lis)
 	if err != nil {
 		return lis.GetName(), listenerUpdate{}, err
 	}
-	lu.Raw = r
+	lu.Raw = rProto.GetValue()
 	return lis.GetName(), *lu, nil
 }
 
@@ -154,8 +162,12 @@ type listenerDecoder struct{}
 
 // Decode deserializes and validates an xDS resource serialized inside the
 // provided `Any` proto, as received from the xDS management server.
-func (listenerDecoder) Decode(resource []byte, _ DecodeOptions) (*DecodeResult, error) {
-	name, listener, err := unmarshalListenerResource(resource)
+func (listenerDecoder) Decode(resource AnyProto, _ DecodeOptions) (*DecodeResult, error) {
+	rProto := &anypb.Any{
+		TypeUrl: resource.TypeURL,
+		Value:   resource.Value,
+	}
+	name, listener, err := unmarshalListenerResource(rProto)
 	switch {
 	case name == "":
 		// Name is unset only when protobuf deserialization fails.

xds/internal/clients/xdsclient/resource_type.go

@@ -51,9 +51,21 @@ type Decoder interface {
 	// Decode deserializes and validates an xDS resource as received from the
 	// xDS management server.
 	//
-	// If deserialization fails or resource validation fails, it returns a
-	// non-nil error. Otherwise, returns a fully populated DecodeResult.
-	Decode(resource []byte, options DecodeOptions) (*DecodeResult, error)
+	// The `resource` parameter may contain a value of the serialized wrapped
+	// resource (i.e. with the type URL
+	// `type.googleapis.com/envoy.service.discovery.v3.Resource`).
+	// Implementations are responsible for unwrapping the underlying resource if
+	// it is wrapped.
+	//
+	// If unmarshalling or validation fails, it returns a non-nil error.
+	// Otherwise, returns a fully populated DecodeResult.
+	Decode(resource AnyProto, options DecodeOptions) (*DecodeResult, error)
+}
+
+// AnyProto contains the type URL and serialized proto data of an xDS resource.
+type AnyProto struct {
+	TypeURL string
+	Value   []byte
 }
 
 // DecodeOptions wraps the options required by ResourceType implementations for

xds/internal/clients/xdsclient/test/helpers_test.go

@@ -34,6 +34,7 @@ import (
 	"google.golang.org/grpc/xds/internal/clients/xdsclient"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
 
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	v3httppb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
@@ -73,17 +74,24 @@ var (
 	}
 )
 
-func unmarshalListenerResource(r []byte) (string, listenerUpdate, error) {
+func unmarshalListenerResource(rProto *anypb.Any) (string, listenerUpdate, error) {
+	rProto, err := xdsresource.UnwrapResource(rProto)
+	if err != nil {
+		return "", listenerUpdate{}, fmt.Errorf("failed to unwrap resource: %v", err)
+	}
+	if !xdsresource.IsListenerResource(rProto.GetTypeUrl()) {
+		return "", listenerUpdate{}, fmt.Errorf("unexpected listener resource type: %q", rProto.GetTypeUrl())
+	}
 	lis := &v3listenerpb.Listener{}
-	if err := proto.Unmarshal(r, lis); err != nil {
+	if err := proto.Unmarshal(rProto.GetValue(), lis); err != nil {
 		return "", listenerUpdate{}, fmt.Errorf("failed to unmarshal resource: %v", err)
 	}
 
 	lu, err := processListener(lis)
 	if err != nil {
 		return lis.GetName(), listenerUpdate{}, err
 	}
-	lu.Raw = r
+	lu.Raw = rProto.GetValue()
 	return lis.GetName(), *lu, nil
 }
 
@@ -165,8 +173,12 @@ type listenerDecoder struct{}
 
 // Decode deserializes and validates an xDS resource serialized inside the
 // provided `Any` proto, as received from the xDS management server.
-func (listenerDecoder) Decode(resource []byte, _ xdsclient.DecodeOptions) (*xdsclient.DecodeResult, error) {
-	name, listener, err := unmarshalListenerResource(resource)
+func (listenerDecoder) Decode(resource xdsclient.AnyProto, _ xdsclient.DecodeOptions) (*xdsclient.DecodeResult, error) {
+	rProto := &anypb.Any{
+		TypeUrl: resource.TypeURL,
+		Value:   resource.Value,
+	}
+	name, listener, err := unmarshalListenerResource(rProto)
 	switch {
 	case name == "":
 		// Name is unset only when protobuf deserialization fails.

xds/internal/xdsclient/pool.go

@@ -25,6 +25,7 @@ import (
 
 	v3statuspb "github.com/envoyproxy/go-control-plane/envoy/service/status/v3"
 	estats "google.golang.org/grpc/experimental/stats"
+	"google.golang.org/grpc/internal/envconfig"
 	istats "google.golang.org/grpc/internal/stats"
 	"google.golang.org/grpc/internal/xds/bootstrap"
 	"google.golang.org/protobuf/proto"
@@ -34,7 +35,10 @@ var (
 	// DefaultPool is the default pool for xDS clients. It is created at init
 	// time and reads bootstrap configuration from env vars to create the xDS
 	// client.
-	DefaultPool = &Pool{clients: make(map[string]*clientImpl)}
+	DefaultPool = &Pool{
+		clients:          make(map[string]*clientImpl),
+		getConfiguration: sync.OnceValues(bootstrap.GetConfiguration),
+	}
 )
 
 // Pool represents a pool of xDS clients that share the same bootstrap
@@ -43,9 +47,12 @@ type Pool struct {
 	// Note that mu should ideally only have to guard clients. But here, we need
 	// it to guard config as well since SetFallbackBootstrapConfig writes to
 	// config.
-	mu      sync.Mutex
-	clients map[string]*clientImpl
-	config  *bootstrap.Config
+	mu             sync.Mutex
+	clients        map[string]*clientImpl
+	fallbackConfig *bootstrap.Config
+	// getConfiguration is a sync.OnceValues that attempts to read the bootstrap
+	// configuration from environment variables once.
+	getConfiguration func() (*bootstrap.Config, error)
 }
 
 // OptionsForTesting contains options to configure xDS client creation for
@@ -78,7 +85,9 @@ type OptionsForTesting struct {
 func NewPool(config *bootstrap.Config) *Pool {
 	return &Pool{
 		clients: make(map[string]*clientImpl),
-		config:  config,
+		getConfiguration: func() (*bootstrap.Config, error) {
+			return config, nil
+		},
 	}
 }
 
@@ -154,12 +163,7 @@ func (p *Pool) GetClientForTesting(name string) (XDSClient, func(), error) {
 func (p *Pool) SetFallbackBootstrapConfig(config *bootstrap.Config) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-
-	if p.config != nil {
-		logger.Error("Attempt to set a bootstrap configuration even though one is already set via environment variables.")
-		return
-	}
-	p.config = config
+	p.fallbackConfig = config
 }
 
 // DumpResources returns the status and contents of all xDS resources.
@@ -191,7 +195,11 @@ func (p *Pool) DumpResources() *v3statuspb.ClientStatusResponse {
 func (p *Pool) BootstrapConfigForTesting() *bootstrap.Config {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	return p.config
+	cfg, _ := p.getConfiguration()
+	if cfg != nil {
+		return cfg
+	}
+	return p.fallbackConfig
 }
 
 // UnsetBootstrapConfigForTesting unsets the bootstrap configuration used by
@@ -201,7 +209,8 @@ func (p *Pool) BootstrapConfigForTesting() *bootstrap.Config {
 func (p *Pool) UnsetBootstrapConfigForTesting() {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	p.config = nil
+	p.fallbackConfig = nil
+	p.getConfiguration = sync.OnceValues(bootstrap.GetConfiguration)
 }
 
 func (p *Pool) clientRefCountedClose(name string) {
@@ -218,7 +227,6 @@ func (p *Pool) clientRefCountedClose(name string) {
 	}
 	delete(p.clients, name)
 
-	client.Close()
 	for _, s := range client.bootstrapConfig.XDSServers() {
 		for _, f := range s.Cleanups() {
 			f()
@@ -233,6 +241,11 @@ func (p *Pool) clientRefCountedClose(name string) {
 	}
 	p.mu.Unlock()
 
+	// This attempts to close the transport to the management server and could
+	// theoretically call back into the xdsclient package again and deadlock.
+	// Hence, this needs to be called without holding the lock.
+	client.Close()
+
 	xdsClientImplCloseHook(name)
 }
 
@@ -243,43 +256,36 @@ func (p *Pool) newRefCounted(name string, metricsRecorder estats.MetricsRecorder
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	if p.config == nil {
-		if len(p.clients) != 0 || p != DefaultPool {
-			// If the current pool `p` already contains xDS clients or it is not
-			// the `DefaultPool`, the bootstrap config should have been already
-			// present in the pool.
-			return nil, nil, fmt.Errorf("xds: bootstrap configuration not set in the pool")
-		}
-		// If the current pool `p` is the `DefaultPool` and has no clients, it
-		// might be the first time an xDS client is being created on it. So,
-		// the bootstrap configuration is read from environment variables.
-		//
-		// DefaultPool is initialized with bootstrap configuration from one of the
-		// supported environment variables. If the environment variables are not
-		// set, then fallback bootstrap configuration should be set before
-		// attempting to create an xDS client, else xDS client creation will fail.
-		config, err := bootstrap.GetConfiguration()
-		if err != nil {
-			return nil, nil, fmt.Errorf("xds: failed to read xDS bootstrap config from env vars:  %v", err)
-		}
-		p.config = config
+	config, err := p.getConfiguration()
+	if err != nil {
+		return nil, nil, fmt.Errorf("xds: failed to read xDS bootstrap config from env vars:  %v", err)
+	}
+
+	if config == nil {
+		// If the environment variables are not set, then fallback bootstrap
+		// configuration should be set before attempting to create an xDS client,
+		// else xDS client creation will fail.
+		config = p.fallbackConfig
+	}
+	if config == nil {
+		return nil, nil, fmt.Errorf("failed to read xDS bootstrap config from env vars: bootstrap environment variables (%q or %q) not defined and fallback config not set", envconfig.XDSBootstrapFileNameEnv, envconfig.XDSBootstrapFileContentEnv)
 	}
 
 	if c := p.clients[name]; c != nil {
 		c.incrRef()
 		return c, sync.OnceFunc(func() { p.clientRefCountedClose(name) }), nil
 	}
 
-	c, err := newClientImpl(p.config, metricsRecorder, name)
+	c, err := newClientImpl(config, metricsRecorder, name)
 	if err != nil {
 		return nil, nil, err
 	}
 	if logger.V(2) {
-		c.logger.Infof("Created client with name %q and bootstrap configuration:\n %s", name, p.config)
+		c.logger.Infof("Created client with name %q and bootstrap configuration:\n %s", name, config)
 	}
 	p.clients[name] = c
 	xdsClientImplCreateHook(name)
 
-	logger.Infof("xDS node ID: %s", p.config.Node().GetId())
+	logger.Infof("xDS node ID: %s", config.Node().GetId())
 	return c, sync.OnceFunc(func() { p.clientRefCountedClose(name) }), nil
 }

xds/internal/xdsclient/pool/pool_test.go

@@ -73,6 +73,10 @@ func (s) TestDefaultPool_LazyLoadBootstrapConfig(t *testing.T) {
 		t.Fatalf("DefaultPool.BootstrapConfigForTesting() = %v, want nil", cfg)
 	}
 
+	// The pool will attempt to read the env vars only once. Reset the pool's
+	// state to make it re-read the env vars during next client creation.
+	xdsclient.DefaultPool.UnsetBootstrapConfigForTesting()
+
 	_, closeFunc, err = xdsclient.DefaultPool.NewClient(t.Name(), &stats.NoopMetricsRecorder{})
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)