feat(bigtable): modularize channel priming behind a ChannelPrimer interface (#20027)

## Summary

Mirrors the pluggable shape that PR #19987 established for the Direct
Access compatibility check. Channel priming had been hard-wired into the
pool via three loose options (`WithInstanceName` / `WithAppProfile` /
`WithFeatureFlagsMetadata`) that the `connectionFactory` had to stitch
back together on every dial.

- New `ChannelPrimer` interface (`channel_primer.go`): `Prime(ctx,
*BigtableConn) error`. The pluggable extension point that future pool
factories (session-based, custom) can swap in.
- `pingAndWarmChannelPrimer` is today's only implementation; it owns the
`(instance, appProfile, featureFlagsMD)` tuple and delegates to
`BigtableConn.Prime` — the existing `PingAndWarm` RPC stays untouched.
- New `WithChannelPrimer` pool option replaces `WithInstanceName` /
`WithAppProfile` / `WithFeatureFlagsMetadata`.
- `connectionFactory` now carries a `ChannelPrimer` instead of three
individual fields. **When the primer is nil, `primeWithRetry` returns
immediately** — the pool dials the channel and puts it straight into
rotation, no `PingAndWarm` sent. The classic channel pool factory still
wires the `pingAndWarmChannelPrimer`, so user-facing default behavior is
unchanged.
- `pingAndWarmDirectAccessChecker` reuses the same primer rather than
duplicating `conn.Prime(ctx, instance, profile, flags)` in two places.
The factory now constructs **one** primer and shares it with both the
pool (via `WithChannelPrimer`) and the checker, eliminating the
three-arg drift that existed across the two consumers.

## Test plan

- [x] `go build ./...` clean
- [x] `go vet ./...` clean
- [x] `golint` clean on touched files
- [x] `go test ./internal/transport/...` passes (existing tests
migrated; new `channel_primer_test.go` covers (a)
`pingAndWarmChannelPrimer.Prime` issues `PingAndWarm` carrying the
configured feature-flag metadata, and (b) `connectionFactory` skips
priming entirely when the primer is nil)

Author: sushanb

bigtable/internal/transport/channel_pool_factory.go

@@ -153,15 +153,17 @@ func CreateBigtableChannelPool(
 
 	fullInstanceName := fmt.Sprintf("projects/%s/instances/%s", project, instance)
 
-	// directAccessMD is the feature-flag metadata used for priming on both
-	// the direct-access and standard-path connection factories — the pool
-	// holds it via WithFeatureFlagsMetadata and applies it to every Prime().
+	// Build a single PingAndWarm primer and share it with both the pool's
+	// connection factory (via WithChannelPrimer) and the direct-access
+	// compatibility checker. Keeping the (instanceName, appProfile,
+	// featureFlagsMD) tuple in one place avoids the three-arg drift between
+	// the two consumers.
+	primer := newPingAndWarmChannelPrimer(fullInstanceName, config.AppProfile, directAccessMD)
+
 	poolOpts := []BigtableChannelPoolOption{
-		WithInstanceName(fullInstanceName),
-		WithAppProfile(config.AppProfile),
 		WithMetricsReporterConfig(btopt.DefaultMetricsReporterConfig()),
 		WithMeterProvider(otelMeterProvider),
-		WithFeatureFlagsMetadata(directAccessMD),
+		WithChannelPrimer(primer),
 	}
 
 	// Pluggable Direct Access strategy: the classic channel pool factory uses
@@ -184,9 +186,7 @@ func CreateBigtableChannelPool(
 		}
 		checker := newPingAndWarmDirectAccessChecker(
 			directAccessDialer,
-			fullInstanceName,
-			config.AppProfile,
-			directAccessMD,
+			primer,
 			otelMeterProvider,
 			nil, // logger plumbed by callers once available
 		)

bigtable/internal/transport/channel_primer.go

@@ -0,0 +1,63 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"context"
+
+	"google.golang.org/grpc/metadata"
+)
+
+// ChannelPrimer warms a freshly-dialed Bigtable channel before it is put
+// into rotation. The pool's connection factory consults the registered
+// primer after every successful dial; a nil ChannelPrimer means the pool
+// skips priming entirely and hands the raw connection straight to the
+// pool.
+//
+// Implementations differ in HOW the channel is warmed:
+//   - pingAndWarmChannelPrimer issues a PingAndWarm against the configured
+//     instance / app profile with the supplied feature-flag metadata
+//     (today's only behavior, used by the classic channel pool factory).
+type ChannelPrimer interface {
+	// Prime warms conn so the next request served by it does not pay the
+	// first-RPC connection-setup cost. The factory wraps Prime in a retry
+	// loop, so transient errors should propagate as-is.
+	Prime(ctx context.Context, conn *BigtableConn) error
+}
+
+// pingAndWarmChannelPrimer primes a channel by issuing a PingAndWarm RPC
+// against the configured instance + app profile, carrying the supplied
+// feature-flag metadata. Stateless aside from the configured identifiers,
+// so a single primer instance is shared across all dials in a pool.
+type pingAndWarmChannelPrimer struct {
+	instanceName   string
+	appProfile     string
+	featureFlagsMD metadata.MD
+}
+
+// newPingAndWarmChannelPrimer constructs the today-default channel primer.
+func newPingAndWarmChannelPrimer(instanceName, appProfile string, featureFlagsMD metadata.MD) *pingAndWarmChannelPrimer {
+	return &pingAndWarmChannelPrimer{
+		instanceName:   instanceName,
+		appProfile:     appProfile,
+		featureFlagsMD: featureFlagsMD,
+	}
+}
+
+// Prime delegates to BigtableConn.Prime, which sends PingAndWarm and
+// records the ALTS / IP-protocol observations on conn as a side effect.
+func (p *pingAndWarmChannelPrimer) Prime(ctx context.Context, conn *BigtableConn) error {
+	return conn.Prime(ctx, p.instanceName, p.appProfile, p.featureFlagsMD)
+}

bigtable/internal/transport/channel_primer_test.go

@@ -0,0 +1,78 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"context"
+	"testing"
+
+	"google.golang.org/grpc/metadata"
+)
+
+// TestPingAndWarmChannelPrimer_Prime verifies the primer delegates to
+// BigtableConn.Prime carrying the configured instance name, app profile,
+// and feature-flag metadata — i.e. that the primer is a thin wrapper that
+// keeps the (instance, profile, flags) tuple in one place.
+func TestPingAndWarmChannelPrimer_Prime(t *testing.T) {
+	fake := &fakeService{}
+	addr := setupTestServer(t, fake)
+	conn, err := dialBigtableserver(addr)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	t.Cleanup(func() { conn.Close() })
+
+	flagsMD := metadata.Pairs("bigtable-features", "primer-test")
+	primer := newPingAndWarmChannelPrimer(testInstanceName, testAppProfile, flagsMD)
+
+	if err := primer.Prime(context.Background(), conn); err != nil {
+		t.Fatalf("Prime returned error: %v", err)
+	}
+
+	if got := fake.getPingCount(); got != 1 {
+		t.Errorf("PingAndWarm call count = %d, want 1", got)
+	}
+
+	gotMD := fake.getPrimeMetadata()
+	if got := gotMD.Get("bigtable-features"); len(got) != 1 || got[0] != "primer-test" {
+		t.Errorf("feature-flag metadata on PingAndWarm = %v, want [primer-test]", got)
+	}
+	if got := gotMD.Get("x-goog-request-params"); len(got) != 1 {
+		t.Errorf("x-goog-request-params on PingAndWarm = %v, want one entry derived from instance/profile", got)
+	}
+}
+
+// TestConnectionFactory_NilPrimerSkipsPriming verifies the contract that a
+// nil ChannelPrimer turns priming off: newEntry dials the channel and
+// returns it without issuing PingAndWarm.
+func TestConnectionFactory_NilPrimerSkipsPriming(t *testing.T) {
+	fake := &fakeService{}
+	addr := setupTestServer(t, fake)
+
+	factory := &connectionFactory{
+		dial:   func() (*BigtableConn, error) { return dialBigtableserver(addr) },
+		primer: nil,
+	}
+
+	entry, err := factory.newEntry(context.Background())
+	if err != nil {
+		t.Fatalf("newEntry returned error: %v", err)
+	}
+	t.Cleanup(func() { entry.conn.Close() })
+
+	if got := fake.getPingCount(); got != 0 {
+		t.Errorf("PingAndWarm call count with nil primer = %d, want 0", got)
+	}
+}

bigtable/internal/transport/connpool.go

@@ -101,13 +101,6 @@ type connPoolStats struct {
 
 var _ Monitor = (*MetricsReporter)(nil)
 
-// WithAppProfile provides the appProfile
-func WithAppProfile(appProfile string) BigtableChannelPoolOption {
-	return func(p *BigtableChannelPool) {
-		p.appProfile = appProfile
-	}
-}
-
 // WithMeterProvider provides the meter provider for writing metrics
 func WithMeterProvider(mp metric.MeterProvider) BigtableChannelPoolOption {
 	return func(p *BigtableChannelPool) {
@@ -129,24 +122,22 @@ func WithDirectAccessChecker(checker DirectAccessChecker) BigtableChannelPoolOpt
 	}
 }
 
-// WithLogger provides the logger for logging events
-func WithLogger(logger *log.Logger) BigtableChannelPoolOption {
+// WithChannelPrimer plugs in the strategy used to warm freshly-dialed
+// channels before they enter rotation. Optional: when no primer is supplied,
+// the pool's connection factory dials the channel and returns it without
+// issuing any prime RPC. The classic channel pool factory wires up a
+// PingAndWarm-based primer; alternative pool factories can swap in a
+// different strategy (e.g. session-based) or pass nothing at all.
+func WithChannelPrimer(primer ChannelPrimer) BigtableChannelPoolOption {
 	return func(p *BigtableChannelPool) {
-		p.logger = logger
+		p.channelPrimer = primer
 	}
 }
 
-// WithInstanceName provides the full instance Name
-func WithInstanceName(instanceName string) BigtableChannelPoolOption {
-	return func(p *BigtableChannelPool) {
-		p.instanceName = instanceName
-	}
-}
-
-// WithFeatureFlagsMetadata provides the feature flags metadata
-func WithFeatureFlagsMetadata(featureFlagsMd metadata.MD) BigtableChannelPoolOption {
+// WithLogger provides the logger for logging events
+func WithLogger(logger *log.Logger) BigtableChannelPoolOption {
 	return func(p *BigtableChannelPool) {
-		p.featureFlagsMD = featureFlagsMd
+		p.logger = logger
 	}
 }
 
@@ -372,10 +363,7 @@ type BigtableChannelPool struct {
 	poolCtx    context.Context    // Context for the pool's background tasks
 	poolCancel context.CancelFunc // Function to cancel the poolCtx
 
-	logger         *log.Logger // logging events
-	appProfile     string
-	instanceName   string
-	featureFlagsMD metadata.MD
+	logger *log.Logger // logging events
 
 	factory *connectionFactory // Use the factory for connection creation
 
@@ -391,6 +379,13 @@ type BigtableChannelPool struct {
 	// direct_access/compatible metric still surfaces the off state.
 	directAccessChecker DirectAccessChecker
 
+	// channelPrimer is the pluggable strategy used to warm freshly-dialed
+	// channels. Optional: when nil, the connection factory skips priming
+	// entirely and hands the raw connection straight to the pool. The
+	// classic channel pool factory wires up a PingAndWarm-based primer; the
+	// future session-pool factory may skip it.
+	channelPrimer ChannelPrimer
+
 	// background monitors
 	monitors []Monitor
 }
@@ -441,9 +436,9 @@ func NewBigtableChannelPool(ctx context.Context, connPoolSize int, strategy btop
 
 	// Default to the standard dialer. The Direct Access checker may swap the
 	// dialer for the direct-access equivalent after a successful compatibility
-	// probe. Feature-flag metadata always comes from pool.featureFlagsMD (set
-	// via WithFeatureFlagsMetadata) — both the direct-access and standard-path
-	// connection factories read it from the same place.
+	// probe. The ChannelPrimer (if any) is the single source of priming
+	// behavior — both the direct-access and standard-path factories run
+	// fresh connections through it before they enter rotation.
 	factoryDial := dial
 
 	var firstConn *BigtableConn
@@ -463,11 +458,9 @@ func NewBigtableChannelPool(ctx context.Context, connPoolSize int, strategy btop
 
 	// Initialize the connectionFactory
 	pool.factory = &connectionFactory{
-		dial:           factoryDial,
-		instanceName:   pool.instanceName,
-		appProfile:     pool.appProfile,
-		featureFlagsMD: pool.featureFlagsMD,
-		logger:         pool.logger,
+		dial:   factoryDial,
+		primer: pool.channelPrimer,
+		logger: pool.logger,
 	}
 
 	// Set the selection function based on the strategy
@@ -980,18 +973,18 @@ func (p *BigtableChannelPool) removeConnections(decreaseDelta, minConns, maxRemo
 
 }
 
-// connectionFactory is responsible for creating and priming new Bigtable connections.
-// TODO remove these members from BigtableConnPool struct
+// connectionFactory is responsible for creating and (optionally) priming
+// new Bigtable connections. When primer is nil the factory dials and
+// returns the connection without warming it.
 type connectionFactory struct {
-	dial           func() (*BigtableConn, error)
-	instanceName   string
-	appProfile     string
-	featureFlagsMD metadata.MD
-	logger         *log.Logger
+	dial   func() (*BigtableConn, error)
+	primer ChannelPrimer
+	logger *log.Logger
 }
 
-// newEntry creates a new connection, primes it, and returns it as a connEntry.
-// Blocks until the connection is successfully primed, or returns an error.
+// newEntry creates a new connection, primes it (if a primer is configured),
+// and returns it as a connEntry. Blocks until the connection is ready, or
+// returns an error.
 func (cf *connectionFactory) newEntry(ctx context.Context) (*connEntry, error) {
 	conn, err := cf.dial()
 	if err != nil {
@@ -1006,8 +999,13 @@ func (cf *connectionFactory) newEntry(ctx context.Context) (*connEntry, error) {
 	return &connEntry{conn: conn}, nil
 }
 
-// primeWithRetry attempts to prime the connection, retrying with exponential backoff.
+// primeWithRetry runs the configured ChannelPrimer with exponential backoff.
+// Returns nil immediately when no primer is configured, so the pool can be
+// used without priming.
 func (cf *connectionFactory) primeWithRetry(ctx context.Context, conn *BigtableConn) error {
+	if cf.primer == nil {
+		return nil
+	}
 	backoffPolicy := gax.Backoff{
 		Initial:    100 * time.Millisecond,
 		Max:        2 * time.Second,
@@ -1022,7 +1020,7 @@ func (cf *connectionFactory) primeWithRetry(ctx context.Context, conn *BigtableC
 			return fmt.Errorf("bigtable_connpool:  error before prime attempt %d: %w", attempt, err)
 		}
 
-		lastErr = conn.Prime(ctx, cf.instanceName, cf.appProfile, cf.featureFlagsMD)
+		lastErr = cf.primer.Prime(ctx, conn)
 		if lastErr == nil {
 			return nil
 		}