Merge #143775 #143933

143775: structlogging: update hot range logging for app tenants r=angles-n-daemons a=angles-n-daemons

structlogging: update hot range logging for app tenants
Hot range logging currently runs as a task on each SQL node in the cluster. Its operative is to call its status server, collect the hot ranges from it, and log those ranges.

This works well for a traditional deployment, since each sql node corresponds to a status server within the same process, so all of the status servers, and therefore all of the nodes with stores area guaranteed to be covered by this operation.

```
*legend*
[] = node
sql = sql server
status = status server

[sql ──────> status]
[sql ──────> status]
[sql ──────> status]
```

For multi-tenant deployments, this assumption goes out the window. In this topology, sql pods randomly connect to kv pods when making calls to the status server, which means that we can no longer guarantee that sql nodes will exhaustively cover the space they're required to cover.

```
[sql] ──────> [status]
[sql] ─┐      [status]
       └────> [status]
              [status]
              [status]
```

To fix this, the changeset proposed changes the hot range logger to a job, which performs a fanout to all the nodes. This solution, while not scaling quite as well, avoids the lossy problem of missing nodes when logging hot ranges.

```
[sql]  ┌────> [status]
[sql] ─┼────> [status]
       ├────> [status]
       ├────> [status]
       └────> [status]
```

Fixes: #143527
Epic: CRDB-43150

Release note (bug fix): fixes an issue where multi-tenant hot range logging did not log all the hot ranges.

143933: storage: use block properties implementation defined in cockroachkvs r=jbowens a=jbowens

The MVCC time interval block properties implementation has been copied into pebble/cockroachkvs so that the Pebble metamorphic tests may make use of it. Use that implementation and remove the pkg/storage copy.

Epic: none
Release note: none

Co-authored-by: Brian Dillmann <[email protected]>
Co-authored-by: Jackson Owens <[email protected]>

Author: 3 people

docs/generated/metrics/metrics.html

@@ -1306,6 +1306,18 @@
 <tr><td>APPLICATION</td><td>jobs.history_retention.resume_completed</td><td>Number of history_retention jobs which successfully resumed to completion</td><td>jobs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>jobs.history_retention.resume_failed</td><td>Number of history_retention jobs which failed with a non-retriable error</td><td>jobs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>jobs.history_retention.resume_retry_error</td><td>Number of history_retention jobs which failed with a retriable error</td><td>jobs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.currently_idle</td><td>Number of hot_ranges_logger jobs currently considered Idle and can be freely shut down</td><td>jobs</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.currently_paused</td><td>Number of hot_ranges_logger jobs currently considered Paused</td><td>jobs</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.currently_running</td><td>Number of hot_ranges_logger jobs currently running in Resume or OnFailOrCancel state</td><td>jobs</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.expired_pts_records</td><td>Number of expired protected timestamp records owned by hot_ranges_logger jobs</td><td>records</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.fail_or_cancel_completed</td><td>Number of hot_ranges_logger jobs which successfully completed their failure or cancelation process</td><td>jobs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.fail_or_cancel_failed</td><td>Number of hot_ranges_logger jobs which failed with a non-retriable error on their failure or cancelation process</td><td>jobs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.fail_or_cancel_retry_error</td><td>Number of hot_ranges_logger jobs which failed with a retriable error on their failure or cancelation process</td><td>jobs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.protected_age_sec</td><td>The age of the oldest PTS record protected by hot_ranges_logger jobs</td><td>seconds</td><td>GAUGE</td><td>SECONDS</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.protected_record_count</td><td>Number of protected timestamp records held by hot_ranges_logger jobs</td><td>records</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.resume_completed</td><td>Number of hot_ranges_logger jobs which successfully resumed to completion</td><td>jobs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.resume_failed</td><td>Number of hot_ranges_logger jobs which failed with a non-retriable error</td><td>jobs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>jobs.hot_ranges_logger.resume_retry_error</td><td>Number of hot_ranges_logger jobs which failed with a retriable error</td><td>jobs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>jobs.import.currently_idle</td><td>Number of import jobs currently considered Idle and can be freely shut down</td><td>jobs</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>APPLICATION</td><td>jobs.import.currently_paused</td><td>Number of import jobs currently considered Paused</td><td>jobs</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>APPLICATION</td><td>jobs.import.currently_running</td><td>Number of import jobs currently running in Resume or OnFailOrCancel state</td><td>jobs</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>

pkg/jobs/jobspb/jobs.proto

@@ -1439,6 +1439,10 @@ message StandbyReadTSPollerProgress {
 
 }
 
+message HotRangesLoggerDetails {
+
+}
+
 message UpdateTableMetadataCacheDetails {}
 message UpdateTableMetadataCacheProgress {
   enum Status {
@@ -1477,6 +1481,10 @@ message SqlActivityFlushProgress {
 
 }
 
+message HotRangesLoggerProgress {
+
+}
+
 message ImportRollbackProgress {}
 
 message Payload {
@@ -1548,6 +1556,7 @@ message Payload {
     UpdateTableMetadataCacheDetails update_table_metadata_cache_details = 49;
     StandbyReadTSPollerDetails standby_read_ts_poller_details = 50;
     SqlActivityFlushDetails sql_activity_flush_details = 51;
+    HotRangesLoggerDetails hot_ranges_logger_details = 52;
   }
   reserved 26;
   // PauseReason is used to describe the reason that the job is currently paused
@@ -1629,6 +1638,7 @@ message Progress {
     UpdateTableMetadataCacheProgress table_metadata_cache = 37;
     StandbyReadTSPollerProgress standby_read_ts_poller = 38;
     SqlActivityFlushProgress sql_activity_flush = 39;
+    HotRangesLoggerProgress hot_ranges_logger = 40;
   }
 
   uint64 trace_id = 21 [(gogoproto.nullable) = false, (gogoproto.customname) = "TraceID", (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb.TraceID"];
@@ -1672,6 +1682,7 @@ enum Type {
   UPDATE_TABLE_METADATA_CACHE = 29 [(gogoproto.enumvalue_customname) = "TypeUpdateTableMetadataCache"];
   STANDBY_READ_TS_POLLER = 30 [(gogoproto.enumvalue_customname) = "TypeStandbyReadTSPoller"];
   SQL_ACTIVITY_FLUSH = 31 [(gogoproto.enumvalue_customname) = "TypeSQLActivityFlush"];
+  HOT_RANGES_LOGGER = 32 [(gogoproto.enumvalue_customname) = "TypeHotRangesLogger"];
 }
 
 message Job {

pkg/jobs/jobspb/wrap.go

@@ -47,6 +47,7 @@ var (
 	_ Details = UpdateTableMetadataCacheDetails{}
 	_ Details = StandbyReadTSPollerDetails{}
 	_ Details = SqlActivityFlushDetails{}
+	_ Details = HotRangesLoggerDetails{}
 )
 
 // ProgressDetails is a marker interface for job progress details proto structs.
@@ -78,6 +79,7 @@ var (
 	_ ProgressDetails = UpdateTableMetadataCacheProgress{}
 	_ ProgressDetails = StandbyReadTSPollerProgress{}
 	_ ProgressDetails = SqlActivityFlushProgress{}
+	_ ProgressDetails = HotRangesLoggerProgress{}
 )
 
 // Type returns the payload's job type and panics if the type is invalid.
@@ -244,6 +246,8 @@ func DetailsType(d isPayload_Details) (Type, error) {
 		return TypeStandbyReadTSPoller, nil
 	case *Payload_SqlActivityFlushDetails:
 		return TypeSQLActivityFlush, nil
+	case *Payload_HotRangesLoggerDetails:
+		return TypeHotRangesLogger, nil
 	default:
 		return TypeUnspecified, errors.Newf("Payload.Type called on a payload with an unknown details type: %T", d)
 	}
@@ -298,6 +302,7 @@ var JobDetailsForEveryJobType = map[Type]Details{
 	TypeUpdateTableMetadataCache:     UpdateTableMetadataCacheDetails{},
 	TypeStandbyReadTSPoller:          StandbyReadTSPollerDetails{},
 	TypeSQLActivityFlush:             SqlActivityFlushDetails{},
+	TypeHotRangesLogger:              HotRangesLoggerDetails{},
 }
 
 // WrapProgressDetails wraps a ProgressDetails object in the protobuf wrapper
@@ -434,6 +439,8 @@ func (p *Payload) UnwrapDetails() Details {
 		return *d.StandbyReadTsPollerDetails
 	case *Payload_SqlActivityFlushDetails:
 		return *d.SqlActivityFlushDetails
+	case *Payload_HotRangesLoggerDetails:
+		return *d.HotRangesLoggerDetails
 	default:
 		return nil
 	}
@@ -501,6 +508,8 @@ func (p *Progress) UnwrapDetails() ProgressDetails {
 		return *d.StandbyReadTsPoller
 	case *Progress_SqlActivityFlush:
 		return *d.SqlActivityFlush
+	case *Progress_HotRangesLogger:
+		return *d.HotRangesLogger
 	default:
 		return nil
 	}
@@ -592,6 +601,8 @@ func WrapPayloadDetails(details Details) interface {
 		return &Payload_StandbyReadTsPollerDetails{StandbyReadTsPollerDetails: &d}
 	case SqlActivityFlushDetails:
 		return &Payload_SqlActivityFlushDetails{SqlActivityFlushDetails: &d}
+	case HotRangesLoggerDetails:
+		return &Payload_HotRangesLoggerDetails{HotRangesLoggerDetails: &d}
 	default:
 		panic(errors.AssertionFailedf("jobs.WrapPayloadDetails: unknown details type %T", d))
 	}
@@ -627,7 +638,7 @@ const (
 func (Type) SafeValue() {}
 
 // NumJobTypes is the number of jobs types.
-const NumJobTypes = 32
+const NumJobTypes = 33
 
 // ChangefeedDetailsMarshaler allows for dependency injection of
 // cloud.SanitizeExternalStorageURI to avoid the dependency from this

pkg/server/server.go

@@ -2316,6 +2316,7 @@ func (s *topLevelServer) AcceptClients(ctx context.Context) error {
 		s.status,
 		*s.sqlServer.internalExecutor,
 		s.ClusterSettings(),
+		nil,
 	); err != nil {
 		return err
 	}

pkg/server/structlogging/BUILD.bazel

@@ -2,10 +2,16 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "structlogging",
-    srcs = ["hot_ranges_log.go"],
+    srcs = [
+        "hot_ranges_log.go",
+        "hot_ranges_log_job.go",
+    ],
     importpath = "github.com/cockroachdb/cockroach/pkg/server/structlogging",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/jobs",
+        "//pkg/jobs/jobspb",
+        "//pkg/multitenant/tenantcapabilities",
         "//pkg/server/serverpb",
         "//pkg/settings",
         "//pkg/settings/cluster",
@@ -16,6 +22,7 @@ go_library(
         "//pkg/util/log/logutil",
         "//pkg/util/stop",
         "//pkg/util/timeutil",
+        "@com_github_cockroachdb_errors//:errors",
     ],
 )
 

pkg/server/structlogging/hot_ranges_log.go

@@ -9,6 +9,9 @@ import (
 	"context"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/multitenant/tenantcapabilities"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
@@ -50,9 +53,12 @@ var TelemetryHotRangesStatsLoggingDelay = settings.RegisterDurationSetting(
 // hotRangesLoggingScheduler is responsible for logging index usage stats
 // on a scheduled interval.
 type hotRangesLoggingScheduler struct {
-	ie      sql.InternalExecutor
-	sServer serverpb.TenantStatusServer
-	st      *cluster.Settings
+	ie          sql.InternalExecutor
+	sServer     serverpb.TenantStatusServer
+	st          *cluster.Settings
+	stopper     *stop.Stopper
+	job         *jobs.Job
+	multiTenant bool
 }
 
 // StartHotRangesLoggingScheduler starts the capture index usage statistics logging scheduler.
@@ -62,42 +68,66 @@ func StartHotRangesLoggingScheduler(
 	sServer serverpb.TenantStatusServer,
 	ie sql.InternalExecutor,
 	st *cluster.Settings,
+	ti *tenantcapabilities.Entry,
 ) error {
+	multiTenant := ti != nil && !ti.TenantID.IsSystem()
 	scheduler := hotRangesLoggingScheduler{
-		ie:      ie,
-		sServer: sServer,
-		st:      st,
+		ie:          ie,
+		sServer:     sServer,
+		st:          st,
+		stopper:     stopper,
+		multiTenant: multiTenant,
 	}
 
-	return scheduler.start(ctx, stopper)
+	if multiTenant {
+		return scheduler.startJob(ctx, stopper)
+	}
+
+	return scheduler.startTask(ctx, stopper)
 }
 
-func (s *hotRangesLoggingScheduler) start(ctx context.Context, stopper *stop.Stopper) error {
+func (s *hotRangesLoggingScheduler) startTask(ctx context.Context, stopper *stop.Stopper) error {
 	return stopper.RunAsyncTask(ctx, "hot-ranges-stats", func(ctx context.Context) {
-		intervalChangedChan := make(chan struct{})
-		// We have to register this callback first. Otherwise we may run into
-		// an unlikely but possible scenario where we've started the ticker,
-		// and the setting is changed before we register the callback and the
-		// ticker will not be reset to the new value.
-		TelemetryHotRangesStatsInterval.SetOnChange(&s.st.SV, func(ctx context.Context) {
-			intervalChangedChan <- struct{}{}
-		})
-
-		ticker := time.NewTicker(TelemetryHotRangesStatsInterval.Get(&s.st.SV))
-
-		for {
-			select {
-			case <-stopper.ShouldQuiesce():
-				return
-			case <-ctx.Done():
-				return
-			case <-ticker.C:
-				s.maybeLogHotRanges(ctx, stopper)
-			case <-intervalChangedChan:
-				ticker.Reset(TelemetryHotRangesStatsInterval.Get(&s.st.SV))
-			}
-		}
+		err := s.start(ctx, stopper)
+		log.Warningf(ctx, "hot ranges stats logging scheduler stopped: %s", err)
+	})
+}
+
+func (s *hotRangesLoggingScheduler) startJob(ctx context.Context, stopper *stop.Stopper) error {
+	jobs.RegisterConstructor(
+		jobspb.TypeHotRangesLogger,
+		func(job *jobs.Job, settings *cluster.Settings) jobs.Resumer {
+			return &hotRangesLoggingScheduler{job: job}
+		},
+		jobs.DisablesTenantCostControl,
+	)
+	return nil
+}
+
+func (s *hotRangesLoggingScheduler) start(ctx context.Context, stopper *stop.Stopper) error {
+	intervalChangedChan := make(chan struct{})
+	// We have to register this callback first. Otherwise we may run into
+	// an unlikely but possible scenario where we've started the ticker,
+	// and the setting is changed before we register the callback and the
+	// ticker will not be reset to the new value.
+	TelemetryHotRangesStatsInterval.SetOnChange(&s.st.SV, func(ctx context.Context) {
+		intervalChangedChan <- struct{}{}
 	})
+
+	ticker := time.NewTicker(TelemetryHotRangesStatsInterval.Get(&s.st.SV))
+
+	for {
+		select {
+		case <-stopper.ShouldQuiesce():
+			return nil
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			s.maybeLogHotRanges(ctx, stopper)
+		case <-intervalChangedChan:
+			ticker.Reset(TelemetryHotRangesStatsInterval.Get(&s.st.SV))
+		}
+	}
 }
 
 // maybeLogHotRanges is a small helper function which couples the
@@ -117,8 +147,12 @@ func (s *hotRangesLoggingScheduler) shouldLog() bool {
 // logHotRanges collects the hot ranges from this node's status server and
 // sends them to the TELEMETRY log channel.
 func (s *hotRangesLoggingScheduler) logHotRanges(ctx context.Context, stopper *stop.Stopper) {
-	resp, err := s.sServer.HotRangesV2(ctx,
-		&serverpb.HotRangesRequest{NodeID: "local", PageSize: ReportTopHottestRanges})
+	req := &serverpb.HotRangesRequest{NodeID: "local", PageSize: ReportTopHottestRanges}
+	// if we are running in single tenant mode, only log the ranges on the status server.
+	if !s.multiTenant {
+		req.NodeID = "local"
+	}
+	resp, err := s.sServer.HotRangesV2(ctx, req)
 	if err != nil {
 		log.Warningf(ctx, "failed to get hot ranges: %s", err)
 		return

pkg/server/structlogging/hot_ranges_log_job.go

@@ -0,0 +1,38 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package structlogging
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/errors"
+)
+
+func (s *hotRangesLoggingScheduler) Resume(ctx context.Context, execCtxI interface{}) error {
+	// This job is a forever running background job, and it is always safe to
+	// terminate the SQL pod whenever the job is running, so mark it as idle.
+	s.job.MarkIdle(true)
+
+	return s.start(ctx, s.stopper)
+}
+
+func (s *hotRangesLoggingScheduler) OnFailOrCancel(
+	ctx context.Context, execCtx interface{}, jobErr error,
+) error {
+	if jobs.HasErrJobCanceled(jobErr) {
+		err := errors.NewAssertionErrorWithWrappedErrf(
+			jobErr, "hot range logging job is not cancelable",
+		)
+		log.Errorf(ctx, "%v", err)
+	}
+	return nil
+}
+
+func (s *hotRangesLoggingScheduler) CollectProfile(ctx context.Context, execCtx interface{}) error {
+	return nil
+}

pkg/server/tenant.go

@@ -980,12 +980,14 @@ func (s *SQLServerWrapper) AcceptClients(ctx context.Context) error {
 		}
 	}
 
+	ti, _ := s.sqlServer.tenantConnect.TenantInfo()
 	if err := structlogging.StartHotRangesLoggingScheduler(
 		ctx,
 		s.stopper,
 		s.sqlServer.tenantConnect,
 		*s.sqlServer.internalExecutor,
 		s.ClusterSettings(),
+		&ti,
 	); err != nil {
 		return err
 	}

pkg/storage/BUILD.bazel

@@ -212,6 +212,7 @@ go_test(
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_errors//oserror",
         "@com_github_cockroachdb_pebble//:pebble",
+        "@com_github_cockroachdb_pebble//cockroachkvs",
         "@com_github_cockroachdb_pebble//objstorage",
         "@com_github_cockroachdb_pebble//objstorage/objstorageprovider",
         "@com_github_cockroachdb_pebble//sstable",