From f9fe10cc49cbdb19f8e0391436f64212f037b8c6 Mon Sep 17 00:00:00 2001 From: Kyle Wong <37189875+kyle-a-wong@users.noreply.github.com> Date: Thu, 26 Feb 2026 17:19:33 -0500 Subject: [PATCH] clustermetrics: add support for stopwatch type Adds support for stopwatch type metrics on the cluster metric read path. Stopwatch metrics are gauges that report the duration of time (in unix nanos) that has passed since it was last updated. Epic: https://cockroachlabs.atlassian.net/browse/CRDB-58342 Resolves: https://cockroachlabs.atlassian.net/browse/CRDB-60768 Release note: None --- pkg/obs/clustermetrics/cmreader/BUILD.bazel | 3 +- .../cmreader/registry_syncer.go | 9 + .../registry_syncer_integration_test.go | 357 ++++++++++++++++-- .../cmreader/registry_syncer_test.go | 153 ++++++-- pkg/obs/clustermetrics/cmwatcher/BUILD.bazel | 1 + .../cmwatcher/cluster_metric_row.go | 24 +- .../cmwatcher/cluster_metric_row_test.go | 41 +- pkg/obs/clustermetrics/utils/test_utils.go | 4 +- 8 files changed, 528 insertions(+), 64 deletions(-) diff --git a/pkg/obs/clustermetrics/cmreader/BUILD.bazel b/pkg/obs/clustermetrics/cmreader/BUILD.bazel index 46f7a3086cb9..96d727c0985b 100644 --- a/pkg/obs/clustermetrics/cmreader/BUILD.bazel +++ b/pkg/obs/clustermetrics/cmreader/BUILD.bazel @@ -36,6 +36,7 @@ go_test( embed = [":cmreader"], deps = [ "//pkg/base", + "//pkg/jobs", "//pkg/obs/clustermetrics", "//pkg/obs/clustermetrics/cmmetrics", "//pkg/obs/clustermetrics/cmwatcher", @@ -49,11 +50,11 @@ go_test( "//pkg/sql", "//pkg/testutils", "//pkg/testutils/serverutils", - "//pkg/testutils/skip", "//pkg/testutils/sqlutils", "//pkg/util/leaktest", "//pkg/util/log", "//pkg/util/metric", + "//pkg/util/timeutil", "@com_github_prometheus_common//expfmt", "@com_github_stretchr_testify//require", ], diff --git a/pkg/obs/clustermetrics/cmreader/registry_syncer.go b/pkg/obs/clustermetrics/cmreader/registry_syncer.go index c044e489f77c..0d1ed139a4ca 100644 --- a/pkg/obs/clustermetrics/cmreader/registry_syncer.go +++ b/pkg/obs/clustermetrics/cmreader/registry_syncer.go @@ -131,6 +131,9 @@ func (s *registrySyncer) deregisterMetricLocked( func (s *registrySyncer) start( ctx context.Context, tableResolver catalog.SystemTableIDResolver, ) error { + if s.knobs != nil && s.knobs.OnRegistrySyncerPreStart != nil { + s.knobs.OnRegistrySyncerPreStart() + } err := s.tableWatcher.Start(ctx, tableResolver) if err != nil { log.Dev.Errorf(ctx, "failed to start cluster metrics rangefeed: %s", err) @@ -174,6 +177,11 @@ func (s *registrySyncer) deregisterMetric(ctx context.Context, row cmwatcher.Clu func (s *registrySyncer) reloadAllMetrics( ctx context.Context, metrics map[int64]cmwatcher.ClusterMetricRow, ) { + defer func() { + if s.knobs != nil && s.knobs.OnReloadComplete != nil { + s.knobs.OnReloadComplete() + } + }() s.mu.Lock() defer s.mu.Unlock() for _, m := range s.mu.trackedMetrics { @@ -184,6 +192,7 @@ func (s *registrySyncer) reloadAllMetrics( for _, row := range metrics { s.updateMetricLocked(ctx, row) } + } type Syncer struct { diff --git a/pkg/obs/clustermetrics/cmreader/registry_syncer_integration_test.go b/pkg/obs/clustermetrics/cmreader/registry_syncer_integration_test.go index 3f670b954097..5cbbe5f05378 100644 --- a/pkg/obs/clustermetrics/cmreader/registry_syncer_integration_test.go +++ b/pkg/obs/clustermetrics/cmreader/registry_syncer_integration_test.go @@ -10,8 +10,10 @@ import ( "fmt" "strings" "testing" + "time" "github.com/cockroachdb/cockroach/pkg/base" + "github.com/cockroachdb/cockroach/pkg/jobs" "github.com/cockroachdb/cockroach/pkg/obs/clustermetrics" "github.com/cockroachdb/cockroach/pkg/obs/clustermetrics/cmmetrics" clustermetricutils "github.com/cockroachdb/cockroach/pkg/obs/clustermetrics/utils" @@ -21,7 +23,6 @@ import ( "github.com/cockroachdb/cockroach/pkg/sql" "github.com/cockroachdb/cockroach/pkg/testutils" "github.com/cockroachdb/cockroach/pkg/testutils/serverutils" - "github.com/cockroachdb/cockroach/pkg/testutils/skip" "github.com/cockroachdb/cockroach/pkg/testutils/sqlutils" "github.com/cockroachdb/cockroach/pkg/util/leaktest" "github.com/cockroachdb/cockroach/pkg/util/log" @@ -30,17 +31,16 @@ import ( "github.com/stretchr/testify/require" ) -// TestRegistrySyncerIntegration starts a real test server, wires the registrySyncer into the +// TestRegistrySyncer starts a real test server, wires the registrySyncer into the // server's cluster metric registry via cmreader.Start, inserts rows into // system.cluster_metrics, and verifies that: // - metrics appear in the registry and respond to inserts, upserts, and deletes // - multiple labeled rows for the same GaugeVec are tracked correctly // - metrics are visible through the /_status/vars prometheus endpoint // - scalar metrics appear in TSDB time series data while labeled (vec) metrics do not -func TestRegistrySyncerIntegration(t *testing.T) { +func TestRegistrySyncer(t *testing.T) { defer leaktest.AfterTest(t)() defer log.Scope(t).Close(t) - skip.UnderStress(t, "test is too slow to run under stress") // Register test metric metadata so ToMetric() can resolve them. defer cmmetrics.TestingRegisterLabeledClusterMetric( @@ -58,16 +58,33 @@ func TestRegistrySyncerIntegration(t *testing.T) { Name: "test.scalar", Help: "A scalar gauge for value verification", })() + defer clustermetrics.TestingRegisterClusterMetric("test.stopwatch", metric.Metadata{ + Name: "test.stopwatch", + Help: "A scalar stopwatch", + })() + defer clustermetrics.TestingRegisterLabeledClusterMetric( + "test.stopwatch_labeled", metric.Metadata{ + Name: "test.stopwatch_labeled", + Help: "A labeled stopwatch", + }, + []string{"store"}, + )() ctx := context.Background() - startedChan := make(chan struct{}) - defer close(startedChan) + preStartChan := make(chan struct{}) + defer close(preStartChan) + fullTableLoadComplete := make(chan struct{}) + defer close(fullTableLoadComplete) srv, db, _ := serverutils.StartServer(t, base.TestServerArgs{ DefaultTestTenant: base.TestIsSpecificToStorageLayerAndNeedsASystemTenant, Knobs: base.TestingKnobs{ + JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(), ClusterMetricsKnobs: &clustermetricutils.TestingKnobs{ - OnRegistrySyncerStart: func() { - startedChan <- struct{}{} + OnRegistrySyncerPreStart: func() { + preStartChan <- struct{}{} + }, + OnReloadComplete: func() { + fullTableLoadComplete <- struct{}{} }, }, }, @@ -81,15 +98,28 @@ func TestRegistrySyncerIntegration(t *testing.T) { // them up via OnRefresh. r.Exec(t, `INSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (100, 'test.gauge_labeled', '{"store": "1"}', 'gauge', 42, 1)`) + VALUES (100, 'test.gauge_labeled', '{"store": "1"}', 'GAUGE', 42, 1)`) r.Exec(t, `INSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (200, 'test.counter', '{}', 'counter', 10, 1)`) + VALUES (200, 'test.counter', '{}', 'COUNTER', 10, 1)`) r.Exec(t, `INSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (300, 'test.scalar', '{}', 'gauge', 50, 1)`) + VALUES (300, 'test.scalar', '{}', 'GAUGE', 50, 1)`) + + // Insert stopwatch metrics with a timestamp 10 seconds in the past so + // the computed elapsed time is non-trivial and easy to assert on. + swTimestamp := time.Now().Add(-10 * time.Second).UnixNano() + r.Exec(t, fmt.Sprintf(`INSERT INTO system.cluster_metrics + (id, name, labels, type, value, node_id) + VALUES (500, 'test.stopwatch', '{}', 'STOPWATCH', %d, 1)`, swTimestamp)) + r.Exec(t, fmt.Sprintf(`INSERT INTO system.cluster_metrics + (id, name, labels, type, value, node_id) + VALUES (600, 'test.stopwatch_labeled', '{"store": "1"}', 'STOPWATCH', %d, 1)`, + swTimestamp)) + + <-preStartChan + <-fullTableLoadComplete - <-startedChan execCfg := ts.ExecutorConfig().(sql.ExecutorConfig) // Get a handle to the cluster metric registry for direct inspection. @@ -100,17 +130,92 @@ func TestRegistrySyncerIntegration(t *testing.T) { requireMetricExists(t, reg, "test.gauge_labeled") requireMetricExists(t, reg, "test.counter") requireMetricExists(t, reg, "test.scalar") + requireMetricExists(t, reg, "test.stopwatch") + requireMetricExists(t, reg, "test.stopwatch_labeled") // Verify initial values for scalar metrics. requireCounterValue(t, reg, "test.counter", 10) requireScalarGaugeValue(t, reg, "test.scalar", 50) + // Verify the scalar stopwatch reports positive elapsed time. + // The stored timestamp is ~10s in the past, so Value() should return + // at least 10 seconds worth of nanoseconds. + requireStopwatchElapsed(t, reg, "test.stopwatch", 10*time.Second) + + // Verify the labeled stopwatch reports positive elapsed time in + // prometheus output (the derived fn is applied during scraping). + requireStopwatchVecElapsed( + t, reg, "test.stopwatch_labeled", + map[string]string{"store": "1"}, 10*time.Second, + ) + + // --------------------------------------------------------------- + // Insert a second label set for the labeled stopwatch, then + // verify both label sets show correct elapsed times. + // --------------------------------------------------------------- + swTimestamp2 := time.Now().Add(-2 * time.Second).UnixNano() + r.Exec(t, fmt.Sprintf(`INSERT INTO system.cluster_metrics + (id, name, labels, type, value, node_id) + VALUES (601, 'test.stopwatch_labeled', '{"store": "2"}', 'STOPWATCH', %d, 1)`, + swTimestamp2)) + + testutils.SucceedsSoon(t, func() error { + return checkStopwatchVecElapsed( + reg, "test.stopwatch_labeled", + map[string]string{"store": "2"}, 2*time.Second, + ) + }) + + // The original label set should still be present and show a longer + // elapsed time than the newly inserted one. + requireStopwatchVecElapsed( + t, reg, "test.stopwatch_labeled", + map[string]string{"store": "1"}, 10*time.Second, + ) + + // --------------------------------------------------------------- + // Upsert the labeled stopwatch's store=1 with a fresh timestamp + // (simulating a stopwatch reset). The elapsed time should drop to + // near zero since the new timestamp is "now". + // --------------------------------------------------------------- + swTimestamp3 := time.Now().UnixNano() + r.Exec(t, fmt.Sprintf(`UPSERT INTO system.cluster_metrics + (id, name, labels, type, value, node_id) + VALUES (600, 'test.stopwatch_labeled', '{"store": "1"}', 'STOPWATCH', %d, 1)`, + swTimestamp3)) + + testutils.SucceedsSoon(t, func() error { + return checkStopwatchVecElapsedLessThan( + reg, "test.stopwatch_labeled", + map[string]string{"store": "1"}, 30*time.Second, + ) + }) + + // --------------------------------------------------------------- + // Upsert the scalar stopwatch with a fresh timestamp (simulating + // a reset). The elapsed time should drop from ~10s to near zero. + // SucceedsSoon polls rapidly until the rangefeed delivers the + // update; the 5s threshold is generous since the new timestamp + // is "now". + // --------------------------------------------------------------- + swTimestamp4 := time.Now().UnixNano() + r.Exec(t, fmt.Sprintf(`UPSERT INTO system.cluster_metrics + (id, name, labels, type, value, node_id) + VALUES (500, 'test.stopwatch', '{}', 'STOPWATCH', %d, 1)`, + swTimestamp4)) + + testutils.SucceedsSoon(t, func() error { + return checkScalarStopwatchElapsedLessThan( + reg, "test.stopwatch", 5*time.Second, + ) + }) + // --------------------------------------------------------------- // Upsert the scalar gauge and verify the updated value. // --------------------------------------------------------------- r.Exec(t, `UPSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (300, 'test.scalar', '{}', 'gauge', 123, 1)`) + VALUES (300, 'test.scalar', '{}', 'GAUGE', 123, 1)`) testutils.SucceedsSoon(t, func() error { return checkScalarGaugeValue(reg, "test.scalar", 123) @@ -122,7 +227,7 @@ func TestRegistrySyncerIntegration(t *testing.T) { // --------------------------------------------------------------- r.Exec(t, `INSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (101, 'test.gauge_labeled', '{"store": "2"}', 'gauge', 77, 1)`) + VALUES (101, 'test.gauge_labeled', '{"store": "2"}', 'GAUGE', 77, 1)`) testutils.SucceedsSoon(t, func() error { return checkGaugeVecValue(reg, "test.gauge_labeled", map[string]string{"store": "2"}, 77) @@ -134,7 +239,7 @@ func TestRegistrySyncerIntegration(t *testing.T) { // Upsert the first label set with a new value. r.Exec(t, `UPSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (100, 'test.gauge_labeled', '{"store": "1"}', 'gauge', 99, 1)`) + VALUES (100, 'test.gauge_labeled', '{"store": "1"}', 'GAUGE', 99, 1)`) testutils.SucceedsSoon(t, func() error { return checkGaugeVecValue(reg, "test.gauge_labeled", map[string]string{"store": "1"}, 99) @@ -150,7 +255,7 @@ func TestRegistrySyncerIntegration(t *testing.T) { r.Exec(t, `INSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (400, 'test.newgauge', '{}', 'gauge', 55, 1)`) + VALUES (400, 'test.newgauge', '{}', 'GAUGE', 55, 1)`) testutils.SucceedsSoon(t, func() error { return checkScalarGaugeValue(reg, "test.newgauge", 55) @@ -168,6 +273,8 @@ func TestRegistrySyncerIntegration(t *testing.T) { require.Contains(t, promOutput, "test_counter", "test.counter should appear in /_status/vars") require.Contains(t, promOutput, "test_scalar", "test.scalar should appear in /_status/vars") require.Contains(t, promOutput, "test_newgauge", "test.newgauge should appear in /_status/vars") + require.Contains(t, promOutput, "test_stopwatch", "test.stopwatch should appear in /_status/vars") + require.Contains(t, promOutput, "test_stopwatch_labeled", "test.stopwatch_labeled should appear in /_status/vars") // Verify labeled metrics include both label sets. require.Contains(t, promOutput, `store="1"`, "store=1 label should appear in /_status/vars") require.Contains(t, promOutput, `store="2"`, "store=2 label should appear in /_status/vars") @@ -212,9 +319,14 @@ func TestRegistrySyncerIntegration(t *testing.T) { require.Equal(t, float64(10), tsNames["cr.cluster.test.counter"]) require.Equal(t, float64(55), tsNames["cr.cluster.test.newgauge"]) + // Scalar stopwatch should appear in TSDB with a positive elapsed value. + require.Contains(t, tsNames, "cr.cluster.test.stopwatch") + require.Greater(t, tsNames["cr.cluster.test.stopwatch"], float64(0)) + // Labeled (vec) metrics should NOT be in TSDB data. The recorder's // extractValue function returns a no-op for PrometheusVector types. require.NotContains(t, tsNames, "cr.cluster.test.gauge_labeled") + require.NotContains(t, tsNames, "cr.cluster.test.stopwatch_labeled") // --------------------------------------------------------------- // DELETE a labeled metric row and verify it's removed from @@ -449,6 +561,176 @@ func checkGaugeVecLabelAbsent( return nil } +// requireStopwatchElapsed asserts that a scalar *metric.Gauge (backed by a +// functional gauge) reports an elapsed time of at least minElapsed. +func requireStopwatchElapsed( + t *testing.T, reg metric.RegistryReader, name string, minElapsed time.Duration, +) { + t.Helper() + var g *metric.Gauge + reg.Each(func(n string, v interface{}) { + if n == name { + if gauge, ok := v.(*metric.Gauge); ok { + g = gauge + } + } + }) + require.NotNilf(t, g, "stopwatch gauge %q not found in registry", name) + elapsedNanos := g.Value() + require.Greater(t, elapsedNanos, int64(0), + "stopwatch %q should report positive elapsed time", name) + require.GreaterOrEqual(t, elapsedNanos, int64(minElapsed), + "stopwatch %q elapsed %s should be >= %s", + name, time.Duration(elapsedNanos), minElapsed) +} + +// checkScalarStopwatchElapsedLessThan returns nil if the scalar stopwatch +// reports a positive elapsed time strictly less than maxElapsed. +func checkScalarStopwatchElapsedLessThan( + reg metric.RegistryReader, name string, maxElapsed time.Duration, +) error { + var g *metric.Gauge + reg.Each(func(n string, v interface{}) { + if n == name { + if gauge, ok := v.(*metric.Gauge); ok { + g = gauge + } + } + }) + if g == nil { + return fmt.Errorf("stopwatch gauge %q not found in registry", name) + } + elapsed := g.Value() + if elapsed <= 0 { + return fmt.Errorf( + "stopwatch %q: expected positive elapsed, got %d", name, elapsed) + } + if elapsed >= int64(maxElapsed) { + return fmt.Errorf( + "stopwatch %q: elapsed %s >= max %s", + name, time.Duration(elapsed), maxElapsed) + } + return nil +} + +// requireStopwatchVecElapsed asserts that a labeled stopwatch (backed by a +// DerivedGaugeVec with a derived fn) reports an elapsed time of at least +// minElapsed for the given label set. The derived fn is applied during +// prometheus scraping. +func requireStopwatchVecElapsed( + t *testing.T, + reg metric.RegistryReader, + name string, + labels map[string]string, + minElapsed time.Duration, +) { + t.Helper() + err := checkStopwatchVecElapsed(reg, name, labels, minElapsed) + require.NoError(t, err) +} + +func checkStopwatchVecElapsed( + reg metric.RegistryReader, name string, labels map[string]string, minElapsed time.Duration, +) error { + actual, err := scrapeStopwatchVecValue(reg, name, labels) + if err != nil { + return err + } + if actual <= 0 { + return fmt.Errorf( + "stopwatch vec %q labels=%v: expected positive elapsed, got %d", + name, labels, actual) + } + if actual < int64(minElapsed) { + return fmt.Errorf( + "stopwatch vec %q labels=%v: elapsed %s < min %s", + name, labels, time.Duration(actual), minElapsed) + } + return nil +} + +// checkStopwatchVecElapsedLessThan returns nil if the labeled stopwatch reports +// a positive elapsed time that is strictly less than maxElapsed. +func checkStopwatchVecElapsedLessThan( + reg metric.RegistryReader, name string, labels map[string]string, maxElapsed time.Duration, +) error { + actual, err := scrapeStopwatchVecValue(reg, name, labels) + if err != nil { + return err + } + if actual <= 0 { + return fmt.Errorf( + "stopwatch vec %q labels=%v: expected positive elapsed, got %d", + name, labels, actual) + } + if actual >= int64(maxElapsed) { + return fmt.Errorf( + "stopwatch vec %q labels=%v: elapsed %s >= max %s", + name, labels, time.Duration(actual), maxElapsed) + } + return nil +} + +// scrapeStopwatchVecValue scrapes the registry for a labeled gauge and returns +// the raw gauge value for the matching label set. +func scrapeStopwatchVecValue( + reg metric.RegistryReader, name string, labels map[string]string, +) (int64, error) { + var gv *metric.GaugeVec + reg.Each(func(n string, v interface{}) { + if n == name { + if vec, ok := v.(*metric.GaugeVec); ok { + gv = vec + } + } + }) + if gv == nil { + return 0, fmt.Errorf("stopwatch vec %q not found in registry", name) + } + + pe := metric.MakePrometheusExporter() + var buf strings.Builder + err := pe.ScrapeAndPrintAsText( + &buf, expfmt.FmtText, func(exporter *metric.PrometheusExporter) { + exporter.ScrapeRegistry(reg) + }) + if err != nil { + return 0, fmt.Errorf("failed to scrape registry: %w", err) + } + + var parser expfmt.TextParser + families, err := parser.TextToMetricFamilies(strings.NewReader(buf.String())) + if err != nil { + return 0, fmt.Errorf("failed to parse prometheus output: %w", err) + } + + exportedName := strings.ReplaceAll(name, ".", "_") + family, ok := families[exportedName] + if !ok { + return 0, fmt.Errorf( + "metric family %q not found in prometheus output", exportedName) + } + + for _, m := range family.GetMetric() { + metricLabels := make(map[string]string, len(m.GetLabel())) + for _, lp := range m.GetLabel() { + metricLabels[lp.GetName()] = lp.GetValue() + } + match := true + for k, v := range labels { + if metricLabels[k] != v { + match = false + break + } + } + if match { + return int64(m.GetGauge().GetValue()), nil + } + } + return 0, fmt.Errorf( + "stopwatch vec %q: no metric found with labels %v", name, labels) +} + // TestRegistrySyncerMultiTenant starts a system tenant and a shared-process // secondary tenant, inserts metrics with the same name into each tenant's // system.cluster_metrics table with different values, and verifies that each @@ -457,7 +739,6 @@ func checkGaugeVecLabelAbsent( func TestRegistrySyncerMultiTenant(t *testing.T) { defer leaktest.AfterTest(t)() defer log.Scope(t).Close(t) - skip.UnderStress(t, "test is too slow to run under stress") defer clustermetrics.TestingRegisterClusterMetric("test.mt_gauge", metric.Metadata{ Name: "test.mt_gauge", @@ -466,29 +747,41 @@ func TestRegistrySyncerMultiTenant(t *testing.T) { ctx := context.Background() - // Use buffered channels so the registrySyncer goroutines do not block if the - // test hasn't started receiving yet. - sysStartedChan := make(chan struct{}, 1) + preStartChan := make(chan struct{}) + defer close(preStartChan) + fullTableLoadComplete := make(chan struct{}) + defer close(fullTableLoadComplete) srv, sysDB, _ := serverutils.StartServer(t, base.TestServerArgs{ DefaultTestTenant: base.TestControlsTenantsExplicitly, Knobs: base.TestingKnobs{ + JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(), ClusterMetricsKnobs: &clustermetricutils.TestingKnobs{ - OnRegistrySyncerStart: func() { - sysStartedChan <- struct{}{} + OnRegistrySyncerPreStart: func() { + preStartChan <- struct{}{} + }, + OnReloadComplete: func() { + fullTableLoadComplete <- struct{}{} }, }, }, }) defer srv.Stopper().Stop(ctx) - tenantStartedChan := make(chan struct{}, 1) + tenantPreStartChan := make(chan struct{}) + defer close(tenantPreStartChan) + tenantFullTableLoadComplete := make(chan struct{}) + defer close(tenantFullTableLoadComplete) tenant, tenantDB := serverutils.StartSharedProcessTenant(t, srv, base.TestSharedProcessTenantArgs{ TenantName: "app", Knobs: base.TestingKnobs{ + JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(), ClusterMetricsKnobs: &clustermetricutils.TestingKnobs{ - OnRegistrySyncerStart: func() { - tenantStartedChan <- struct{}{} + OnRegistrySyncerPreStart: func() { + tenantPreStartChan <- struct{}{} + }, + OnReloadComplete: func() { + tenantFullTableLoadComplete <- struct{}{} }, }, }, @@ -501,14 +794,16 @@ func TestRegistrySyncerMultiTenant(t *testing.T) { // system.cluster_metrics table, but with different values. sysRunner.Exec(t, `INSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (100, 'test.mt_gauge', '{}', 'gauge', 42, 1)`) + VALUES (100, 'test.mt_gauge', '{}', 'GAUGE', 42, 1)`) tenantRunner.Exec(t, `INSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (100, 'test.mt_gauge', '{}', 'gauge', 99, 1)`) + VALUES (100, 'test.mt_gauge', '{}', 'GAUGE', 99, 1)`) // Wait for both registry syncers to complete their initial scan. - <-sysStartedChan - <-tenantStartedChan + <-preStartChan + <-fullTableLoadComplete + <-tenantPreStartChan + <-tenantFullTableLoadComplete tenantExecCfg := tenant.ExecutorConfig().(sql.ExecutorConfig) tenantID := tenantExecCfg.Codec.TenantID @@ -529,7 +824,7 @@ func TestRegistrySyncerMultiTenant(t *testing.T) { // --------------------------------------------------------------- sysRunner.Exec(t, `UPSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (100, 'test.mt_gauge', '{}', 'gauge', 100, 1)`) + VALUES (100, 'test.mt_gauge', '{}', 'GAUGE', 100, 1)`) testutils.SucceedsSoon(t, func() error { return checkScalarGaugeValue(sysReg, "test.mt_gauge", 100) @@ -542,7 +837,7 @@ func TestRegistrySyncerMultiTenant(t *testing.T) { // --------------------------------------------------------------- tenantRunner.Exec(t, `UPSERT INTO system.cluster_metrics (id, name, labels, type, value, node_id) - VALUES (100, 'test.mt_gauge', '{}', 'gauge', 200, 1)`) + VALUES (100, 'test.mt_gauge', '{}', 'GAUGE', 200, 1)`) testutils.SucceedsSoon(t, func() error { return checkScalarGaugeValue(tenantReg, "test.mt_gauge", 200) diff --git a/pkg/obs/clustermetrics/cmreader/registry_syncer_test.go b/pkg/obs/clustermetrics/cmreader/registry_syncer_test.go index b1b53b8de116..0de96262e249 100644 --- a/pkg/obs/clustermetrics/cmreader/registry_syncer_test.go +++ b/pkg/obs/clustermetrics/cmreader/registry_syncer_test.go @@ -8,12 +8,14 @@ package cmreader import ( "context" "testing" + "time" "github.com/cockroachdb/cockroach/pkg/obs/clustermetrics/cmmetrics" "github.com/cockroachdb/cockroach/pkg/obs/clustermetrics/cmwatcher" "github.com/cockroachdb/cockroach/pkg/util/leaktest" "github.com/cockroachdb/cockroach/pkg/util/log" "github.com/cockroachdb/cockroach/pkg/util/metric" + "github.com/cockroachdb/cockroach/pkg/util/timeutil" "github.com/stretchr/testify/require" ) @@ -46,7 +48,7 @@ func TestUpdateMetricLocked(t *testing.T) { }) }, rows: []cmwatcher.ClusterMetricRow{{ - ID: 1, Name: "test.gauge", Type: "gauge", Value: 42, + ID: 1, Name: "test.gauge", Type: "GAUGE", Value: 42, }}, verify: func(t *testing.T, u *registrySyncer, reg *registry) { require.Contains(t, u.mu.trackedMetrics, "test.gauge") @@ -73,9 +75,9 @@ func TestUpdateMetricLocked(t *testing.T) { }) }, rows: []cmwatcher.ClusterMetricRow{{ - ID: 1, Name: "test.gauge", Type: "gauge", Value: 10, + ID: 1, Name: "test.gauge", Type: "GAUGE", Value: 10, }, { - ID: 1, Name: "test.gauge", Type: "gauge", Value: 99, + ID: 1, Name: "test.gauge", Type: "GAUGE", Value: 99, }}, verify: func(t *testing.T, u *registrySyncer, _ *registry) { g := u.mu.trackedMetrics["test.gauge"].(*metric.Gauge) @@ -91,9 +93,9 @@ func TestUpdateMetricLocked(t *testing.T) { }) }, rows: []cmwatcher.ClusterMetricRow{{ - ID: 1, Name: "test.counter", Type: "counter", Value: 0, + ID: 1, Name: "test.counter", Type: "COUNTER", Value: 0, }, { - ID: 1, Name: "test.counter", Type: "counter", Value: 42, + ID: 1, Name: "test.counter", Type: "COUNTER", Value: 42, }}, verify: func(t *testing.T, u *registrySyncer, _ *registry) { c := u.mu.trackedMetrics["test.counter"].(*metric.Counter) @@ -103,7 +105,7 @@ func TestUpdateMetricLocked(t *testing.T) { name: "not found", setup: func() func() { return func() {} }, rows: []cmwatcher.ClusterMetricRow{{ - ID: 999, Name: "nonexistent.metric", Type: "gauge", Value: 42, + ID: 999, Name: "nonexistent.metric", Type: "GAUGE", Value: 42, }}, verify: func(t *testing.T, u *registrySyncer, _ *registry) { require.Empty(t, u.mu.trackedMetrics) @@ -121,7 +123,7 @@ func TestUpdateMetricLocked(t *testing.T) { rows: []cmwatcher.ClusterMetricRow{{ ID: 1, Name: "test.gaugevec", Labels: map[string]string{"store": "1"}, - Type: "gauge", Value: 42, + Type: "GAUGE", Value: 42, }}, verify: func(t *testing.T, u *registrySyncer, reg *registry) { require.Contains(t, u.mu.trackedMetrics, "test.gaugevec") @@ -149,11 +151,11 @@ func TestUpdateMetricLocked(t *testing.T) { rows: []cmwatcher.ClusterMetricRow{{ ID: 1, Name: "test.gaugevec", Labels: map[string]string{"store": "1"}, - Type: "gauge", Value: 10, + Type: "GAUGE", Value: 10, }, { ID: 2, Name: "test.gaugevec", Labels: map[string]string{"store": "2"}, - Type: "gauge", Value: 20, + Type: "GAUGE", Value: 20, }}, verify: func(t *testing.T, u *registrySyncer, _ *registry) { require.Contains(t, u.mu.trackedRows, int64(1)) @@ -173,11 +175,11 @@ func TestUpdateMetricLocked(t *testing.T) { rows: []cmwatcher.ClusterMetricRow{{ ID: 1, Name: "test.gaugevec", Labels: map[string]string{"store": "1"}, - Type: "gauge", Value: 10, + Type: "GAUGE", Value: 10, }, { ID: 1, Name: "test.gaugevec", Labels: map[string]string{"store": "1"}, - Type: "gauge", Value: 99, + Type: "GAUGE", Value: 99, }}, verify: func(t *testing.T, u *registrySyncer, _ *registry) { require.Contains(t, u.mu.trackedRows, int64(1)) @@ -196,15 +198,15 @@ func TestUpdateMetricLocked(t *testing.T) { rows: []cmwatcher.ClusterMetricRow{{ ID: 1, Name: "test.countervec", Labels: map[string]string{"store": "1"}, - Type: "counter", Value: 10, + Type: "COUNTER", Value: 10, }, { ID: 2, Name: "test.countervec", Labels: map[string]string{"store": "2"}, - Type: "counter", Value: 20, + Type: "COUNTER", Value: 20, }, { ID: 1, Name: "test.countervec", Labels: map[string]string{"store": "1"}, - Type: "counter", Value: 50, + Type: "COUNTER", Value: 50, }}, verify: func(t *testing.T, u *registrySyncer, _ *registry) { require.Contains(t, u.mu.trackedMetrics, "test.countervec") @@ -216,6 +218,115 @@ func TestUpdateMetricLocked(t *testing.T) { require.Equal(t, int64(20), cv.Count(map[string]string{"store": "2"})) }, + }, { + name: "new stopwatch", + setup: func() func() { + return cmmetrics.TestingRegisterClusterMetric( + "test.stopwatch", metric.Metadata{ + Name: "test.stopwatch", + Help: "A test stopwatch", + }) + }, + rows: []cmwatcher.ClusterMetricRow{{ + ID: 1, Name: "test.stopwatch", Type: "STOPWATCH", + Value: timeutil.Now().Add(-10 * time.Second).UnixNano(), + }}, + verify: func(t *testing.T, u *registrySyncer, reg *registry) { + require.Contains(t, u.mu.trackedMetrics, "test.stopwatch") + require.Contains(t, u.mu.trackedRows, int64(1)) + + var found bool + reg.Each(func(name string, _ interface{}) { + if name == "test.stopwatch" { + found = true + } + }) + require.True(t, found, "metric should be in registry") + + // A DerivedGauge is a type alias for *metric.Gauge. Its Value() + // method applies the timeElapsedSince callback, so the result + // should be at least 10s of nanoseconds. + g := u.mu.trackedMetrics["test.stopwatch"].(*metric.Gauge) + require.GreaterOrEqual(t, g.Value(), int64(10*time.Second)) + }, + }, { + name: "existing stopwatch update", + setup: func() func() { + return cmmetrics.TestingRegisterClusterMetric( + "test.stopwatch", metric.Metadata{ + Name: "test.stopwatch", + Help: "A test stopwatch", + }) + }, + rows: []cmwatcher.ClusterMetricRow{{ + ID: 1, Name: "test.stopwatch", Type: "STOPWATCH", + Value: timeutil.Now().Add(-10 * time.Second).UnixNano(), + }, { + // Simulate a stopwatch reset: update with a recent timestamp. + ID: 1, Name: "test.stopwatch", Type: "STOPWATCH", + Value: timeutil.Now().UnixNano(), + }}, + verify: func(t *testing.T, u *registrySyncer, _ *registry) { + g := u.mu.trackedMetrics["test.stopwatch"].(*metric.Gauge) + // After reset the elapsed time should be much less than 10s. + require.Less(t, g.Value(), int64(5*time.Second)) + require.Greater(t, g.Value(), int64(0)) + }, + }, { + name: "new stopwatch vec", + setup: func() func() { + return cmmetrics.TestingRegisterLabeledClusterMetric( + "test.stopwatchvec", metric.Metadata{ + Name: "test.stopwatchvec", + Help: "A test stopwatch vec", + }, []string{"store"}) + }, + rows: []cmwatcher.ClusterMetricRow{{ + ID: 1, Name: "test.stopwatchvec", + Labels: map[string]string{"store": "1"}, + Type: "STOPWATCH", + Value: timeutil.Now().Add(-10 * time.Second).UnixNano(), + }}, + verify: func(t *testing.T, u *registrySyncer, reg *registry) { + require.Contains(t, u.mu.trackedMetrics, "test.stopwatchvec") + _, ok := u.mu.trackedMetrics["test.stopwatchvec"].(*metric.GaugeVec) + require.True(t, ok, "expected *metric.GaugeVec") + require.Contains(t, u.mu.trackedRows, int64(1)) + + var found bool + reg.Each(func(name string, _ interface{}) { + if name == "test.stopwatchvec" { + found = true + } + }) + require.True(t, found, "metric should be in registry") + }, + }, { + name: "stopwatch vec second label set", + setup: func() func() { + return cmmetrics.TestingRegisterLabeledClusterMetric( + "test.stopwatchvec", metric.Metadata{ + Name: "test.stopwatchvec", + Help: "A test stopwatch vec", + }, []string{"store"}) + }, + rows: []cmwatcher.ClusterMetricRow{{ + ID: 1, Name: "test.stopwatchvec", + Labels: map[string]string{"store": "1"}, + Type: "STOPWATCH", + Value: timeutil.Now().Add(-10 * time.Second).UnixNano(), + }, { + ID: 2, Name: "test.stopwatchvec", + Labels: map[string]string{"store": "2"}, + Type: "STOPWATCH", + Value: timeutil.Now().UnixNano(), + }}, + verify: func(t *testing.T, u *registrySyncer, _ *registry) { + require.Contains(t, u.mu.trackedRows, int64(1)) + require.Contains(t, u.mu.trackedRows, int64(2)) + require.Len(t, u.mu.trackedMetrics, 1) + require.Contains(t, u.mu.trackedMetrics, "test.stopwatchvec") + }, }} for _, tt := range tests { @@ -256,7 +367,7 @@ func TestDeleteMetricLocked(t *testing.T) { }) }, insertRows: []cmwatcher.ClusterMetricRow{{ - ID: 1, Name: "test.gauge", Type: "gauge", Value: 42, + ID: 1, Name: "test.gauge", Type: "GAUGE", Value: 42, }}, deleteRow: cmwatcher.ClusterMetricRow{ID: 1}, verify: func(t *testing.T, u *registrySyncer, reg *registry) { @@ -279,7 +390,7 @@ func TestDeleteMetricLocked(t *testing.T) { insertRows: []cmwatcher.ClusterMetricRow{{ ID: 1, Name: "test.vec", Labels: map[string]string{"store": "1"}, - Type: "gauge", Value: 42, + Type: "GAUGE", Value: 42, }}, deleteRow: cmwatcher.ClusterMetricRow{ID: 1}, verify: func(t *testing.T, u *registrySyncer, reg *registry) { @@ -344,10 +455,10 @@ func TestStop(t *testing.T) { u.mu.Lock() u.updateMetricLocked(ctx, cmwatcher.ClusterMetricRow{ - ID: 1, Name: "gauge_one", Type: "gauge", Value: 1, + ID: 1, Name: "gauge_one", Type: "GAUGE", Value: 1, }) u.updateMetricLocked(ctx, cmwatcher.ClusterMetricRow{ - ID: 2, Name: "gauge_two", Type: "gauge", Value: 2, + ID: 2, Name: "gauge_two", Type: "GAUGE", Value: 2, }) u.mu.Unlock() @@ -385,7 +496,7 @@ func TestOnRefresh(t *testing.T) { // Pre-populate with an old metric. u.mu.Lock() u.updateMetricLocked(ctx, cmwatcher.ClusterMetricRow{ - ID: 1, Name: "gauge.old", Type: "gauge", Value: 1, + ID: 1, Name: "gauge.old", Type: "GAUGE", Value: 1, }) u.mu.Unlock() @@ -393,8 +504,8 @@ func TestOnRefresh(t *testing.T) { // Simulate a full refresh with a different set of rows. refreshRows := map[int64]cmwatcher.ClusterMetricRow{ - 10: {ID: 10, Name: "gauge.new", Type: "gauge", Value: 100}, - 20: {ID: 20, Name: "counter.new", Type: "counter", Value: 200}, + 10: {ID: 10, Name: "gauge.new", Type: "GAUGE", Value: 100}, + 20: {ID: 20, Name: "counter.new", Type: "COUNTER", Value: 200}, } u.reloadAllMetrics(ctx, refreshRows) diff --git a/pkg/obs/clustermetrics/cmwatcher/BUILD.bazel b/pkg/obs/clustermetrics/cmwatcher/BUILD.bazel index f7d1d702fd98..68bf6e6f526b 100644 --- a/pkg/obs/clustermetrics/cmwatcher/BUILD.bazel +++ b/pkg/obs/clustermetrics/cmwatcher/BUILD.bazel @@ -27,6 +27,7 @@ go_library( "//pkg/util/metric", "//pkg/util/startup", "//pkg/util/stop", + "//pkg/util/timeutil", "@com_github_cockroachdb_errors//:errors", ], ) diff --git a/pkg/obs/clustermetrics/cmwatcher/cluster_metric_row.go b/pkg/obs/clustermetrics/cmwatcher/cluster_metric_row.go index 3b7a912e6c31..c33b16bc7cd0 100644 --- a/pkg/obs/clustermetrics/cmwatcher/cluster_metric_row.go +++ b/pkg/obs/clustermetrics/cmwatcher/cluster_metric_row.go @@ -18,6 +18,7 @@ import ( "github.com/cockroachdb/cockroach/pkg/sql/rowenc/valueside" "github.com/cockroachdb/cockroach/pkg/sql/sem/tree" "github.com/cockroachdb/cockroach/pkg/util/metric" + "github.com/cockroachdb/cockroach/pkg/util/timeutil" "github.com/cockroachdb/errors" ) @@ -54,14 +55,18 @@ func (cmr ClusterMetricRow) toLabeledMetric( metadata metric.Metadata, labels []string, ) (metric.Iterable, error) { switch cmr.Type { - case "gauge": + case "GAUGE": vec := metric.NewExportedGaugeVec(metadata, labels) vec.Update(cmr.Labels, cmr.Value) return vec, nil - case "counter": + case "COUNTER": vec := metric.NewExportedCounterVec(metadata, labels) vec.Update(cmr.Labels, cmr.Value) return vec, nil + case "STOPWATCH": + vec := metric.NewDerivedExportedGaugeVec(metadata, labels, timeElapsedSince) + vec.Update(cmr.Labels, cmr.Value) + return vec, nil default: return nil, errors.Newf("unknown metric type %s for exported metric %s", cmr.Type, cmr.Name) } @@ -69,14 +74,18 @@ func (cmr ClusterMetricRow) toLabeledMetric( func (cmr ClusterMetricRow) toMetric(metadata metric.Metadata) (metric.Iterable, error) { switch cmr.Type { - case "gauge": + case "GAUGE": gauge := metric.NewGauge(metadata) gauge.Update(cmr.Value) return gauge, nil - case "counter": + case "COUNTER": counter := metric.NewCounter(metadata) counter.Update(cmr.Value) return counter, nil + case "STOPWATCH": + sw := metric.NewDerivedGauge(metadata, timeElapsedSince) + sw.Update(cmr.Value) + return sw, nil default: return nil, errors.Newf("unknown metric type %s for metric %s", cmr.Type, cmr.Name) } @@ -157,3 +166,10 @@ func (d *RowDecoder) DecodeRow(kv roachpb.KeyValue) (_ ClusterMetricRow, tombsto return row, false, nil } + +func timeElapsedSince(val int64) int64 { + if val == 0 { + return 0 + } + return timeutil.Now().UnixNano() - val +} diff --git a/pkg/obs/clustermetrics/cmwatcher/cluster_metric_row_test.go b/pkg/obs/clustermetrics/cmwatcher/cluster_metric_row_test.go index 68dc49398531..87f199a74757 100644 --- a/pkg/obs/clustermetrics/cmwatcher/cluster_metric_row_test.go +++ b/pkg/obs/clustermetrics/cmwatcher/cluster_metric_row_test.go @@ -99,13 +99,21 @@ func TestToMetric(t *testing.T) { gaugeMeta := metric.Metadata{Name: "test.gauge", Help: "A test gauge"} counterMeta := metric.Metadata{Name: "test.counter", Help: "A test counter"} + stopwatchMeta := metric.Metadata{Name: "test.stopwatch", Help: "A test stopwatch"} labeledGaugeMeta := metric.Metadata{Name: "test.labeled.gauge", Help: "A labeled gauge"} labeledCounterMeta := metric.Metadata{Name: "test.labeled.counter", Help: "A labeled counter"} + labeledStopwatchMeta := metric.Metadata{ + Name: "test.labeled.stopwatch", Help: "A labeled stopwatch", + } cleanupGauge := clustermetrics.TestingRegisterClusterMetric("test.gauge", gaugeMeta) defer cleanupGauge() cleanupCounter := clustermetrics.TestingRegisterClusterMetric("test.counter", counterMeta) defer cleanupCounter() + cleanupStopwatch := clustermetrics.TestingRegisterClusterMetric( + "test.stopwatch", stopwatchMeta, + ) + defer cleanupStopwatch() cleanupLabeledGauge := clustermetrics.TestingRegisterLabeledClusterMetric( "test.labeled.gauge", labeledGaugeMeta, []string{"store"}, ) @@ -114,6 +122,10 @@ func TestToMetric(t *testing.T) { "test.labeled.counter", labeledCounterMeta, []string{"node"}, ) defer cleanupLabeledCounter() + cleanupLabeledStopwatch := clustermetrics.TestingRegisterLabeledClusterMetric( + "test.labeled.stopwatch", labeledStopwatchMeta, []string{"store"}, + ) + defer cleanupLabeledStopwatch() tests := []struct { name string @@ -124,7 +136,7 @@ func TestToMetric(t *testing.T) { name: "gauge without labels", row: cmwatcher.ClusterMetricRow{ Name: "test.gauge", - Type: "gauge", + Type: "GAUGE", Value: 42, }, wantType: &metric.Gauge{}, @@ -132,7 +144,7 @@ func TestToMetric(t *testing.T) { name: "counter without labels", row: cmwatcher.ClusterMetricRow{ Name: "test.counter", - Type: "counter", + Type: "COUNTER", Value: 100, }, wantType: &metric.Counter{}, @@ -141,7 +153,7 @@ func TestToMetric(t *testing.T) { row: cmwatcher.ClusterMetricRow{ Name: "test.labeled.gauge", Labels: map[string]string{"store": "1"}, - Type: "gauge", + Type: "GAUGE", Value: 7, }, wantType: &metric.GaugeVec{}, @@ -150,10 +162,27 @@ func TestToMetric(t *testing.T) { row: cmwatcher.ClusterMetricRow{ Name: "test.labeled.counter", Labels: map[string]string{"node": "3"}, - Type: "counter", + Type: "COUNTER", Value: 55, }, wantType: &metric.CounterVec{}, + }, { + name: "stopwatch without labels", + row: cmwatcher.ClusterMetricRow{ + Name: "test.stopwatch", + Type: "STOPWATCH", + Value: 1000, + }, + wantType: &metric.Gauge{}, + }, { + name: "labeled stopwatch", + row: cmwatcher.ClusterMetricRow{ + Name: "test.labeled.stopwatch", + Labels: map[string]string{"store": "1"}, + Type: "STOPWATCH", + Value: 2000, + }, + wantType: &metric.GaugeVec{}, }, { name: "unknown type without labels", row: cmwatcher.ClusterMetricRow{ @@ -175,7 +204,7 @@ func TestToMetric(t *testing.T) { name: "unregistered metric without labels", row: cmwatcher.ClusterMetricRow{ Name: "nonexistent.metric", - Type: "gauge", + Type: "GAUGE", Value: 1, }, wantErr: "no metadata found for metric nonexistent.metric", @@ -184,7 +213,7 @@ func TestToMetric(t *testing.T) { row: cmwatcher.ClusterMetricRow{ Name: "nonexistent.labeled", Labels: map[string]string{"k": "v"}, - Type: "gauge", + Type: "GAUGE", Value: 1, }, wantErr: "no metadata found for metric nonexistent.labeled", diff --git a/pkg/obs/clustermetrics/utils/test_utils.go b/pkg/obs/clustermetrics/utils/test_utils.go index 20a12e4676ff..6e0374dd5e0b 100644 --- a/pkg/obs/clustermetrics/utils/test_utils.go +++ b/pkg/obs/clustermetrics/utils/test_utils.go @@ -6,7 +6,9 @@ package clustermetricutils type TestingKnobs struct { - OnRegistrySyncerStart func() + OnRegistrySyncerPreStart func() + OnRegistrySyncerStart func() + OnReloadComplete func() } // ModuleTestingKnobs implements base.ModuleTestingKnobs interface.