changefeedccl: add progress skew metrics

This patch adds two new changefeed metrics for tracking the max skew
between a changefeed's slowest and fastest span/table. The metrics are
gauge metrics with the names `changefeed.progress_skew.{span,table}`.

Release note (ops change): Two new changefeed metrics for tracking
the max skew between a changefeed's slowest and fastest span/table
have been added. The metrics are gauge metrics with the names
`changefeed.progress_skew.{span,table}`.

Author: andyyang890

docs/generated/metrics/metrics.yaml

@@ -1725,6 +1725,22 @@ layers:
       unit: NANOSECONDS
       aggregation: AVG
       derivative: NONE
+    - name: changefeed.progress_skew.span
+      exported_name: changefeed_progress_skew_span
+      description: The time difference between the fastest and slowest span's resolved timestamp
+      y_axis_label: Nanoseconds
+      type: GAUGE
+      unit: NANOSECONDS
+      aggregation: AVG
+      derivative: NONE
+    - name: changefeed.progress_skew.table
+      exported_name: changefeed_progress_skew_table
+      description: The time difference between the fastest and slowest table's resolved timestamp
+      y_axis_label: Nanoseconds
+      type: GAUGE
+      unit: NANOSECONDS
+      aggregation: AVG
+      derivative: NONE
     - name: changefeed.queue_time_nanos
       exported_name: changefeed_queue_time_nanos
       description: Time KV event spent waiting to be processed

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1736,6 +1736,8 @@ func (cf *changeFrontier) noteAggregatorProgress(ctx context.Context, d rowenc.E
 		}
 	}
 
+	cf.updateProgressSkewMetrics()
+
 	return nil
 }
 
@@ -2263,6 +2265,20 @@ func (cf *changeFrontier) maybeEmitResolved(ctx context.Context, newResolved hlc
 	return nil
 }
 
+// updateProgressSkewMetrics updates the progress skew metrics.
+func (cf *changeFrontier) updateProgressSkewMetrics() {
+	maxSpanTS := cf.frontier.LatestTS()
+	maxTableTS := cf.frontier.Frontier()
+	for _, f := range cf.frontier.Frontiers() {
+		tableTS := f.Frontier()
+		if tableTS.After(maxTableTS) {
+			maxTableTS = tableTS
+		}
+	}
+
+	cf.sliMetrics.setFastestTS(cf.sliMetricsID, maxSpanTS, maxTableTS)
+}
+
 func frontierIsBehind(frontier hlc.Timestamp, sv *settings.Values) bool {
 	if frontier.IsEmpty() {
 		// During backfills we consider ourselves "behind" for the purposes of

pkg/ccl/changefeedccl/changefeed_progress_test.go

@@ -7,6 +7,9 @@ package changefeedccl
 
 import (
 	"context"
+	"fmt"
+	"math"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -17,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
@@ -40,7 +44,8 @@ func TestChangefeedFrontierPersistence(t *testing.T) {
 		ctx := context.Background()
 
 		// Set a short interval for frontier persistence.
-		sqlDB.Exec(t, "SET CLUSTER SETTING changefeed.progress.frontier_persistence.interval = '5s'")
+		changefeedbase.FrontierPersistenceInterval.Override(ctx,
+			&s.Server.ClusterSettings().SV, 5*time.Second)
 
 		// Get frontier persistence metric.
 		registry := s.Server.JobRegistry().(*jobs.Registry)
@@ -179,3 +184,128 @@ RETURNING cluster_logical_timestamp()`).Scan(&tsStr)
 
 	cdcTest(t, testFn, feedTestEnterpriseSinks)
 }
+
+func TestChangefeedProgressSkewMetrics(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testutils.RunTrueAndFalse(t, "per-table tracking", func(t *testing.T, perTableTracking bool) {
+		testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+			sqlDB := sqlutils.MakeSQLRunner(s.DB)
+			ctx := context.Background()
+
+			// Enable/disable per-table tracking.
+			changefeedbase.TrackPerTableProgress.Override(ctx,
+				&s.Server.ClusterSettings().SV, perTableTracking)
+
+			registry := s.Server.JobRegistry().(*jobs.Registry)
+			aggMetrics := registry.MetricsStruct().Changefeed.(*Metrics).AggMetrics
+			const scope = "skew"
+			scopedMetrics, err := aggMetrics.getOrCreateScope(scope)
+			require.NoError(t, err)
+
+			// Progress skew metrics should start at zero.
+			require.Zero(t, aggMetrics.SpanProgressSkew.Value())
+			require.Zero(t, aggMetrics.TableProgressSkew.Value())
+			require.Zero(t, scopedMetrics.SpanProgressSkew.Value())
+			require.Zero(t, scopedMetrics.TableProgressSkew.Value())
+
+			// Create two tables and insert some initial data.
+			sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY)`)
+			sqlDB.Exec(t, `CREATE TABLE bar (b INT PRIMARY KEY)`)
+			sqlDB.Exec(t, `INSERT INTO foo VALUES (1), (2), (3)`)
+			sqlDB.Exec(t, `INSERT INTO bar VALUES (1), (2), (3)`)
+
+			// Set up testing knobs to block all progress updates for bar.
+			var blockBarProgress atomic.Bool
+			blockBarProgress.Store(true)
+			{
+				barTableSpan := desctestutils.
+					TestingGetPublicTableDescriptor(s.Server.DB(), s.Codec, "d", "bar").
+					PrimaryIndexSpan(s.Codec)
+
+				knobs := s.TestingKnobs.
+					DistSQL.(*execinfra.TestingKnobs).
+					Changefeed.(*TestingKnobs)
+
+				knobs.FilterSpanWithMutation = func(rs *jobspb.ResolvedSpan) (bool, error) {
+					if blockBarProgress.Load() && barTableSpan.Contains(rs.Span) {
+						return true, nil
+					}
+					return false, nil
+				}
+			}
+
+			// Create changefeed for both tables with no initial scan.
+			feed := feed(t, f, fmt.Sprintf(`CREATE CHANGEFEED FOR foo, bar
+WITH no_initial_scan, min_checkpoint_frequency='1s', resolved, metrics_label='%s'`, scope))
+			defer closeFeed(t, feed)
+
+			assertSpanSkewInRange := func(start int64, end int64) int64 {
+				var spanSkew int64
+				testutils.SucceedsSoon(t, func() error {
+					spanSkew = aggMetrics.SpanProgressSkew.Value()
+					scopedSpanSkew := scopedMetrics.SpanProgressSkew.Value()
+					if spanSkew != scopedSpanSkew {
+						return errors.Newf("aggregate and scoped span skew don't match")
+					}
+					if spanSkew < start {
+						return errors.Newf("expected span skew to be at least %d, got %d", start, spanSkew)
+					}
+					if spanSkew >= end {
+						return errors.Newf("expected span skew to be less than %d, got %d", end, spanSkew)
+					}
+					return nil
+				})
+				return spanSkew
+			}
+			assertTableSkewInRange := func(start int64, end int64) int64 {
+				var tableSkew int64
+				testutils.SucceedsSoon(t, func() error {
+					tableSkew = aggMetrics.TableProgressSkew.Value()
+					scopedTableSkew := scopedMetrics.TableProgressSkew.Value()
+					if tableSkew != scopedTableSkew {
+						return errors.Newf("aggregate and scoped table skew don't match")
+					}
+					if !perTableTracking {
+						if tableSkew != 0 {
+							return errors.Newf("expected table skew to be 0, got %d", tableSkew)
+						}
+						return nil
+					}
+					if tableSkew < start {
+						return errors.Newf("expected table skew to be at least %d, got %d", start, tableSkew)
+					}
+					if tableSkew >= end {
+						return errors.Newf("expected table skew to be less than %d, got %d", end, tableSkew)
+					}
+					return nil
+				})
+				return tableSkew
+			}
+
+			// Verify that progress skew metrics show a non-negligible amount of lag
+			// since bar progress is blocked. Some amount of skew is often unavoidable
+			// due to the fact the aggregator processes the rangefeed checkpoints for
+			// different spans separately and at the time of a flush, may have only
+			// processed a portion of the checkpoints for a specific closed timestamp.
+			// The duration of 5s has been chosen given the default closed timestamp
+			// interval is 3s.
+			startingSpanSkew := assertSpanSkewInRange(int64(5*time.Second), math.MaxInt64)
+			startingTableSkew := assertTableSkewInRange(int64(5*time.Second), math.MaxInt64)
+
+			// Verify that skew continues to increase since bar progress is still blocked.
+			assertSpanSkewInRange(startingSpanSkew+int64(5*time.Second), math.MaxInt64)
+			assertTableSkewInRange(startingTableSkew+int64(5*time.Second), math.MaxInt64)
+
+			// Re-enable progress updates for bar.
+			blockBarProgress.Store(false)
+
+			// Verify that skew drops below the skew observed at the start.
+			assertSpanSkewInRange(0, startingSpanSkew)
+			assertTableSkewInRange(0, startingTableSkew)
+		}
+
+		cdcTest(t, testFn)
+	})
+}

pkg/ccl/changefeedccl/metrics.go

@@ -88,6 +88,8 @@ type AggMetrics struct {
 	KafkaThrottlingNanos        *aggmetric.AggHistogram
 	SinkErrors                  *aggmetric.AggCounter
 	MaxBehindNanos              *aggmetric.AggGauge
+	SpanProgressSkew            *aggmetric.AggGauge
+	TableProgressSkew           *aggmetric.AggGauge
 
 	Timers *timers.Timers
 
@@ -175,14 +177,18 @@ type sliMetrics struct {
 	KafkaThrottlingNanos        *aggmetric.Histogram
 	SinkErrors                  *aggmetric.Counter
 	MaxBehindNanos              *aggmetric.Gauge
+	SpanProgressSkew            *aggmetric.Gauge
+	TableProgressSkew           *aggmetric.Gauge
 
 	Timers *timers.ScopedTimers
 
 	mu struct {
 		syncutil.Mutex
-		id         int64
-		resolved   map[int64]hlc.Timestamp
-		checkpoint map[int64]hlc.Timestamp
+		id           int64
+		resolved     map[int64]hlc.Timestamp
+		checkpoint   map[int64]hlc.Timestamp
+		fastestSpan  map[int64]hlc.Timestamp
+		fastestTable map[int64]hlc.Timestamp
 	}
 	NetMetrics *cidr.NetMetrics
 
@@ -196,6 +202,8 @@ func (m *sliMetrics) closeId(id int64) {
 	defer m.mu.Unlock()
 	delete(m.mu.checkpoint, id)
 	delete(m.mu.resolved, id)
+	delete(m.mu.fastestSpan, id)
+	delete(m.mu.fastestTable, id)
 }
 
 // setResolved writes a resolved timestamp entry for the given id.
@@ -216,6 +224,18 @@ func (m *sliMetrics) setCheckpoint(id int64, ts hlc.Timestamp) {
 	}
 }
 
+// setFastestTS saves the fastest span/table timestamps for a given id.
+func (m *sliMetrics) setFastestTS(id int64, spanTS hlc.Timestamp, tableTS hlc.Timestamp) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if _, ok := m.mu.fastestSpan[id]; ok {
+		m.mu.fastestSpan[id] = spanTS
+	}
+	if _, ok := m.mu.fastestTable[id]; ok {
+		m.mu.fastestTable[id] = tableTS
+	}
+}
+
 // claimId claims a unique ID.
 func (m *sliMetrics) claimId() int64 {
 	m.mu.Lock()
@@ -225,6 +245,8 @@ func (m *sliMetrics) claimId() int64 {
 	// ignored until a nonzero timestamp is written.
 	m.mu.checkpoint[id] = hlc.Timestamp{}
 	m.mu.resolved[id] = hlc.Timestamp{}
+	m.mu.fastestSpan[id] = hlc.Timestamp{}
+	m.mu.fastestTable[id] = hlc.Timestamp{}
 	m.mu.id++
 	return id
 }
@@ -1035,6 +1057,18 @@ func newAggregateMetrics(histogramWindow time.Duration, lookup *cidr.Lookup) *Ag
 		Measurement: "Nanoseconds",
 		Unit:        metric.Unit_NANOSECONDS,
 	}
+	metaChangefeedSpanProgressSkew := metric.Metadata{
+		Name:        "changefeed.progress_skew.span",
+		Help:        "The time difference between the fastest and slowest span's resolved timestamp",
+		Measurement: "Nanoseconds",
+		Unit:        metric.Unit_NANOSECONDS,
+	}
+	metaChangefeedTableProgressSkew := metric.Metadata{
+		Name:        "changefeed.progress_skew.table",
+		Help:        "The time difference between the fastest and slowest table's resolved timestamp",
+		Measurement: "Nanoseconds",
+		Unit:        metric.Unit_NANOSECONDS,
+	}
 
 	functionalGaugeMinFn := func(childValues []int64) int64 {
 		var min int64
@@ -1152,6 +1186,8 @@ func newAggregateMetrics(histogramWindow time.Duration, lookup *cidr.Lookup) *Ag
 		}),
 		SinkErrors:        b.Counter(metaSinkErrors),
 		MaxBehindNanos:    b.FunctionalGauge(metaChangefeedMaxBehindNanos, functionalGaugeMaxFn),
+		SpanProgressSkew:  b.FunctionalGauge(metaChangefeedSpanProgressSkew, functionalGaugeMaxFn),
+		TableProgressSkew: b.FunctionalGauge(metaChangefeedTableProgressSkew, functionalGaugeMaxFn),
 		Timers:            timers.New(histogramWindow),
 		NetMetrics:        lookup.MakeNetMetrics(metaNetworkBytesOut, metaNetworkBytesIn, "sink"),
 		CheckpointMetrics: checkpoint.NewAggMetrics(b),
@@ -1236,6 +1272,8 @@ func (a *AggMetrics) getOrCreateScope(scope string) (*sliMetrics, error) {
 	}
 	sm.mu.resolved = make(map[int64]hlc.Timestamp)
 	sm.mu.checkpoint = make(map[int64]hlc.Timestamp)
+	sm.mu.fastestSpan = make(map[int64]hlc.Timestamp)
+	sm.mu.fastestTable = make(map[int64]hlc.Timestamp)
 	sm.mu.id = 1 // start the first id at 1 so we can detect intiialization
 
 	minTimestampGetter := func(m map[int64]hlc.Timestamp) func() int64 {
@@ -1266,9 +1304,34 @@ func (a *AggMetrics) getOrCreateScope(scope string) (*sliMetrics, error) {
 		}
 	}
 
+	maxTimestampSkewGetter := func(
+		base map[int64]hlc.Timestamp, ahead map[int64]hlc.Timestamp,
+	) func() int64 {
+		return func() int64 {
+			sm.mu.Lock()
+			defer sm.mu.Unlock()
+			var maxSkew int64
+			for id, b := range base {
+				a := ahead[id]
+				if a.IsEmpty() || b.IsEmpty() {
+					continue
+				}
+				skew := a.WallTime - b.WallTime
+				if skew > maxSkew {
+					maxSkew = skew
+				}
+			}
+			return maxSkew
+		}
+	}
+
 	sm.AggregatorProgress = a.AggregatorProgress.AddFunctionalChild(minTimestampGetter(sm.mu.resolved), scope)
 	sm.CheckpointProgress = a.CheckpointProgress.AddFunctionalChild(minTimestampGetter(sm.mu.checkpoint), scope)
 	sm.MaxBehindNanos = a.MaxBehindNanos.AddFunctionalChild(maxBehindNanosGetter(sm.mu.resolved), scope)
+	sm.SpanProgressSkew = a.SpanProgressSkew.AddFunctionalChild(
+		maxTimestampSkewGetter(sm.mu.checkpoint, sm.mu.fastestSpan), scope)
+	sm.TableProgressSkew = a.TableProgressSkew.AddFunctionalChild(
+		maxTimestampSkewGetter(sm.mu.checkpoint, sm.mu.fastestTable), scope)
 
 	a.mu.sliMetrics[scope] = sm
 	return sm, nil

pkg/ccl/changefeedccl/resolvedspan/frontier.go

@@ -354,6 +354,11 @@ func (f *resolvedSpanFrontier) HasLaggingSpans(sv *settings.Values) bool {
 	return frontier.Add(lagThresholdNanos, 0).Less(f.latestTS)
 }
 
+// LatestTS returns the latest timestamp in the frontier.
+func (f *resolvedSpanFrontier) LatestTS() hlc.Timestamp {
+	return f.latestTS
+}
+
 // All returns an iterator over the resolved spans in the frontier.
 func (f *resolvedSpanFrontier) All() iter.Seq[jobspb.ResolvedSpan] {
 	return func(yield func(jobspb.ResolvedSpan) bool) {