address review comments: compare full structures in Equal and tests

Author: mbissa

internal/xds/bootstrap/bootstrap_test.go

@@ -1749,54 +1749,75 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) {
 	tests := []struct {
 		name string
 		json string
-		// The identity of the built channel credentials must match the first
-		// supported channel-creds entry from the bootstrap JSON.
-		wantSelectedChannelCredsType   string
-		wantSelectedChannelCredsConfig json.RawMessage
-		wantSideCallCreds              int
+		// want carries the expected target and credential identities;
+		// comparisons use the Equal methods via cmp.Diff.
+		want AllowedGRPCServices
 	}{
 		{
-			name:                         "insecure_channel_creds",
-			json:                         `{"dns:///sharding-service:443": {"channel_creds": [{"type": "insecure"}]}}`,
-			wantSelectedChannelCredsType: "insecure",
-			wantSideCallCreds:            0,
+			name: "insecure_channel_creds",
+			json: `{"dns:///sharding-service:443": {"channel_creds": [{"type": "insecure"}]}}`,
+			want: AllowedGRPCServices{target: {
+				targetURI:        target,
+				sideChannelCreds: xdscreds.NewChannelCreds(nil, xdscreds.Identity{Type: "insecure"}, nil),
+			}},
 		},
 		{
-			name:                         "with_call_creds",
-			json:                         `{"dns:///sharding-service:443": {"channel_creds": [{"type": "insecure"}], "call_creds": [{"type": "jwt_token_file", "config": {"jwt_token_file": "/var/run/secrets/tokens/istio-token"}}]}}`,
-			wantSelectedChannelCredsType: "insecure",
-			// One call credential is built for the supported call-creds
-			// config.
-			wantSideCallCreds: 1,
+			name: "with_call_creds",
+			json: `{"dns:///sharding-service:443": {"channel_creds": [{"type": "insecure"}], "call_creds": [{"type": "jwt_token_file", "config": {"jwt_token_file": "/var/run/secrets/tokens/istio-token"}}]}}`,
+			want: AllowedGRPCServices{target: {
+				targetURI:        target,
+				sideChannelCreds: xdscreds.NewChannelCreds(nil, xdscreds.Identity{Type: "insecure"}, nil),
+				sideCallCreds: []*xdscreds.CallCreds{xdscreds.NewCallCreds(nil, xdscreds.Identity{
+					Type: "jwt_token_file",
+					Data: json.RawMessage(`{"jwt_token_file": "/var/run/secrets/tokens/istio-token"}`),
+				}, nil)},
+			}},
 		},
 		{
-			name:                         "unsupported_call_creds_skipped",
-			json:                         `{"dns:///sharding-service:443": {"channel_creds": [{"type": "insecure"}], "call_creds": [{"type": "unsupported_call_creds_type"}]}}`,
-			wantSelectedChannelCredsType: "insecure",
-			// Unsupported call-creds types are skipped without error, so
-			// no call credentials are built.
-			wantSideCallCreds: 0,
+			// Unsupported call-creds types are skipped without error, so no
+			// call credentials are built.
+			name: "unsupported_call_creds_skipped",
+			json: `{"dns:///sharding-service:443": {"channel_creds": [{"type": "insecure"}], "call_creds": [{"type": "unsupported_call_creds_type"}]}}`,
+			want: AllowedGRPCServices{target: {
+				targetURI:        target,
+				sideChannelCreds: xdscreds.NewChannelCreds(nil, xdscreds.Identity{Type: "insecure"}, nil),
+			}},
 		},
 		{
-			name:                         "multiple_supported_call_creds",
-			json:                         `{"dns:///sharding-service:443": {"channel_creds": [{"type": "insecure"}], "call_creds": [{"type": "jwt_token_file", "config": {"jwt_token_file": "/tokens/token-one"}}, {"type": "jwt_token_file", "config": {"jwt_token_file": "/tokens/token-two"}}]}}`,
-			wantSelectedChannelCredsType: "insecure",
 			// One call credential is built for each supported call-creds
-			// config.
-			wantSideCallCreds: 2,
+			// config, preserving order.
+			name: "multiple_supported_call_creds",
+			json: `{"dns:///sharding-service:443": {"channel_creds": [{"type": "insecure"}], "call_creds": [{"type": "jwt_token_file", "config": {"jwt_token_file": "/tokens/token-one"}}, {"type": "jwt_token_file", "config": {"jwt_token_file": "/tokens/token-two"}}]}}`,
+			want: AllowedGRPCServices{target: {
+				targetURI:        target,
+				sideChannelCreds: xdscreds.NewChannelCreds(nil, xdscreds.Identity{Type: "insecure"}, nil),
+				sideCallCreds: []*xdscreds.CallCreds{
+					xdscreds.NewCallCreds(nil, xdscreds.Identity{
+						Type: "jwt_token_file",
+						Data: json.RawMessage(`{"jwt_token_file": "/tokens/token-one"}`),
+					}, nil),
+					xdscreds.NewCallCreds(nil, xdscreds.Identity{
+						Type: "jwt_token_file",
+						Data: json.RawMessage(`{"jwt_token_file": "/tokens/token-two"}`),
+					}, nil),
+				},
+			}},
 		},
 		{
-			name:                           "tls_channel_creds",
-			json:                           `{"dns:///sharding-service:443": {"channel_creds": [{"type": "tls", "config": {}}]}}`,
-			wantSelectedChannelCredsType:   "tls",
-			wantSelectedChannelCredsConfig: json.RawMessage("{}"),
-			wantSideCallCreds:              0,
+			name: "tls_channel_creds",
+			json: `{"dns:///sharding-service:443": {"channel_creds": [{"type": "tls", "config": {}}]}}`,
+			want: AllowedGRPCServices{target: {
+				targetURI:        target,
+				sideChannelCreds: xdscreds.NewChannelCreds(nil, xdscreds.Identity{Type: "tls", Data: json.RawMessage("{}")}, nil),
+			}},
 		},
 		{
-			name:                         "skips_unsupported_channel_creds",
-			json:                         `{"dns:///sharding-service:443": {"channel_creds": [{"type": "unsupported_cred_type"}, {"type": "insecure"}]}}`,
-			wantSelectedChannelCredsType: "insecure",
-			wantSideCallCreds:            0,
+			name: "skips_unsupported_channel_creds",
+			json: `{"dns:///sharding-service:443": {"channel_creds": [{"type": "unsupported_cred_type"}, {"type": "insecure"}]}}`,
+			want: AllowedGRPCServices{target: {
+				targetURI:        target,
+				sideChannelCreds: xdscreds.NewChannelCreds(nil, xdscreds.Identity{Type: "insecure"}, nil),
+			}},
 		},
 	}
 
@@ -1806,24 +1827,15 @@ func (s) TestAllowedGRPCServices_UnmarshalJSON(t *testing.T) {
 			if err := json.Unmarshal([]byte(test.json), &got); err != nil {
 				t.Fatalf("AllowedGRPCServices unmarshal failed: %v", err)
 			}
-			svc, ok := got[target]
-			if !ok {
-				t.Fatalf("AllowedGRPCServices missing key %q", target)
-			}
-			if svc.TargetURI() != target {
-				t.Errorf("TargetURI() = %q, want %q", svc.TargetURI(), target)
+			if diff := cmp.Diff(test.want, got); diff != "" {
+				t.Errorf("AllowedGRPCServices unmarshal returned unexpected diff (-want +got):\n%s", diff)
 			}
-			chanCreds, callCreds := svc.SideChannelCredentials()
+			// Equal compares credentials by identity only, so it cannot tell
+			// a built bundle from a nil one; verify the bundle was built.
+			chanCreds, _ := got[target].SideChannelCredentials()
 			if chanCreds == nil || chanCreds.Bundle() == nil {
 				t.Error("SideChannelCredentials() returned no built channel credentials")
 			}
-			wantIdentity := xdscreds.Identity{Type: test.wantSelectedChannelCredsType, Data: test.wantSelectedChannelCredsConfig}
-			if wantChanCreds := xdscreds.NewChannelCreds(nil, wantIdentity, nil); !chanCreds.Equal(wantChanCreds) {
-				t.Errorf("SideChannelCredentials() channel credentials = %+v, want identity %+v", chanCreds, wantIdentity)
-			}
-			if got := len(callCreds); got != test.wantSideCallCreds {
-				t.Errorf("len(SideChannelCredentials() call creds) = %d, want %d", got, test.wantSideCallCreds)
-			}
 		})
 	}
 }

internal/xds/grpcservice/grpcservice.go

@@ -22,6 +22,7 @@ package grpcservice
 
 import (
 	"fmt"
+	"maps"
 	"net/url"
 	"slices"
 	"strings"
@@ -62,15 +63,14 @@ type Config struct {
 	CallCredentials []*xdscreds.CallCreds
 }
 
-// Equal reports whether c and other describe the same side channel: the same
-// target with the same channel and call credential identities. Timeout and
-// initial metadata are applied per-RPC and intentionally do not affect
-// channel sharing.
+// Equal reports whether c and other are equal.
 func (c *Config) Equal(other *Config) bool {
 	targetEqual := c.TargetURI == other.TargetURI
+	timeoutEqual := c.Timeout == other.Timeout
+	metadataEqual := maps.EqualFunc(c.InitialMetadata, other.InitialMetadata, slices.Equal)
 	channelCredsEqual := c.ChannelCredentials.Equal(other.ChannelCredentials)
 	callCredsEqual := slices.EqualFunc(c.CallCredentials, other.CallCredentials, (*xdscreds.CallCreds).Equal)
-	return targetEqual && channelCredsEqual && callCredsEqual
+	return targetEqual && timeoutEqual && metadataEqual && channelCredsEqual && callCredsEqual
 }
 
 // Close releases the credentials owned by the config. It is idempotent, and a
@@ -84,10 +84,7 @@ func (c *Config) Close() {
 }
 
 // Dial creates a channel to the side-channel service, using the channel and
-// call credentials from the config along with the provided dial options. The
-// call credentials are attached to the channel; since they are part of the
-// config's identity, configs whose call credentials differ do not share a
-// channel.
+// call credentials from the config along with the provided dial options.
 //
 // Dial does not take ownership of the config: the caller releases the
 // config's credentials via Close when the config is no longer needed, after

internal/xds/grpcservice/grpcservice_test.go

@@ -22,6 +22,7 @@ import (
 	"encoding/json"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"google.golang.org/grpc/internal/envconfig"
@@ -215,13 +216,43 @@ func (s) TestParse(t *testing.T) {
 			config:  bootstrapConfig(t, "{}"),
 			wantErr: "target_uri must be non-empty",
 		},
+		{
+			name:   "valid_timeout",
+			gs:     googleGrpcService(target, []*anypb.Any{insecurePlugin}, nil, durationpb.New(10*time.Second)),
+			sc:     trustedServerConfig(t),
+			config: bootstrapConfig(t, "{}"),
+			want: &Config{
+				TargetURI:          target,
+				Timeout:            10 * time.Second,
+				ChannelCredentials: xdscreds.NewChannelCreds(nil, protoIdentity(insecurePlugin), nil),
+			},
+		},
 		{
 			name:    "zero_timeout_rejected",
 			gs:      googleGrpcService(target, []*anypb.Any{insecurePlugin}, nil, durationpb.New(0)),
 			sc:      trustedServerConfig(t),
 			config:  bootstrapConfig(t, "{}"),
 			wantErr: "timeout must be strictly positive",
 		},
+		{
+			name: "initial_metadata",
+			gs: func() *v3corepb.GrpcService {
+				gs := googleGrpcService(target, []*anypb.Any{insecurePlugin}, nil, nil)
+				gs.InitialMetadata = []*v3corepb.HeaderValue{
+					{Key: "key-b", Value: "b"},
+					// raw_value takes precedence over the legacy value field.
+					{Key: "key-a", Value: "legacy", RawValue: []byte("raw-a")},
+				}
+				return gs
+			}(),
+			sc:     trustedServerConfig(t),
+			config: bootstrapConfig(t, "{}"),
+			want: &Config{
+				TargetURI:          target,
+				InitialMetadata:    metadata.MD{"key-b": {"b"}, "key-a": {"raw-a"}},
+				ChannelCredentials: xdscreds.NewChannelCreds(nil, protoIdentity(insecurePlugin), nil),
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -266,22 +297,6 @@ func (s) TestConfigDial(t *testing.T) {
 	}
 }
 
-func (s) TestParseInitialMetadata(t *testing.T) {
-	gs := googleGrpcService(target, []*anypb.Any{{TypeUrl: insecureCredsTypeURL}}, nil, nil)
-	gs.InitialMetadata = []*v3corepb.HeaderValue{
-		{Key: "key-b", Value: "b"},
-		{Key: "key-a", Value: "legacy", RawValue: []byte("raw-a")},
-	}
-	got, err := Parse(gs, bootstrapConfig(t, "{}"), trustedServerConfig(t))
-	if err != nil {
-		t.Fatalf("Parse() returned unexpected error: %v", err)
-	}
-	want := metadata.MD{"key-b": []string{"b"}, "key-a": []string{"raw-a"}}
-	if diff := cmp.Diff(want, got.InitialMetadata); diff != "" {
-		t.Errorf("Parse() InitialMetadata mismatch (-want +got):\n%s", diff)
-	}
-}
-
 func (s) TestConfigEqual(t *testing.T) {
 	insecurePlugin := &anypb.Any{TypeUrl: insecureCredsTypeURL}
 	protoInsecure := xdscreds.NewChannelCreds(nil, protoIdentity(insecurePlugin), nil)
@@ -293,16 +308,22 @@ func (s) TestConfigEqual(t *testing.T) {
 		want bool
 	}{
 		{
-			name: "equal_identities_share",
-			a:    &Config{TargetURI: target, ChannelCredentials: protoInsecure},
-			b:    &Config{TargetURI: target, ChannelCredentials: xdscreds.NewChannelCreds(nil, protoIdentity(insecurePlugin), nil)},
+			name: "equal",
+			a:    &Config{TargetURI: target, ChannelCredentials: protoInsecure, Timeout: 1, InitialMetadata: metadata.Pairs("k", "v")},
+			b:    &Config{TargetURI: target, ChannelCredentials: xdscreds.NewChannelCreds(nil, protoIdentity(insecurePlugin), nil), Timeout: 1, InitialMetadata: metadata.Pairs("k", "v")},
 			want: true,
 		},
 		{
-			name: "timeout_and_metadata_do_not_affect_sharing",
+			name: "different_timeouts",
 			a:    &Config{TargetURI: target, ChannelCredentials: protoInsecure, Timeout: 1},
+			b:    &Config{TargetURI: target, ChannelCredentials: protoInsecure, Timeout: 2},
+			want: false,
+		},
+		{
+			name: "different_initial_metadata",
+			a:    &Config{TargetURI: target, ChannelCredentials: protoInsecure},
 			b:    &Config{TargetURI: target, ChannelCredentials: protoInsecure, InitialMetadata: metadata.Pairs("k", "v")},
-			want: true,
+			want: false,
 		},
 		{
 			name: "different_targets",

internal/xds/httpfilter/extproc/config_test.go

@@ -129,6 +129,55 @@ func overrideCreateExtProcChannel(t *testing.T, failTarget string) {
 	t.Cleanup(func() { iextproc.CreateExtProcChannel = origCreateExtProcChannel })
 }
 
+// Tests that channel sharing considers the target and the credential
+// identities, and ignores the per-RPC timeout and initial metadata.
+func (s) TestSharesChannel(t *testing.T) {
+	insecureCreds := func() *xdscreds.ChannelCreds {
+		return xdscreds.NewChannelCreds(nil, xdscreds.Identity{Type: "insecure"}, nil)
+	}
+	const target = "dns:///proc-server:443"
+
+	tests := []struct {
+		name string
+		a, b *grpcservice.Config
+		want bool
+	}{
+		{
+			name: "equal_identities_share_despite_timeout_and_metadata",
+			a:    &grpcservice.Config{TargetURI: target, ChannelCredentials: insecureCreds(), Timeout: time.Second},
+			b:    &grpcservice.Config{TargetURI: target, ChannelCredentials: insecureCreds(), InitialMetadata: metadata.Pairs("k", "v")},
+			want: true,
+		},
+		{
+			name: "different_targets",
+			a:    &grpcservice.Config{TargetURI: target, ChannelCredentials: insecureCreds()},
+			b:    &grpcservice.Config{TargetURI: "dns:///other:443", ChannelCredentials: insecureCreds()},
+			want: false,
+		},
+		{
+			name: "different_channel_creds",
+			a:    &grpcservice.Config{TargetURI: target, ChannelCredentials: insecureCreds()},
+			b:    &grpcservice.Config{TargetURI: target, ChannelCredentials: xdscreds.NewChannelCreds(nil, xdscreds.Identity{Type: "other"}, nil)},
+			want: false,
+		},
+		{
+			name: "different_call_creds",
+			a:    &grpcservice.Config{TargetURI: target, ChannelCredentials: insecureCreds()},
+			b: &grpcservice.Config{TargetURI: target, ChannelCredentials: insecureCreds(), CallCredentials: []*xdscreds.CallCreds{
+				xdscreds.NewCallCreds(nil, xdscreds.Identity{Type: "access_token"}, nil),
+			}},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := sharesChannel(tt.a, tt.b); got != tt.want {
+				t.Errorf("sharesChannel() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 var cmpOpts = []cmp.Option{
 	cmp.AllowUnexported(
 		baseConfig{},

internal/xds/httpfilter/extproc/ext_proc.go

@@ -39,6 +39,7 @@ import (
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/optional"
 	"google.golang.org/grpc/internal/resolver"
+	xdscreds "google.golang.org/grpc/internal/xds/credentials"
 	"google.golang.org/grpc/internal/xds/grpcservice"
 	"google.golang.org/grpc/internal/xds/httpfilter"
 	iextproc "google.golang.org/grpc/internal/xds/httpfilter/extproc/internal"
@@ -270,28 +271,41 @@ type clientFilter struct {
 
 func (*clientFilter) Close() {}
 
-// getProcChannel returns an existing refcounted client for an equal server
-// config if present and its refcount is incremented successfully.
+// sharesChannel reports whether two GrpcService configs may share a channel
+// to the external processor server: the same target with the same channel
+// and call credential identities. Timeout and initial metadata are applied
+// per-RPC and do not affect sharing; call credentials do, because Dial
+// attaches them to the channel.
+func sharesChannel(a, b *grpcservice.Config) bool {
+	targetEqual := a.TargetURI == b.TargetURI
+	channelCredsEqual := a.ChannelCredentials.Equal(b.ChannelCredentials)
+	callCredsEqual := slices.EqualFunc(a.CallCredentials, b.CallCredentials, (*xdscreds.CallCreds).Equal)
+	return targetEqual && channelCredsEqual && callCredsEqual
+}
+
+// getProcChannel returns an existing refcounted client for a channel-sharing
+// compatible server config if present and its refcount is incremented
+// successfully.
 func (cf *clientFilter) getProcChannel(server *grpcservice.Config) *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient] {
 	cf.mu.Lock()
 	defer cf.mu.Unlock()
 	for _, e := range cf.procChannels {
-		if e.server.Equal(server) && e.rc.TryIncrement() {
+		if sharesChannel(&e.server, server) && e.rc.TryIncrement() {
 			return e.rc
 		}
 	}
 	return nil
 }
 
 // storeProcChannel stores the created channel entry if no valid channel
-// exists for an equal server config. If another goroutine already stored one
-// while unlocked, it increments the existing channel's refcount and returns
-// it.
+// exists for a channel-sharing compatible server config. If another goroutine
+// already stored one while unlocked, it increments the existing channel's
+// refcount and returns it.
 func (cf *clientFilter) storeProcChannel(entry *procChannelEntry) *grpcsync.RefCounted[v3procservicegrpc.ExternalProcessorClient] {
 	cf.mu.Lock()
 	defer cf.mu.Unlock()
 	for _, e := range cf.procChannels {
-		if e.server.Equal(&entry.server) && e.rc.TryIncrement() {
+		if sharesChannel(&e.server, &entry.server) && e.rc.TryIncrement() {
 			return e.rc
 		}
 	}