Skip to content

Commit

Permalink
spanconfig: introduce the spanconfig.SQLWatcher
Browse files Browse the repository at this point in the history 
This patch introduces the SQLWatcher, which is intended to incrementally
watch for updates to system.zones and system.descriptors. It does so by
establishing rangefeeds at a given timestamp.

The SQLWatcher invokes a callback from time to time  with a list of
updates that have been observed in the window
(previous checkpointTS, checkpointTS]. The checkpointTS is also
provided to the callback.

Internally, the SQLWatcher uses a buffer to keep track of events
generated by the SQLWatcher's rangefeeds. It also tracks the individual
frontier timestamps of both the rangefeeds. This helps to maintain the
notion of the combined frontier timestamp, which is computed as the
minimum of the two. This combined frontier timestamp serves as the
checkpoint to the SQLWatcher's callback function.

This interface isn't hooked up to anything yet. It'll be used by the
sponconfig.Reconciler soon to perform partial reconciliation once
full reconciliation is done. It is intended that the IDs from the
updates produced by the SQLWatcher will be fed into the SQLTranslator.

References #67679
Carved from #69661

Release note: None
  • Loading branch information
@arulajmani
arulajmani committed Nov 10, 2021
1 parent f29118b commit 24465b8
Show file tree
Hide file tree
Showing 22 changed files with 1,523 additions and 56 deletions.
1 change: 1 addition & 0 deletions pkg/BUILD.bazel
View file
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,7 @@ ALL_TESTS = [
"//pkg/spanconfig/spanconfigkvsubscriber:spanconfigkvsubscriber_test",
"//pkg/spanconfig/spanconfigmanager:spanconfigmanager_test",
"//pkg/spanconfig/spanconfigsqltranslator:spanconfigsqltranslator_test",
"//pkg/spanconfig/spanconfigsqlwatcher:spanconfigsqlwatcher_test",
"//pkg/spanconfig/spanconfigstore:spanconfigstore_test",
"//pkg/spanconfig/spanconfigtestutils:spanconfigtestutils_test",
"//pkg/sql/catalog/catalogkeys:catalogkeys_test",
Expand Down
53 changes: 27 additions & 26 deletions pkg/ccl/changefeedccl/helpers_tenant_shim_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,32 +51,33 @@ func (t *testServerShim) ServingSQLAddr() string {
return t.SQLAddr()
}

func (t *testServerShim) Stopper() *stop.Stopper { panic(unsupportedShimMethod) }
func (t *testServerShim) Start(context.Context) error { panic(unsupportedShimMethod) }
func (t *testServerShim) Node() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) NodeID() roachpb.NodeID { panic(unsupportedShimMethod) }
func (t *testServerShim) ClusterID() uuid.UUID { panic(unsupportedShimMethod) }
func (t *testServerShim) ServingRPCAddr() string { panic(unsupportedShimMethod) }
func (t *testServerShim) RPCAddr() string { panic(unsupportedShimMethod) }
func (t *testServerShim) DB() *kv.DB { panic(unsupportedShimMethod) }
func (t *testServerShim) RPCContext() *rpc.Context { panic(unsupportedShimMethod) }
func (t *testServerShim) LeaseManager() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) InternalExecutor() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) ExecutorConfig() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) TracerI() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) GossipI() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) RangeFeedFactory() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) Clock() *hlc.Clock { panic(unsupportedShimMethod) }
func (t *testServerShim) DistSenderI() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) MigrationServer() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SpanConfigAccessor() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SpanConfigSQLTranslator() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SQLServer() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SQLLivenessProvider() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) StartupMigrationsManager() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) NodeLiveness() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) HeartbeatNodeLiveness() error { panic(unsupportedShimMethod) }
func (t *testServerShim) NodeDialer() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) Stopper() *stop.Stopper { panic(unsupportedShimMethod) }
func (t *testServerShim) Start(context.Context) error { panic(unsupportedShimMethod) }
func (t *testServerShim) Node() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) NodeID() roachpb.NodeID { panic(unsupportedShimMethod) }
func (t *testServerShim) ClusterID() uuid.UUID { panic(unsupportedShimMethod) }
func (t *testServerShim) ServingRPCAddr() string { panic(unsupportedShimMethod) }
func (t *testServerShim) RPCAddr() string { panic(unsupportedShimMethod) }
func (t *testServerShim) DB() *kv.DB { panic(unsupportedShimMethod) }
func (t *testServerShim) RPCContext() *rpc.Context { panic(unsupportedShimMethod) }
func (t *testServerShim) LeaseManager() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) InternalExecutor() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) ExecutorConfig() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) TracerI() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) GossipI() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) RangeFeedFactory() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) Clock() *hlc.Clock { panic(unsupportedShimMethod) }
func (t *testServerShim) DistSenderI() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) MigrationServer() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SpanConfigAccessor() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SpanConfigSQLTranslator() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SpanConfigSQLWatcherFactory() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SQLServer() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SQLLivenessProvider() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) StartupMigrationsManager() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) NodeLiveness() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) HeartbeatNodeLiveness() error { panic(unsupportedShimMethod) }
func (t *testServerShim) NodeDialer() interface{} { panic(unsupportedShimMethod) }
func (t *testServerShim) SetDistSQLSpanResolver(spanResolver interface{}) {
panic(unsupportedShimMethod)
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/kv/kvclient/rangefeed/rangefeedbuffer/buffer.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ func New(limit int) *Buffer {
}

// Add adds the given entry to the buffer.
func (b *Buffer) Add(ctx context.Context, ev Event) error {
func (b *Buffer) Add(ev Event) error {
b.mu.Lock()
defer b.mu.Unlock()

Expand Down
26 changes: 13 additions & 13 deletions pkg/kv/kvclient/rangefeed/rangefeedbuffer/buffer_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,11 +48,11 @@ func TestBuffer(t *testing.T) {
}

{ // Flushing at a timestamp lower than buffered events should return nothing.
require.NoError(t, buffer.Add(ctx, makeEvent("a", ts(13)))) // a@13
require.NoError(t, buffer.Add(ctx, makeEvent("b", ts(11)))) // b@11
require.NoError(t, buffer.Add(ctx, makeEvent("c", ts(15)))) // c@15
require.NoError(t, buffer.Add(ctx, makeEvent("d", ts(12)))) // d@12
require.NoError(t, buffer.Add(ctx, makeEvent("e", ts(18)))) // e@18
require.NoError(t, buffer.Add(makeEvent("a", ts(13)))) // a@13
require.NoError(t, buffer.Add(makeEvent("b", ts(11)))) // b@11
require.NoError(t, buffer.Add(makeEvent("c", ts(15)))) // c@15
require.NoError(t, buffer.Add(makeEvent("d", ts(12)))) // d@12
require.NoError(t, buffer.Add(makeEvent("e", ts(18)))) // e@18

events := buffer.Flush(ctx, ts(10))
require.True(t, len(events) == 0)
Expand All @@ -75,18 +75,18 @@ func TestBuffer(t *testing.T) {
}

{ // Adding events with timestamps <= the last flush are discarded.
require.NoError(t, buffer.Add(ctx, makeEvent("a", ts(13)))) // a@13
require.NoError(t, buffer.Add(ctx, makeEvent("b", ts(11)))) // b@11
require.NoError(t, buffer.Add(ctx, makeEvent("c", ts(15)))) // c@15
require.NoError(t, buffer.Add(ctx, makeEvent("d", ts(12)))) // d@12
require.NoError(t, buffer.Add(makeEvent("a", ts(13)))) // a@13
require.NoError(t, buffer.Add(makeEvent("b", ts(11)))) // b@11
require.NoError(t, buffer.Add(makeEvent("c", ts(15)))) // c@15
require.NoError(t, buffer.Add(makeEvent("d", ts(12)))) // d@12

events := buffer.Flush(ctx, ts(15))
require.True(t, len(events) == 0)
}

{ // Additional events are flushed out at appropriate points.
require.NoError(t, buffer.Add(ctx, makeEvent("f", ts(19)))) // f@19
require.NoError(t, buffer.Add(ctx, makeEvent("g", ts(21)))) // g@21
require.NoError(t, buffer.Add(makeEvent("f", ts(19)))) // f@19
require.NoError(t, buffer.Add(makeEvent("g", ts(21)))) // g@21

events := buffer.Flush(ctx, ts(20))
require.True(t, len(events) == 2)
Expand All @@ -107,10 +107,10 @@ func TestBuffer(t *testing.T) {

{ // Ensure that buffer limits are respected.
for i := 0; i < limit; i++ {
require.NoError(t, buffer.Add(ctx, makeEvent("x", ts(101)))) // x@101
require.NoError(t, buffer.Add(makeEvent("x", ts(101)))) // x@101
}

err := buffer.Add(ctx, makeEvent("x", ts(101)))
err := buffer.Add(makeEvent("x", ts(101)))
require.ErrorIs(t, err, rangefeedbuffer.ErrBufferLimitExceeded)
}
}
Expand Down
1 change: 1 addition & 0 deletions pkg/server/BUILD.bazel
View file
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,7 @@ go_library(
"//pkg/spanconfig/spanconfigkvsubscriber",
"//pkg/spanconfig/spanconfigmanager",
"//pkg/spanconfig/spanconfigsqltranslator",
"//pkg/spanconfig/spanconfigsqlwatcher",
"//pkg/sql",
"//pkg/sql/catalog/bootstrap",
"//pkg/sql/catalog/catalogkeys",
Expand Down
10 changes: 10 additions & 0 deletions pkg/server/server_sql.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ import (
"github.com/cockroachdb/cockroach/pkg/spanconfig"
"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigmanager"
"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigsqltranslator"
"github.com/cockroachdb/cockroach/pkg/spanconfig/spanconfigsqlwatcher"
"github.com/cockroachdb/cockroach/pkg/sql"
"github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkeys"
"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
Expand Down Expand Up @@ -842,13 +843,22 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
// only do it if COCKROACH_EXPERIMENTAL_SPAN_CONFIGS is set.
spanConfigKnobs, _ := cfg.TestingKnobs.SpanConfig.(*spanconfig.TestingKnobs)
sqlTranslator := spanconfigsqltranslator.New(execCfg, codec)
sqlWatcherFactory := spanconfigsqlwatcher.NewFactory(
codec,
cfg.Settings,
cfg.rangeFeedFactory,
1<<20, /* 1 MB bufferMemLimit */
cfg.stopper,
spanConfigKnobs,
)
spanConfigMgr = spanconfigmanager.New(
cfg.db,
jobRegistry,
cfg.circularInternalExecutor,
cfg.stopper,
cfg.Settings,
cfg.spanConfigAccessor,
sqlWatcherFactory,
sqlTranslator,
spanConfigKnobs,
)
Expand Down
10 changes: 10 additions & 0 deletions pkg/server/testserver.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -961,6 +961,16 @@ func (ts *TestServer) SpanConfigSQLTranslator() interface{} {
return ts.sqlServer.spanconfigMgr.SQLTranslator
}

// SpanConfigSQLWatcherFactory is part of TestServerInterface.
func (ts *TestServer) SpanConfigSQLWatcherFactory() interface{} {
if ts.sqlServer.spanconfigMgr == nil {
panic(
"span config manager uninitialized; see EnableSpanConfigs testing knob to use span configs",
)
}
return ts.sqlServer.spanconfigMgr.SQLWatcherFactory
}

// SQLServer is part of TestServerInterface.
func (ts *TestServer) SQLServer() interface{} {
return ts.PGServer().SQLServer
Expand Down
1 change: 1 addition & 0 deletions pkg/spanconfig/BUILD.bazel
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ go_library(
"//pkg/base",
"//pkg/keys",
"//pkg/roachpb:with-mocks",
"//pkg/sql/catalog",
"//pkg/sql/catalog/descpb",
"//pkg/util/hlc",
],
Expand Down
63 changes: 56 additions & 7 deletions pkg/spanconfig/spanconfig.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import (

"github.com/cockroachdb/cockroach/pkg/keys"
"github.com/cockroachdb/cockroach/pkg/roachpb"
"github.com/cockroachdb/cockroach/pkg/sql/catalog"
"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
"github.com/cockroachdb/cockroach/pkg/util/hlc"
)
Expand Down Expand Up @@ -117,6 +118,59 @@ func FullTranslate(
return s.Translate(ctx, descpb.IDs{keys.RootNamespaceID})
}

// SQLWatcher watches for events on system.zones and system.descriptors.
type SQLWatcher interface {
// WatchForSQLUpdates watches for updates to zones and descriptors starting at
// the given timestamp (exclusive), informing callers using the handler
// callback.
//
// The handler callback[1] is invoked from time to time with a list of updates
// and a checkpointTS. Invocations of the handler callback provide the
// following semantics:
// 1. Calls to the handler are serial.
// 2. The timestamp supplied to the handler is monotonically increasing.
// 3. The list of DescriptorUpdates supplied to handler includes all events
// in the window (prevInvocationCheckpointTS, checkpointTS].
// 4. No further calls to the handler are made if a call to the handler
// returns an error.
//
// These guarantees mean that users of this interface are free to persist the
// checkpointTS and later use it to re-establish a SQLWatcher without missing
// any updates.
//
// WatchForSQLUpdates can only ever be called once, effectively making the
// SQLWatcher a single use interface.
//
// WatchForSQLUpdates may run out of memory and return an error if it is
// tracking too many events between two checkpoints.
//
// [1] Users of this interface should not intend to do expensive work in the
// handler callback.
// TODO(arul): Possibly get rid of this limitation.
WatchForSQLUpdates(
ctx context.Context,
startTS hlc.Timestamp,
handler func(ctx context.Context, updates []DescriptorUpdate, checkpointTS hlc.Timestamp) error,
) error
}

// DescriptorUpdate captures the ID and type of a descriptor or zone that the
// SQLWatcher has observed updated.
type DescriptorUpdate struct {
// ID of the descriptor/zone that has been updated.
ID descpb.ID

// DescriptorType of the descriptor/zone that has been updated. Could be either
// the specific type or catalog.Any if no information is available.
DescriptorType catalog.DescriptorType
}

// SQLWatcherFactory is used to construct new SQLWatchers.
type SQLWatcherFactory interface {
// New returns a new SQLWatcher.
New() SQLWatcher
}

// ReconciliationDependencies captures what's needed by the span config
// reconciliation job to perform its task. The job is responsible for
// reconciling a tenant's zone configurations with the clusters span
Expand All @@ -126,11 +180,7 @@ type ReconciliationDependencies interface {

SQLTranslator

// TODO(arul): We'll also want access to a "SQLWatcher", something that
// watches for changes to system.{descriptors, zones} to feed IDs to the
// SQLTranslator. These interfaces will be used by the "Reconciler to perform
// full/partial reconciliation, checkpoint the span config job, and update KV
// with the tenants span config state.
SQLWatcherFactory
}

// Store is a data structure used to store spans and their corresponding
Expand Down Expand Up @@ -235,8 +285,7 @@ func (u Update) Deletion() bool {
return u.Config.IsEmpty()
}

// Addition returns true if the update corresponds to a span config being
// added.
// Addition returns true if the update corresponds to a span config being added.
func (u Update) Addition() bool {
return !u.Deletion()
}
2 changes: 1 addition & 1 deletion pkg/spanconfig/spanconfigkvsubscriber/kvsubscriber.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -260,7 +260,7 @@ func (s *KVSubscriber) run(ctx context.Context) error {
update.Config = entry.Config
}

if err := buffer.Add(ctx, &bufferEvent{update, ev.Value.Timestamp}); err != nil {
if err := buffer.Add(&bufferEvent{update, ev.Value.Timestamp}); err != nil {
select {
case <-ctx.Done():
// The context is canceled when the rangefeed is closed by the
Expand Down
19 changes: 11 additions & 8 deletions pkg/spanconfig/spanconfigmanager/manager.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ type Manager struct {
knobs *spanconfig.TestingKnobs

spanconfig.KVAccessor
spanconfig.SQLWatcherFactory
spanconfig.SQLTranslator
}

Expand All @@ -73,21 +74,23 @@ func New(
stopper *stop.Stopper,
settings *cluster.Settings,
kvAccessor spanconfig.KVAccessor,
sqlWatcherFactory spanconfig.SQLWatcherFactory,
sqlTranslator spanconfig.SQLTranslator,
knobs *spanconfig.TestingKnobs,
) *Manager {
if knobs == nil {
knobs = &spanconfig.TestingKnobs{}
}
return &Manager{
db: db,
jr: jr,
ie: ie,
stopper: stopper,
settings: settings,
KVAccessor: kvAccessor,
SQLTranslator: sqlTranslator,
knobs: knobs,
db: db,
jr: jr,
ie: ie,
stopper: stopper,
settings: settings,
KVAccessor: kvAccessor,
SQLWatcherFactory: sqlWatcherFactory,
SQLTranslator: sqlTranslator,
knobs: knobs,
}
}

Expand Down
3 changes: 3 additions & 0 deletions pkg/spanconfig/spanconfigmanager/manager_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@ func TestManagerConcurrentJobCreation(t *testing.T) {
ts.Stopper(),
ts.ClusterSettings(),
ts.SpanConfigAccessor().(spanconfig.KVAccessor),
ts.SpanConfigSQLWatcherFactory().(spanconfig.SQLWatcherFactory),
ts.SpanConfigSQLTranslator().(spanconfig.SQLTranslator),
&spanconfig.TestingKnobs{
ManagerCreatedJobInterceptor: func(jobI interface{}) {
Expand Down Expand Up @@ -162,6 +163,7 @@ func TestManagerStartsJobIfFailed(t *testing.T) {
ts.Stopper(),
ts.ClusterSettings(),
ts.SpanConfigAccessor().(spanconfig.KVAccessor),
ts.SpanConfigSQLWatcherFactory().(spanconfig.SQLWatcherFactory),
ts.SpanConfigSQLTranslator().(spanconfig.SQLTranslator),
&spanconfig.TestingKnobs{
ManagerAfterCheckedReconciliationJobExistsInterceptor: func(exists bool) {
Expand Down Expand Up @@ -237,6 +239,7 @@ func TestManagerCheckJobConditions(t *testing.T) {
ts.Stopper(),
ts.ClusterSettings(),
ts.SpanConfigAccessor().(spanconfig.KVAccessor),
ts.SpanConfigSQLWatcherFactory().(spanconfig.SQLWatcherFactory),
ts.SpanConfigSQLTranslator().(spanconfig.SQLTranslator),
&spanconfig.TestingKnobs{
ManagerDisableJobCreation: true,
Expand Down
Loading

0 comments on commit 24465b8

Please sign in to comment.