Merge #149540 #149712

149540: metric: Report aggregate metric when no child metrics are present r=dhartunian a=alyshanjahani-crl

SQLMetrics are PrometheusIterable, they are the first and only instance of metrics that dynamically add and remove childset metrics.

This commit fixes a bug where the includeAggregate flag being false (cluster setting disabled) resulting in these SQLMetrics not being reported.

When a PrometheusIterable has no child metrics it should report the aggregate regardless of the cluster setting.

Fixes: #149481

Release note (bug fix): When child metrics are enabled, include_aggregate is disabled, and the sql.metric.application_name/database_name are disabled, a handful of `sql` metrics were not being reported.

149712: storage: aggressively separate values in range ID keyspace r=sumeerbhola a=jbowens

Adapt Cockroach's implementation of pebble.SpanPolicyFunc to mark the range ID keyspace with ValueStorageLatencyTolerant. This setting will cause Pebble to separate values into blob files as long as they're large enough that a pointer to an external value is likely to be smaller than the value itself. Separating values improves write amplification by avoiding rewriting values during some compactions.

We target the range ID keyspace because it contains the raft log, which is written during all user writes but rarely read.

Tangentially related to #16624.

Epic: none
Release note (performance improvement): Reduces write amplification by storing raft log values in separate blob files. This reduces write bandwidth, especially on stores with many replicas. This in turn can increase throughput and reduce latency. This behavior is active as long as the storage.value_separation.enabled cluster setting is enabled.

Co-authored-by: Alyshan Jahani <[email protected]>
Co-authored-by: Jackson Owens <[email protected]>

Author: 3 people

pkg/server/status/recorder.go

@@ -98,6 +98,16 @@ var includeAggregateMetricsEnabled = settings.RegisterBoolSetting(
 	true,
 	settings.WithPublic)
 
+// bugfix149481Enabled is a (temporary) cluster setting that fixes
+// https://github.com/cockroachdb/cockroach/issues/149481. It is true (enabled) by default on master,
+// and false on backports.
+var bugfix149481Enabled = settings.RegisterBoolSetting(
+	settings.ApplicationLevel, "server.child_metrics.bugfix_missing_sql_metrics_149481.enabled",
+	"fixes bug where certain SQL metrics are not being reported, see: "+
+		"https://github.com/cockroachdb/cockroach/issues/149481",
+	true,
+	settings.WithVisibility(settings.Reserved))
+
 // MetricsRecorder is used to periodically record the information in a number of
 // metric registries.
 //
@@ -348,10 +358,12 @@ func (mr *MetricsRecorder) ScrapeIntoPrometheusWithStaticLabels(
 
 		includeChildMetrics := ChildMetricsEnabled.Get(&mr.settings.SV)
 		includeAggregateMetrics := includeAggregateMetricsEnabled.Get(&mr.settings.SV)
+		reinitialisableBugFixEnabled := bugfix149481Enabled.Get(&mr.settings.SV)
 		scrapeOptions := []metric.ScrapeOption{
 			metric.WithIncludeChildMetrics(includeChildMetrics),
 			metric.WithIncludeAggregateMetrics(includeAggregateMetrics),
 			metric.WithUseStaticLabels(useStaticLabels),
+			metric.WithReinitialisableBugFixEnabled(reinitialisableBugFixEnabled),
 		}
 		if mr.mu.nodeRegistry == nil {
 			// We haven't yet processed initialization information; output nothing.

pkg/storage/pebble.go

@@ -549,28 +549,7 @@ func DefaultPebbleOptions() *pebble.Options {
 	opts.TargetByteDeletionRate = 128 << 20 // 128 MB
 	opts.Experimental.ShortAttributeExtractor = shortAttributeExtractorForValues
 
-	lockTableStartKey := EncodeMVCCKey(MVCCKey{Key: keys.LocalRangeLockTablePrefix})
-	lockTableEndKey := EncodeMVCCKey(MVCCKey{Key: keys.LocalRangeLockTablePrefix.PrefixEnd()})
-	localEndKey := EncodeMVCCKey(MVCCKey{Key: keys.LocalPrefix.PrefixEnd()})
-	opts.Experimental.SpanPolicyFunc = func(startKey []byte) (policy pebble.SpanPolicy, endKey []byte, _ error) {
-		if !bytes.HasPrefix(startKey, keys.LocalPrefix) {
-			return pebble.SpanPolicy{}, nil, nil
-		}
-		// Prefer fast compression for all local keys, since they shouldn't take up
-		// a significant part of the space.
-		policy.PreferFastCompression = true
-
-		// We also disable value separation for lock keys.
-		if cockroachkvs.Compare(startKey, lockTableEndKey) >= 0 {
-			return policy, localEndKey, nil
-		}
-		if cockroachkvs.Compare(startKey, lockTableStartKey) < 0 {
-			return policy, lockTableStartKey, nil
-		}
-		policy.DisableValueSeparationBySuffix = true
-		policy.ValueStoragePolicy = pebble.ValueStorageLowReadLatency
-		return policy, lockTableEndKey, nil
-	}
+	opts.Experimental.SpanPolicyFunc = spanPolicyFunc
 	opts.Experimental.UserKeyCategories = userKeyCategories
 
 	opts.Levels[0] = pebble.LevelOptions{
@@ -607,6 +586,55 @@ func DefaultPebbleOptions() *pebble.Options {
 	return opts
 }
 
+var (
+	spanPolicyLocalRangeIDEndKey = EncodeMVCCKey(MVCCKey{Key: keys.LocalRangeIDPrefix.AsRawKey().PrefixEnd()})
+	spanPolicyLockTableStartKey  = EncodeMVCCKey(MVCCKey{Key: keys.LocalRangeLockTablePrefix})
+	spanPolicyLockTableEndKey    = EncodeMVCCKey(MVCCKey{Key: keys.LocalRangeLockTablePrefix.PrefixEnd()})
+	spanPolicyLocalEndKey        = EncodeMVCCKey(MVCCKey{Key: keys.LocalPrefix.PrefixEnd()})
+)
+
+// spanPolicyFunc is a pebble.SpanPolicyFunc that applies special policies for
+// the CockroachDB keyspace.
+func spanPolicyFunc(startKey []byte) (policy pebble.SpanPolicy, endKey []byte, _ error) {
+	// There's no special policy for non-local keys.
+	if !bytes.HasPrefix(startKey, keys.LocalPrefix) {
+		return pebble.SpanPolicy{}, nil, nil
+	}
+	// Prefer fast compression for all local keys, since they shouldn't take up
+	// a significant part of the space.
+	policy.PreferFastCompression = true
+
+	// The first section of the local keyspace is the Range-ID keyspace. It
+	// extends from the beginning of the keyspace to the Range Local keys. The
+	// Range-ID keyspace includes the raft log, which is rarely read and
+	// receives ~half the writes.
+	if cockroachkvs.Compare(startKey, spanPolicyLocalRangeIDEndKey) < 0 {
+		if !bytes.HasPrefix(startKey, keys.LocalRangeIDPrefix) {
+			return pebble.SpanPolicy{}, nil, errors.AssertionFailedf("startKey %s is not a Range-ID key", startKey)
+		}
+		policy.ValueStoragePolicy = pebble.ValueStorageLatencyTolerant
+		return policy, spanPolicyLocalRangeIDEndKey, nil
+	}
+
+	// We also disable value separation for lock keys.
+	if cockroachkvs.Compare(startKey, spanPolicyLockTableEndKey) >= 0 {
+		// Not a lock key, so use default value separation within sstable (by
+		// suffix) and into blob files.
+		// NB: there won't actually be a suffix in these local keys.
+		return policy, spanPolicyLocalEndKey, nil
+	}
+	if cockroachkvs.Compare(startKey, spanPolicyLockTableStartKey) < 0 {
+		// Not a lock key, so use default value separation within sstable (by
+		// suffix) and into blob files.
+		// NB: there won't actually be a suffix in these local keys.
+		return policy, spanPolicyLockTableStartKey, nil
+	}
+	// Lock key. Disable value separation.
+	policy.DisableValueSeparationBySuffix = true
+	policy.ValueStoragePolicy = pebble.ValueStorageLowReadLatency
+	return policy, spanPolicyLockTableEndKey, nil
+}
+
 func shortAttributeExtractorForValues(
 	key []byte, keyPrefixLen int, value []byte,
 ) (pebble.ShortAttribute, error) {

pkg/storage/pebble_test.go

@@ -1762,3 +1762,66 @@ func TestPebbleCompactCancellation(t *testing.T) {
 	bfs.WaitForBlockAndUnblock()
 	wg.Wait()
 }
+
+func TestPebbleSpanPolicyFunc(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	type testCase struct {
+		startKey   roachpb.Key
+		wantPolicy pebble.SpanPolicy
+		wantEndKey []byte
+	}
+	cases := []testCase{
+		{
+			startKey: keys.RaftHardStateKey(1),
+			wantPolicy: pebble.SpanPolicy{
+				PreferFastCompression: true,
+				ValueStoragePolicy:    pebble.ValueStorageLatencyTolerant,
+			},
+			wantEndKey: spanPolicyLocalRangeIDEndKey,
+		},
+		{
+			startKey: keys.RaftLogKey(9, 2),
+			wantPolicy: pebble.SpanPolicy{
+				PreferFastCompression: true,
+				ValueStoragePolicy:    pebble.ValueStorageLatencyTolerant,
+			},
+			wantEndKey: spanPolicyLocalRangeIDEndKey,
+		},
+		{
+			startKey: keys.RangeDescriptorKey(roachpb.RKey("a")),
+			wantPolicy: pebble.SpanPolicy{
+				PreferFastCompression: true,
+			},
+			wantEndKey: spanPolicyLockTableStartKey,
+		},
+		{
+			startKey: func() roachpb.Key {
+				k, _ := keys.LockTableSingleKey(roachpb.Key("a"), nil)
+				return k
+			}(),
+			wantPolicy: pebble.SpanPolicy{
+				PreferFastCompression:          true,
+				DisableValueSeparationBySuffix: true,
+				ValueStoragePolicy:             pebble.ValueStorageLowReadLatency,
+			},
+			wantEndKey: spanPolicyLockTableEndKey,
+		},
+		{
+			startKey:   keys.SystemSQLCodec.IndexPrefix(1, 2),
+			wantPolicy: pebble.SpanPolicy{},
+			wantEndKey: nil,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(fmt.Sprintf("%x", tc.startKey), func(t *testing.T) {
+			ek := EngineKey{Key: tc.startKey}.Encode()
+			policy, endKey, err := spanPolicyFunc(ek)
+			require.NoError(t, err)
+			require.Equal(t, tc.wantPolicy, policy)
+			require.Equal(t, tc.wantEndKey, endKey)
+		})
+	}
+}

pkg/util/metric/aggmetric/agg_metric_test.go

@@ -33,40 +33,87 @@ func TestExcludeAggregateMetrics(t *testing.T) {
 	counter.AddChild("xyz")
 	r.AddMetric(counter)
 
-	// Don't include child metrics, so only report the aggregate.
-	// Flipping the includeAggregateMetrics flag should have no effect.
-	testutils.RunTrueAndFalse(t, "includeChildMetrics=false,includeAggregateMetrics", func(t *testing.T, includeAggregateMetrics bool) {
+	testutils.RunTrueAndFalse(t, "reinitialisableBugFixEnabled", func(t *testing.T, includeBugFix bool) {
+		// Don't include child metrics, so only report the aggregate.
+		// Flipping the includeAggregateMetrics flag should have no effect.
+		testutils.RunTrueAndFalse(t, "includeChildMetrics=false,includeAggregateMetrics", func(t *testing.T, includeAggregateMetrics bool) {
+			pe := metric.MakePrometheusExporter()
+			pe.ScrapeRegistry(r, metric.WithIncludeChildMetrics(false), metric.WithIncludeAggregateMetrics(includeAggregateMetrics), metric.WithReinitialisableBugFixEnabled(includeBugFix))
+			families, err := pe.Gather()
+			require.NoError(t, err)
+			require.Equal(t, 1, len(families))
+			require.Equal(t, "counter", families[0].GetName())
+			require.Equal(t, 1, len(families[0].GetMetric()))
+			require.Equal(t, 0, len(families[0].GetMetric()[0].GetLabel()))
+		})
+
+		testutils.RunTrueAndFalse(t, "includeChildMetrics=true,includeAggregateMetrics", func(t *testing.T, includeAggregateMetrics bool) {
+			pe := metric.MakePrometheusExporter()
+			pe.ScrapeRegistry(r, metric.WithIncludeChildMetrics(true), metric.WithIncludeAggregateMetrics(includeAggregateMetrics), metric.WithReinitialisableBugFixEnabled(includeBugFix))
+			families, err := pe.Gather()
+			require.NoError(t, err)
+			require.Equal(t, 1, len(families))
+			require.Equal(t, "counter", families[0].GetName())
+			var labelPair *prometheusgo.LabelPair
+			if includeAggregateMetrics {
+				require.Equal(t, 2, len(families[0].GetMetric()))
+				require.Equal(t, 0, len(families[0].GetMetric()[0].GetLabel()))
+				require.Equal(t, 1, len(families[0].GetMetric()[1].GetLabel()))
+				labelPair = families[0].GetMetric()[1].GetLabel()[0]
+			} else {
+				require.Equal(t, 1, len(families[0].GetMetric()))
+				require.Equal(t, 1, len(families[0].GetMetric()[0].GetLabel()))
+				labelPair = families[0].GetMetric()[0].GetLabel()[0]
+			}
+			require.Equal(t, "child_label", *labelPair.Name)
+			require.Equal(t, "xyz", *labelPair.Value)
+		})
+	})
+
+	t.Run("reinitialisable metrics", func(t *testing.T) {
+		r := metric.NewRegistry()
+		counter := NewSQLCounter(metric.Metadata{Name: "counter"})
+		r.AddMetric(counter)
+
+		// Aggregate disabled, but no children, so the aggregate is still reported.
 		pe := metric.MakePrometheusExporter()
-		pe.ScrapeRegistry(r, metric.WithIncludeChildMetrics(false), metric.WithIncludeAggregateMetrics(includeAggregateMetrics))
+		pe.ScrapeRegistry(r, metric.WithIncludeChildMetrics(true), metric.WithIncludeAggregateMetrics(false), metric.WithReinitialisableBugFixEnabled(true))
 		families, err := pe.Gather()
 		require.NoError(t, err)
 		require.Equal(t, 1, len(families))
-		require.Equal(t, "counter", families[0].GetName())
 		require.Equal(t, 1, len(families[0].GetMetric()))
-		require.Equal(t, 0, len(families[0].GetMetric()[0].GetLabel()))
-	})
-
-	testutils.RunTrueAndFalse(t, "includeChildMetrics=true,includeAggregateMetrics", func(t *testing.T, includeAggregateMetrics bool) {
-		pe := metric.MakePrometheusExporter()
-		pe.ScrapeRegistry(r, metric.WithIncludeChildMetrics(true), metric.WithIncludeAggregateMetrics(includeAggregateMetrics))
-		families, err := pe.Gather()
+		require.Equal(t, 0, len(families[0].GetMetric()[0].GetLabel())) // The aggregate has no labels.
+
+		// Add children (app and db label). Now only the childset metric is reported.
+		r.ReinitialiseChildMetrics(true, true)
+		counter.Inc(1, "db_foo", "app_foo")
+		pe = metric.MakePrometheusExporter()
+		pe.ScrapeRegistry(r, metric.WithIncludeChildMetrics(true), metric.WithIncludeAggregateMetrics(false), metric.WithReinitialisableBugFixEnabled(true))
+		families, err = pe.Gather()
 		require.NoError(t, err)
 		require.Equal(t, 1, len(families))
-		require.Equal(t, "counter", families[0].GetName())
-		var labelPair *prometheusgo.LabelPair
-		if includeAggregateMetrics {
-			require.Equal(t, 2, len(families[0].GetMetric()))
-			require.Equal(t, 0, len(families[0].GetMetric()[0].GetLabel()))
-			require.Equal(t, 1, len(families[0].GetMetric()[1].GetLabel()))
-			labelPair = families[0].GetMetric()[1].GetLabel()[0]
-		} else {
-			require.Equal(t, 1, len(families[0].GetMetric()))
-			require.Equal(t, 1, len(families[0].GetMetric()[0].GetLabel()))
-			labelPair = families[0].GetMetric()[0].GetLabel()[0]
-		}
-		require.Equal(t, "child_label", *labelPair.Name)
-		require.Equal(t, "xyz", *labelPair.Value)
+		require.Equal(t, 1, len(families[0].GetMetric()))
+		// Childset metric labels.
+		labelPair := families[0].GetMetric()[0].GetLabel()[0]
+		require.Equal(t, "database", *labelPair.Name)
+		labelPair = families[0].GetMetric()[0].GetLabel()[1]
+		require.Equal(t, "application_name", *labelPair.Name)
+
+		// Enable aggregate. Now reporting two metrics (the aggregate and the childset).
+		pe = metric.MakePrometheusExporter()
+		pe.ScrapeRegistry(r, metric.WithIncludeChildMetrics(true), metric.WithIncludeAggregateMetrics(true), metric.WithReinitialisableBugFixEnabled(true))
+		families, err = pe.Gather()
+		require.NoError(t, err)
+		require.Equal(t, 1, len(families))
+		require.Equal(t, 2, len(families[0].GetMetric()))
+		require.Equal(t, 0, len(families[0].GetMetric()[0].GetLabel())) // The aggregate has no labels.
+		// Childset metric labels.
+		labelPair = families[0].GetMetric()[1].GetLabel()[0]
+		require.Equal(t, "database", *labelPair.Name)
+		labelPair = families[0].GetMetric()[1].GetLabel()[1]
+		require.Equal(t, "application_name", *labelPair.Name)
 	})
+
 }
 
 func TestAggMetric(t *testing.T) {

pkg/util/metric/prometheus_exporter.go

@@ -78,9 +78,10 @@ func (pm *PrometheusExporter) findOrCreateFamily(
 }
 
 type scrapeOptions struct {
-	includeChildMetrics     bool
-	includeAggregateMetrics bool
-	useStaticLabels         bool
+	includeChildMetrics          bool
+	includeAggregateMetrics      bool
+	useStaticLabels              bool
+	reinitialisableBugFixEnabled bool
 }
 
 // ScrapeOption is a function that modifies scrapeOptions
@@ -107,6 +108,13 @@ func WithUseStaticLabels(use bool) ScrapeOption {
 	}
 }
 
+// WithReinitialisableBugFixEnabled returns an option to set whether the reinitialisable bug fix is enabled
+func WithReinitialisableBugFixEnabled(enabled bool) ScrapeOption {
+	return func(o *scrapeOptions) {
+		o.reinitialisableBugFixEnabled = enabled
+	}
+}
+
 // applyScrapeOptions creates a new scrapeOptions with the given options applied
 func applyScrapeOptions(options ...ScrapeOption) *scrapeOptions {
 	opts := &scrapeOptions{
@@ -125,6 +133,7 @@ func applyScrapeOptions(options ...ScrapeOption) *scrapeOptions {
 func (pm *PrometheusExporter) ScrapeRegistry(registry *Registry, options ...ScrapeOption) {
 	o := applyScrapeOptions(options...)
 	labels := registry.GetLabels()
+
 	f := func(name string, v interface{}) {
 		switch prom := v.(type) {
 		case PrometheusVector:
@@ -138,6 +147,34 @@ func (pm *PrometheusExporter) ScrapeRegistry(registry *Registry, options ...Scra
 			}
 
 		case PrometheusExportable:
+			if _, ok := v.(PrometheusReinitialisable); ok && o.reinitialisableBugFixEnabled {
+				m := prom.ToPrometheusMetric()
+				// Set registry and metric labels.
+				m.Label = append(labels, prom.GetLabels(o.useStaticLabels)...)
+				family := pm.findOrCreateFamily(prom, o)
+
+				if o.includeAggregateMetrics {
+					family.Metric = append(family.Metric, m)
+				}
+
+				promIter, ok := v.(PrometheusIterable)
+				numChildren := 0
+				if ok && o.includeChildMetrics {
+					promIter.Each(m.Label, func(metric *prometheusgo.Metric) {
+						family.Metric = append(family.Metric, metric)
+						numChildren += 1
+					})
+				}
+
+				// PrometheusReinitialisable metrics (like SQLMetric) dynamically
+				// add child metrics. If no child metrics are present we want to ensure
+				// we report the aggregate regardless of the respective cluster setting.
+				if numChildren == 0 && !o.includeAggregateMetrics {
+					family.Metric = append(family.Metric, m)
+				}
+				return
+			}
+
 			m := prom.ToPrometheusMetric()
 			// Set registry and metric labels.
 			m.Label = append(labels, prom.GetLabels(o.useStaticLabels)...)