streamingest: add a replication lag metric

Add a metric to track the lag of the replication frontier, in seconds.

Informs: cockroachdb#92959

Epic: CRDB-18752

Release note: None

pkg/ccl/streamingccl/streamingest/metrics.go

@@ -92,6 +92,12 @@ var (
 		Measurement: "Resolved Spans",
 		Unit:        metric.Unit_COUNT,
 	}
+	metaFrontierLagSeconds = metric.Metadata{
+		Name:        "streaming.frontier_lag_seconds",
+		Help:        "Time the replication frontier lags",
+		Measurement: "Seconds",
+		Unit:        metric.Unit_SECONDS,
+	}
 	metaJobProgressUpdates = metric.Metadata{
 		Name:        "streaming.job_progress_updates",
 		Help:        "Total number of updates to the ingestion job progress",
@@ -115,6 +121,7 @@ type Metrics struct {
 	LatestDataCheckpointSpan    *metric.Gauge
 	DataCheckpointSpanCount     *metric.Gauge
 	FrontierCheckpointSpanCount *metric.Gauge
+	FrontierLagSeconds          *metric.GaugeFloat64
 }
 
 // MetricStruct implements the metric.Struct interface.
@@ -139,6 +146,7 @@ func MakeMetrics(histogramWindow time.Duration) metric.Struct {
 		LatestDataCheckpointSpan:    metric.NewGauge(metaLatestDataCheckpointSpan),
 		DataCheckpointSpanCount:     metric.NewGauge(metaDataCheckpointSpanCount),
 		FrontierCheckpointSpanCount: metric.NewGauge(metaFrontierCheckpointSpanCount),
+		FrontierLagSeconds:          metric.NewGaugeFloat64(metaFrontierLagSeconds),
 	}
 	return m
 }

pkg/ccl/streamingccl/streamingest/stream_ingestion_frontier_processor.go

@@ -467,6 +467,7 @@ func (sf *streamIngestionFrontier) maybeUpdatePartitionProgress() error {
 	sf.metrics.JobProgressUpdates.Inc(1)
 	sf.persistedHighWater = f.Frontier()
 	sf.metrics.FrontierCheckpointSpanCount.Update(int64(len(frontierResolvedSpans)))
+	sf.metrics.FrontierLagSeconds.Update(timeutil.Since(sf.persistedHighWater.GoTime()).Seconds())
 
 	return nil
 }

pkg/ts/catalog/chart_catalog.go

@@ -1623,6 +1623,10 @@ var charts = []sectionDescription{
 				Title:   "Frontier Checkpoint Span Count",
 				Metrics: []string{"streaming.frontier_checkpoint_span_count"},
 			},
+			{
+				Title:   "Frontier Lag",
+				Metrics: []string{"streaming.frontier_lag_seconds"},
+			},
 			{
 				Title:   "Job Progress Updates",
 				Metrics: []string{"streaming.job_progress_updates"},