Merge #148471

148471: changefeedccl: record metrics for pts management r=asg0451,andyyang890 a=aerfrei

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.create_pts_hist_nanos and the
speed of advancing the PTS records when the frontier
progresses in changefeed.manage_pts_hist_nanos.

Epic: none
Fixes: #147779
Informs: #147780

Release note: This change adds new metrics:
changefeed.create_pts_hist_nanos
and changefeed.manage_pts_hist_nanos to measure the
performance of managing protected ts records.

Co-authored-by: Aerin Freilich <[email protected]>

Author: craig[bot]

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

@@ -1928,6 +1928,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
@@ -1983,6 +1995,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

@@ -11497,14 +11497,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()
@@ -11549,6 +11563,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) {
@@ -11562,6 +11581,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
 }