kvserver: allow wrapping the exising reader

This commit allows the testing StoreKVSubscriberOverride to be made more
flexible. Specifically it passes the original to the test to allow
wrapping it rather than just creating a brand new one. The main benefit
of this change is to allow overriding of a single span rather than some
of the other ways of injecting spans in.

Epic: none

Release note: None

pkg/kv/kvserver/client_spanconfigs_test.go

@@ -55,7 +55,7 @@ func TestSpanConfigUpdateAppliedToReplica(t *testing.T) {
 				DisableGCQueue:    true,
 			},
 			SpanConfig: &spanconfig.TestingKnobs{
-				StoreKVSubscriberOverride: mockSubscriber,
+				StoreKVSubscriberOverride: func(spanconfig.KVSubscriber) spanconfig.KVSubscriber { return mockSubscriber },
 			},
 		},
 	}
@@ -120,7 +120,7 @@ func TestFallbackSpanConfigOverride(t *testing.T) {
 				DisableGCQueue:    true,
 			},
 			SpanConfig: &spanconfig.TestingKnobs{
-				StoreKVSubscriberOverride: mockSubscriber,
+				StoreKVSubscriberOverride: func(spanconfig.KVSubscriber) spanconfig.KVSubscriber { return mockSubscriber },
 			},
 		},
 	}

pkg/server/server.go

@@ -753,41 +753,41 @@ func NewServer(cfg Config, stopper *stop.Stopper) (serverctl.ServerStartupInterf
 		kvAccessorForTenantRecords spanconfig.KVAccessor
 	}
 	spanConfigKnobs, _ := cfg.TestingKnobs.SpanConfig.(*spanconfig.TestingKnobs)
+	// We use the span configs infra to control whether rangefeeds are
+	// enabled on a given range. At the moment this only applies to
+	// system tables (on both host and secondary tenants). We need to
+	// consider two things:
+	// - The sql-side reconciliation process runs asynchronously. When
+	//   the config for a given range is requested, we might not yet have
+	//   it, thus falling back to the static config below.
+	// - Various internal subsystems rely on rangefeeds to function.
+	//
+	// Consequently, we configure our static fallback config to actually
+	// allow rangefeeds. As the sql-side reconciliation process kicks
+	// off, it'll install the actual configs that we'll later consult.
+	// For system table ranges we install configs that allow for
+	// rangefeeds. Until then, we simply allow rangefeeds when a more
+	// targeted config is not found.
+	fallbackConf := cfg.DefaultZoneConfig.AsSpanConfig()
+	fallbackConf.RangefeedEnabled = true
+	// We do the same for opting out of strict GC enforcement; it
+	// really only applies to user table ranges
+	fallbackConf.GCPolicy.IgnoreStrictEnforcement = true
+
+	spanConfig.subscriber = spanconfigkvsubscriber.New(
+		clock,
+		rangeFeedFactory,
+		keys.SpanConfigurationsTableID,
+		4<<20, /* 4 MB */
+		fallbackConf,
+		cfg.Settings,
+		spanconfigstore.NewBoundsReader(tenantCapabilitiesWatcher),
+		spanConfigKnobs,
+		nodeRegistry,
+	)
+
 	if spanConfigKnobs != nil && spanConfigKnobs.StoreKVSubscriberOverride != nil {
-		spanConfig.subscriber = spanConfigKnobs.StoreKVSubscriberOverride
-	} else {
-		// We use the span configs infra to control whether rangefeeds are
-		// enabled on a given range. At the moment this only applies to
-		// system tables (on both host and secondary tenants). We need to
-		// consider two things:
-		// - The sql-side reconciliation process runs asynchronously. When
-		//   the config for a given range is requested, we might not yet have
-		//   it, thus falling back to the static config below.
-		// - Various internal subsystems rely on rangefeeds to function.
-		//
-		// Consequently, we configure our static fallback config to actually
-		// allow rangefeeds. As the sql-side reconciliation process kicks
-		// off, it'll install the actual configs that we'll later consult.
-		// For system table ranges we install configs that allow for
-		// rangefeeds. Until then, we simply allow rangefeeds when a more
-		// targeted config is not found.
-		fallbackConf := cfg.DefaultZoneConfig.AsSpanConfig()
-		fallbackConf.RangefeedEnabled = true
-		// We do the same for opting out of strict GC enforcement; it
-		// really only applies to user table ranges
-		fallbackConf.GCPolicy.IgnoreStrictEnforcement = true
-
-		spanConfig.subscriber = spanconfigkvsubscriber.New(
-			clock,
-			rangeFeedFactory,
-			keys.SpanConfigurationsTableID,
-			4<<20, /* 4 MB */
-			fallbackConf,
-			cfg.Settings,
-			spanconfigstore.NewBoundsReader(tenantCapabilitiesWatcher),
-			spanConfigKnobs,
-			nodeRegistry,
-		)
+		spanConfig.subscriber = spanConfigKnobs.StoreKVSubscriberOverride(spanConfig.subscriber)
 	}
 
 	scKVAccessor := spanconfigkvaccessor.New(

pkg/spanconfig/testing_knobs.go

@@ -62,7 +62,7 @@ type TestingKnobs struct {
 
 	// StoreKVSubscriberOverride is used to override the KVSubscriber used when
 	// setting up a new store.
-	StoreKVSubscriberOverride KVSubscriber
+	StoreKVSubscriberOverride func(KVSubscriber) KVSubscriber
 
 	// KVAccessorPaginationInterceptor, if set, is invoked on every pagination
 	// event.