server: adopt for settings rangefeed-backed settingswatcher, remove g…

pkg/server/BUILD.bazel

@@ -38,7 +38,6 @@ go_library(
         "server_sql.go",
         "server_systemlog_gc.go",
         "settings_cache.go",
-        "settingsworker.go",
         "sql_stats.go",
         "statement_diagnostics_requests.go",
         "statements.go",
@@ -297,7 +296,7 @@ go_test(
         "server_systemlog_gc_test.go",
         "server_test.go",
         "settings_cache_test.go",
-        "settingsworker_test.go",
+        "settings_test.go",
         "statements_test.go",
         "stats_test.go",
         "status_test.go",
@@ -321,6 +320,7 @@ go_test(
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/kv/kvserver",
+        "//pkg/kv/kvserver/closedts",
         "//pkg/kv/kvserver/closedts/ctpb",
         "//pkg/kv/kvserver/kvserverbase",
         "//pkg/kv/kvserver/kvserverpb",

pkg/server/server.go

@@ -785,6 +785,7 @@ func NewServer(cfg Config, stopper *stop.Stopper) (*Server, error) {
 	})
 	registry.AddMetricStruct(kvProber.Metrics())
 
+	settingsWriter := &settingsCacheWriter{eng: engines[0]}
 	sqlServer, err := newSQLServer(ctx, sqlServerArgs{
 		sqlServerOptionalKVArgs: sqlServerOptionalKVArgs{
 			nodesStatusServer:        serverpb.MakeOptionalNodesStatusServer(sStatus),
@@ -824,6 +825,7 @@ func NewServer(cfg Config, stopper *stop.Stopper) (*Server, error) {
 		regionsServer:            sStatus,
 		tenantUsageServer:        tenantUsage,
 		monitorAndMetrics:        sqlMonitorAndMetrics,
+		settingsStorage:          settingsWriter,
 	})
 	if err != nil {
 		return nil, err
@@ -1545,7 +1547,9 @@ func (s *Server) PreStart(ctx context.Context) error {
 
 	// Apply any cached initial settings (and start the gossip listener) as early
 	// as possible, to avoid spending time with stale settings.
-	if err := s.refreshSettings(state.initialSettingsKVs); err != nil {
+	if err := initializeCachedSettings(
+		ctx, keys.SystemSQLCodec, s.st.MakeUpdater(), state.initialSettingsKVs,
+	); err != nil {
 		return errors.Wrap(err, "during initializing settings updater")
 	}
 

pkg/server/server_sql.go

@@ -302,6 +302,10 @@ type sqlServerArgs struct {
 
 	// monitorAndMetrics contains the return value of newRootSQLMemoryMonitor.
 	monitorAndMetrics monitorAndMetrics
+
+	// settingsStorage is an optional interface to drive storing of settings
+	// data on disk to provide a fresh source of settings upon next startup.
+	settingsStorage settingswatcher.Storage
 }
 
 type monitorAndMetrics struct {
@@ -936,12 +940,9 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 		reporter.TestingKnobs = &cfg.TestingKnobs.Server.(*TestingKnobs).DiagnosticsTestingKnobs
 	}
 
-	var settingsWatcher *settingswatcher.SettingsWatcher
-	if !codec.ForSystemTenant() {
-		settingsWatcher = settingswatcher.New(
-			cfg.clock, codec, cfg.Settings, cfg.rangeFeedFactory, cfg.stopper,
-		)
-	}
+	settingsWatcher := settingswatcher.New(
+		cfg.clock, codec, cfg.Settings, cfg.rangeFeedFactory, cfg.stopper, cfg.settingsStorage,
+	)
 
 	return &SQLServer{
 		ambientCtx:              cfg.BaseConfig.AmbientCtx,
@@ -1146,10 +1147,8 @@ func (s *SQLServer) preStart(
 		bootstrapVersion = roachpb.Version{Major: 20, Minor: 1, Internal: 1}
 	}
 
-	if s.settingsWatcher != nil {
-		if err := s.settingsWatcher.Start(ctx); err != nil {
-			return errors.Wrap(err, "initializing settings")
-		}
+	if err := s.settingsWatcher.Start(ctx); err != nil {
+		return errors.Wrap(err, "initializing settings")
 	}
 
 	// Run startup migrations (note: these depend on jobs subsystem running).

pkg/server/settings_cache.go

@@ -15,12 +15,26 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/server/settingswatcher"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
+	"github.com/cockroachdb/errors"
 )
 
+type settingsCacheWriter struct {
+	eng storage.Engine
+}
+
+func (s settingsCacheWriter) WriteKVs(ctx context.Context, kvs []roachpb.KeyValue) error {
+	return storeCachedSettingsKVs(ctx, s.eng, kvs)
+}
+
+var _ settingswatcher.Storage = (*settingsCacheWriter)(nil)
+
 // storeCachedSettingsKVs stores or caches node's settings locally.
 // This helps in restoring the node restart with the at least the same settings with which it died.
 func storeCachedSettingsKVs(ctx context.Context, eng storage.Engine, kvs []roachpb.KeyValue) error {
@@ -64,3 +78,22 @@ func loadCachedSettingsKVs(_ context.Context, eng storage.Engine) ([]roachpb.Key
 	}
 	return settingsKVs, nil
 }
+
+func initializeCachedSettings(
+	ctx context.Context, codec keys.SQLCodec, updater settings.Updater, kvs []roachpb.KeyValue,
+) error {
+	dec := settingswatcher.MakeRowDecoder(codec)
+	for _, kv := range kvs {
+		settings, val, valType, _, err := dec.DecodeRow(kv)
+		if err != nil {
+			return errors.Wrap(err, `while decoding settings data
+-this likely indicates the settings table structure or encoding has been altered;
+-skipping settings updates`)
+		}
+		if err := updater.Set(ctx, settings, val, valType); err != nil {
+			log.Warningf(ctx, "setting %q to %q failed: %+v", settings, val, err)
+		}
+	}
+	updater.ResetRemaining(ctx)
+	return nil
+}

pkg/server/settings_cache_test.go

@@ -14,9 +14,11 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
@@ -73,9 +75,10 @@ func TestCachedSettingsServerRestart(t *testing.T) {
 			},
 		},
 	}
-
 	var settingsCache []roachpb.KeyValue
 	testServer, _, _ := serverutils.StartServer(t, serverArgs)
+	closedts.TargetDuration.Override(ctx, &testServer.ClusterSettings().SV, 10*time.Millisecond)
+	closedts.SideTransportCloseInterval.Override(ctx, &testServer.ClusterSettings().SV, 10*time.Millisecond)
 	testutils.SucceedsSoon(t, func() error {
 		store, err := testServer.GetStores().(*kvserver.Stores).GetStore(1)
 		if err != nil {

pkg/server/settingswatcher/BUILD.bazel

@@ -12,6 +12,7 @@ go_library(
         "//pkg/clusterversion",
         "//pkg/keys",
         "//pkg/kv/kvclient/rangefeed:with-mocks",
+        "//pkg/kv/kvclient/rangefeed/rangefeedbuffer",
         "//pkg/roachpb:with-mocks",
         "//pkg/settings",
         "//pkg/settings/cluster",
@@ -28,6 +29,8 @@ go_library(
         "//pkg/util/log",
         "//pkg/util/protoutil",
         "//pkg/util/stop",
+        "//pkg/util/syncutil",
+        "//pkg/util/syncutil/singleflight",
         "@com_github_cockroachdb_errors//:errors",
     ],
 )
@@ -43,6 +46,7 @@ go_test(
         ":settingswatcher",
         "//pkg/base",
         "//pkg/keys",
+        "//pkg/kv",
         "//pkg/roachpb:with-mocks",
         "//pkg/security",
         "//pkg/security/securitytest",
@@ -56,6 +60,7 @@ go_test(
         "//pkg/testutils/testcluster",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
+        "//pkg/util/syncutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
     ],