Merge #158506

158506: sql/fingerprint: add new job fingerprint spans job r=dt a=dt

Currently unused outside of tests.

This new job computes the kv fingerprint of the configured spans using
the KV ExportRequest fingerprint mechanism.

Release note: none.
Epic: none.


Co-authored-by: David Taylor <[email protected]>

Author: craig[bot]

docs/generated/metrics/metrics.yaml

@@ -4430,6 +4430,105 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.fingerprint.currently_idle
+      exported_name: jobs_fingerprint_currently_idle
+      labeled_name: 'jobs{type: fingerprint, status: currently_idle}'
+      description: Number of fingerprint jobs currently considered Idle and can be freely shut down
+      y_axis_label: jobs
+      type: GAUGE
+      unit: COUNT
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.fingerprint.currently_paused
+      exported_name: jobs_fingerprint_currently_paused
+      labeled_name: 'jobs{name: fingerprint, status: currently_paused}'
+      description: Number of fingerprint jobs currently considered Paused
+      y_axis_label: jobs
+      type: GAUGE
+      unit: COUNT
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.fingerprint.currently_running
+      exported_name: jobs_fingerprint_currently_running
+      labeled_name: 'jobs{type: fingerprint, status: currently_running}'
+      description: Number of fingerprint jobs currently running in Resume or OnFailOrCancel state
+      y_axis_label: jobs
+      type: GAUGE
+      unit: COUNT
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.fingerprint.expired_pts_records
+      exported_name: jobs_fingerprint_expired_pts_records
+      labeled_name: 'jobs.expired_pts_records{type: fingerprint}'
+      description: Number of expired protected timestamp records owned by fingerprint jobs
+      y_axis_label: records
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.fingerprint.fail_or_cancel_completed
+      exported_name: jobs_fingerprint_fail_or_cancel_completed
+      labeled_name: 'jobs.fail_or_cancel{name: fingerprint, status: completed}'
+      description: Number of fingerprint jobs which successfully completed their failure or cancelation process
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.fingerprint.fail_or_cancel_retry_error
+      exported_name: jobs_fingerprint_fail_or_cancel_retry_error
+      labeled_name: 'jobs.fail_or_cancel{name: fingerprint, status: retry_error}'
+      description: Number of fingerprint jobs which failed with a retriable error on their failure or cancelation process
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.fingerprint.protected_age_sec
+      exported_name: jobs_fingerprint_protected_age_sec
+      labeled_name: 'jobs.protected_age_sec{type: fingerprint}'
+      description: The age of the oldest PTS record protected by fingerprint jobs
+      y_axis_label: seconds
+      type: GAUGE
+      unit: SECONDS
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.fingerprint.protected_record_count
+      exported_name: jobs_fingerprint_protected_record_count
+      labeled_name: 'jobs.protected_record_count{type: fingerprint}'
+      description: Number of protected timestamp records held by fingerprint jobs
+      y_axis_label: records
+      type: GAUGE
+      unit: COUNT
+      aggregation: AVG
+      derivative: NONE
+    - name: jobs.fingerprint.resume_completed
+      exported_name: jobs_fingerprint_resume_completed
+      labeled_name: 'jobs.resume{name: fingerprint, status: completed}'
+      description: Number of fingerprint jobs which successfully resumed to completion
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.fingerprint.resume_failed
+      exported_name: jobs_fingerprint_resume_failed
+      labeled_name: 'jobs.resume{name: fingerprint, status: failed}'
+      description: Number of fingerprint jobs which failed with a non-retriable error
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.fingerprint.resume_retry_error
+      exported_name: jobs_fingerprint_resume_retry_error
+      labeled_name: 'jobs.resume{name: fingerprint, status: retry_error}'
+      description: Number of fingerprint jobs which failed with a retriable error
+      y_axis_label: jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: jobs.history_retention.currently_idle
       exported_name: jobs_history_retention_currently_idle
       labeled_name: 'jobs{type: history_retention, status: currently_idle}'

pkg/BUILD.bazel

@@ -500,6 +500,7 @@ ALL_TESTS = [
     "//pkg/sql/execstats:execstats_test",
     "//pkg/sql/export:export_test",
     "//pkg/sql/exprutil:exprutil_test",
+    "//pkg/sql/fingerprint:fingerprint_test",
     "//pkg/sql/flowinfra:flowinfra_disallowed_imports_test",
     "//pkg/sql/flowinfra:flowinfra_test",
     "//pkg/sql/gcjob/gcjobnotifier:gcjobnotifier_test",
@@ -2064,6 +2065,8 @@ GO_TARGETS = [
     "//pkg/sql/exprutil:exprutil",
     "//pkg/sql/exprutil:exprutil_test",
     "//pkg/sql/faketreeeval:faketreeeval",
+    "//pkg/sql/fingerprint:fingerprint",
+    "//pkg/sql/fingerprint:fingerprint_test",
     "//pkg/sql/flowinfra:flowinfra",
     "//pkg/sql/flowinfra:flowinfra_test",
     "//pkg/sql/gcjob/gcjobnotifier:gcjobnotifier",

pkg/jobs/job_info_utils.go

@@ -8,12 +8,14 @@ package jobs
 import (
 	"bytes"
 	"context"
+	"encoding/binary"
 	"fmt"
 	"io"
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
+	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/errors"
 	"github.com/klauspost/compress/gzip"
 )
@@ -108,3 +110,47 @@ func ReadChunkedFileToJobInfo(
 
 	return buf.Bytes(), nil
 }
+
+// WriteUint64 writes a uint64 value to the info storage.
+func (i InfoStorage) WriteUint64(ctx context.Context, infoKey string, value uint64) error {
+	buf := make([]byte, 8)
+	binary.LittleEndian.PutUint64(buf, value)
+	return i.Write(ctx, infoKey, buf)
+}
+
+// GetUint64 retrieves a uint64 value from the info storage.
+// Returns (value, found, error).
+func (i InfoStorage) GetUint64(ctx context.Context, infoKey string) (uint64, bool, error) {
+	data, found, err := i.Get(ctx, "get-uint64", infoKey)
+	if err != nil || !found {
+		return 0, found, err
+	}
+	if len(data) != 8 {
+		return 0, false, errors.Newf("invalid uint64 data length: %d", len(data))
+	}
+	return binary.LittleEndian.Uint64(data), true, nil
+}
+
+// WriteProto writes a protobuf message to the info storage.
+func (i InfoStorage) WriteProto(ctx context.Context, infoKey string, msg protoutil.Message) error {
+	data, err := protoutil.Marshal(msg)
+	if err != nil {
+		return errors.Wrap(err, "failed to marshal proto")
+	}
+	return i.Write(ctx, infoKey, data)
+}
+
+// GetProto retrieves a protobuf message from the info storage.
+// Returns (found, error). If found, the message is unmarshaled into msg.
+func (i InfoStorage) GetProto(
+	ctx context.Context, infoKey string, msg protoutil.Message,
+) (bool, error) {
+	data, found, err := i.Get(ctx, "get-proto", infoKey)
+	if err != nil || !found {
+		return found, err
+	}
+	if err := protoutil.Unmarshal(data, msg); err != nil {
+		return false, errors.Wrap(err, "failed to unmarshal proto")
+	}
+	return true, nil
+}

pkg/jobs/jobspb/jobs.proto

@@ -1534,6 +1534,18 @@ message InspectDetails {
   ];
 }
 
+message FingerprintDetails {
+  repeated roachpb.Span spans = 1 [(gogoproto.nullable) = false];
+  util.hlc.Timestamp as_of = 2 [(gogoproto.nullable) = false];
+  util.hlc.Timestamp start = 3 [(gogoproto.nullable) = false];
+  bool stripped = 4;
+  bytes pts = 5 [(gogoproto.customname) = "PTS", (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID", (gogoproto.nullable) = false];
+}
+
+message FingerprintProgress {
+ // Not used: progress is stored in its own info key(s) and frontier.
+}
+
 message UpdateTableMetadataCacheDetails {}
 message UpdateTableMetadataCacheProgress {
   enum Status {
@@ -1659,6 +1671,7 @@ message Payload {
     SqlActivityFlushDetails sql_activity_flush_details = 51;
     HotRangesLoggerDetails hot_ranges_logger_details = 52;
     InspectDetails inspect_details = 53;
+    FingerprintDetails fingerprint_details = 54;
   }
   reserved 26;
   // PauseReason is used to describe the reason that the job is currently paused
@@ -1743,6 +1756,7 @@ message Progress {
     SqlActivityFlushProgress sql_activity_flush = 39;
     HotRangesLoggerProgress hot_ranges_logger = 40;
     InspectProgress inspect = 41;
+    FingerprintProgress fingerprint = 42;
   }
 
   uint64 trace_id = 21 [(gogoproto.nullable) = false, (gogoproto.customname) = "TraceID", (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb.TraceID"];
@@ -1790,6 +1804,7 @@ enum Type {
   SQL_ACTIVITY_FLUSH = 31 [(gogoproto.enumvalue_customname) = "TypeSQLActivityFlush"];
   HOT_RANGES_LOGGER = 32 [(gogoproto.enumvalue_customname) = "TypeHotRangesLogger"];
   INSPECT = 33 [(gogoproto.enumvalue_customname) = "TypeInspect"];
+  FINGERPRINT = 34 [(gogoproto.enumvalue_customname) = "TypeFingerprint"];
 }
 
 message Job {

pkg/jobs/jobspb/wrap.go

@@ -49,6 +49,7 @@ var (
 	_ Details = SqlActivityFlushDetails{}
 	_ Details = HotRangesLoggerDetails{}
 	_ Details = InspectDetails{}
+	_ Details = FingerprintDetails{}
 )
 
 // ProgressDetails is a marker interface for job progress details proto structs.
@@ -82,6 +83,7 @@ var (
 	_ ProgressDetails = SqlActivityFlushProgress{}
 	_ ProgressDetails = HotRangesLoggerProgress{}
 	_ ProgressDetails = InspectProgress{}
+	_ ProgressDetails = FingerprintProgress{}
 )
 
 // Type returns the payload's job type and panics if the type is invalid.
@@ -249,6 +251,8 @@ func DetailsType(d isPayload_Details) (Type, error) {
 		return TypeHotRangesLogger, nil
 	case *Payload_InspectDetails:
 		return TypeInspect, nil
+	case *Payload_FingerprintDetails:
+		return TypeFingerprint, nil
 	default:
 		return TypeUnspecified, errors.Newf("Payload.Type called on a payload with an unknown details type: %T", d)
 	}
@@ -305,6 +309,7 @@ var JobDetailsForEveryJobType = map[Type]Details{
 	TypeSQLActivityFlush:             SqlActivityFlushDetails{},
 	TypeHotRangesLogger:              HotRangesLoggerDetails{},
 	TypeInspect:                      InspectDetails{},
+	TypeFingerprint:                  FingerprintDetails{},
 }
 
 // WrapProgressDetails wraps a ProgressDetails object in the protobuf wrapper
@@ -378,6 +383,8 @@ func WrapProgressDetails(details ProgressDetails) interface {
 		return &Progress_HotRangesLogger{HotRangesLogger: &d}
 	case InspectProgress:
 		return &Progress_Inspect{Inspect: &d}
+	case FingerprintProgress:
+		return &Progress_Fingerprint{Fingerprint: &d}
 	default:
 		panic(errors.AssertionFailedf("WrapProgressDetails: unknown progress type %T", d))
 	}
@@ -449,6 +456,8 @@ func (p *Payload) UnwrapDetails() Details {
 		return *d.HotRangesLoggerDetails
 	case *Payload_InspectDetails:
 		return *d.InspectDetails
+	case *Payload_FingerprintDetails:
+		return *d.FingerprintDetails
 	default:
 		return nil
 	}
@@ -520,6 +529,8 @@ func (p *Progress) UnwrapDetails() ProgressDetails {
 		return *d.HotRangesLogger
 	case *Progress_Inspect:
 		return d.Inspect
+	case *Progress_Fingerprint:
+		return d.Fingerprint
 	default:
 		return nil
 	}
@@ -615,6 +626,8 @@ func WrapPayloadDetails(details Details) interface {
 		return &Payload_HotRangesLoggerDetails{HotRangesLoggerDetails: &d}
 	case InspectDetails:
 		return &Payload_InspectDetails{InspectDetails: &d}
+	case FingerprintDetails:
+		return &Payload_FingerprintDetails{FingerprintDetails: &d}
 	default:
 		panic(errors.AssertionFailedf("jobs.WrapPayloadDetails: unknown details type %T", d))
 	}
@@ -650,7 +663,7 @@ const (
 func (Type) SafeValue() {}
 
 // NumJobTypes is the number of jobs types.
-const NumJobTypes = 34
+const NumJobTypes = 35
 
 // ChangefeedDetailsMarshaler allows for dependency injection of
 // cloud.SanitizeExternalStorageURI to avoid the dependency from this

pkg/roachprod/agents/opentelemetry/files/cockroachdb_metrics.yaml

@@ -842,6 +842,17 @@ jobs_create_stats_fail_or_cancel_retry_error: jobs.create_stats.fail_or_cancel_r
 jobs_create_stats_resume_completed: jobs.create_stats.resume_completed
 jobs_create_stats_resume_failed: jobs.create_stats.resume_failed
 jobs_create_stats_resume_retry_error: jobs.create_stats.resume_retry_error
+jobs_fingerprint_currently_idle: jobs.fingerprint.currently_idle
+jobs_fingerprint_currently_paused: jobs.fingerprint.currently_paused
+jobs_fingerprint_currently_running: jobs.fingerprint.currently_running
+jobs_fingerprint_expired_pts_records: jobs.fingerprint.expired_pts_records
+jobs_fingerprint_fail_or_cancel_completed: jobs.fingerprint.fail_or_cancel_completed
+jobs_fingerprint_fail_or_cancel_retry_error: jobs.fingerprint.fail_or_cancel_retry_error
+jobs_fingerprint_protected_age_sec: jobs.fingerprint.protected_age_sec
+jobs_fingerprint_protected_record_count: jobs.fingerprint.protected_record_count
+jobs_fingerprint_resume_completed: jobs.fingerprint.resume_completed
+jobs_fingerprint_resume_failed: jobs.fingerprint.resume_failed
+jobs_fingerprint_resume_retry_error: jobs.fingerprint.resume_retry_error
 jobs_history_retention_currently_idle: jobs.history_retention.currently_idle
 jobs_history_retention_currently_paused: jobs.history_retention.currently_paused
 jobs_history_retention_currently_running: jobs.history_retention.currently_running

pkg/sql/fingerprint/BUILD.bazel

@@ -0,0 +1,68 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "fingerprint",
+    srcs = [
+        "fingerprint_job.go",
+        "gatherer.go",
+    ],
+    importpath = "github.com/cockroachdb/cockroach/pkg/sql/fingerprint",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/ccl/kvccl/kvfollowerreadsccl",
+        "//pkg/jobs",
+        "//pkg/jobs/joberror",
+        "//pkg/jobs/jobfrontier",
+        "//pkg/jobs/jobspb",
+        "//pkg/kv",
+        "//pkg/kv/kvpb",
+        "//pkg/kv/kvserver/protectedts",
+        "//pkg/roachpb",
+        "//pkg/settings/cluster",
+        "//pkg/sql",
+        "//pkg/sql/isql",
+        "//pkg/util/admission/admissionpb",
+        "//pkg/util/ctxgroup",
+        "//pkg/util/hlc",
+        "//pkg/util/log",
+        "//pkg/util/retry",
+        "//pkg/util/span",
+        "//pkg/util/syncutil",
+        "//pkg/util/timeutil",
+        "//pkg/util/tracing",
+        "//pkg/util/tracing/tracingpb",
+        "//pkg/util/uuid",
+        "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_redact//:redact",
+    ],
+)
+
+go_test(
+    name = "fingerprint_test",
+    srcs = [
+        "fingerprint_job_test.go",
+        "gatherer_test.go",
+        "main_test.go",
+    ],
+    embed = [":fingerprint"],
+    deps = [
+        "//pkg/base",
+        "//pkg/jobs",
+        "//pkg/jobs/jobspb",
+        "//pkg/roachpb",
+        "//pkg/security/securityassets",
+        "//pkg/security/securitytest",
+        "//pkg/security/username",
+        "//pkg/server",
+        "//pkg/sql",
+        "//pkg/testutils/jobutils",
+        "//pkg/testutils/serverutils",
+        "//pkg/testutils/sqlutils",
+        "//pkg/testutils/testcluster",
+        "//pkg/util/hlc",
+        "//pkg/util/leaktest",
+        "//pkg/util/log",
+        "//pkg/util/span",
+        "@com_github_stretchr_testify//require",
+    ],
+)