Merge #142996 #143467

142996: structlogging: restructure hot range logger for testability r=angles-n-daemons a=angles-n-daemons

structlogging: restructure hot range logger for testability

This change does a few things to improve the testability of the hot
ranges logger. The includes:

__Logger__:
 * The introduction of a shouldLog function, which determines whether
         the system should log or not.
 * The breakout of the logging action into its own function.

__Tests__:
 * The addition of a setup and teardown utility for the hot ranage
         logger.
 * The breakout of the default case and a timed case.
 * Tests for hot ranges which exist in the system to start.

Fixes: #142995
Epic: CRDB-43150


143467: changefeedccl: update TestChangefeedLaggingSpanCheckpointing r=aerfrei,asg0451 a=andyyang890

This patch updates `TestChangefeedLaggingSpanCheckpointing` to ensure
we checkpoint spans at at least two different timestamps.

This patch also fixes the issue where the `OnRangeFeedStart` kvfeed
testing knob was not being invoked in production code.

Informs #137692
Fixes #143377

Release note: None

Co-authored-by: Brian Dillmann <[email protected]>
Co-authored-by: Andy Yang <[email protected]>

Author: 3 people

pkg/ccl/changefeedccl/changefeed_test.go

@@ -2354,8 +2354,8 @@ func TestChangefeedSchemaChangeNoBackfill(t *testing.T) {
 	}
 }
 
-// Test checkpointing when the highwater does not move due to some issues with
-// specific spans lagging behind
+// TestChangefeedLaggingSpanCheckpointing tests checkpointing when the highwater
+// does not advance due to specific spans lagging behind.
 func TestChangefeedLaggingSpanCheckpointing(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -2369,7 +2369,7 @@ func TestChangefeedLaggingSpanCheckpointing(t *testing.T) {
 		DistSQL.(*execinfra.TestingKnobs).
 		Changefeed.(*TestingKnobs)
 
-	// Initialize table with multiple ranges.
+	// Initialize table with 20 ranges.
 	sqlDB.Exec(t, `
   CREATE TABLE foo (key INT PRIMARY KEY);
   INSERT INTO foo (key) SELECT * FROM generate_series(1, 1000);
@@ -2381,28 +2381,39 @@ func TestChangefeedLaggingSpanCheckpointing(t *testing.T) {
 	changefeedbase.SpanCheckpointInterval.Override(
 		context.Background(), &s.ClusterSettings().SV, 10*time.Millisecond)
 	changefeedbase.SpanCheckpointMaxBytes.Override(
-		context.Background(), &s.ClusterSettings().SV, 100<<20)
+		context.Background(), &s.ClusterSettings().SV, 100<<20 /* 100 MiB */)
 	changefeedbase.SpanCheckpointLagThreshold.Override(
 		context.Background(), &s.ClusterSettings().SV, 10*time.Millisecond)
 
-	// We'll start changefeed with the cursor.
+	// We'll start the changefeed with the cursor set to the current time (not insert time).
+	// NB: The changefeed created in this test doesn't actually send any message events.
 	var tsStr string
 	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp() from foo`).Scan(&tsStr)
 	cursor := parseTimeToHLC(t, tsStr)
+	t.Logf("cursor: %v", cursor)
 
 	// Rangefeed will skip some of the checkpoints to simulate lagging spans.
 	var laggingSpans roachpb.SpanGroup
-	numLagging := 0
+	nonLaggingSpans := make(map[string]int)
+	var numLagging, numNonLagging int
 	knobs.FeedKnobs.ShouldSkipCheckpoint = func(checkpoint *kvpb.RangeFeedCheckpoint) bool {
-		// Skip spans that were skipped before; otherwise skip some spans.
-		seenBefore := laggingSpans.Encloses(checkpoint.Span)
-		if seenBefore || (numLagging < 5 && rnd.Int()%3 == 0) {
-			if !seenBefore {
-				laggingSpans.Add(checkpoint.Span)
-				numLagging++
-			}
+		// Skip spans that we already picked to be lagging.
+		if laggingSpans.Encloses(checkpoint.Span) {
+			return true /* skip */
+		}
+		// Skip additional updates for some non-lagging spans so that we can
+		// have more than one timestamp in the checkpoint.
+		if i, ok := nonLaggingSpans[checkpoint.Span.String()]; ok {
+			return i%3 == 0
+		}
+		// Ensure we have a few spans that are lagging at the cursor.
+		if numLagging == 0 || (numLagging < 5 && rnd.Int()%3 == 0) {
+			laggingSpans.Add(checkpoint.Span)
+			numLagging++
 			return true /* skip */
 		}
+		nonLaggingSpans[checkpoint.Span.String()] = numNonLagging
+		numNonLagging++
 		return false
 	}
 
@@ -2419,64 +2430,83 @@ func TestChangefeedLaggingSpanCheckpointing(t *testing.T) {
 		return job.Progress()
 	}
 
-	// Should eventually checkpoint all spans around the lagging span
+	// We should eventually checkpoint some spans that are ahead of the highwater.
+	// We'll wait until we have two unique timestamps.
 	testutils.SucceedsSoon(t, func() error {
 		progress := loadProgress()
-		if loadCheckpoint(t, progress) != nil {
+		cp := maps.Collect(loadCheckpoint(t, progress).All())
+		if len(cp) >= 2 {
 			return nil
 		}
-		return errors.New("waiting for checkpoint")
+		return errors.New("waiting for checkpoint with two different timestamps")
 	})
 
 	sqlDB.Exec(t, "PAUSE JOB $1", jobID)
 	waitForJobState(sqlDB, t, jobID, jobs.StatePaused)
 
 	// We expect highwater to be 0 (because we skipped some spans) or exactly cursor
-	// (this is mostly due to racy updates sent from aggregators to the frontier.
+	// (this is mostly due to racy updates sent from aggregators to the frontier).
 	// However, the checkpoint timestamp should be at least at the cursor.
 	progress := loadProgress()
-	require.True(t, progress.GetHighWater().IsEmpty() || *progress.GetHighWater() == cursor,
-		"expected empty highwater or %s,  found %s", cursor, progress.GetHighWater())
+	require.True(t, progress.GetHighWater().IsEmpty() || progress.GetHighWater().Equal(cursor),
+		"expected empty highwater or %s, found %s", cursor, progress.GetHighWater())
 	spanLevelCheckpoint := loadCheckpoint(t, progress)
 	require.NotNil(t, spanLevelCheckpoint)
-	minCheckpointTS := spanLevelCheckpoint.MinTimestamp()
-	require.True(t, cursor.LessEq(minCheckpointTS))
+	require.True(t, cursor.LessEq(spanLevelCheckpoint.MinTimestamp()))
+
+	// Construct a reverse index from spans to timestamps.
+	spanTimestamps := make(map[string]hlc.Timestamp)
+	for ts, spans := range spanLevelCheckpoint.All() {
+		for _, s := range spans {
+			spanTimestamps[s.String()] = ts
+		}
+	}
 
+	var rangefeedStarted bool
 	var incorrectCheckpointErr error
 	knobs.FeedKnobs.OnRangeFeedStart = func(spans []kvcoord.SpanTimePair) {
+		rangefeedStarted = true
+
 		setErr := func(stp kvcoord.SpanTimePair, expectedTS hlc.Timestamp) {
 			incorrectCheckpointErr = errors.Newf(
 				"rangefeed for span %s expected to start @%s, started @%s instead",
 				stp.Span, expectedTS, stp.StartAfter)
 		}
 
+		// Verify that the start time for each span is correct.
 		for _, sp := range spans {
-			if laggingSpans.Encloses(sp.Span) {
-				if !sp.StartAfter.Equal(cursor) {
-					setErr(sp, cursor)
+			if checkpointTS := spanTimestamps[sp.Span.String()]; checkpointTS.IsSet() {
+				// Any span in the checkpoint should be resumed at its checkpoint timestamp.
+				if !sp.StartAfter.Equal(checkpointTS) {
+					setErr(sp, checkpointTS)
 				}
 			} else {
-				if !sp.StartAfter.Equal(minCheckpointTS) {
-					setErr(sp, minCheckpointTS)
+				// Any spans not in the checkpoint should be at the cursor.
+				if !sp.StartAfter.Equal(cursor) {
+					setErr(sp, cursor)
 				}
 			}
 		}
 	}
+	knobs.FeedKnobs.ShouldSkipCheckpoint = nil
 
 	sqlDB.Exec(t, "RESUME JOB $1", jobID)
 	waitForJobState(sqlDB, t, jobID, jobs.StateRunning)
 
 	// Wait until highwater advances past cursor.
 	testutils.SucceedsSoon(t, func() error {
 		progress := loadProgress()
-		if hw := progress.GetHighWater(); hw != nil && cursor.LessEq(*hw) {
+		if hw := progress.GetHighWater(); hw != nil && cursor.Less(*hw) {
 			return nil
 		}
 		return errors.New("waiting for checkpoint advance")
 	})
 
 	sqlDB.Exec(t, "PAUSE JOB $1", jobID)
 	waitForJobState(sqlDB, t, jobID, jobs.StatePaused)
+	// Verify the rangefeed started. This guards against the testing knob
+	// not being called, which was happening in earlier versions of the code.
+	require.True(t, rangefeedStarted)
 	// Verify we didn't see incorrect timestamps when resuming.
 	require.NoError(t, incorrectCheckpointErr)
 }

pkg/ccl/changefeedccl/helpers_test.go

@@ -11,6 +11,7 @@ import (
 	gosql "database/sql"
 	gojson "encoding/json"
 	"fmt"
+	"maps"
 	"math"
 	"math/rand"
 	"net/url"
@@ -868,7 +869,7 @@ func loadCheckpoint(t *testing.T, progress jobspb.Progress) *jobspb.TimestampSpa
 	if spanLevelCheckpoint.IsEmpty() {
 		return nil
 	}
-	t.Logf("found checkpoint: %#v", spanLevelCheckpoint)
+	t.Logf("found checkpoint: %v", maps.Collect(spanLevelCheckpoint.All()))
 	return spanLevelCheckpoint
 }
 

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

@@ -99,12 +99,17 @@ func (p rangefeedFactory) Run(ctx context.Context, sink kvevent.Writer, cfg rang
 	if cfg.RangeObserver != nil {
 		rfOpts = append(rfOpts, kvcoord.WithRangeObserver(cfg.RangeObserver))
 	}
-	rfOpts = append(rfOpts, kvcoord.WithConsumerID(cfg.ConsumerID))
+	if cfg.ConsumerID != 0 {
+		rfOpts = append(rfOpts, kvcoord.WithConsumerID(cfg.ConsumerID))
+	}
 	if len(cfg.Knobs.RangefeedOptions) != 0 {
 		rfOpts = append(rfOpts, cfg.Knobs.RangefeedOptions...)
 	}
 
 	g.GoCtx(func(ctx context.Context) error {
+		if cfg.Knobs.OnRangeFeedStart != nil {
+			cfg.Knobs.OnRangeFeedStart(cfg.Spans)
+		}
 		return p(ctx, cfg.Spans, eventCh, rfOpts...)
 	})
 	return g.Wait()

pkg/server/structlogging/BUILD.bazel

@@ -32,25 +32,20 @@ go_test(
     deps = [
         ":structlogging",
         "//pkg/base",
-        "//pkg/keys",
         "//pkg/kv/kvserver",
         "//pkg/kv/kvserver/allocator/plan",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
         "//pkg/server",
-        "//pkg/sql/catalog/desctestutils",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/skip",
-        "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/log/eventpb",
         "//pkg/util/log/logpb",
         "//pkg/util/randutil",
         "//pkg/util/syncutil",
-        "@com_github_cockroachdb_errors//:errors",
-        "@com_github_stretchr_testify//assert",
     ],
 )

pkg/server/structlogging/hot_ranges_log.go

@@ -84,7 +84,6 @@ func (s *hotRangesLoggingScheduler) start(ctx context.Context, stopper *stop.Sto
 		})
 
 		ticker := time.NewTicker(TelemetryHotRangesStatsInterval.Get(&s.st.SV))
-		defer ticker.Stop()
 
 		for {
 			select {
@@ -93,43 +92,60 @@ func (s *hotRangesLoggingScheduler) start(ctx context.Context, stopper *stop.Sto
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
-				if !TelemetryHotRangesStatsEnabled.Get(&s.st.SV) {
-					continue
-				}
-				resp, err := s.sServer.HotRangesV2(ctx,
-					&serverpb.HotRangesRequest{NodeID: "local", PageSize: ReportTopHottestRanges})
-				if err != nil {
-					log.Warningf(ctx, "failed to get hot ranges: %s", err)
-					continue
-				}
-				var events []logpb.EventPayload
-				ts := timeutil.Now().UnixNano()
-
-				for _, r := range resp.Ranges {
-					hrEvent := &eventpb.HotRangesStats{
-						RangeID:             int64(r.RangeID),
-						Qps:                 r.QPS,
-						Databases:           r.Databases,
-						Tables:              r.Tables,
-						Indexes:             r.Indexes,
-						SchemaName:          r.SchemaName,
-						CPUTimePerSecond:    r.CPUTimePerSecond,
-						ReadBytesPerSecond:  r.ReadBytesPerSecond,
-						WriteBytesPerSecond: r.WriteBytesPerSecond,
-						ReadsPerSecond:      r.ReadsPerSecond,
-						WritesPerSecond:     r.WritesPerSecond,
-						LeaseholderNodeID:   int32(r.LeaseholderNodeID),
-						CommonEventDetails: logpb.CommonEventDetails{
-							Timestamp: ts,
-						},
-					}
-					events = append(events, hrEvent)
-				}
-				logutil.LogEventsWithDelay(ctx, events, stopper, TelemetryHotRangesStatsLoggingDelay.Get(&s.st.SV))
-
+				s.maybeLogHotRanges(ctx, stopper)
 			case <-intervalChangedChan:
 				ticker.Reset(TelemetryHotRangesStatsInterval.Get(&s.st.SV))
 			}
 		}
 	})
 }
+
+// maybeLogHotRanges is a small helper function which couples the
+// functionality of checking whether to log and logging.
+func (s *hotRangesLoggingScheduler) maybeLogHotRanges(ctx context.Context, stopper *stop.Stopper) {
+	if s.shouldLog() {
+		s.logHotRanges(ctx, stopper)
+	}
+}
+
+// shouldLog checks the below conditions to see whether it should emit logs.
+//   - Is the cluster setting server.telemetry.hot_ranges_stats.enabled true?
+func (s *hotRangesLoggingScheduler) shouldLog() bool {
+	return TelemetryHotRangesStatsEnabled.Get(&s.st.SV)
+}
+
+// logHotRanges collects the hot ranges from this node's status server and
+// sends them to the TELEMETRY log channel.
+func (s *hotRangesLoggingScheduler) logHotRanges(ctx context.Context, stopper *stop.Stopper) {
+	resp, err := s.sServer.HotRangesV2(ctx,
+		&serverpb.HotRangesRequest{NodeID: "local", PageSize: ReportTopHottestRanges})
+	if err != nil {
+		log.Warningf(ctx, "failed to get hot ranges: %s", err)
+		return
+	}
+
+	var events []logpb.EventPayload
+	ts := timeutil.Now().UnixNano()
+
+	for _, r := range resp.Ranges {
+		hrEvent := &eventpb.HotRangesStats{
+			RangeID:             int64(r.RangeID),
+			Qps:                 r.QPS,
+			Databases:           r.Databases,
+			Tables:              r.Tables,
+			Indexes:             r.Indexes,
+			SchemaName:          r.SchemaName,
+			CPUTimePerSecond:    r.CPUTimePerSecond,
+			ReadBytesPerSecond:  r.ReadBytesPerSecond,
+			WriteBytesPerSecond: r.WriteBytesPerSecond,
+			ReadsPerSecond:      r.ReadsPerSecond,
+			WritesPerSecond:     r.WritesPerSecond,
+			LeaseholderNodeID:   int32(r.LeaseholderNodeID),
+			CommonEventDetails: logpb.CommonEventDetails{
+				Timestamp: ts,
+			},
+		}
+		events = append(events, hrEvent)
+	}
+	logutil.LogEventsWithDelay(ctx, events, stopper, TelemetryHotRangesStatsLoggingDelay.Get(&s.st.SV))
+}