Merge #88370

88370: changefeedccl: add metric for size based flushes r=jayshrivastava a=jayshrivastava Add metric to track when the cloud storage sink has to flush data due to the file size limit being reached. Fixes: #84435 Release note: None Co-authored-by: Jayant Shrivastava <jayants@cockroachlabs.com>
cockroachdb · Sep 21, 2022 · 6450a9d · 6450a9d
2 parents 31754ac + dcba53b
commit 6450a9d
Show file tree

Hide file tree

Showing 3 changed files with 38 additions and 6 deletions.
diff --git a/pkg/ccl/changefeedccl/metrics.go b/pkg/ccl/changefeedccl/metrics.go
@@ -51,6 +51,7 @@ type AggMetrics struct {
 	BatchHistNanos            *aggmetric.AggHistogram
 	Flushes                   *aggmetric.AggCounter
 	FlushHistNanos            *aggmetric.AggHistogram
+	SizeBasedFlushes          *aggmetric.AggCounter
 	CommitLatency             *aggmetric.AggHistogram
 	BackfillCount             *aggmetric.AggGauge
 	BackfillPendingRanges     *aggmetric.AggGauge
@@ -85,6 +86,7 @@ type metricsRecorder interface {
 	recordFlushRequestCallback() func()
 	getBackfillCallback() func() func()
 	getBackfillRangeCallback() func(int64) (func(), func())
+	recordSizeBasedFlush()
 }
 
 var _ metricsRecorder = (*sliMetrics)(nil)
@@ -102,6 +104,7 @@ type sliMetrics struct {
 	BatchHistNanos            *aggmetric.Histogram
 	Flushes                   *aggmetric.Counter
 	FlushHistNanos            *aggmetric.Histogram
+	SizeBasedFlushes          *aggmetric.Counter
 	CommitLatency             *aggmetric.Histogram
 	ErrorRetries              *aggmetric.Counter
 	AdmitLatency              *aggmetric.Histogram
@@ -224,6 +227,15 @@ func (m *sliMetrics) getBackfillRangeCallback() func(int64) (func(), func()) {
 	}
 }
 
+// Record size-based flush.
+func (m *sliMetrics) recordSizeBasedFlush() {
+	if m == nil {
+		return
+	}
+
+	m.SizeBasedFlushes.Inc(1)
+}
+
 type wrappingCostController struct {
 	ctx      context.Context
 	inner    metricsRecorder
@@ -284,6 +296,11 @@ func (w *wrappingCostController) getBackfillRangeCallback() func(int64) (func(),
 	return w.inner.getBackfillRangeCallback()
 }
 
+// Record size-based flush.
+func (w *wrappingCostController) recordSizeBasedFlush() {
+	w.inner.recordSizeBasedFlush()
+}
+
 var (
 	metaChangefeedForwardedResolvedMessages = metric.Metadata{
 		Name:        "changefeed.forwarded_resolved_messages",
@@ -369,6 +386,12 @@ func newAggregateMetrics(histogramWindow time.Duration) *AggMetrics {
 		Measurement: "Flushes",
 		Unit:        metric.Unit_COUNT,
 	}
+	metaSizeBasedFlushes := metric.Metadata{
+		Name:        "changefeed.size_based_flushes",
+		Help:        "Total size based flushes across all feeds",
+		Measurement: "Flushes",
+		Unit:        metric.Unit_COUNT,
+	}
 	metaChangefeedBatchHistNanos := metric.Metadata{
 		Name:        "changefeed.sink_batch_hist_nanos",
 		Help:        "Time spent batched in the sink buffer before being flushed and acknowledged",
@@ -440,12 +463,13 @@ func newAggregateMetrics(histogramWindow time.Duration) *AggMetrics {
 	// retain significant figures of 2.
 	b := aggmetric.MakeBuilder("scope")
 	a := &AggMetrics{
-		ErrorRetries:    b.Counter(metaChangefeedErrorRetries),
-		EmittedMessages: b.Counter(metaChangefeedEmittedMessages),
-		MessageSize:     b.Histogram(metaMessageSize, histogramWindow, metric.DataSize16MBBuckets),
-		EmittedBytes:    b.Counter(metaChangefeedEmittedBytes),
-		FlushedBytes:    b.Counter(metaChangefeedFlushedBytes),
-		Flushes:         b.Counter(metaChangefeedFlushes),
+		ErrorRetries:     b.Counter(metaChangefeedErrorRetries),
+		EmittedMessages:  b.Counter(metaChangefeedEmittedMessages),
+		MessageSize:      b.Histogram(metaMessageSize, histogramWindow, metric.DataSize16MBBuckets),
+		EmittedBytes:     b.Counter(metaChangefeedEmittedBytes),
+		FlushedBytes:     b.Counter(metaChangefeedFlushedBytes),
+		Flushes:          b.Counter(metaChangefeedFlushes),
+		SizeBasedFlushes: b.Counter(metaSizeBasedFlushes),
 
 		BatchHistNanos:            b.Histogram(metaChangefeedBatchHistNanos, histogramWindow, metric.BatchProcessLatencyBuckets),
 		FlushHistNanos:            b.Histogram(metaChangefeedFlushHistNanos, histogramWindow, metric.BatchProcessLatencyBuckets),
@@ -507,6 +531,7 @@ func (a *AggMetrics) getOrCreateScope(scope string) (*sliMetrics, error) {
 		BatchHistNanos:            a.BatchHistNanos.AddChild(scope),
 		Flushes:                   a.Flushes.AddChild(scope),
 		FlushHistNanos:            a.FlushHistNanos.AddChild(scope),
+		SizeBasedFlushes:          a.SizeBasedFlushes.AddChild(scope),
 		CommitLatency:             a.CommitLatency.AddChild(scope),
 		ErrorRetries:              a.ErrorRetries.AddChild(scope),
 		AdmitLatency:              a.AdmitLatency.AddChild(scope),

diff --git a/pkg/ccl/changefeedccl/sink_cloudstorage.go b/pkg/ccl/changefeedccl/sink_cloudstorage.go
@@ -480,6 +480,7 @@ func (s *cloudStorageSink) EmitRow(
 	}
 
 	if int64(file.buf.Len()) > s.targetMaxFileSize {
+		s.metrics.recordSizeBasedFlush()
 		if err := s.flushTopicVersions(ctx, file.topic, file.schemaID); err != nil {
 			return err
 		}

diff --git a/pkg/ts/catalog/chart_catalog.go b/pkg/ts/catalog/chart_catalog.go
@@ -1385,6 +1385,12 @@ var charts = []sectionDescription{
 					"changefeed.flushes",
 				},
 			},
+			{
+				Title: "Size Based Flushes",
+				Metrics: []string{
+					"changefeed.size_based_flushes",
+				},
+			},
 			{
 				Title: "Max Behind Nanos",
 				Metrics: []string{