spanconfig: introduce spanconfig.{KVWatcher,StoreWriter}

The KVWatcher is a per-store listener over `system.span_configurations`.
It listens in on updates to the global span configurations state, and
uses these updates to maintain a (new) in-memory spanconfig.StoreWriter
datastructure.

This datastructure satisfies the spanconfig.Store interface, and as such
can be used to replace the gossiped config.SystemConfig.

Release note: None
Release justification: non-production code changes

pkg/BUILD.bazel

@@ -179,7 +179,9 @@ ALL_TESTS = [
     "//pkg/server:server_test",
     "//pkg/settings:settings_test",
     "//pkg/spanconfig/spanconfigkvaccessor:spanconfigkvaccessor_test",
+    "//pkg/spanconfig/spanconfigkvwatcher:spanconfigkvwatcher_test",
     "//pkg/spanconfig/spanconfigmanager:spanconfigmanager_test",
+    "//pkg/spanconfig/spanconfigstore:spanconfigstore_test",
     "//pkg/sql/catalog/catalogkeys:catalogkeys_test",
     "//pkg/sql/catalog/catalogkv:catalogkv_test",
     "//pkg/sql/catalog/catformat:catformat_test",

pkg/base/test_server_args.go

@@ -130,6 +130,9 @@ type TestServerArgs struct {
 	// IF set, the demo login endpoint will be enabled.
 	EnableDemoLoginEndpoint bool
 
+	// If set, the experimental span configs infrastructure will be enabled.
+	EnableExperimentalSpanConfigs bool
+
 	// If set, a TraceDir is initialized at the provided path.
 	TraceDir string
 

pkg/config/zonepb/zone_test.go

@@ -1429,3 +1429,8 @@ func TestZoneConfigToSpanConfigConversion(t *testing.T) {
 		require.Equal(t, tc.expectSpanConfig, spanConfig)
 	}
 }
+
+func TestDefaultZoneAndSpanConfigs(t *testing.T) {
+	converted := DefaultZoneConfigRef().AsSpanConfig()
+	require.True(t, converted.Equal(roachpb.TestingDefaultSpanConfig()))
+}

pkg/keys/constants.go

@@ -360,10 +360,11 @@ const (
 	ZonesTableConfigColumnID = 2
 	ZonesTableConfigColFamID = 2
 
-	DescriptorTablePrimaryKeyIndexID  = 1
-	DescriptorTableDescriptorColID    = 2
-	DescriptorTableDescriptorColFamID = 2
-	TenantsTablePrimaryKeyIndexID     = 1
+	DescriptorTablePrimaryKeyIndexID         = 1
+	DescriptorTableDescriptorColID           = 2
+	DescriptorTableDescriptorColFamID        = 2
+	TenantsTablePrimaryKeyIndexID            = 1
+	SpanConfigurationsTablePrimaryKeyIndexID = 1
 
 	// Reserved IDs for other system tables. Note that some of these IDs refer
 	// to "Ranges" instead of a Table - these IDs are needed to store custom

pkg/kv/kvclient/kvcoord/split_test.go

@@ -175,7 +175,7 @@ func TestRangeSplitsWithWritePressure(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 	// Override default span config.
-	cfg := kvserver.TestingDefaultSpanConfig()
+	cfg := roachpb.TestingDefaultSpanConfig()
 	cfg.RangeMaxBytes = 1 << 18
 
 	// Manually create the local test cluster so that the split queue

pkg/kv/kvserver/BUILD.bazel

@@ -151,6 +151,7 @@ go_library(
         "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/spanconfig",
+        "//pkg/spanconfig/spanconfigstore",
         "//pkg/sql/sessiondata",
         "//pkg/sql/sqlutil",
         "//pkg/storage",

pkg/kv/kvserver/queue_concurrency_test.go

@@ -71,7 +71,7 @@ func TestBaseQueueConcurrent(t *testing.T) {
 		cfg: StoreConfig{
 			Clock:             hlc.NewClock(hlc.UnixNano, time.Second),
 			AmbientCtx:        log.AmbientContext{Tracer: tracing.NewTracer()},
-			DefaultSpanConfig: TestingDefaultSpanConfig(),
+			DefaultSpanConfig: roachpb.TestingDefaultSpanConfig(),
 		},
 	}
 

pkg/kv/kvserver/split_queue_test.go

@@ -89,7 +89,7 @@ func TestSplitQueueShouldQueue(t *testing.T) {
 		repl.mu.Lock()
 		repl.mu.state.Stats = &enginepb.MVCCStats{KeyBytes: test.bytes}
 		repl.mu.Unlock()
-		conf := TestingDefaultSpanConfig()
+		conf := roachpb.TestingDefaultSpanConfig()
 		conf.RangeMaxBytes = test.maxBytes
 		repl.SetSpanConfig(conf)
 

pkg/kv/kvserver/store.go

@@ -53,6 +53,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/spanconfig"
+	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigstore"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
@@ -215,7 +216,6 @@ func testStoreConfig(clock *hlc.Clock, version roachpb.Version) StoreConfig {
 	st := cluster.MakeTestingClusterSettingsWithVersions(version, version, true)
 	sc := StoreConfig{
 		DefaultSpanConfig:           zonepb.DefaultZoneConfigRef().AsSpanConfig(),
-		DefaultSystemSpanConfig:     zonepb.DefaultSystemZoneConfigRef().AsSpanConfig(),
 		Settings:                    st,
 		AmbientCtx:                  log.AmbientContext{Tracer: st.Tracer},
 		Clock:                       clock,
@@ -649,6 +649,7 @@ type Store struct {
 
 	computeInitialMetrics              sync.Once
 	systemConfigUpdateQueueRateLimiter *quotapool.RateLimiter
+	spanConfigUpdateQueueRateLimiter   *quotapool.RateLimiter
 }
 
 var _ kv.Sender = &Store{}
@@ -661,18 +662,17 @@ type StoreConfig struct {
 	AmbientCtx log.AmbientContext
 	base.RaftConfig
 
-	DefaultSpanConfig       roachpb.SpanConfig
-	DefaultSystemSpanConfig roachpb.SpanConfig
-	Settings                *cluster.Settings
-	Clock                   *hlc.Clock
-	DB                      *kv.DB
-	Gossip                  *gossip.Gossip
-	NodeLiveness            *liveness.NodeLiveness
-	StorePool               *StorePool
-	Transport               *RaftTransport
-	NodeDialer              *nodedialer.Dialer
-	RPCContext              *rpc.Context
-	RangeDescriptorCache    *rangecache.RangeCache
+	DefaultSpanConfig    roachpb.SpanConfig
+	Settings             *cluster.Settings
+	Clock                *hlc.Clock
+	DB                   *kv.DB
+	Gossip               *gossip.Gossip
+	NodeLiveness         *liveness.NodeLiveness
+	StorePool            *StorePool
+	Transport            *RaftTransport
+	NodeDialer           *nodedialer.Dialer
+	RPCContext           *rpc.Context
+	RangeDescriptorCache *rangecache.RangeCache
 
 	ClosedTimestampSender   *sidetransport.Sender
 	ClosedTimestampReceiver sidetransportReceiver
@@ -757,6 +757,10 @@ type StoreConfig struct {
 	// SpanConfigsEnabled determines whether we're able to use the span configs
 	// infrastructure.
 	SpanConfigsEnabled bool
+	// Used to watch for span configuration changes.
+	SpanConfigWatcher spanconfig.KVWatcher
+	// SpanConfigStore is used to store and retrieve span configs.
+	SpanConfigStore spanconfig.Store
 }
 
 // ConsistencyTestingKnobs is a BatchEvalTestingKnobs struct used to control the
@@ -1592,6 +1596,44 @@ func (s *Store) Start(ctx context.Context, stopper *stop.Stopper) error {
 		s.cfg.NodeLiveness.RegisterCallback(s.nodeIsLiveCallback)
 	}
 
+	if s.cfg.SpanConfigsEnabled {
+		// When toggling between the system config span and the span configs
+		// infrastructure, we want to re-apply all span configs on all
+		// replicas from whatever the new source is.
+		spanconfigstore.EnabledSetting.SetOnChange(&s.ClusterSettings().SV, func(ctx context.Context) {
+			enabled := spanconfigstore.EnabledSetting.Get(&s.ClusterSettings().SV)
+			if !enabled {
+				if s.cfg.Gossip != nil && s.cfg.Gossip.GetSystemConfig() != nil {
+					s.systemGossipUpdate(s.cfg.Gossip.GetSystemConfig())
+				}
+
+				return
+			}
+
+			now := s.cfg.Clock.NowAsClockTimestamp()
+			shouldQueue := s.spanConfigUpdateQueueRateLimiter.AdmitN(1)
+			newStoreReplicaVisitor(s).Visit(func(repl *Replica) bool {
+				replCtx := repl.AnnotateCtx(ctx)
+				key := repl.Desc().StartKey
+				conf, err := s.cfg.SpanConfigStore.GetSpanConfigForKey(replCtx, key)
+				if err != nil {
+					log.Fatalf(ctx, "%v", err)
+				}
+
+				repl.SetSpanConfig(conf)
+				if shouldQueue {
+					s.splitQueue.Async(replCtx, "span config update", true /* wait */, func(ctx context.Context, h queueHelper) {
+						h.MaybeAdd(ctx, repl, now)
+					})
+					s.mergeQueue.Async(replCtx, "span config update", true /* wait */, func(ctx context.Context, h queueHelper) {
+						h.MaybeAdd(ctx, repl, now)
+					})
+				}
+				return true // more
+			})
+		})
+	}
+
 	// Gossip is only ever nil while bootstrapping a cluster and
 	// in unittests.
 	if s.cfg.Gossip != nil {
@@ -1630,6 +1672,32 @@ func (s *Store) Start(ctx context.Context, stopper *stop.Stopper) error {
 		})
 	}
 
+	// SpanConfigWatcher is nil unless COCKROACH_EXPERIMENTAL_SPAN_CONFIGS
+	// is set, and in unit tests.
+	if s.cfg.SpanConfigsEnabled && s.cfg.SpanConfigWatcher != nil && !s.cfg.TestingKnobs.DisableSpanConfigWatcher {
+		// Start the watcher process to listen in on span config updates and
+		// propagate said updates to the underlying store.
+		spanConfigUpdateC, err := s.cfg.SpanConfigWatcher.WatchForKVUpdates(ctx)
+		if err != nil {
+			return err
+		}
+		if err := s.stopper.RunAsyncTask(ctx, "spanconfig-watcher", func(context.Context) {
+			for {
+				select {
+				case update := <-spanConfigUpdateC:
+					if interceptor := s.TestingKnobs().SpanConfigUpdateInterceptor; interceptor != nil {
+						interceptor(update)
+					}
+					s.onSpanConfigUpdate(ctx, update)
+				case <-s.stopper.ShouldQuiesce():
+					return
+				}
+			}
+		}); err != nil {
+			return err
+		}
+	}
+
 	if !s.cfg.TestingKnobs.DisableAutomaticLeaseRenewal {
 		s.startLeaseRenewer(ctx)
 	}
@@ -1787,6 +1855,10 @@ func (s *Store) GetConfReader() (spanconfig.StoreReader, error) {
 		return nil, errSysCfgUnavailable
 	}
 
+	if s.cfg.SpanConfigsEnabled && spanconfigstore.EnabledSetting.Get(&s.ClusterSettings().SV) {
+		return spanconfigstore.NewShadowReader(s.cfg.SpanConfigStore, sysCfg), nil
+	}
+
 	return sysCfg, nil
 }
 
@@ -1968,6 +2040,67 @@ func (s *Store) systemGossipUpdate(sysCfg *config.SystemConfig) {
 	})
 }
 
+// onSpanConfigUpdate is the callback invoked whenever this store learns of a
+// span config update.
+func (s *Store) onSpanConfigUpdate(ctx context.Context, update spanconfig.Update) {
+	if log.ExpensiveLogEnabled(ctx, 1) {
+		log.Infof(ctx, "received update span=%s conf=%s deleted=%t", update.Span, update.Config.String(), update.Deletion())
+	}
+
+	s.cfg.SpanConfigStore.Apply(ctx, update, false)
+
+	if !spanconfigstore.EnabledSetting.Get(&s.ClusterSettings().SV) {
+		return
+	}
+
+	// We'll want to offer all replicas to the split and merge queues. Be a
+	// little careful about not spawning too many individual goroutines.
+
+	now := s.cfg.Clock.NowAsClockTimestamp()
+	shouldQueue := s.spanConfigUpdateQueueRateLimiter.AdmitN(1)
+	newStoreReplicaVisitor(s).Visit(func(repl *Replica) bool {
+		replCtx := repl.AnnotateCtx(ctx)
+		replicaSpan := repl.Desc().RSpan().AsRawSpanWithNoLocals()
+
+		if !update.Span.Overlaps(replicaSpan) {
+			return true // more
+		}
+
+		// TODO(irfansharif): It's possible for a config to be applied over an
+		// entire range when it only pertains to the first half of the range.
+		// This will be corrected shortly -- we enqueue the range for a split
+		// below where we then apply the right config on each half. But still,
+		// it's surprising behavior and gets in the way of a desirable
+		// consistency guarantee: a key's config at any point in time is one
+		// that was explicitly declared over it, or the default config. We can
+		// do better, we can skip applying the config entirely and enqueue the
+		// split, then relying on the split to install the right configs on each
+		// half. The current structure is as it is to maintain parity with the
+		// system config span variant.
+		key := repl.Desc().StartKey
+		conf, err := s.cfg.SpanConfigStore.GetSpanConfigForKey(replCtx, key)
+		if err != nil {
+			log.Fatalf(replCtx, "%v", err)
+		}
+
+		repl.SetSpanConfig(conf)
+
+		// TODO(irfansharif): For symmetry with the system config span variant,
+		// we queue blindly; we could instead only queue it if we knew the
+		// range's keyspans has a split in there somewhere, or was now part of a
+		// larger range and eligible for a merge.
+		if shouldQueue {
+			s.splitQueue.Async(replCtx, "span config update", true /* wait */, func(ctx context.Context, h queueHelper) {
+				h.MaybeAdd(ctx, repl, now)
+			})
+			s.mergeQueue.Async(replCtx, "span config update", true /* wait */, func(ctx context.Context, h queueHelper) {
+				h.MaybeAdd(ctx, repl, now)
+			})
+		}
+		return true // more
+	})
+}
+
 func (s *Store) asyncGossipStore(ctx context.Context, reason string, useCached bool) {
 	if err := s.stopper.RunAsyncTask(
 		ctx, fmt.Sprintf("storage.Store: gossip on %s", reason),
@@ -2972,8 +3105,3 @@ func min(a, b int) int {
 	}
 	return b
 }
-
-// TestingDefaultSpanConfig exposes the default span config for testing purposes.
-func TestingDefaultSpanConfig() roachpb.SpanConfig {
-	return zonepb.DefaultZoneConfigRef().AsSpanConfig()
-}

pkg/kv/kvserver/testing_knobs.go

@@ -18,6 +18,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/tenantrate"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/txnwait"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/spanconfig"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
@@ -91,6 +92,8 @@ type StoreTestingKnobs struct {
 	// DisableAutomaticLeaseRenewal enables turning off the background worker
 	// that attempts to automatically renew expiration-based leases.
 	DisableAutomaticLeaseRenewal bool
+	// DisableSpanConfigWatcher disables the span config watcher process.
+	DisableSpanConfigWatcher bool
 	// LeaseRequestEvent, if set, is called when replica.requestLeaseLocked() is
 	// called to acquire a new lease. This can be used to assert that a request
 	// triggers a lease acquisition.
@@ -327,6 +330,9 @@ type StoreTestingKnobs struct {
 	// PurgeOutdatedReplicasInterceptor intercepts attempts to purge outdated
 	// replicas in the store.
 	PurgeOutdatedReplicasInterceptor func()
+	// SpanConfigUpdateInterceptor is called after the store hears about a span
+	// config update.
+	SpanConfigUpdateInterceptor func(spanconfig.Update)
 	// If set, use the given version as the initial replica version when
 	// bootstrapping ranges. This is used for testing the migration
 	// infrastructure.

pkg/roachpb/span_config.go

@@ -98,3 +98,23 @@ func (c ConstraintsConjunction) String() string {
 	}
 	return sb.String()
 }
+
+// TestingDefaultSpanConfig exports the default span config for testing purposes.
+func TestingDefaultSpanConfig() SpanConfig {
+	return SpanConfig{
+		RangeMinBytes: 128 << 20, // 128 MB
+		RangeMaxBytes: 512 << 20, // 512 MB
+		// Use 25 hours instead of the previous 24 to make users successful by
+		// default. Users desiring to take incremental backups every 24h may
+		// incorrectly assume that the previous default 24h was sufficient to do
+		// that. But the equation for incremental backups is:
+		// 	GC TTLSeconds >= (desired backup interval) + (time to perform incremental backup)
+		// We think most new users' incremental backups will complete within an
+		// hour, and larger clusters will have more experienced operators and will
+		// understand how to change these settings if needed.
+		GCPolicy: GCPolicy{
+			TTLSeconds: 25 * 60 * 60,
+		},
+		NumReplicas: 3,
+	}
+}

pkg/server/BUILD.bazel

@@ -113,7 +113,9 @@ go_library(
         "//pkg/spanconfig",
         "//pkg/spanconfig/spanconfigjob",
         "//pkg/spanconfig/spanconfigkvaccessor",
+        "//pkg/spanconfig/spanconfigkvwatcher",
         "//pkg/spanconfig/spanconfigmanager",
+        "//pkg/spanconfig/spanconfigstore",
         "//pkg/sql",
         "//pkg/sql/catalog/bootstrap",
         "//pkg/sql/catalog/catalogkeys",

pkg/server/server.go

@@ -71,6 +71,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/spanconfig"
 	_ "github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigjob" // register jobs declared outside of pkg/sql
 	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigkvaccessor"
+	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigkvwatcher"
+	"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigstore"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/systemschema"
 	"github.com/cockroachdb/cockroach/pkg/sql/contention"
@@ -607,6 +609,8 @@ func NewServer(cfg Config, stopper *stop.Stopper) (*Server, error) {
 	var spanConfigAccessor spanconfig.KVAccessor
 	if cfg.SpanConfigsEnabled {
 		storeCfg.SpanConfigsEnabled = true
+		storeCfg.SpanConfigWatcher = spanconfigkvwatcher.New(stopper, db, clock, rangeFeedFactory, keys.SpanConfigurationsTableID)
+		storeCfg.SpanConfigStore = spanconfigstore.New(storeCfg.DefaultSpanConfig)
 		spanConfigAccessor = spanconfigkvaccessor.New(
 			db, internalExecutor, cfg.Settings,
 			systemschema.SpanConfigurationsTableName.FQString(),