feat(spanner): complete location-aware routing resilience and observability (#14418)

## Summary

This change expands the location-aware routing path in the Spanner Go
client with endpoint lifecycle management,
stronger endpoint health handling, retry-aware exclusion, latency-aware
replica selection, and route-selection tracing.

  ## What Changed

- enabled location-aware routing automatically for experimental-host
configurations while preserving env-var override
    behavior
- added endpoint cache support for GetIfPresent, Evict, and
DefaultChannel
- switched endpoint health evaluation to real gRPC connectivity state,
including transient-failure detection
  - added an endpoint lifecycle manager with:
      - background probing
      - idle eviction
      - endpoint recreation requests
      - recent transient-failure eviction tracking
- integrated lifecycle handling into routing and affinity fallback paths
- added request-id keyed one-shot endpoint exclusion for
RESOURCE_EXHAUSTED retries
- expanded skipped_tablet_uid reporting for transient failures and
recent transient-failure evictions, with
    deduplication
  - added OpenTelemetry route-selection span attributes and events
  - updated focused tests and goldens for the new routing behavior

  ## Behavioral Notes

- stopped or transient-failure endpoints are now treated differently
from simply non-ready endpoints during tablet
    selection
- retry attempts can avoid previously selected routed endpoints after
RESOURCE_EXHAUSTED
- leader-preferred requests still favor close leaders, but can fall back
to a closer non-leader replica when the leader
    is too far away
- route selection is now observable through tracing attributes and
events

Author: rahul2393

spanner/batch_test.go

@@ -389,6 +389,14 @@ func TestBatchExecute_Query_PreparesRoutingHint(t *testing.T) {
 	client.locationRouter.observePartialResultSet(&sppb.PartialResultSet{
 		CacheUpdate: &sppb.CacheUpdate{DatabaseId: 7},
 	})
+	waitForAsyncRoutingUpdate(t, func() bool {
+		req := &sppb.ExecuteSqlRequest{
+			Sql: "SELECT 1",
+		}
+		client.locationRouter.prepareExecuteSQLRequest(context.Background(), req)
+		hint := req.GetRoutingHint()
+		return hint != nil && hint.GetDatabaseId() == 7
+	})
 
 	partitions, err := txn.PartitionQuery(ctx, NewStatement(SelectSingerIDAlbumIDAlbumTitleFromAlbums), PartitionOptions{MaxPartitions: 1})
 	if err != nil {
@@ -447,6 +455,16 @@ func TestBatchExecute_Read_PreparesRoutingHint(t *testing.T) {
 	client.locationRouter.observePartialResultSet(&sppb.PartialResultSet{
 		CacheUpdate: &sppb.CacheUpdate{DatabaseId: 9},
 	})
+	waitForAsyncRoutingUpdate(t, func() bool {
+		req := &sppb.ReadRequest{
+			Table:   "Albums",
+			Columns: []string{"SingerId"},
+			KeySet:  &sppb.KeySet{},
+		}
+		client.locationRouter.prepareReadRequest(context.Background(), req)
+		hint := req.GetRoutingHint()
+		return hint != nil && hint.GetDatabaseId() == 9
+	})
 
 	partitions, err := txn.PartitionRead(ctx, "Albums", KeySets(Key{"foo"}), []string{"SingerId", "AlbumId", "AlbumTitle"}, PartitionOptions{MaxPartitions: 1})
 	if err != nil {

spanner/channel_finder.go

@@ -20,8 +20,11 @@ import (
 	"context"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	sppb "cloud.google.com/go/spanner/apiv1/spannerpb"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protoreflect"
 )
 
 type channelFinder struct {
@@ -30,26 +33,86 @@ type channelFinder struct {
 	databaseID  atomic.Uint64
 	recipeCache *keyRecipeCache
 	rangeCache  *keyRangeCache
+
+	coalescingMu    sync.Mutex
+	pendingUpdates  []*sppb.CacheUpdate
+	flushScheduled  bool
+	coalescingDelay time.Duration
+	scheduleFlush   func(time.Duration, func())
+}
+
+const cacheUpdateCoalescingWindow = 5 * time.Millisecond
+
+func defaultChannelFinderFlushScheduler(delay time.Duration, fn func()) {
+	time.AfterFunc(delay, fn)
 }
 
 func newChannelFinder(endpointCache channelEndpointCache) *channelFinder {
 	return &channelFinder{
-		recipeCache: newKeyRecipeCache(),
-		rangeCache:  newKeyRangeCache(endpointCache),
+		recipeCache:     newKeyRecipeCache(),
+		rangeCache:      newKeyRangeCache(endpointCache),
+		coalescingDelay: cacheUpdateCoalescingWindow,
+		scheduleFlush:   defaultChannelFinderFlushScheduler,
 	}
 }
 
 func (f *channelFinder) useDeterministicRandom() {
 	f.rangeCache.useDeterministicRandom()
 }
 
+func (f *channelFinder) setLifecycleManager(lifecycleManager *endpointLifecycleManager) {
+	if f == nil {
+		return
+	}
+	f.rangeCache.setLifecycleManager(lifecycleManager)
+}
+
+func (f *channelFinder) setCoalescingDelayForTest(delay time.Duration) {
+	if f == nil {
+		return
+	}
+	f.coalescingMu.Lock()
+	defer f.coalescingMu.Unlock()
+	f.coalescingDelay = delay
+}
+
+func (f *channelFinder) setFlushSchedulerForTest(schedule func(time.Duration, func())) {
+	if f == nil {
+		return
+	}
+	if schedule == nil {
+		schedule = defaultChannelFinderFlushScheduler
+	}
+	f.coalescingMu.Lock()
+	defer f.coalescingMu.Unlock()
+	f.scheduleFlush = schedule
+}
+
 func (f *channelFinder) update(update *sppb.CacheUpdate) {
 	if update == nil {
 		return
 	}
 	f.updateMu.Lock()
 	defer f.updateMu.Unlock()
+	f.applyUpdateLocked(update)
+}
+
+func (f *channelFinder) applyUpdates(updates []*sppb.CacheUpdate) {
+	if len(updates) == 0 {
+		return
+	}
+	f.updateMu.Lock()
+	defer f.updateMu.Unlock()
 
+	for _, update := range updates {
+		f.applyUpdateLocked(update)
+	}
+}
+
+func (f *channelFinder) applyUpdateLocked(update *sppb.CacheUpdate) {
+	if update == nil {
+		return
+	}
 	currentID := f.databaseID.Load()
 	if currentID != update.GetDatabaseId() {
 		if currentID != 0 {
@@ -64,13 +127,103 @@ func (f *channelFinder) update(update *sppb.CacheUpdate) {
 	f.rangeCache.addRanges(update)
 }
 
+func (f *channelFinder) updateAsync(update *sppb.CacheUpdate) {
+	if !f.shouldProcessUpdate(update) {
+		return
+	}
+	f.enqueueCoalescedUpdate(update)
+}
+
+func (f *channelFinder) shouldProcessUpdate(update *sppb.CacheUpdate) bool {
+	if update == nil {
+		return false
+	}
+	// Apply any material cache update and let applyUpdateLocked handle a database
+	// switch by clearing stale state before storing the new database ID. For
+	// database-ID-only messages, only process them when they indicate that this
+	// finder has switched to a different database; database IDs are treated as
+	// identity values, not an ordered sequence.
+	if f.isMaterialUpdate(update) {
+		return true
+	}
+	updateDatabaseID := update.GetDatabaseId()
+	return updateDatabaseID != 0 && f.databaseID.Load() != updateDatabaseID
+}
+
+func (*channelFinder) isMaterialUpdate(update *sppb.CacheUpdate) bool {
+	if update == nil {
+		return false
+	}
+	return len(update.GetGroup()) > 0 ||
+		len(update.GetRange()) > 0 ||
+		(update.GetKeyRecipes() != nil && len(update.GetKeyRecipes().GetRecipe()) > 0)
+}
+
+func (f *channelFinder) enqueueCoalescedUpdate(update *sppb.CacheUpdate) {
+	if f == nil || update == nil {
+		return
+	}
+
+	f.coalescingMu.Lock()
+	f.pendingUpdates = append(f.pendingUpdates, cloneCacheUpdate(update))
+	if f.flushScheduled {
+		f.coalescingMu.Unlock()
+		return
+	}
+	f.flushScheduled = true
+	delay := f.coalescingDelay
+	scheduleFlush := f.scheduleFlush
+	f.coalescingMu.Unlock()
+
+	scheduleFlush(delay, f.flushCoalescedUpdates)
+}
+
+func (f *channelFinder) flushCoalescedUpdates() {
+	if f == nil {
+		return
+	}
+	f.coalescingMu.Lock()
+	updates := f.pendingUpdates
+	f.pendingUpdates = nil
+	f.flushScheduled = false
+	f.coalescingMu.Unlock()
+
+	f.applyUpdates(updates)
+}
+
+func cloneCacheUpdate(update *sppb.CacheUpdate) *sppb.CacheUpdate {
+	if update == nil {
+		return nil
+	}
+	return cloneProto(update)
+}
+
+func cloneProto[M interface{ ProtoReflect() protoreflect.Message }](msg M) M {
+	if any(msg) == nil {
+		var zero M
+		return zero
+	}
+	return proto.Clone(msg).(M)
+}
+
 func (f *channelFinder) findServerRead(ctx context.Context, req *sppb.ReadRequest, preferLeader bool) channelEndpoint {
+	return f.findServerReadWithExclusions(ctx, req, preferLeader, nil)
+}
+
+func (f *channelFinder) findServerReadWithExclusions(ctx context.Context, req *sppb.ReadRequest, preferLeader bool, excludedEndpoints endpointExcluder) channelEndpoint {
+	endpoint, _ := f.findServerReadWithExclusionsAndDetails(ctx, req, preferLeader, excludedEndpoints)
+	return endpoint
+}
+
+func (f *channelFinder) findServerReadWithExclusionsAndDetails(ctx context.Context, req *sppb.ReadRequest, preferLeader bool, excludedEndpoints endpointExcluder) (channelEndpoint, routeSelectionDetails) {
+	details := newRouteSelectionDetails()
 	if req == nil {
-		return nil
+		details.defaultReasonCode = routeReasonRangeCacheMiss
+		return nil, details
 	}
 	f.recipeCache.computeReadKeys(req)
 	hint := ensureReadRoutingHint(req)
-	return f.fillRoutingHint(ctx, preferLeader, rangeModeCoveringSplit, req.GetDirectedReadOptions(), hint)
+	return f.fillRoutingHintWithExclusionsAndDetails(ctx, preferLeader, rangeModeCoveringSplit, req.GetDirectedReadOptions(), hint, excludedEndpoints)
 }
 
 func (f *channelFinder) findServerReadWithTransaction(ctx context.Context, req *sppb.ReadRequest) channelEndpoint {
@@ -81,12 +234,23 @@ func (f *channelFinder) findServerReadWithTransaction(ctx context.Context, req *
 }
 
 func (f *channelFinder) findServerExecuteSQL(ctx context.Context, req *sppb.ExecuteSqlRequest, preferLeader bool) channelEndpoint {
+	return f.findServerExecuteSQLWithExclusions(ctx, req, preferLeader, nil)
+}
+
+func (f *channelFinder) findServerExecuteSQLWithExclusions(ctx context.Context, req *sppb.ExecuteSqlRequest, preferLeader bool, excludedEndpoints endpointExcluder) channelEndpoint {
+	endpoint, _ := f.findServerExecuteSQLWithExclusionsAndDetails(ctx, req, preferLeader, excludedEndpoints)
+	return endpoint
+}
+
+func (f *channelFinder) findServerExecuteSQLWithExclusionsAndDetails(ctx context.Context, req *sppb.ExecuteSqlRequest, preferLeader bool, excludedEndpoints endpointExcluder) (channelEndpoint, routeSelectionDetails) {
+	details := newRouteSelectionDetails()
 	if req == nil {
-		return nil
+		details.defaultReasonCode = routeReasonRangeCacheMiss
+		return nil, details
 	}
 	f.recipeCache.computeQueryKeys(req)
 	hint := ensureExecuteSQLRoutingHint(req)
-	return f.fillRoutingHint(ctx, preferLeader, rangeModePickRandom, req.GetDirectedReadOptions(), hint)
+	return f.fillRoutingHintWithExclusionsAndDetails(ctx, preferLeader, rangeModePickRandom, req.GetDirectedReadOptions(), hint, excludedEndpoints)
 }
 
 func (f *channelFinder) findServerExecuteSQLWithTransaction(ctx context.Context, req *sppb.ExecuteSqlRequest) channelEndpoint {
@@ -97,46 +261,93 @@ func (f *channelFinder) findServerExecuteSQLWithTransaction(ctx context.Context,
 }
 
 func (f *channelFinder) findServerBeginTransaction(ctx context.Context, req *sppb.BeginTransactionRequest) channelEndpoint {
+	return f.findServerBeginTransactionWithExclusions(ctx, req, nil)
+}
+
+func (f *channelFinder) findServerBeginTransactionWithExclusions(ctx context.Context, req *sppb.BeginTransactionRequest, excludedEndpoints endpointExcluder) channelEndpoint {
+	endpoint, _ := f.findServerBeginTransactionWithExclusionsAndDetails(ctx, req, excludedEndpoints)
+	return endpoint
+}
+
+func (f *channelFinder) findServerBeginTransactionWithExclusionsAndDetails(ctx context.Context, req *sppb.BeginTransactionRequest, excludedEndpoints endpointExcluder) (channelEndpoint, routeSelectionDetails) {
+	details := newRouteSelectionDetails()
 	if req == nil || req.GetMutationKey() == nil {
-		return nil
+		details.defaultReasonCode = routeReasonRangeCacheMiss
+		return nil, details
 	}
-	return f.routeMutation(ctx, req.GetMutationKey(), preferLeaderFromTransactionOptions(req.GetOptions()), ensureBeginTransactionRoutingHint(req))
+	return f.routeMutationWithExclusionsAndDetails(ctx, req.GetMutationKey(), preferLeaderFromTransactionOptions(req.GetOptions()), ensureBeginTransactionRoutingHint(req), excludedEndpoints)
 }
 
 func (f *channelFinder) fillCommitRoutingHint(ctx context.Context, req *sppb.CommitRequest) channelEndpoint {
+	return f.fillCommitRoutingHintWithExclusions(ctx, req, nil)
+}
+
+func (f *channelFinder) fillCommitRoutingHintWithExclusions(ctx context.Context, req *sppb.CommitRequest, excludedEndpoints endpointExcluder) channelEndpoint {
+	endpoint, _ := f.fillCommitRoutingHintWithExclusionsAndDetails(ctx, req, excludedEndpoints)
+	return endpoint
+}
+
+func (f *channelFinder) fillCommitRoutingHintWithExclusionsAndDetails(ctx context.Context, req *sppb.CommitRequest, excludedEndpoints endpointExcluder) (channelEndpoint, routeSelectionDetails) {
+	details := newRouteSelectionDetails()
 	if req == nil {
-		return nil
+		details.defaultReasonCode = routeReasonRangeCacheMiss
+		return nil, details
 	}
 	mutation := selectMutationProtoForRouting(req.GetMutations())
 	if mutation == nil {
-		return nil
+		details.defaultReasonCode = routeReasonRangeCacheMiss
+		return nil, details
 	}
-	return f.routeMutation(ctx, mutation, true, ensureCommitRoutingHint(req))
+	return f.routeMutationWithExclusionsAndDetails(ctx, mutation, true, ensureCommitRoutingHint(req), excludedEndpoints)
 }
 
 func (f *channelFinder) routeMutation(ctx context.Context, mutation *sppb.Mutation, preferLeader bool, hint *sppb.RoutingHint) channelEndpoint {
+	return f.routeMutationWithExclusions(ctx, mutation, preferLeader, hint, nil)
+}
+
+func (f *channelFinder) routeMutationWithExclusions(ctx context.Context, mutation *sppb.Mutation, preferLeader bool, hint *sppb.RoutingHint, excludedEndpoints endpointExcluder) channelEndpoint {
+	endpoint, _ := f.routeMutationWithExclusionsAndDetails(ctx, mutation, preferLeader, hint, excludedEndpoints)
+	return endpoint
+}
+
+func (f *channelFinder) routeMutationWithExclusionsAndDetails(ctx context.Context, mutation *sppb.Mutation, preferLeader bool, hint *sppb.RoutingHint, excludedEndpoints endpointExcluder) (channelEndpoint, routeSelectionDetails) {
+	details := newRouteSelectionDetails()
 	if mutation == nil || hint == nil {
-		return nil
+		details.defaultReasonCode = routeReasonRangeCacheMiss
+		return nil, details
 	}
 	f.recipeCache.applySchemaGeneration(hint)
 	target := f.recipeCache.mutationToTargetRange(mutation)
 	if target == nil {
-		return nil
+		details.defaultReasonCode = routeReasonRangeCacheMiss
+		return nil, details
 	}
 	f.recipeCache.applyTargetRange(hint, target)
-	return f.fillRoutingHint(ctx, preferLeader, rangeModeCoveringSplit, &sppb.DirectedReadOptions{}, hint)
+	return f.fillRoutingHintWithExclusionsAndDetails(ctx, preferLeader, rangeModeCoveringSplit, &sppb.DirectedReadOptions{}, hint, excludedEndpoints)
 }
 
 func (f *channelFinder) fillRoutingHint(ctx context.Context, preferLeader bool, mode rangeMode, directedReadOptions *sppb.DirectedReadOptions, hint *sppb.RoutingHint) channelEndpoint {
+	return f.fillRoutingHintWithExclusions(ctx, preferLeader, mode, directedReadOptions, hint, nil)
+}
+
+func (f *channelFinder) fillRoutingHintWithExclusions(ctx context.Context, preferLeader bool, mode rangeMode, directedReadOptions *sppb.DirectedReadOptions, hint *sppb.RoutingHint, excludedEndpoints endpointExcluder) channelEndpoint {
+	endpoint, _ := f.fillRoutingHintWithExclusionsAndDetails(ctx, preferLeader, mode, directedReadOptions, hint, excludedEndpoints)
+	return endpoint
+}
+
+func (f *channelFinder) fillRoutingHintWithExclusionsAndDetails(ctx context.Context, preferLeader bool, mode rangeMode, directedReadOptions *sppb.DirectedReadOptions, hint *sppb.RoutingHint, excludedEndpoints endpointExcluder) (channelEndpoint, routeSelectionDetails) {
+	details := newRouteSelectionDetails()
 	if hint == nil {
-		return nil
+		details.defaultReasonCode = routeReasonRangeCacheMiss
+		return nil, details
 	}
 	databaseID := f.databaseID.Load()
 	if databaseID == 0 {
-		return nil
+		details.defaultReasonCode = routeReasonRangeCacheMiss
+		return nil, details
 	}
 	hint.DatabaseId = databaseID
-	return f.rangeCache.fillRoutingHint(ctx, preferLeader, mode, directedReadOptions, hint)
+	return f.rangeCache.fillRoutingHintWithExclusionsAndDetails(ctx, preferLeader, mode, directedReadOptions, hint, excludedEndpoints)
 }
 
 func preferLeaderFromSelector(selector *sppb.TransactionSelector) bool {

spanner/client.go

@@ -654,13 +654,22 @@ func newClientWithConfig(ctx context.Context, database string, config ClientConf
 	sc.mu.Unlock()
 
 	var locationRouter *locationRouter
-	if isExperimentalLocationAPIEnabled() {
+	if isExperimentalLocationAPIEnabledForConfig(config) {
 		sc.baseClientOpts = endpointClientOpts
+		defaultEndpointAddress := ""
 		if conn := pool.Conn(); conn != nil {
 			sc.endpointAuthority = normalizeAuthorityTarget(conn.Target())
+			defaultEndpointAddress = sc.endpointAuthority
 		}
-		epCache := newEndpointClientCache(sc.createEndpointClient)
+		// Some transport wrappers, such as GCPMultiEndpoint and GCPFallback,
+		// intentionally do not expose a concrete default *grpc.ClientConn and
+		// return nil from Conn(). In that case location-aware routing remains
+		// enabled; we only skip deriving the default endpoint metadata used for
+		// authority preservation and default-endpoint diagnostics.
+		epCache := newEndpointClientCacheWithDefaultAddress(sc.createEndpointClient, defaultEndpointAddress)
 		locationRouter = newLocationRouter(epCache)
+		locationRouter.lifecycleManager = newEndpointLifecycleManager(epCache)
+		locationRouter.finder.setLifecycleManager(locationRouter.lifecycleManager)
 	}
 
 	// Create a session manager.