changefeedccl: record metrics for pts management

Before this commit, we did not measure the performance
of protected timestamp creation or management. Since
DB-level changefeeds will potentially create many more
PTS records per job (one per watched table), we will
use these metrics to establish a baseline of how long
these operations take and to alert us to issues if
the new DB-level feeds overwhelm the PTS record table.

We will measure the speed of both creating the initial
PTS records in changefeed.stage.pts.create.latency and the
speed of advancing the PTS records when the frontier
progresses in changefeed.stage.pts.manage.latency.

We also include changefeed.stage.pts.manage_error.latency
for cases where the pts management errors.

Epic: none
Fixes: #147779
Informs: #147780

Release note (general): This change adds new metrics:
changefeed.stage.pts.create.latency,
changefeed.stage.pts.manage.latency,
changefeed.stage.pts.manage_error.latency,
to measure the performance of managing protected ts records.

Author: aerfrei

docs/generated/metrics/metrics.yaml

@@ -1211,7 +1211,7 @@ layers:
     - name: changefeed.checkpoint_hist_nanos
       exported_name: changefeed_checkpoint_hist_nanos
       description: Time spent checkpointing changefeed progress
-      y_axis_label: Changefeeds
+      y_axis_label: Nanoseconds
       type: HISTOGRAM
       unit: NANOSECONDS
       aggregation: AVG
@@ -1536,6 +1536,30 @@ layers:
       unit: NANOSECONDS
       aggregation: AVG
       derivative: NONE
+    - name: changefeed.stage.pts.create.latency
+      exported_name: changefeed_stage_pts_create_latency
+      description: 'Latency of the changefeed stage: Time spent creating protected timestamp records on changefeed creation'
+      y_axis_label: Latency
+      type: HISTOGRAM
+      unit: NANOSECONDS
+      aggregation: AVG
+      derivative: NONE
+    - name: changefeed.stage.pts.manage.latency
+      exported_name: changefeed_stage_pts_manage_latency
+      description: 'Latency of the changefeed stage: Time spent successfully managing protected timestamp records on highwater advance, including time spent creating new protected timestamps when needed'
+      y_axis_label: Latency
+      type: HISTOGRAM
+      unit: NANOSECONDS
+      aggregation: AVG
+      derivative: NONE
+    - name: changefeed.stage.pts.manage_error.latency
+      exported_name: changefeed_stage_pts_manage_error_latency
+      description: 'Latency of the changefeed stage: Time spent managing protected timestamp when we eventually error'
+      y_axis_label: Latency
+      type: HISTOGRAM
+      unit: NANOSECONDS
+      aggregation: AVG
+      derivative: NONE
     - name: changefeed.stage.rangefeed_buffer_checkpoint.latency
       exported_name: changefeed_stage_rangefeed_buffer_checkpoint_latency
       description: 'Latency of the changefeed stage: buffering rangefeed checkpoint events'

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1919,6 +1919,18 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 		return false, nil
 	}
 
+	recordPTSMetricsTime := cf.sliMetrics.Timers.PTSManage.Start()
+	recordPTSMetricsErrorTime := cf.sliMetrics.Timers.PTSManageError.Start()
+	defer func() {
+		if err != nil {
+			recordPTSMetricsErrorTime()
+			return
+		}
+		if updated {
+			recordPTSMetricsTime()
+		}
+	}()
+
 	pts := cf.FlowCtx.Cfg.ProtectedTimestampProvider.WithTxn(txn)
 
 	// Create / advance the protected timestamp record to the highwater mark
@@ -1974,6 +1986,10 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 		return false, nil
 	}
 
+	if cf.knobs.ManagePTSError != nil {
+		return false, cf.knobs.ManagePTSError()
+	}
+
 	log.VEventf(ctx, 2, "updating protected timestamp %v at %v", progress.ProtectedTimestampRecord, highWater)
 	return true, pts.UpdateTimestamp(ctx, progress.ProtectedTimestampRecord, highWater)
 }

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -302,6 +302,15 @@ func changefeedPlanHook(
 		jobID := p.ExecCfg().JobRegistry.MakeJobID()
 		jr.JobID = jobID
 		{
+			metrics := p.ExecCfg().JobRegistry.MetricsStruct().Changefeed.(*Metrics)
+			scope, _ := opts.GetMetricScope()
+			sliMetrics, err := metrics.getSLIMetrics(scope)
+			if err != nil {
+				return err
+			}
+
+			recordPTSMetricsTime := sliMetrics.Timers.PTSCreate.Start()
+
 			var ptr *ptpb.Record
 			codec := p.ExecCfg().Codec
 			ptr = createProtectedTimestampRecord(
@@ -358,6 +367,7 @@ func changefeedPlanHook(
 				}
 				return err
 			}
+			recordPTSMetricsTime()
 		}
 
 		// Start the job.

pkg/ccl/changefeedccl/changefeed_test.go

@@ -11474,14 +11474,28 @@ func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 
 		sqlDB.Exec(t, `CREATE TABLE foo (id INT)`)
 
+		registry := s.Server.JobRegistry().(*jobs.Registry)
+		metrics := registry.MetricsStruct().Changefeed.(*Metrics)
+		createPtsCount, _ := metrics.AggMetrics.Timers.PTSCreate.WindowedSnapshot().Total()
+		managePtsCount, _ := metrics.AggMetrics.Timers.PTSManage.WindowedSnapshot().Total()
+		managePTSErrorCount, _ := metrics.AggMetrics.Timers.PTSManageError.WindowedSnapshot().Total()
+		require.Equal(t, int64(0), createPtsCount)
+		require.Equal(t, int64(0), managePtsCount)
+		require.Equal(t, int64(0), managePTSErrorCount)
+
 		createStmt := `CREATE CHANGEFEED FOR foo WITH resolved='10ms', no_initial_scan`
 		testFeed := feed(t, f, createStmt)
 		defer closeFeed(t, testFeed)
 
+		createPtsCount, _ = metrics.AggMetrics.Timers.PTSCreate.WindowedSnapshot().Total()
+		managePtsCount, _ = metrics.AggMetrics.Timers.PTSManage.WindowedSnapshot().Total()
+		require.Equal(t, int64(1), createPtsCount)
+		require.Equal(t, int64(0), managePtsCount)
+
 		eFeed, ok := testFeed.(cdctest.EnterpriseTestFeed)
 		require.True(t, ok)
 
-		// Wait for the changefeed to checkpoint.
+		// Wait for the changefeed to checkpoint and update PTS at least once.
 		var lastHWM hlc.Timestamp
 		checkHWM := func() error {
 			hwm, err := eFeed.HighWaterMark()
@@ -11526,6 +11540,11 @@ func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 		sqlDB.QueryRow(t, ptsQry).Scan(&ts2)
 		require.NoError(t, err)
 		require.Less(t, ts, ts2)
+
+		managePtsCount, _ = metrics.AggMetrics.Timers.PTSManage.WindowedSnapshot().Total()
+		managePTSErrorCount, _ = metrics.AggMetrics.Timers.PTSManageError.WindowedSnapshot().Total()
+		require.GreaterOrEqual(t, managePtsCount, int64(2))
+		require.Equal(t, int64(0), managePTSErrorCount)
 	}
 
 	withTxnRetries := withArgsFn(func(args *base.TestServerArgs) {
@@ -11539,6 +11558,66 @@ func TestChangefeedProtectedTimestampUpdate(t *testing.T) {
 	cdcTest(t, testFn, feedTestForceSink("kafka"), withTxnRetries)
 }
 
+func TestChangefeedProtectedTimestampUpdateError(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		// Checkpoint and trigger potential protected timestamp updates frequently.
+		// Make the protected timestamp lag long enough that it shouldn't be
+		// immediately updated after a restart.
+		changefeedbase.SpanCheckpointInterval.Override(
+			context.Background(), &s.Server.ClusterSettings().SV, 10*time.Millisecond)
+		changefeedbase.ProtectTimestampInterval.Override(
+			context.Background(), &s.Server.ClusterSettings().SV, 10*time.Millisecond)
+		changefeedbase.ProtectTimestampLag.Override(
+			context.Background(), &s.Server.ClusterSettings().SV, 10*time.Hour)
+
+		sqlDB.Exec(t, `CREATE TABLE foo (id INT)`)
+
+		registry := s.Server.JobRegistry().(*jobs.Registry)
+		metrics := registry.MetricsStruct().Changefeed.(*Metrics)
+		createPtsCount, _ := metrics.AggMetrics.Timers.PTSCreate.WindowedSnapshot().Total()
+		managePtsCount, _ := metrics.AggMetrics.Timers.PTSManage.WindowedSnapshot().Total()
+		managePTSErrorCount, _ := metrics.AggMetrics.Timers.PTSManageError.WindowedSnapshot().Total()
+		require.Equal(t, int64(0), createPtsCount)
+		require.Equal(t, int64(0), managePtsCount)
+		require.Equal(t, int64(0), managePTSErrorCount)
+
+		knobs := s.TestingKnobs.
+			DistSQL.(*execinfra.TestingKnobs).
+			Changefeed.(*TestingKnobs)
+
+		knobs.ManagePTSError = func() error {
+			return errors.New("test error")
+		}
+
+		createStmt := `CREATE CHANGEFEED FOR foo WITH resolved='10ms', no_initial_scan`
+		testFeed := feed(t, f, createStmt)
+		defer closeFeed(t, testFeed)
+
+		createPtsCount, _ = metrics.AggMetrics.Timers.PTSCreate.WindowedSnapshot().Total()
+		require.Equal(t, int64(1), createPtsCount)
+		managePTSErrorCount, _ = metrics.AggMetrics.Timers.PTSManageError.WindowedSnapshot().Total()
+		require.Equal(t, int64(0), managePTSErrorCount)
+
+		// Lower the PTS lag to trigger a PTS update.
+		changefeedbase.ProtectTimestampLag.Override(
+			context.Background(), &s.Server.ClusterSettings().SV, 10*time.Millisecond)
+
+		testutils.SucceedsSoon(t, func() error {
+			managePTSErrorCount, _ = metrics.AggMetrics.Timers.PTSManageError.WindowedSnapshot().Total()
+			if managePTSErrorCount > 0 {
+				fmt.Println("manage protected timestamps test: manage pts error count", managePTSErrorCount)
+				return nil
+			}
+			return errors.New("waiting for manage pts error")
+		})
+	}
+	cdcTest(t, testFn, feedTestForceSink("kafka"))
+}
+
 func TestCDCQuerySelectSingleRow(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/changefeedccl/metrics.go

@@ -731,7 +731,7 @@ var (
 	metaChangefeedCheckpointHistNanos = metric.Metadata{
 		Name:        "changefeed.checkpoint_hist_nanos",
 		Help:        "Time spent checkpointing changefeed progress",
-		Measurement: "Changefeeds",
+		Measurement: "Nanoseconds",
 		Unit:        metric.Unit_NANOSECONDS,
 	}
 

pkg/ccl/changefeedccl/testing_knobs.go

@@ -90,6 +90,9 @@ type TestingKnobs struct {
 	// its PTS record to include all required targets.
 	PreservePTSTargets func() bool
 
+	// ManagePTSError is used to return an error when managing protected timestamps.
+	ManagePTSError func() error
+
 	// PulsarClientSkipCreation skips creating the sink client when
 	// dialing.
 	PulsarClientSkipCreation bool

pkg/ccl/changefeedccl/timers/timers.go

@@ -23,6 +23,9 @@ type Timers struct {
 	KVFeedBuffer              *aggmetric.AggHistogram
 	RangefeedBufferValue      *aggmetric.AggHistogram
 	RangefeedBufferCheckpoint *aggmetric.AggHistogram
+	PTSManage                 *aggmetric.AggHistogram
+	PTSManageError            *aggmetric.AggHistogram
+	PTSCreate                 *aggmetric.AggHistogram
 }
 
 func (*Timers) MetricStruct() {}
@@ -54,6 +57,9 @@ func New(histogramWindow time.Duration) *Timers {
 		KVFeedBuffer:              b.Histogram(histogramOptsFor("changefeed.stage.kv_feed_buffer.latency", "Latency of the changefeed stage: waiting to buffer kv events")),
 		RangefeedBufferValue:      b.Histogram(histogramOptsFor("changefeed.stage.rangefeed_buffer_value.latency", "Latency of the changefeed stage: buffering rangefeed value events")),
 		RangefeedBufferCheckpoint: b.Histogram(histogramOptsFor("changefeed.stage.rangefeed_buffer_checkpoint.latency", "Latency of the changefeed stage: buffering rangefeed checkpoint events")),
+		PTSManage:                 b.Histogram(histogramOptsFor("changefeed.stage.pts.manage.latency", "Latency of the changefeed stage: Time spent successfully managing protected timestamp records on highwater advance, including time spent creating new protected timestamps when needed")),
+		PTSManageError:            b.Histogram(histogramOptsFor("changefeed.stage.pts.manage_error.latency", "Latency of the changefeed stage: Time spent managing protected timestamp when we eventually error")),
+		PTSCreate:                 b.Histogram(histogramOptsFor("changefeed.stage.pts.create.latency", "Latency of the changefeed stage: Time spent creating protected timestamp records on changefeed creation")),
 	}
 }
 
@@ -67,6 +73,9 @@ func (ts *Timers) GetOrCreateScopedTimers(scope string) *ScopedTimers {
 		KVFeedBuffer:              &timer{ts.KVFeedBuffer.AddChild(scope)},
 		RangefeedBufferValue:      &timer{ts.RangefeedBufferValue.AddChild(scope)},
 		RangefeedBufferCheckpoint: &timer{ts.RangefeedBufferCheckpoint.AddChild(scope)},
+		PTSManage:                 &timer{ts.PTSManage.AddChild(scope)},
+		PTSManageError:            &timer{ts.PTSManageError.AddChild(scope)},
+		PTSCreate:                 &timer{ts.PTSCreate.AddChild(scope)},
 	}
 }
 
@@ -77,6 +86,9 @@ type ScopedTimers struct {
 	DownstreamClientSend      *timer
 	KVFeedWaitForTableEvent   *timer
 	KVFeedBuffer              *timer
+	PTSCreate                 *timer
+	PTSManage                 *timer
+	PTSManageError            *timer
 	RangefeedBufferValue      *timer
 	RangefeedBufferCheckpoint *timer
 }