feat(bigtable): modularize Direct Access compatibility check (#19987)

## Summary
- Extract the Direct Access compatibility decision out of
`BigtableChannelPool` behind a new `DirectAccessChecker` interface so
each channel pool factory can plug in its own strategy.
- Today's only implementation, `pingAndWarmDirectAccessChecker`,
preserves current behavior (PingAndWarm probe + ALTS check + async
failure investigation + `direct_access/compatible` metric). The upcoming
session pool factory will plug in a `GetClientConfiguration`-based
checker (driven by `ClientConfigurationManager`) without touching the
pool.
- A small `disabledDirectAccessChecker` keeps the
`direct_access/compatible{reason=manually_disabled}` reading when the
env/config disables Direct Access — the env var check moves up to the
factory so the pool no longer re-reads `CBT_ENABLE_DIRECTPATH`.

## Changes
- **New** `internal/transport/direct_access_checker.go`:
`DirectAccessChecker` interface, `pingAndWarmDirectAccessChecker`,
`disabledDirectAccessChecker`, and the moved helpers
(`xdsCdsURITemplate`, `checkIPPlumbing`, `checkKernelRoutes`,
`probeSingleEndpoint`).
- `internal/transport/connpool.go`: remove `directAccessDialer`,
`directAccessFeatureFlagsMD`, `daEligibleGauge` fields, the
`WithDirectAccessDialer` / `WithDirectAccessFeatureFlagsMetadata`
options, the in-pool probe + investigation chain, the duplicate
`CBT_ENABLE_DIRECTPATH` check, and the metric reporting helpers. Add
`WithDirectAccessChecker`. `NewBigtableChannelPool` now just delegates
to the checker.
- `internal/transport/channel_pool_factory.go`: when
`isDirectAccessEnabled(config)` is true, wire a
`pingAndWarmDirectAccessChecker`; otherwise wire a
`disabledDirectAccessChecker`. Both passed via
`WithDirectAccessChecker`.
- `internal/transport/connpool_test.go`: six `TestDirectAccessLogic`
subtests updated to construct the appropriate checker. The env-disabled
subtest becomes `DirectAccess_DisabledChecker` since the env-var gate
has moved to the factory layer.

## Test plan
- [x] `cd bigtable && go build ./...`
- [x] `cd bigtable && go vet ./...`
- [x] `cd bigtable && go test ./internal/transport/... -count=1` (full
transport suite, including all six `TestDirectAccessLogic` subtests,
`TestCreateAndStartManagedChannelPool*`, and
`TestManagedChannelPool_Close`)

Author: sushanb

bigtable/internal/transport/channel_pool_factory.go

@@ -153,15 +153,22 @@ 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().
 	poolOpts := []BigtableChannelPoolOption{
 		WithInstanceName(fullInstanceName),
 		WithAppProfile(config.AppProfile),
-		WithFeatureFlagsMetadata(directAccessMD),
 		WithMetricsReporterConfig(btopt.DefaultMetricsReporterConfig()),
 		WithMeterProvider(otelMeterProvider),
-		WithDirectAccessFeatureFlagsMetadata(directAccessMD),
+		WithFeatureFlagsMetadata(directAccessMD),
 	}
 
+	// Pluggable Direct Access strategy: the classic channel pool factory uses
+	// a PingAndWarm probe; when the env/config disables Direct Access we still
+	// wire a "disabled" checker so the direct_access/compatible metric keeps
+	// surfacing the off state with reason=manually_disabled. The future
+	// session pool factory will swap in a GetClientConfiguration-based checker.
 	if isDirectAccessEnabled(config) {
 		directAccessDialerOptions := make([]option.ClientOption, len(o))
 		copy(directAccessDialerOptions, o)
@@ -175,7 +182,17 @@ func CreateBigtableChannelPool(
 			}
 			return NewBigtableConn(grpcConn), nil
 		}
-		poolOpts = append(poolOpts, WithDirectAccessDialer(directAccessDialer))
+		checker := newPingAndWarmDirectAccessChecker(
+			directAccessDialer,
+			fullInstanceName,
+			config.AppProfile,
+			directAccessMD,
+			otelMeterProvider,
+			nil, // logger plumbed by callers once available
+		)
+		poolOpts = append(poolOpts, WithDirectAccessChecker(checker))
+	} else {
+		poolOpts = append(poolOpts, WithDirectAccessChecker(newDisabledDirectAccessChecker(otelMeterProvider, nil)))
 	}
 
 	return NewBigtableChannelPool(ctx,

bigtable/internal/transport/conn_recycler_test.go

@@ -38,7 +38,7 @@ func TestConnectionRecycler_CheckRecycle(t *testing.T) {
 			MaxJitter: 0,
 		}
 
-		pool, err := NewBigtableChannelPool(ctx, 1, btopt.RoundRobin, dialFunc, time.Now())
+		pool, err := NewBigtableChannelPool(ctx, 1, btopt.RoundRobin, dialFunc, time.Now(), WithDirectAccessChecker(newDisabledDirectAccessChecker(nil, nil)))
 		if err != nil {
 			t.Fatalf("Failed to create pool: %v", err)
 		}
@@ -70,7 +70,7 @@ func TestConnectionRecycler_CheckRecycle(t *testing.T) {
 			MaxJitter: 0,
 		}
 
-		pool, err := NewBigtableChannelPool(ctx, 1, btopt.RoundRobin, dialFunc, time.Now())
+		pool, err := NewBigtableChannelPool(ctx, 1, btopt.RoundRobin, dialFunc, time.Now(), WithDirectAccessChecker(newDisabledDirectAccessChecker(nil, nil)))
 		if err != nil {
 			t.Fatalf("Failed to create pool: %v", err)
 		}
@@ -99,7 +99,7 @@ func TestConnectionRecycler_CheckRecycle(t *testing.T) {
 		}
 		// 5 conns
 		poolSize := 5
-		pool, err := NewBigtableChannelPool(ctx, poolSize, btopt.RoundRobin, dialFunc, time.Now())
+		pool, err := NewBigtableChannelPool(ctx, poolSize, btopt.RoundRobin, dialFunc, time.Now(), WithDirectAccessChecker(newDisabledDirectAccessChecker(nil, nil)))
 		if err != nil {
 			t.Fatalf("Failed to create pool: %v", err)
 		}