storage: add cluster settings for compaction concurrency

Add two cluster settings that control the compaction concurrency
limits: `storage.compaction_concurrency` and
`storage.max_compaction_concurrency`.

Note that the `COCKROACH_CONCURRENT_COMPACTIONS` env var is still
supported: when it is used together with the cluster setting, we take
the max of the two values.

Fixes: #144963
Release note: None

Author: RaduBerinde

pkg/storage/mvcc.go

@@ -103,7 +103,21 @@ var TargetBytesPerLockConflictError = settings.RegisterIntSetting(
 
 var defaultMaxConcurrentCompactions = getDefaultMaxConcurrentCompactions()
 
+// By default, we use up to min(GOMAXPROCS-1, 3) threads for background
+// compactions per store (reserving the final process for flushes).
 func getDefaultMaxConcurrentCompactions() int {
+	const def = 3
+	if n := runtime.GOMAXPROCS(0); n-1 < def {
+		return max(n-1, 1)
+	}
+	return def
+}
+
+// envMaxConcurrentCompactions is not zero if this node has an env var override
+// for the concurrency.
+var envMaxConcurrentCompactions = getMaxConcurrentCompactionsFromEnv()
+
+func getMaxConcurrentCompactionsFromEnv() int {
 	if v := envutil.EnvOrDefaultInt("COCKROACH_CONCURRENT_COMPACTIONS", 0); v > 0 {
 		return v
 	}
@@ -119,14 +133,26 @@ func getDefaultMaxConcurrentCompactions() int {
 	if oldV := envutil.EnvOrDefaultInt("COCKROACH_ROCKSDB_CONCURRENCY", 0); oldV > 0 {
 		return max(oldV-1, 1)
 	}
+	return 0
+}
 
-	// By default use up to min(GOMAXPROCS-1, 3) threads for background
-	// compactions per store (reserving the final process for flushes).
-	const upperLimit = 3
-	if n := runtime.GOMAXPROCS(0); n-1 < upperLimit {
-		return max(n-1, 1)
+// determineMaxConcurrentCompactions determines the upper limit on compaction
+// concurrency.
+//
+// Normally, we use the default limit of min(3, numCPU-1). This limit can be
+// changed via an environment variable or via a cluster setting. If both of
+// those are used, the maximum of them is taken.
+func determineMaxConcurrentCompactions(defaultValue int, envValue int, clusterSetting int) int {
+	if envValue > 0 {
+		if clusterSetting > 0 {
+			return max(envValue, clusterSetting)
+		}
+		return envValue
+	}
+	if clusterSetting > 0 {
+		return clusterSetting
 	}
-	return upperLimit
+	return defaultValue
 }
 
 // l0SubLevelCompactionConcurrency is the sub-level threshold at which to

pkg/storage/pebble.go

@@ -282,6 +282,50 @@ var walFailoverUnhealthyOpThreshold = settings.RegisterDurationSetting(
 	settings.WithPublic,
 )
 
+// This cluster setting controls the baseline compaction concurrency (which
+// Pebble can dynamically increase up to the max concurrency; see
+// CompactionConcurrencyRange).
+//
+// When the value of this cluster setting is larger than the max concurrency
+// (see below), this cluster setting takes precedence (i.e. the max concurrency
+// will also equal this value).
+//
+// The baseline compaction concurrency is temporarily overridden while
+// crdb_internal.set_compaction_concurrency is running (which uses
+// SetCompactionConcurrency).
+var compactionConcurrencyLower = settings.RegisterIntSetting(
+	settings.ApplicationLevel,
+	"storage.compaction_concurrency",
+	"the baseline number of concurrent compactions",
+	1,
+	settings.IntWithMinimum(1),
+)
+
+// The maximum concurrency can be configured via an env var (which allows
+// per-node control) and/or this cluster setting (which applies to the entire
+// cluster).
+//
+// The environment variable is COCKROACH_CONCURRENT_COMPACTIONS (we also support
+// a deprecated variable, see getMaxConcurrentCompactionsFromEnv()).
+//
+// In the absence of any configuration, we use min(GOMAXPROCS-1, 3).
+//
+// If both the env variable and cluster setting are set, we take the maximum of
+// the two.
+//
+// In all cases, the maximum concurrency is temporarily overridden while
+// crdb_internal.set_compaction_concurrency is running (which uses
+// SetCompactionConcurrency).
+var compactionConcurrencyUpper = settings.RegisterIntSetting(
+	settings.ApplicationLevel,
+	"storage.max_compaction_concurrency",
+	"the maximum number of concurrent compactions (0 = default); the default value is "+
+		"min(3,numCPUs-1) or what the COCKROACH_CONCURRENT_COMPACTIONS env var specifies; "+
+		"the env var also takes precedence over the cluster setting when the latter is lower",
+	0,
+	settings.NonNegativeInt,
+)
+
 // TODO(ssd): This could be SystemOnly but we currently init pebble
 // engines for temporary storage. Temporary engines shouldn't really
 // care about download compactions, but they do currently simply
@@ -672,12 +716,18 @@ func newPebble(ctx context.Context, cfg engineConfig) (p *Pebble, err error) {
 			return getCompressionAlgorithm(ctx, cfg.settings, CompressionAlgorithmStorage)
 		}
 	}
-	// Note: the MaxConcurrentCompactions function will be wrapped below to allow
-	// overriding dynamically via overriding the max at runtime through
+	// Note: the CompactionConcurrencyRange function will be wrapped below to
+	// allow overriding the lower and upper values at runtime through
 	// Engine.SetCompactionConcurrency.
 	if cfg.opts.CompactionConcurrencyRange == nil {
 		cfg.opts.CompactionConcurrencyRange = func() (lower, upper int) {
-			return 1, defaultMaxConcurrentCompactions
+			lower = int(compactionConcurrencyLower.Get(&cfg.settings.SV))
+			upper = determineMaxConcurrentCompactions(
+				defaultMaxConcurrentCompactions,
+				envMaxConcurrentCompactions,
+				int(compactionConcurrencyUpper.Get(&cfg.settings.SV)),
+			)
+			return lower, max(lower, upper)
 		}
 	}
 	if cfg.opts.MaxConcurrentDownloads == nil {
@@ -790,8 +840,8 @@ func newPebble(ctx context.Context, cfg engineConfig) (p *Pebble, err error) {
 		diskWriteStatsCollector: cfg.DiskWriteStatsCollector,
 	}
 
-	// Wrap the MaxConcurrentCompactions function to allow overriding dynamically via
-	// overriding the max at runtime through Engine.SetCompactionConcurrency.
+	// Wrap the CompactionConcurrencyRange function to allow overriding the lower
+	// and upper values at runtime through Engine.SetCompactionConcurrency.
 	cfg.opts.CompactionConcurrencyRange = p.cco.Wrap(cfg.opts.CompactionConcurrencyRange)
 
 	// NB: The ordering of the event listeners passed to TeeEventListener is

pkg/storage/pebble_test.go

@@ -1505,69 +1505,86 @@ func TestCompactionConcurrencyEnvVars(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	maxProcsBefore := runtime.GOMAXPROCS(0)
-	defer runtime.GOMAXPROCS(maxProcsBefore)
-
 	type testCase struct {
-		maxProcs             int
-		rocksDBConcurrency   string
-		cockroachConcurrency string
-		want                 int
+		cockroachConcurrentCompactions string
+		cockroachRocksDBConcurrency    string
+		want                           int
 	}
 	cases := []testCase{
-		// Defaults
-		{32, "", "", 3},
-		{4, "", "", 3},
-		{3, "", "", 2},
-		{2, "", "", 1},
-		{1, "", "", 1},
-		// Old COCKROACH_ROCKSDB_CONCURRENCY env var is set. The user-provided
-		// value includes 1 slot for flushes, so resulting compaction
-		// concurrency is n-1.
-		{32, "4", "", 3},
-		{4, "4", "", 3},
-		{2, "4", "", 3},
-		{1, "4", "", 3},
-		{32, "8", "", 7},
-		{4, "8", "", 7},
-		{2, "8", "", 7},
-		{1, "8", "", 7},
-		// New COCKROACH_CONCURRENT_COMPACTIONS env var is set.
-		{32, "", "4", 4},
-		{4, "", "4", 4},
-		{2, "", "4", 4},
-		{1, "", "4", 4},
-		{32, "", "8", 8},
-		{4, "", "8", 8},
-		{2, "", "8", 8},
-		{1, "", "8", 8},
-		// Both settings are set; COCKROACH_CONCURRENT_COMPACTIONS supersedes
-		// COCKROACH_ROCKSDB_CONCURRENCY.
-		{32, "8", "4", 4},
-		{4, "1", "4", 4},
-		{2, "2", "4", 4},
-		{1, "5", "4", 4},
-		{32, "1", "8", 8},
-		{4, "2", "8", 8},
-		{2, "4", "8", 8},
-		{1, "1", "8", 8},
+		// Default.
+		{want: 0},
+		{cockroachConcurrentCompactions: "5", want: 5},
+		{cockroachConcurrentCompactions: "1", want: 1},
+		{cockroachConcurrentCompactions: "0", want: 0},
+		// COCKROACH_ROCKSDB_CONCURRENCY is deprecated but we still support it. Note
+		// that for this env var, the compaction concurrency is the value minus 1.
+		{cockroachRocksDBConcurrency: "5", want: 4},
+		{cockroachRocksDBConcurrency: "2", want: 1},
+		{cockroachRocksDBConcurrency: "1", want: 1},
+		{cockroachRocksDBConcurrency: "0", want: 0},
+		// The non-deprecated var takes precedence.
+		{cockroachConcurrentCompactions: "10", cockroachRocksDBConcurrency: "5", want: 10},
 	}
 	for _, tc := range cases {
-		t.Run(fmt.Sprintf("GOMAXPROCS=%d,old=%q,new=%q", tc.maxProcs, tc.rocksDBConcurrency, tc.cockroachConcurrency),
+		t.Run(fmt.Sprintf("old=%q,new=%q", tc.cockroachRocksDBConcurrency, tc.cockroachConcurrentCompactions),
 			func(t *testing.T) {
-				runtime.GOMAXPROCS(tc.maxProcs)
-
-				if tc.rocksDBConcurrency == "" {
+				if tc.cockroachRocksDBConcurrency == "" {
 					defer envutil.TestUnsetEnv(t, "COCKROACH_ROCKSDB_CONCURRENCY")()
 				} else {
-					defer envutil.TestSetEnv(t, "COCKROACH_ROCKSDB_CONCURRENCY", tc.rocksDBConcurrency)()
+					defer envutil.TestSetEnv(t, "COCKROACH_ROCKSDB_CONCURRENCY", tc.cockroachRocksDBConcurrency)()
 				}
-				if tc.cockroachConcurrency == "" {
+				if tc.cockroachConcurrentCompactions == "" {
 					defer envutil.TestUnsetEnv(t, "COCKROACH_CONCURRENT_COMPACTIONS")()
 				} else {
-					defer envutil.TestSetEnv(t, "COCKROACH_CONCURRENT_COMPACTIONS", tc.cockroachConcurrency)()
+					defer envutil.TestSetEnv(t, "COCKROACH_CONCURRENT_COMPACTIONS", tc.cockroachConcurrentCompactions)()
 				}
-				require.Equal(t, tc.want, getDefaultMaxConcurrentCompactions())
+				require.Equal(t, tc.want, getMaxConcurrentCompactionsFromEnv())
+			})
+	}
+}
+
+func TestDetermineMaxConcurrentCompactions(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	maxProcsBefore := runtime.GOMAXPROCS(0)
+	defer runtime.GOMAXPROCS(maxProcsBefore)
+
+	type testCase struct {
+		maxProcs       int
+		envValue       int
+		clusterSetting int
+		want           int
+	}
+	cases := []testCase{
+		// Defaults.
+		{maxProcs: 32, want: 3},
+		{maxProcs: 8, want: 3},
+		{maxProcs: 4, want: 3},
+		{maxProcs: 2, want: 1},
+		{maxProcs: 1, want: 1},
+		// Env value override.
+		{maxProcs: 32, envValue: 10, want: 10},
+		{maxProcs: 1, envValue: 10, want: 10},
+		// Cluster setting override.
+		{maxProcs: 32, clusterSetting: 10, want: 10},
+		{maxProcs: 1, clusterSetting: 10, want: 10},
+		// Env value and cluster setting override.
+		{maxProcs: 32, envValue: 15, clusterSetting: 10, want: 15},
+		{maxProcs: 32, envValue: 10, clusterSetting: 15, want: 15},
+		{maxProcs: 1, envValue: 15, clusterSetting: 10, want: 15},
+		{maxProcs: 1, envValue: 10, clusterSetting: 15, want: 15},
+	}
+	for _, tc := range cases {
+		t.Run(fmt.Sprintf("GOMAXPROCS=%d,env=%d,setting=%d", tc.maxProcs, tc.envValue, tc.clusterSetting),
+			func(t *testing.T) {
+				runtime.GOMAXPROCS(tc.maxProcs)
+				actual := determineMaxConcurrentCompactions(
+					getDefaultMaxConcurrentCompactions(),
+					tc.envValue,
+					tc.clusterSetting,
+				)
+				require.Equal(t, tc.want, actual)
 			})
 	}
 }