sql/inspect: add job metrics for INSPECT jobs

Implements three metrics for INSPECT jobs:
- jobs.inspect.runs: Total number of INSPECT job executions
- jobs.inspect.runs_with_issues: Count of jobs that found
  inconsistencies
- jobs.inspect.issues_found: Aggregate count of issues found across all
  processors

Closes #153884

Epic: CRDB-30356
Release note: none

Author: spilchen

docs/generated/metrics/metrics.yaml

@@ -5290,6 +5290,14 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.issues_found
+      exported_name: jobs_inspect_issues_found
+      description: Total count of issues found by INSPECT jobs
+      y_axis_label: Issues
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: jobs.inspect.protected_age_sec
       exported_name: jobs_inspect_protected_age_sec
       labeled_name: 'jobs.protected_age_sec{type: inspect}'
@@ -5335,6 +5343,22 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.runs
+      exported_name: jobs_inspect_runs
+      description: Number of INSPECT jobs executed
+      y_axis_label: Jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.runs_with_issues
+      exported_name: jobs_inspect_runs_with_issues
+      description: Number of INSPECT jobs that found at least one issue
+      y_axis_label: Jobs
+      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/jobs/metrics.go

@@ -42,6 +42,7 @@ type Metrics struct {
 	Changefeed   metric.Struct
 	StreamIngest metric.Struct
 	Backup       metric.Struct
+	Inspect      metric.Struct
 
 	// AdoptIterations counts the number of adopt loops executed by Registry.
 	AdoptIterations *metric.Counter
@@ -402,6 +403,9 @@ func (m *Metrics) init(histogramWindowInterval time.Duration, lookup *cidr.Looku
 	if MakeBackupMetricsHook != nil {
 		m.Backup = MakeBackupMetricsHook(histogramWindowInterval)
 	}
+	if MakeInspectMetricsHook != nil {
+		m.Inspect = MakeInspectMetricsHook(histogramWindowInterval)
+	}
 
 	m.AdoptIterations = metric.NewCounter(metaAdoptIterations)
 	m.ClaimedJobs = metric.NewCounter(metaClaimedJobs)
@@ -482,6 +486,9 @@ var MakeRowLevelTTLMetricsHook func(time.Duration) metric.Struct
 // MakeBackupMetricsHook allows for registration of backup metrics.
 var MakeBackupMetricsHook func(time.Duration) metric.Struct
 
+// MakeInspectMetricsHook allows for registration of inspect metrics.
+var MakeInspectMetricsHook func(time.Duration) metric.Struct
+
 // JobTelemetryMetrics is a telemetry metrics for individual job types.
 type JobTelemetryMetrics struct {
 	Successful telemetry.Counter

pkg/sql/inspect/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "index_consistency_check.go",
         "inspect_job.go",
         "inspect_logger.go",
+        "inspect_metrics.go",
         "inspect_processor.go",
         "issue.go",
         "log_sink.go",
@@ -48,6 +49,7 @@ go_library(
         "//pkg/util/ctxgroup",
         "//pkg/util/hlc",
         "//pkg/util/log",
+        "//pkg/util/metric",
         "//pkg/util/protoutil",
         "//pkg/util/span",
         "//pkg/util/syncutil",
@@ -64,6 +66,7 @@ go_test(
     srcs = [
         "index_consistency_check_test.go",
         "inspect_job_test.go",
+        "inspect_metrics_test.go",
         "inspect_processor_test.go",
         "issue_test.go",
         "main_test.go",

pkg/sql/inspect/inspect_job.go

@@ -38,6 +38,8 @@ func (c *inspectResumer) Resume(ctx context.Context, execCtx interface{}) error
 	jobExecCtx := execCtx.(sql.JobExecContext)
 	execCfg := jobExecCtx.ExecCfg()
 
+	execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics).Runs.Inc(1)
+
 	if err := c.maybeRunOnJobStartHook(execCfg); err != nil {
 		return err
 	}
@@ -93,6 +95,11 @@ func (c *inspectResumer) OnFailOrCancel(
 	jobExecCtx := execCtx.(sql.JobExecContext)
 	execCfg := jobExecCtx.ExecCfg()
 	c.maybeCleanupProtectedTimestamp(ctx, execCfg)
+
+	// Record RunsWithIssues metric if the job failed due to finding inconsistencies.
+	if jobErr != nil && errors.Is(jobErr, errInspectFoundInconsistencies) {
+		execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics).RunsWithIssues.Inc(1)
+	}
 	return nil
 }
 

pkg/sql/inspect/inspect_logger.go

@@ -7,7 +7,9 @@ package inspect
 
 import (
 	"context"
+	"sync/atomic"
 
+	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/errors"
 )
 
@@ -45,3 +47,23 @@ func (l inspectLoggers) hasIssues() bool {
 	}
 	return false
 }
+
+// metricsLogger increments metrics when issues are logged.
+type metricsLogger struct {
+	foundIssue     atomic.Bool
+	issuesFoundCtr *metric.Counter
+}
+
+var _ inspectLogger = &metricsLogger{}
+
+// logIssue implements the inspectLogger interface.
+func (m *metricsLogger) logIssue(ctx context.Context, issue *inspectIssue) error {
+	m.foundIssue.Store(true)
+	m.issuesFoundCtr.Inc(1)
+	return nil
+}
+
+// hasIssues implements the inspectLogger interface.
+func (m *metricsLogger) hasIssues() bool {
+	return m.foundIssue.Load()
+}

pkg/sql/inspect/inspect_metrics.go

@@ -0,0 +1,59 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package inspect
+
+import (
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/util/metric"
+)
+
+// InspectMetrics holds metrics for INSPECT job operations.
+type InspectMetrics struct {
+	Runs           *metric.Counter
+	RunsWithIssues *metric.Counter
+	IssuesFound    *metric.Counter
+}
+
+var _ metric.Struct = (*InspectMetrics)(nil)
+
+// MetricStruct implements the metric.Struct interface.
+func (InspectMetrics) MetricStruct() {}
+
+var (
+	metaInspectRuns = metric.Metadata{
+		Name:        "jobs.inspect.runs",
+		Help:        "Number of INSPECT jobs executed",
+		Measurement: "Jobs",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaInspectRunsWithIssues = metric.Metadata{
+		Name:        "jobs.inspect.runs_with_issues",
+		Help:        "Number of INSPECT jobs that found at least one issue",
+		Measurement: "Jobs",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaInspectIssuesFound = metric.Metadata{
+		Name:        "jobs.inspect.issues_found",
+		Help:        "Total count of issues found by INSPECT jobs",
+		Measurement: "Issues",
+		Unit:        metric.Unit_COUNT,
+	}
+)
+
+// MakeInspectMetrics instantiates the metrics for INSPECT jobs.
+func MakeInspectMetrics(histogramWindow time.Duration) metric.Struct {
+	return &InspectMetrics{
+		Runs:           metric.NewCounter(metaInspectRuns),
+		RunsWithIssues: metric.NewCounter(metaInspectRunsWithIssues),
+		IssuesFound:    metric.NewCounter(metaInspectIssuesFound),
+	}
+}
+
+func init() {
+	jobs.MakeInspectMetricsHook = MakeInspectMetrics
+}

pkg/sql/inspect/inspect_metrics_test.go

@@ -0,0 +1,101 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package inspect
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/require"
+)
+
+// TestInspectMetrics verifies that INSPECT job metrics are correctly updated
+// when jobs run.
+func TestInspectMetrics(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{
+		DefaultTestTenant: base.TestIsSpecificToStorageLayerAndNeedsASystemTenant,
+	})
+	defer s.Stopper().Stop(ctx)
+
+	runner := sqlutils.MakeSQLRunner(db)
+	runner.Exec(t, `
+		CREATE DATABASE db;
+		SET enable_scrub_job = true;
+		CREATE TABLE db.t (
+			id INT PRIMARY KEY,
+			val INT
+		);
+		CREATE INDEX i1 on db.t (val);
+		INSERT INTO db.t VALUES (1, 2), (2, 3);
+	`)
+
+	execCfg := s.ApplicationLayer().ExecutorConfig().(sql.ExecutorConfig)
+	metrics := execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics)
+
+	initialRuns := metrics.Runs.Count()
+	initialRunsWithIssues := metrics.RunsWithIssues.Count()
+	initialIssuesFound := metrics.IssuesFound.Count()
+
+	// First run: no corruption, should succeed without issues
+	runner.Exec(t, "EXPERIMENTAL SCRUB 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")
+
+	// Create corruption: delete a secondary index entry for row (1, 2)
+	// This creates a "missing_secondary_index_entry" issue - the primary key exists
+	// but the corresponding secondary index entry is missing.
+	kvDB := s.DB()
+	codec := s.Codec()
+	tableDesc := desctestutils.TestingGetPublicTableDescriptor(kvDB, codec, "db", "t")
+	secIndex := tableDesc.PublicNonPrimaryIndexes()[0] // i1 index on (val)
+	row := []tree.Datum{
+		tree.NewDInt(1), // id
+		tree.NewDInt(2), // val
+	}
+	err := deleteSecondaryIndexEntry(ctx, row, kvDB, tableDesc, secIndex)
+	require.NoError(t, err)
+
+	// Second run: with corruption, should detect the missing index entry.
+	_, err = db.Exec("EXPERIMENTAL SCRUB TABLE db.t")
+	require.Error(t, err, "INSPECT should fail when corruption is detected")
+	require.Contains(t, err.Error(), "INSPECT found inconsistencies")
+	require.Equal(t, initialRuns+2, metrics.Runs.Count(),
+		"Runs counter should increment for each job execution")
+	require.Equal(t, initialRunsWithIssues+1, metrics.RunsWithIssues.Count(),
+		"RunsWithIssues should increment when issues are found")
+	require.Equal(t, initialIssuesFound+1, metrics.IssuesFound.Count(),
+		"IssuesFound should increment when issues are detected")
+
+	// Third run: on a different clean table to verify RunsWithIssues doesn't increment again.
+	runner.Exec(t, `
+		CREATE TABLE db.t2 (
+			id INT PRIMARY KEY,
+			val INT
+		);
+		CREATE INDEX i2 on db.t2 (val);
+		INSERT INTO db.t2 VALUES (1, 2), (2, 3);
+	`)
+	runner.Exec(t, "EXPERIMENTAL SCRUB TABLE db.t2")
+	require.Equal(t, initialRuns+3, metrics.Runs.Count(),
+		"Runs counter should increment for third job execution")
+	require.Equal(t, initialRunsWithIssues+1, metrics.RunsWithIssues.Count(),
+		"RunsWithIssues should NOT increment for successful job")
+	require.Equal(t, initialIssuesFound+1, metrics.IssuesFound.Count(),
+		"IssuesFound should NOT increment for successful job")
+}

pkg/sql/inspect/inspect_processor.go

@@ -29,6 +29,10 @@ import (
 )
 
 var (
+	// errInspectFoundInconsistencies is a sentinel error used to mark errors
+	// returned when INSPECT jobs find data inconsistencies.
+	errInspectFoundInconsistencies = pgerror.New(pgcode.DataException, "INSPECT found inconsistencies")
+
 	processorConcurrencyOverride = settings.RegisterIntSetting(
 		settings.ApplicationLevel,
 		"sql.inspect.processor_concurrency",
@@ -176,7 +180,7 @@ func (p *inspectProcessor) runInspect(ctx context.Context, output execinfra.RowR
 
 	log.Dev.Infof(ctx, "INSPECT processor completed processorID=%d issuesFound=%t", p.processorID, p.logger.hasIssues())
 	if p.logger.hasIssues() {
-		return pgerror.Newf(pgcode.DataException, "INSPECT found inconsistencies")
+		return errInspectFoundInconsistencies
 	}
 	return nil
 }
@@ -194,15 +198,21 @@ func getProcessorConcurrency(flowCtx *execinfra.FlowCtx) int {
 
 // getInspectLogger returns a logger for the inspect processor.
 func getInspectLogger(flowCtx *execinfra.FlowCtx, jobID jobspb.JobID) inspectLogger {
+	execCfg := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig)
+	metrics := execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics)
+
 	loggers := inspectLoggers{
 		&logSink{},
 		&tableSink{
 			db:    flowCtx.Cfg.DB,
 			jobID: jobID,
 		},
+		&metricsLogger{
+			issuesFoundCtr: metrics.IssuesFound,
+		},
 	}
 
-	knobs := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig).InspectTestingKnobs
+	knobs := execCfg.InspectTestingKnobs
 	if knobs != nil && knobs.InspectIssueLogger != nil {
 		loggers = append(loggers, knobs.InspectIssueLogger.(inspectLogger))
 	}