Merge #156384

156384: sql/inspect: record spans processed metric r=spilchen a=spilchen

This add a new INSPECT metric to track a counter of the number spans completed. This will be useful to see the rate of progress for INSPECT.

Closes #155484
Epic: CRDB-55075
Release note: none

Co-authored-by: Matt Spilchen <[email protected]>

Author: craig[bot]

docs/generated/metrics/metrics.yaml

@@ -5137,6 +5137,14 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.spans_processed
+      exported_name: jobs_inspect_spans_processed
+      description: Number of spans processed by INSPECT jobs
+      y_axis_label: Spans
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: jobs.key_visualizer.currently_idle
       exported_name: jobs_key_visualizer_currently_idle
       labeled_name: 'jobs{type: key_visualizer, status: currently_idle}'

pkg/roachprod/agents/opentelemetry/cockroachdb_metrics.go

@@ -902,6 +902,7 @@ var cockroachdbMetrics = map[string]string{
     "jobs_inspect_resume_retry_error": "jobs.inspect.resume_retry_error",
     "jobs_inspect_runs": "jobs.inspect.runs",
     "jobs_inspect_runs_with_issues": "jobs.inspect.runs_with_issues",
+    "jobs_inspect_spans_processed": "jobs.inspect.spans_processed",
     "jobs_key_visualizer_currently_idle": "jobs.key_visualizer.currently_idle",
     "jobs_key_visualizer_currently_paused": "jobs.key_visualizer.currently_paused",
     "jobs_key_visualizer_currently_running": "jobs.key_visualizer.currently_running",

pkg/sql/inspect/BUILD.bazel

@@ -85,6 +85,7 @@ go_test(
         "//build/toolchains:is_heavy": {"test.Pool": "large"},
         "//conditions:default": {"test.Pool": "default"},
     }),
+    shard_count = 4,
     deps = [
         "//pkg/base",
         "//pkg/jobs",

pkg/sql/inspect/inspect_metrics.go

@@ -17,6 +17,7 @@ type InspectMetrics struct {
 	Runs           *metric.Counter
 	RunsWithIssues *metric.Counter
 	IssuesFound    *metric.Counter
+	SpansProcessed *metric.Counter
 }
 
 var _ metric.Struct = (*InspectMetrics)(nil)
@@ -43,6 +44,12 @@ var (
 		Measurement: "Issues",
 		Unit:        metric.Unit_COUNT,
 	}
+	metaInspectSpansProcessed = metric.Metadata{
+		Name:        "jobs.inspect.spans_processed",
+		Help:        "Number of spans processed by INSPECT jobs",
+		Measurement: "Spans",
+		Unit:        metric.Unit_COUNT,
+	}
 )
 
 // MakeInspectMetrics instantiates the metrics for INSPECT jobs.
@@ -51,6 +58,7 @@ func MakeInspectMetrics(histogramWindow time.Duration) metric.Struct {
 		Runs:           metric.NewCounter(metaInspectRuns),
 		RunsWithIssues: metric.NewCounter(metaInspectRunsWithIssues),
 		IssuesFound:    metric.NewCounter(metaInspectIssuesFound),
+		SpansProcessed: metric.NewCounter(metaInspectSpansProcessed),
 	}
 }
 

pkg/sql/inspect/inspect_metrics_test.go

@@ -51,12 +51,15 @@ func TestInspectMetrics(t *testing.T) {
 	initialRuns := metrics.Runs.Count()
 	initialRunsWithIssues := metrics.RunsWithIssues.Count()
 	initialIssuesFound := metrics.IssuesFound.Count()
+	initialSpansProcessed := metrics.SpansProcessed.Count()
 
 	// First run: no corruption, should succeed without issues
 	runner.Exec(t, "INSPECT TABLE db.t")
 	require.Equal(t, initialRuns+1, metrics.Runs.Count(), "Runs counter should increment")
 	require.Equal(t, initialRunsWithIssues, metrics.RunsWithIssues.Count(), "RunsWithIssues should not increment")
 	require.Equal(t, initialIssuesFound, metrics.IssuesFound.Count(), "IssuesFound should not increment")
+	require.Equal(t, initialSpansProcessed+1, metrics.SpansProcessed.Count(),
+		"SpansProcessed should increment by 1 (one secondary index i1)")
 
 	// Create corruption: delete a secondary index entry for row (1, 2)
 	// This creates a "missing_secondary_index_entry" issue - the primary key exists
@@ -86,6 +89,8 @@ func TestInspectMetrics(t *testing.T) {
 		"RunsWithIssues should increment when issues are found")
 	require.Equal(t, initialIssuesFound+1, metrics.IssuesFound.Count(),
 		"IssuesFound should increment when issues are detected")
+	require.Equal(t, initialSpansProcessed+2, metrics.SpansProcessed.Count(),
+		"SpansProcessed should increment by 2 total (1 span per INSPECT × 2 runs)")
 
 	// Third run: on a different clean table to verify RunsWithIssues doesn't increment again.
 	runner.Exec(t, `
@@ -103,4 +108,6 @@ func TestInspectMetrics(t *testing.T) {
 		"RunsWithIssues should NOT increment for successful job")
 	require.Equal(t, initialIssuesFound+1, metrics.IssuesFound.Count(),
 		"IssuesFound should NOT increment for successful job")
+	require.Equal(t, initialSpansProcessed+3, metrics.SpansProcessed.Count(),
+		"SpansProcessed should increment by 3 total (1 span per INSPECT × 3 runs)")
 }

pkg/sql/inspect/inspect_processor.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
@@ -83,16 +84,17 @@ func getInspectQoS(sv *settings.Values) sessiondatapb.QoSLevel {
 type inspectCheckFactory func(asOf hlc.Timestamp) inspectCheck
 
 type inspectProcessor struct {
-	processorID    int32
-	flowCtx        *execinfra.FlowCtx
-	spec           execinfrapb.InspectSpec
-	cfg            *execinfra.ServerConfig
-	checkFactories []inspectCheckFactory
-	spanSrc        spanSource
-	loggerBundle   *inspectLoggerBundle
-	concurrency    int
-	clock          *hlc.Clock
-	mu             struct {
+	processorID       int32
+	flowCtx           *execinfra.FlowCtx
+	spec              execinfrapb.InspectSpec
+	cfg               *execinfra.ServerConfig
+	checkFactories    []inspectCheckFactory
+	spanSrc           spanSource
+	loggerBundle      *inspectLoggerBundle
+	concurrency       int
+	clock             *hlc.Clock
+	spansProcessedCtr *metric.Counter
+	mu                struct {
 		// Guards calls to output.Push because DistSQLReceiver.Push is not
 		// concurrency safe and progress metadata can be emitted from multiple
 		// worker goroutines.
@@ -365,6 +367,9 @@ func (p *inspectProcessor) getTimestampForSpan() hlc.Timestamp {
 func (p *inspectProcessor) sendSpanCompletionProgress(
 	ctx context.Context, output execinfra.RowReceiver, completedSpan roachpb.Span, finished bool,
 ) error {
+	if p.spansProcessedCtr != nil {
+		p.spansProcessedCtr.Inc(1)
+	}
 	progressMsg := &jobspb.InspectProcessorProgress{
 		ChecksCompleted: 0, // No additional checks completed, just marking span done
 		Finished:        finished,
@@ -414,17 +419,23 @@ func newInspectProcessor(
 	if err != nil {
 		return nil, err
 	}
+	var spansProcessedCtr *metric.Counter
+	if execCfg, ok := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig); ok {
+		inspectMetrics := execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics)
+		spansProcessedCtr = inspectMetrics.SpansProcessed
+	}
 
 	return &inspectProcessor{
-		spec:           spec,
-		processorID:    processorID,
-		flowCtx:        flowCtx,
-		checkFactories: checkFactories,
-		cfg:            flowCtx.Cfg,
-		spanSrc:        newSliceSpanSource(spec.Spans),
-		loggerBundle:   getInspectLogger(flowCtx, spec.JobID),
-		concurrency:    getProcessorConcurrency(flowCtx),
-		clock:          flowCtx.Cfg.DB.KV().Clock(),
+		spec:              spec,
+		processorID:       processorID,
+		flowCtx:           flowCtx,
+		checkFactories:    checkFactories,
+		cfg:               flowCtx.Cfg,
+		spanSrc:           newSliceSpanSource(spec.Spans),
+		loggerBundle:      getInspectLogger(flowCtx, spec.JobID),
+		concurrency:       getProcessorConcurrency(flowCtx),
+		clock:             flowCtx.Cfg.DB.KV().Clock(),
+		spansProcessedCtr: spansProcessedCtr,
 	}, nil
 }