Merge #157226

157226: ts: introduce a new 1 minute poller to write into tsdb r=jasonlmfong a=jasonlmfong

This change adds a new 1 minute resolution and the machinery around rollup and storage.
Also adds a new poller at server start-up time, this poller currently performs a no-op every 1 minute.

Epic: CRDB-55079
Release: None

Co-authored-by: Jason Fong <[email protected]>

Author: craig[bot]

docs/generated/settings/settings-for-tenants.txt

@@ -418,6 +418,7 @@ storage.sstable.compression_algorithm	enumeration	fastest	determines the compres
 storage.sstable.compression_algorithm_backup_storage	enumeration	fastest	determines the compression algorithm to use when compressing sstable data blocks for backup row data storage (fast,balanced,good are experimental); [snappy = 1, zstd = 2, none = 3, minlz = 4, fastest = 5, fast = 6, balanced = 7, good = 8]	system-visible
 storage.sstable.compression_algorithm_backup_transport	enumeration	fastest	determines the compression algorithm to use when compressing sstable data blocks for backup transport (fast,balanced,good are experimental); [snappy = 1, zstd = 2, none = 3, minlz = 4, fastest = 5, fast = 6, balanced = 7, good = 8]	system-visible
 timeseries.storage.resolution_10s.ttl	duration	240h0m0s	the maximum age of time series data stored at the 10 second resolution. Data older than this is subject to rollup and deletion.	system-visible
+timeseries.storage.resolution_1m.ttl	duration	240h0m0s	the maximum age of time series data stored at the 1 minute resolution. Data older than this is subject to rollup and deletion.	system-visible
 timeseries.storage.resolution_30m.ttl	duration	2160h0m0s	the maximum age of time series data stored at the 30 minute resolution. Data older than this is subject to deletion.	system-visible
 trace.debug_http_endpoint.enabled (alias: trace.debug.enable)	boolean	false	if set, traces for recent requests can be seen at https://<ui>/debug/requests	application
 trace.opentelemetry.collector	string		address of an OpenTelemetry trace collector to receive traces using the otel gRPC protocol, as <host>:<port>. If no port is specified, 4317 will be used.	application

docs/generated/settings/settings.html

@@ -377,6 +377,7 @@
 <tr><td><div id="setting-storage-wal-failover-unhealthy-op-threshold" class="anchored"><code>storage.wal_failover.unhealthy_op_threshold</code></div></td><td>duration</td><td><code>100ms</code></td><td>the latency of a WAL write considered unhealthy and triggers a failover to a secondary WAL location</td><td>Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-timeseries-storage-enabled" class="anchored"><code>timeseries.storage.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>if set, periodic timeseries data is stored within the cluster; disabling is not recommended unless you are storing the data elsewhere</td><td>Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-timeseries-storage-resolution-10s-ttl" class="anchored"><code>timeseries.storage.resolution_10s.ttl</code></div></td><td>duration</td><td><code>240h0m0s</code></td><td>the maximum age of time series data stored at the 10 second resolution. Data older than this is subject to rollup and deletion.</td><td>Advanced/Self-hosted (read-write); Basic/Standard (read-only)</td></tr>
+<tr><td><div id="setting-timeseries-storage-resolution-1m-ttl" class="anchored"><code>timeseries.storage.resolution_1m.ttl</code></div></td><td>duration</td><td><code>240h0m0s</code></td><td>the maximum age of time series data stored at the 1 minute resolution. Data older than this is subject to rollup and deletion.</td><td>Advanced/Self-hosted (read-write); Basic/Standard (read-only)</td></tr>
 <tr><td><div id="setting-timeseries-storage-resolution-30m-ttl" class="anchored"><code>timeseries.storage.resolution_30m.ttl</code></div></td><td>duration</td><td><code>2160h0m0s</code></td><td>the maximum age of time series data stored at the 30 minute resolution. Data older than this is subject to deletion.</td><td>Advanced/Self-hosted (read-write); Basic/Standard (read-only)</td></tr>
 <tr><td><div id="setting-trace-debug-enable" class="anchored"><code>trace.debug_http_endpoint.enabled<br />(alias: trace.debug.enable)</code></div></td><td>boolean</td><td><code>false</code></td><td>if set, traces for recent requests can be seen at https://&lt;ui&gt;/debug/requests</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-trace-opentelemetry-collector" class="anchored"><code>trace.opentelemetry.collector</code></div></td><td>string</td><td><code></code></td><td>address of an OpenTelemetry trace collector to receive traces using the otel gRPC protocol, as &lt;host&gt;:&lt;port&gt;. If no port is specified, 4317 will be used.</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>

pkg/base/config.go

@@ -88,6 +88,10 @@ const (
 	// contributes meaningfully to the average, while earlier measurements have
 	// diminishing impact.
 	DefaultCPUUsageMovingAverageAge = 20
+
+	// DefaultHighCardinalityMetricsSampleInterval is the default interval for
+	// sampling low-frequency high-cardinality metrics.
+	DefaultHighCardinalityMetricsSampleInterval = time.Minute
 )
 
 // DefaultCertsDirectory is the default value for the cert directory flag.

pkg/server/server.go

@@ -2013,7 +2013,12 @@ func (s *topLevelServer) PreStart(ctx context.Context) error {
 	// The writes will be async; we'll wait for the first one to go through
 	// later in this method, using the returned channel.
 	firstTSDBPollDone := s.tsDB.PollSource(
-		s.cfg.AmbientCtx, s.recorder, base.DefaultMetricsSampleInterval, ts.Resolution10s, s.stopper,
+		s.cfg.AmbientCtx, s.recorder, base.DefaultMetricsSampleInterval, ts.Resolution10s, s.stopper, false,
+	)
+
+	// high cardinality child metrics collector, we don't need to wait for the first one
+	s.tsDB.PollSource(
+		s.cfg.AmbientCtx, s.recorder, base.DefaultHighCardinalityMetricsSampleInterval, ts.Resolution1m, s.stopper, true,
 	)
 
 	// Export statistics to graphite, if enabled by configuration.

pkg/server/status/recorder.go

@@ -108,6 +108,17 @@ var bugfix149481Enabled = settings.RegisterBoolSetting(
 	true,
 	settings.WithVisibility(settings.Reserved))
 
+// ChildMetricsStorageEnabled controls whether to record high-cardinality child metrics
+// into the time series database. This is separate from ChildMetricsEnabled which controls
+// Prometheus exports, allowing independent control of child metrics recording vs export.
+// This setting enables debugging of changefeeds and should not be considered functionality
+// to expand support for.
+var ChildMetricsStorageEnabled = settings.RegisterBoolSetting(
+	settings.ApplicationLevel, "timeseries.child_metrics.enabled",
+	"enables the collection of high-cardinality child metrics into the time series database",
+	false,
+	settings.WithVisibility(settings.Reserved))
+
 // MetricsRecorder is used to periodically record the information in a number of
 // metric registries.
 //
@@ -428,7 +439,7 @@ func (mr *MetricsRecorder) ExportToGraphite(
 // GetTimeSeriesData serializes registered metrics for consumption by
 // CockroachDB's time series system. GetTimeSeriesData implements the DataSource
 // interface of the ts package.
-func (mr *MetricsRecorder) GetTimeSeriesData() []tspb.TimeSeriesData {
+func (mr *MetricsRecorder) GetTimeSeriesData(childMetrics bool) []tspb.TimeSeriesData {
 	mr.mu.RLock()
 	defer mr.mu.RUnlock()
 
@@ -440,6 +451,13 @@ func (mr *MetricsRecorder) GetTimeSeriesData() []tspb.TimeSeriesData {
 		return nil
 	}
 
+	if childMetrics {
+		if !ChildMetricsStorageEnabled.Get(&mr.settings.SV) {
+			return nil
+		}
+		return nil // TODO(jasonlmfong): to be implemented
+	}
+
 	lastDataCount := atomic.LoadInt64(&mr.lastDataCount)
 	data := make([]tspb.TimeSeriesData, 0, lastDataCount)
 

pkg/server/status/recorder_test.go

@@ -250,7 +250,7 @@ func TestMetricsRecorderLabels(t *testing.T) {
 		},
 	}
 
-	actualData := recorder.GetTimeSeriesData()
+	actualData := recorder.GetTimeSeriesData(false)
 
 	// compare actual vs expected values
 	sort.Sort(byTimeAndName(actualData))
@@ -650,7 +650,7 @@ func TestMetricsRecorder(t *testing.T) {
 	// ========================================
 	// Verify time series data
 	// ========================================
-	actual := recorder.GetTimeSeriesData()
+	actual := recorder.GetTimeSeriesData(false)
 
 	// Actual comparison is simple: sort the resulting arrays by time and name,
 	// and use reflect.DeepEqual.
@@ -725,7 +725,7 @@ func TestMetricsRecorder(t *testing.T) {
 				t.Error(err)
 			}
 			_ = recorder.PrintAsText(io.Discard, expfmt.FmtText, false)
-			_ = recorder.GetTimeSeriesData()
+			_ = recorder.GetTimeSeriesData(false)
 			wg.Done()
 		}()
 	}

pkg/ts/db.go

@@ -29,6 +29,7 @@ var (
 	// version upgrade, the rollup threshold is used instead.
 	deprecatedResolution10sDefaultPruneThreshold = 30 * 24 * time.Hour
 	resolution10sDefaultRollupThreshold          = 10 * 24 * time.Hour
+	resolution1mDefaultRollupThreshold           = 10 * 24 * time.Hour
 	resolution30mDefaultPruneThreshold           = 90 * 24 * time.Hour
 	resolution50nsDefaultPruneThreshold          = 1 * time.Millisecond
 	storeDataTimeout                             = 1 * time.Minute
@@ -54,6 +55,17 @@ var Resolution10sStorageTTL = settings.RegisterDurationSetting(
 	resolution10sDefaultRollupThreshold,
 	settings.WithPublic)
 
+// Resolution1mStorageTTL defines the maximum age of data that will be retained
+// at the 1 minute resolution. Data older than this is subject to being "rolled
+// up" into the 30 minute resolution and then deleted.
+var Resolution1mStorageTTL = settings.RegisterDurationSetting(
+	settings.SystemVisible, // currently used in DB Console.
+	"timeseries.storage.resolution_1m.ttl",
+	"the maximum age of time series data stored at the 1 minute resolution. Data older than this "+
+		"is subject to rollup and deletion.",
+	resolution1mDefaultRollupThreshold,
+	settings.WithPublic)
+
 // Resolution30mStorageTTL defines the maximum age of data that will be
 // retained at the 30 minute resolution. Data older than this is subject to
 // deletion.
@@ -89,6 +101,7 @@ func NewDB(db *kv.DB, settings *cluster.Settings) *DB {
 			return Resolution10sStorageTTL.Get(&settings.SV).Nanoseconds()
 		},
 		Resolution30m:  func() int64 { return Resolution30mStorageTTL.Get(&settings.SV).Nanoseconds() },
+		Resolution1m:   func() int64 { return Resolution1mStorageTTL.Get(&settings.SV).Nanoseconds() },
 		resolution1ns:  func() int64 { return resolution1nsDefaultRollupThreshold.Nanoseconds() },
 		resolution50ns: func() int64 { return resolution50nsDefaultPruneThreshold.Nanoseconds() },
 	}
@@ -102,7 +115,7 @@ func NewDB(db *kv.DB, settings *cluster.Settings) *DB {
 
 // A DataSource can be queried for a slice of time series data.
 type DataSource interface {
-	GetTimeSeriesData() []tspb.TimeSeriesData
+	GetTimeSeriesData(childMetrics bool) []tspb.TimeSeriesData
 }
 
 // poller maintains information for a polling process started by PollSource().
@@ -113,6 +126,9 @@ type poller struct {
 	frequency time.Duration
 	r         Resolution
 	stopper   *stop.Stopper
+	// When childMetrics is set to true, the polling process calls alternate behaviour in metrics recorder,
+	// returning child metrics to be stored.
+	childMetrics bool
 }
 
 // PollSource begins a Goroutine which periodically queries the supplied
@@ -125,6 +141,7 @@ func (db *DB) PollSource(
 	frequency time.Duration,
 	r Resolution,
 	stopper *stop.Stopper,
+	childMetrics bool,
 ) (firstDone <-chan struct{}) {
 	ambient.AddLogTag("ts-poll", nil)
 	p := &poller{
@@ -134,6 +151,7 @@ func (db *DB) PollSource(
 		frequency:      frequency,
 		r:              r,
 		stopper:        stopper,
+		childMetrics:   childMetrics,
 	}
 	return p.start()
 }
@@ -179,7 +197,7 @@ func (p *poller) poll(ctx context.Context) {
 	}
 
 	if err := p.stopper.RunTask(ctx, "ts.poller: poll", func(ctx context.Context) {
-		data := p.source.GetTimeSeriesData()
+		data := p.source.GetTimeSeriesData(p.childMetrics)
 		if len(data) == 0 {
 			return
 		}

pkg/ts/db_test.go

@@ -667,7 +667,7 @@ type modelDataSource struct {
 // set of TimeSeriesData to subsequent calls. It stores each TimeSeriesData
 // object in the test model before returning it. If all TimeSeriesData objects
 // have been returned, this method will stop the provided Stopper.
-func (mds *modelDataSource) GetTimeSeriesData() []tspb.TimeSeriesData {
+func (mds *modelDataSource) GetTimeSeriesData(childMetrics bool) []tspb.TimeSeriesData {
 	if len(mds.datasets) == 0 {
 		// Stop on goroutine to prevent deadlock.
 		go mds.once.Do(func() { mds.stopper.Stop(context.Background()) })
@@ -762,7 +762,7 @@ func TestPollSource(t *testing.T) {
 		}
 
 		ambient := log.MakeTestingAmbientContext(tr)
-		tm.DB.PollSource(ambient, &testSource, time.Millisecond, Resolution10s, testSource.stopper)
+		tm.DB.PollSource(ambient, &testSource, time.Millisecond, Resolution10s, testSource.stopper, false)
 		<-testSource.stopper.IsStopped()
 		if a, e := testSource.calledCount, 2; a != e {
 			t.Errorf("testSource was called %d times, expected %d", a, e)
@@ -815,7 +815,7 @@ func TestDisableStorage(t *testing.T) {
 		}
 
 		ambient := log.MakeTestingAmbientCtxWithNewTracer()
-		tm.DB.PollSource(ambient, &testSource, time.Millisecond, Resolution10s, testSource.stopper)
+		tm.DB.PollSource(ambient, &testSource, time.Millisecond, Resolution10s, testSource.stopper, false)
 		select {
 		case <-testSource.stopper.IsStopped():
 			t.Error("testSource data exhausted when polling should have been enabled")

pkg/ts/keys_test.go

@@ -48,6 +48,30 @@ func TestDataKeys(t *testing.T) {
 			12,
 			"/System/tsd//10s/1924-09-18T08:00:00Z/",
 		},
+		{
+			"test.metric",
+			"testsource",
+			0,
+			Resolution1m,
+			30,
+			"/System/tsd/test.metric/1m/1970-01-01T00:00:00Z/testsource",
+		},
+		{
+			"test.no.source",
+			"",
+			1429114700000000000,
+			Resolution1m,
+			26,
+			"/System/tsd/test.no.source/1m/2015-04-15T16:00:00Z/",
+		},
+		{
+			"",
+			"",
+			-1429114700000000000,
+			Resolution1m,
+			12,
+			"/System/tsd//1m/1924-09-18T08:00:00Z/",
+		},
 	}
 
 	for i, tc := range testCases {

pkg/ts/memory_test.go

@@ -146,6 +146,94 @@ func TestGetMaxTimespan(t *testing.T) {
 			math.MaxInt64,
 			"",
 		},
+		// Resolution1m - room for exactly one hour of query
+		{
+			Resolution1m,
+			QueryMemoryOptions{
+				BudgetBytes:             2 * (sizeOfTimeSeriesData + sizeOfSample*60),
+				EstimatedSources:        1,
+				InterpolationLimitNanos: 0,
+			},
+			(1 * time.Hour).Nanoseconds(),
+			"",
+		},
+		// Resolution1m - insufficient memory
+		{
+			Resolution1m,
+			QueryMemoryOptions{
+				BudgetBytes:             sizeOfTimeSeriesData + sizeOfSample*60,
+				EstimatedSources:        1,
+				InterpolationLimitNanos: 0,
+			},
+			0,
+			"insufficient",
+		},
+		// Resolution1m - 3 sources, room for 1 hour.
+		{
+			Resolution1m,
+			QueryMemoryOptions{
+				BudgetBytes:             6 * (sizeOfTimeSeriesData + sizeOfSample*60),
+				EstimatedSources:        3,
+				InterpolationLimitNanos: 0,
+			},
+			(1 * time.Hour).Nanoseconds(),
+			"",
+		},
+		// Resolution1m - 3 sources, room for 2 hours.
+		{
+			Resolution1m,
+			QueryMemoryOptions{
+				BudgetBytes:             9 * (sizeOfTimeSeriesData + sizeOfSample*60),
+				EstimatedSources:        3,
+				InterpolationLimitNanos: 0,
+			},
+			(2 * time.Hour).Nanoseconds(),
+			"",
+		},
+		// Not enough room due to interpolation buffer.
+		{
+			Resolution1m,
+			QueryMemoryOptions{
+				BudgetBytes:             6 * (sizeOfTimeSeriesData + sizeOfSample*60),
+				EstimatedSources:        3,
+				InterpolationLimitNanos: 1,
+			},
+			0,
+			"insufficient",
+		},
+		// Sufficient room even with interpolation buffer.
+		{
+			Resolution1m,
+			QueryMemoryOptions{
+				BudgetBytes:             9 * (sizeOfTimeSeriesData + sizeOfSample*60),
+				EstimatedSources:        3,
+				InterpolationLimitNanos: 1,
+			},
+			(1 * time.Hour).Nanoseconds(),
+			"",
+		},
+		// Insufficient room for interpolation buffer (due to straddling)
+		{
+			Resolution1m,
+			QueryMemoryOptions{
+				BudgetBytes:             9 * (sizeOfTimeSeriesData + sizeOfSample*60),
+				EstimatedSources:        3,
+				InterpolationLimitNanos: int64(float64(Resolution1m.SlabDuration()) * 0.75),
+			},
+			0,
+			"insufficient",
+		},
+		// Sufficient room even with interpolation buffer.
+		{
+			Resolution1m,
+			QueryMemoryOptions{
+				BudgetBytes:             12 * (sizeOfTimeSeriesData + sizeOfSample*60),
+				EstimatedSources:        3,
+				InterpolationLimitNanos: int64(float64(Resolution1m.SlabDuration()) * 0.75),
+			},
+			(1 * time.Hour).Nanoseconds(),
+			"",
+		},
 	} {
 		t.Run("", func(t *testing.T) {
 			mem := QueryMemoryContext{