xds/gcp_authn: implement GCP Authentication filter (gRFC A83).

credentials/google/gcp_service_account_identity_credentials.go

@@ -36,28 +36,37 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-// preemptiveRefresh is the window before a token's actual expiration during
-// which the token is considered stale. Requests using a stale but still valid
-// token will trigger a background asynchronous refresh. This avoids blocking
-// the current RPC, preventing periodic latency spikes during token refresh.
-const preemptiveRefresh = 1 * time.Minute
+const (
+	// preemptiveRefresh is the window before a token's actual expiration during
+	// which the token is considered stale. Requests using a stale but still
+	// valid token will trigger a background asynchronous refresh. This avoids
+	// blocking the current RPC, preventing periodic latency spikes during token
+	// refresh.
+	preemptiveRefresh = 1 * time.Minute
+
+	// metadataTimeout is the timeout for the call to the metadata server to
+	// prevent the background token fetch from hanging indefinitely in case
+	// of connection issues.
+	metadataTimeout = 60 * time.Second
+)
 
 type gcpServiceAccountIdentityCallCreds struct {
 	// The following fields are initialized at creation time and are read-only
 	// after that.
+	ctx      context.Context
 	audience string
 	creds    *auth.Credentials
 	backoff  backoff.Strategy
 
 	// The following fields are protected by mu.
 	mu                     sync.Mutex
 	token                  *auth.Token
-	tokenExpiry            time.Time // timestamp after which the cached token is considered invalid
-	preemptiveTokenRefresh time.Time // timestamp after which background preemptive refresh is triggered
-	fetching               bool      // true if a background token fetch is in progress
-	nextRetryTime          time.Time // timestamp after which we can attempt the next token fetch
-	retryAttempt           int       // consecutive fetch failure count used to compute backoff delay
-	lastErr                error     // cached error returned from the most recent token fetch attempt
+	tokenExpiry            time.Time     // timestamp after which the cached token is considered invalid
+	preemptiveTokenRefresh time.Time     // timestamp after which background preemptive refresh is triggered
+	fetching               chan struct{} // used to deduplicate concurrent background token fetches
+	nextRetryTime          time.Time     // timestamp after which we can attempt the next token fetch
+	retryAttempt           int           // consecutive fetch failure count used to compute backoff delay
+	lastErr                error         // cached error returned from the most recent token fetch attempt
 }
 
 func init() {
@@ -79,7 +88,7 @@ func init() {
 //
 // Notice: This API is EXPERIMENTAL and may be changed or removed in a
 // later release.
-func NewServiceAccountIdentityCredentials(audience string) (credentials.PerRPCCredentials, error) {
+func NewServiceAccountIdentityCredentials(ctx context.Context, audience string) (credentials.PerRPCCredentials, error) {
 	if audience == "" {
 		return nil, fmt.Errorf("credentials: audience cannot be empty")
 	}
@@ -90,6 +99,7 @@ func NewServiceAccountIdentityCredentials(audience string) (credentials.PerRPCCr
 	}
 
 	return &gcpServiceAccountIdentityCallCreds{
+		ctx:      ctx,
 		audience: audience,
 		creds:    creds,
 		backoff:  internal.BackoffStrategy,
@@ -111,33 +121,48 @@ func (c *gcpServiceAccountIdentityCallCreds) GetRequestMetadata(ctx context.Cont
 	}
 
 	c.mu.Lock()
-	defer c.mu.Unlock()
 
 	// If token is valid, return it. If it's also stale, trigger a background
 	// refresh if not already running and return the current token.
 	if c.token != nil && c.isTokenValidLocked() {
-		if c.isTokenStaleLocked() && !c.fetching {
-			c.fetching = true
+		if c.isTokenStaleLocked() && c.fetching == nil {
+			c.fetching = make(chan struct{})
 			go c.startFetch()
 		}
+		defer c.mu.Unlock()
 		return map[string]string{
 			"authorization": "Bearer " + c.token.Value,
 		}, nil
 	}
 
 	if c.lastErr != nil && time.Now().Before(c.nextRetryTime) {
+		c.mu.Unlock()
 		return nil, c.lastErr
 	}
 
-	token, err := c.creds.TokenProvider.Token(context.Background())
-	c.updateStateLocked(token, err)
-	if err != nil {
-		return nil, c.lastErr
+	if c.fetching == nil {
+		c.fetching = make(chan struct{})
+		go c.startFetch()
+	}
+	wait := c.fetching
+	c.mu.Unlock()
+
+	select {
+	case <-wait:
+		c.mu.Lock()
+		defer c.mu.Unlock()
+		if c.token != nil && c.isTokenValidLocked() {
+			return map[string]string{
+				"authorization": "Bearer " + c.token.Value,
+			}, nil
+		}
+		if c.lastErr != nil {
+			return nil, c.lastErr
+		}
+		return nil, status.Error(codes.Unavailable, "credentials: fetched token is expired")
+	case <-ctx.Done():
+		return nil, ctx.Err()
 	}
-
-	return map[string]string{
-		"authorization": "Bearer " + c.token.Value,
-	}, nil
 }
 
 // RequireTransportSecurity indicates whether the credentials requires
@@ -161,12 +186,15 @@ func (c *gcpServiceAccountIdentityCallCreds) isTokenValidLocked() bool {
 // startFetch initiates a token fetch and updates the credential
 // state upon completion.
 func (c *gcpServiceAccountIdentityCallCreds) startFetch() {
-	token, err := c.creds.TokenProvider.Token(context.Background())
+	ctx, cancel := context.WithTimeout(c.ctx, metadataTimeout)
+	defer cancel()
+	token, err := c.creds.TokenProvider.Token(ctx)
 
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
-	c.fetching = false
+	close(c.fetching)
+	c.fetching = nil
 	c.updateStateLocked(token, err)
 }
 

credentials/google/gcp_service_account_identity_credentials_test.go

@@ -129,7 +129,9 @@ func setupTestGCPServiceAccountIdentityCreds(t *testing.T, stubToken *stubTokenP
 	}
 	t.Cleanup(func() { internal.NewIDTokenCredentials = origNewIDTokenCredentials })
 
-	creds, err := google.NewServiceAccountIdentityCredentials("audience")
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	creds, err := google.NewServiceAccountIdentityCredentials(ctx, "audience")
 	if err != nil {
 		t.Fatalf("NewServiceAccountIdentityCredentials() failed: %v", err)
 	}
@@ -141,7 +143,9 @@ func setupTestGCPServiceAccountIdentityCreds(t *testing.T, stubToken *stubTokenP
 // when called with empty audience.
 func (s) TestNewServiceAccountIdentityCredentials_EmptyAudience(t *testing.T) {
 	const wantErr = "credentials: audience cannot be empty"
-	if _, err := google.NewServiceAccountIdentityCredentials(""); err == nil || !strings.Contains(err.Error(), wantErr) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	if _, err := google.NewServiceAccountIdentityCredentials(ctx, ""); err == nil || !strings.Contains(err.Error(), wantErr) {
 		t.Fatalf("NewServiceAccountIdentityCredentials() returned error = %v, want error containing %q", err, wantErr)
 	}
 }

internal/xds/balancer/clustermanager/picker.go

@@ -43,7 +43,7 @@ func newPickerGroup(idToPickerState map[string]*subBalancerState) *pickerGroup {
 }
 
 func (pg *pickerGroup) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
-	cluster := getPickedCluster(info.Ctx)
+	cluster := PickedCluster(info.Ctx)
 	if p := pg.pickers[cluster]; p != nil {
 		return p.Pick(info)
 	}
@@ -52,17 +52,12 @@ func (pg *pickerGroup) Pick(info balancer.PickInfo) (balancer.PickResult, error)
 
 type clusterKey struct{}
 
-func getPickedCluster(ctx context.Context) string {
+// PickedCluster returns the cluster name stored in the context.
+func PickedCluster(ctx context.Context) string {
 	cluster, _ := ctx.Value(clusterKey{}).(string)
 	return cluster
 }
 
-// GetPickedClusterForTesting returns the cluster in the context; to be used
-// for testing only.
-func GetPickedClusterForTesting(ctx context.Context) string {
-	return getPickedCluster(ctx)
-}
-
 // SetPickedCluster adds the selected cluster to the context for the
 // xds_cluster_manager LB policy to pick.
 func SetPickedCluster(ctx context.Context, cluster string) context.Context {

internal/xds/httpfilter/extproc/config_test.go

@@ -735,7 +735,7 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			builder := builder{}
-			filter := builder.BuildClientFilter()
+			filter := builder.BuildClientFilter("")
 			defer filter.Close()
 
 			intptr, err := filter.BuildClientInterceptor(tc.cfg, tc.override)
@@ -818,7 +818,7 @@ func (s) TestBuildClientInterceptor_Failure(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			builder := builder{}
-			filter := builder.BuildClientFilter()
+			filter := builder.BuildClientFilter("")
 			defer filter.Close()
 
 			_, err := filter.BuildClientInterceptor(tc.cfg, tc.override)

internal/xds/httpfilter/extproc/ext_proc.go

@@ -190,7 +190,7 @@ func (builder) IsTerminal() bool {
 	return false
 }
 
-func (builder) BuildClientFilter() httpfilter.ClientFilter {
+func (builder) BuildClientFilter(string) httpfilter.ClientFilter {
 	return clientFilter{}
 }
 

internal/xds/httpfilter/fault/fault.go

@@ -105,7 +105,7 @@ func (builder) IsTerminal() bool {
 	return false
 }
 
-func (builder) BuildClientFilter() httpfilter.ClientFilter {
+func (builder) BuildClientFilter(string) httpfilter.ClientFilter {
 	return clientFilter{}
 }