sql/fingerprint: add new job fingerprint spans job

This just adds the job itself; it is currently unused outside of its
direct tests included here (i.e. there is no code path to create these
jobs wired up yet).

This new job computes the kv fingerprint of the configured spans using
the KV ExportRequest fingerprint mechanism, and stores progress and the
result in the job's job_info rows.

This KV fingerprint currently only fingerprints as of a given timestamp;
it does not fingerprint all revisions prior to that timestamp. This
means it does not need to support representing range deletions in its
fingerprint, as those are not directly observed: you either observe the
keys they deleted if you read prior to them, or don't observe them if
you read after the deletion and reflect those keys or their absense in
the fingerprint accordingly, but do not need to repesent the deletion
itself in the fingerprint. This substantially simplifies the collection
of the fingerprint from separate ExportRequests, as we do not need to
accumulate and defragment partial range deletions across requests.

The KV mode added here is useful for comparing the result of any sort of
byte-for-byte copying of data, such as done by a RESTOTRE or PCR. A
future "SQL" fingerprint implementation could be an alternative to the
KV fingerprinter used here currently, which would use SQL-level scans
and schema-aware datum decoding to compute fingerprints over decoded,
logical values. While more expensive, this can be useful for cases where
logically eqiuivalent data may have different on-disk representations
(e.g. composite types), making such a mode ideal for comparing data
copied by something like LDR.

Release note: none.
Epic: none.

Author: dt

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/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

@@ -2,39 +2,68 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "fingerprint",
-    srcs = ["gatherer.go"],
+    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",
         "//pkg/testutils/serverutils",
+        "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
+        "//pkg/util/log",
         "//pkg/util/span",
+        "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
     ],
 )