From 0fc9386f5b7103144878e7c888380419aac47ef2 Mon Sep 17 00:00:00 2001
From: Andrei Matei <andrei@cockroachlabs.com>
Date: Fri, 12 Mar 2021 15:09:06 -0500
Subject: [PATCH] roachtest/cdc: update status when watching changefeed latency

When watching the latency, we were logging every second. This patch
reduces the frequency to 1/10s and makes the test also update its status
so that the current latency high-water mark is visible in the web UI.

Release note: None
---
 pkg/cmd/roachtest/cdc.go | 21 +++++++++++++++++----
 1 file changed, 17 insertions(+), 4 deletions(-)

diff --git a/pkg/cmd/roachtest/cdc.go b/pkg/cmd/roachtest/cdc.go
index 772801ffda15..344afe1cf4cf 100644
--- a/pkg/cmd/roachtest/cdc.go
+++ b/pkg/cmd/roachtest/cdc.go
@@ -33,6 +33,7 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
@@ -141,6 +142,7 @@ func cdcBasicTest(ctx context.Context, t *test, c *cluster, args cdcTestArgs) {
 		args.targetInitialScanLatency,
 		args.targetSteadyLatency,
 		changefeedLogger,
+		t.Status,
 		args.crdbChaos,
 	)
 	defer verifier.maybeLogLatencyHist()
@@ -1278,9 +1280,11 @@ type latencyVerifier struct {
 	targetInitialScanLatency time.Duration
 	tolerateErrors           bool
 	logger                   *logger
+	setTestStatus            func(...interface{})
 
 	initialScanLatency   time.Duration
 	maxSeenSteadyLatency time.Duration
+	maxSeenSteadyEveryN  log.EveryN
 	latencyBecameSteady  bool
 
 	latencyHist *hdrhistogram.Histogram
@@ -1290,6 +1294,7 @@ func makeLatencyVerifier(
 	targetInitialScanLatency time.Duration,
 	targetSteadyLatency time.Duration,
 	l *logger,
+	setTestStatus func(...interface{}),
 	tolerateErrors bool,
 ) *latencyVerifier {
 	const sigFigs, minLatency, maxLatency = 1, 100 * time.Microsecond, 100 * time.Second
@@ -1298,8 +1303,10 @@ func makeLatencyVerifier(
 		targetInitialScanLatency: targetInitialScanLatency,
 		targetSteadyLatency:      targetSteadyLatency,
 		logger:                   l,
+		setTestStatus:            setTestStatus,
 		latencyHist:              hist,
 		tolerateErrors:           tolerateErrors,
+		maxSeenSteadyEveryN:      log.Every(10 * time.Second),
 	}
 }
 
@@ -1323,8 +1330,11 @@ func (lv *latencyVerifier) noteHighwater(highwaterTime time.Time) {
 		// tracking the max latency once we seen a latency
 		// that's less than the max allowed. Verify at the end
 		// of the test that this happens at some point.
-		lv.logger.Printf("end-to-end latency %s not yet below target steady latency %s\n",
-			latency, lv.targetSteadyLatency)
+		if lv.maxSeenSteadyEveryN.ShouldLog() {
+			lv.setTestStatus(fmt.Sprintf(
+				"watching changefeed: end-to-end latency %s not yet below target steady latency %s",
+				latency.Truncate(time.Millisecond), lv.targetSteadyLatency.Truncate(time.Millisecond)))
+		}
 		return
 	}
 	if err := lv.latencyHist.RecordValue(latency.Nanoseconds()); err != nil {
@@ -1333,8 +1343,11 @@ func (lv *latencyVerifier) noteHighwater(highwaterTime time.Time) {
 	if latency > lv.maxSeenSteadyLatency {
 		lv.maxSeenSteadyLatency = latency
 	}
-	lv.logger.Printf("end-to-end steady latency %s; max steady latency so far %s\n",
-		latency, lv.maxSeenSteadyLatency)
+	if lv.maxSeenSteadyEveryN.ShouldLog() {
+		lv.setTestStatus(fmt.Sprintf(
+			"watching changefeed: end-to-end steady latency %s; max steady latency so far %s",
+			latency.Truncate(time.Millisecond), lv.maxSeenSteadyLatency.Truncate(time.Millisecond)))
+	}
 }
 
 func (lv *latencyVerifier) pollLatency(