Merge pull request #159190 from jasonlmfong/blathers/backport-release-26.1-158716

release-26.1: pkg/cli: allow tsdump to also dump the labeled child metrics

Author: jasonlmfong

pkg/cli/tsdump_upload.go

@@ -321,6 +321,12 @@ func (d *datadogWriter) dump(kv *roachpb.KeyValue) (*datadogV2.MetricSeries, err
 	if err := kv.Value.GetProto(&idata); err != nil {
 		return nil, err
 	}
+	braceIdx := strings.IndexByte(name, '{')
+	if braceIdx != -1 {
+		// has child labels
+		// TODO(jasonlmfong): support uploading child labels
+		return nil, errors.Errorf("timeseries key with child labels: %s", name)
+	}
 
 	series := &datadogV2.MetricSeries{
 		Metric:   name,

pkg/server/status/recorder.go

@@ -813,25 +813,6 @@ type registryRecorder struct {
 	childMetricNameCache *syncutil.Map[uint64, cacheEntry]
 }
 
-// allowedChangefeedMetrics is the list of changefeed metrics that should have
-// child metrics collected and recorded to TSDB. This is a curated list to prevent
-// unbounded cardinality while still capturing the most important per-changefeed metrics.
-var allowedChangefeedMetrics = map[string]struct{}{
-	"changefeed.max_behind_nanos":                     {},
-	"changefeed.error_retries":                        {},
-	"changefeed.internal_retry_message_count":         {},
-	"changefeed.stage.downstream_client_send.latency": {},
-	"changefeed.emitted_messages":                     {},
-	"changefeed.sink_backpressure_nanos":              {},
-	"changefeed.backfill_pending_ranges":              {},
-	"changefeed.sink_io_inflight":                     {},
-	"changefeed.lagging_ranges":                       {},
-	"changefeed.aggregator_progress":                  {},
-	"changefeed.checkpoint_progress":                  {},
-	"changefeed.emitted_batch_sizes":                  {},
-	"changefeed.total_ranges":                         {},
-}
-
 // extractValue extracts the metric value(s) for the given metric and passes it, along with the metric name, to the
 // provided callback function.
 func extractValue(name string, mtr interface{}, fn func(string, float64)) error {
@@ -1023,7 +1004,7 @@ func (rr registryRecorder) recordChangefeedChildMetrics(dest *[]tspb.TimeSeriesD
 
 	labels := rr.registry.GetLabels()
 	rr.registry.Each(func(name string, v interface{}) {
-		if _, allowed := allowedChangefeedMetrics[name]; !allowed {
+		if _, allowed := tsutil.AllowedChildMetrics[name]; !allowed {
 			return
 		}
 		// Check if the metric has child collection enabled in its metadata

pkg/server/status/recorder_test.go

@@ -1283,8 +1283,8 @@ func BenchmarkRecordChangefeedChildMetrics(b *testing.B) {
 	enableChildCollection := true
 
 	// Get metrics from the allowed list and convert to slice for indexing
-	allowedMetricsList := make([]string, 0, len(allowedChangefeedMetrics))
-	for metricName := range allowedChangefeedMetrics {
+	allowedMetricsList := make([]string, 0, len(tsutil.AllowedChildMetrics))
+	for metricName := range tsutil.AllowedChildMetrics {
 		allowedMetricsList = append(allowedMetricsList, metricName)
 	}
 

pkg/ts/server.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/ts/tspb"
+	"github.com/cockroachdb/cockroach/pkg/ts/tsutil"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
@@ -481,12 +482,32 @@ func dumpImpl(
 				ResolutionFromProto(res),
 				req.StartNanos,
 				req.EndNanos,
+				false,
 				d,
 			); err != nil {
 				return err
 			}
 		}
 	}
+
+	// Dump child metrics only for allowed metrics at 1M resolution
+	for _, seriesName := range req.Names {
+		if !tsutil.IsAllowedChildMetric(seriesName) {
+			continue
+		}
+		if err := dumpTimeseriesAllSources(
+			ctx,
+			db,
+			seriesName,
+			Resolution1m,
+			req.StartNanos,
+			req.EndNanos,
+			true,
+			d,
+		); err != nil {
+			return err
+		}
+	}
 	return nil
 }
 
@@ -536,6 +557,7 @@ func dumpTimeseriesAllSources(
 	seriesName string,
 	diskResolution Resolution,
 	startNanos, endNanos int64,
+	includeChildMetrics bool,
 	dump func(*roachpb.KeyValue) error,
 ) error {
 	if endNanos == 0 {
@@ -548,12 +570,20 @@ func dumpTimeseriesAllSources(
 		endNanos += delta
 	}
 
+	var endKeyName string
+	if includeChildMetrics {
+		// Create a span that covers the metric's children.
+		endKeyName = seriesName + string(rune(0x7C)) // '|' is the next char after '{'
+	} else {
+		endKeyName = seriesName
+	}
+
 	span := &roachpb.Span{
 		Key: MakeDataKey(
 			seriesName, "" /* source */, diskResolution, startNanos,
 		),
 		EndKey: MakeDataKey(
-			seriesName, "" /* source */, diskResolution, endNanos,
+			endKeyName, "" /* source */, diskResolution, endNanos,
 		),
 	}
 

pkg/ts/server_test.go

@@ -11,6 +11,7 @@ import (
 	"io"
 	"reflect"
 	"sort"
+	"strings"
 	"testing"
 	"unsafe"
 
@@ -877,6 +878,137 @@ func TestServerDump(t *testing.T) {
 	}
 }
 
+func TestServerDumpChildMetrics(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+
+	s := serverutils.StartServerOnly(t, base.TestServerArgs{
+		// For now, direct access to the tsdb is reserved to the storage layer.
+		DefaultTestTenant: base.TestIsSpecificToStorageLayerAndNeedsASystemTenant,
+
+		Knobs: base.TestingKnobs{
+			Store: &kvserver.StoreTestingKnobs{
+				DisableTimeSeriesMaintenanceQueue: true,
+			},
+		},
+	})
+	defer s.Stopper().Stop(ctx)
+
+	tsdb := s.TsDB().(*ts.DB)
+
+	// Store parent metric (without labels)
+	parentMetric := "cr.node.changefeed.emitted_messages"
+	if err := tsdb.StoreData(ctx, ts.Resolution10s, []tspb.TimeSeriesData{
+		{
+			Name:   parentMetric,
+			Source: "1",
+			Datapoints: []tspb.TimeSeriesDatapoint{
+				{TimestampNanos: 100 * 1e9, Value: 1000.0},
+				{TimestampNanos: 200 * 1e9, Value: 2000.0},
+			},
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Store child metrics (with labels encoded in name)
+	childMetric1 := fmt.Sprintf(`%s{feed_id="123",scope="default"}`, parentMetric)
+	childMetric2 := fmt.Sprintf(`%s{feed_id="456",scope="system"}`, parentMetric)
+	if err := tsdb.StoreData(ctx, ts.Resolution1m, []tspb.TimeSeriesData{
+		{
+			Name:   childMetric1,
+			Source: "1",
+			Datapoints: []tspb.TimeSeriesDatapoint{
+				{TimestampNanos: 100 * 1e9, Value: 500.0},
+				{TimestampNanos: 200 * 1e9, Value: 1500.0},
+			},
+		},
+		{
+			Name:   childMetric2,
+			Source: "1",
+			Datapoints: []tspb.TimeSeriesDatapoint{
+				{TimestampNanos: 100 * 1e9, Value: 300.0},
+				{TimestampNanos: 200 * 1e9, Value: 800.0},
+			},
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	conn := s.RPCClientConn(t, username.RootUserName())
+	client := conn.NewTimeSeriesClient()
+
+	t.Run("includes child metrics", func(t *testing.T) {
+		dumpClient, err := client.Dump(ctx, &tspb.DumpRequest{
+			Names:      []string{parentMetric},
+			StartNanos: 100 * 1e9,
+			EndNanos:   300 * 1e9,
+		})
+		require.NoError(t, err)
+
+		resultMap := make(map[string][]tspb.TimeSeriesDatapoint)
+		for {
+			msg, err := dumpClient.Recv()
+			if err == io.EOF {
+				break
+			}
+			require.NoError(t, err)
+			resultMap[msg.Name] = append(resultMap[msg.Name], msg.Datapoints...)
+		}
+
+		// Should have parent metric AND both child metrics
+		require.Contains(t, resultMap, parentMetric, "parent metric should be included")
+		require.Contains(t, resultMap, childMetric1, "child metric 1 should be included")
+		require.Contains(t, resultMap, childMetric2, "child metric 2 should be included")
+		require.Equal(t, 3, len(resultMap), "should have parent and both child metrics")
+
+		// Verify data correctness for parent metric
+		require.Len(t, resultMap[parentMetric], 2, "parent metric should have 2 datapoints")
+		require.Equal(t, 1000.0, resultMap[parentMetric][0].Value)
+		require.Equal(t, 2000.0, resultMap[parentMetric][1].Value)
+
+		// Verify data correctness for child metrics
+		require.Len(t, resultMap[childMetric1], 2, "child metric 1 should have 2 datapoints")
+		require.Equal(t, 500.0, resultMap[childMetric1][0].Value)
+		require.Equal(t, 1500.0, resultMap[childMetric1][1].Value)
+
+		require.Len(t, resultMap[childMetric2], 2, "child metric 2 should have 2 datapoints")
+		require.Equal(t, 300.0, resultMap[childMetric2][0].Value)
+		require.Equal(t, 800.0, resultMap[childMetric2][1].Value)
+	})
+
+	t.Run("DumpRaw sees child metrics", func(t *testing.T) {
+		dumpRawClient, err := client.DumpRaw(ctx, &tspb.DumpRequest{
+			Names:      []string{parentMetric},
+			StartNanos: 100 * 1e9,
+			EndNanos:   300 * 1e9,
+		})
+		require.NoError(t, err)
+
+		var kvCount int
+		seenMetrics := make(map[string]bool)
+		for {
+			kv, err := dumpRawClient.Recv()
+			if err == io.EOF {
+				break
+			}
+			require.NoError(t, err)
+			kvCount++
+			// Decode the key to verify it contains child metrics
+			// The key contains the encoded metric name
+			keyStr := string(kv.Key)
+			if strings.Contains(keyStr, "{") {
+				seenMetrics["child"] = true
+			}
+		}
+
+		require.Greater(t, kvCount, 0, "should have raw KVs")
+		require.True(t, seenMetrics["child"], "should have seen child metrics in raw dump")
+	})
+}
+
 func BenchmarkServerQuery(b *testing.B) {
 	defer log.Scope(b).Close(b)
 

pkg/ts/tsutil/util.go

@@ -14,6 +14,38 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ts/tspb"
 )
 
+// AllowedChildMetrics is the list of metrics that should have child metrics
+// collected and recorded to TSDB. This is a curated list to prevent unbounded
+// cardinality while still capturing the most important per-changefeed metrics.
+var AllowedChildMetrics = map[string]struct{}{
+	"changefeed.max_behind_nanos":                     {},
+	"changefeed.error_retries":                        {},
+	"changefeed.internal_retry_message_count":         {},
+	"changefeed.stage.downstream_client_send.latency": {},
+	"changefeed.emitted_messages":                     {},
+	"changefeed.sink_backpressure_nanos":              {},
+	"changefeed.backfill_pending_ranges":              {},
+	"changefeed.sink_io_inflight":                     {},
+	"changefeed.lagging_ranges":                       {},
+	"changefeed.aggregator_progress":                  {},
+	"changefeed.checkpoint_progress":                  {},
+	"changefeed.emitted_batch_sizes":                  {},
+	"changefeed.total_ranges":                         {},
+}
+
+// IsAllowedChildMetric checks if a metric name matches one of the allowed child metrics.
+func IsAllowedChildMetric(name string) bool {
+	metricName := name
+	for _, prefix := range []string{"cr.node.", "cr.store."} {
+		if strings.HasPrefix(metricName, prefix) {
+			metricName = strings.TrimPrefix(metricName, prefix)
+			break
+		}
+	}
+	_, ok := AllowedChildMetrics[metricName]
+	return ok
+}
+
 // DumpRawTo is a helper that gob-encodes all messages received from the
 // source stream to the given WriteCloser.
 func DumpRawTo(src tspb.RPCTimeSeries_DumpRawClient, out io.Writer) error {