changefeedccl: add parallel io workers metric

Add a gauge metric to track the number of workers in ParallelIO.

Release note(ops change): Add metric changefeed.parallel_io_workers to
track the number of workers in ParallelIO.

Author: KeithCh

docs/generated/metrics/metrics.yaml

@@ -1780,7 +1780,7 @@ layers:
     - name: changefeed.parallel_io_pending_rows
       exported_name: changefeed_parallel_io_pending_rows
       description: Number of rows which are blocked from being sent due to conflicting in-flight keys
-      y_axis_label: Keys
+      y_axis_label: Messages
       type: GAUGE
       unit: COUNT
       aggregation: AVG
@@ -1801,6 +1801,14 @@ layers:
       unit: NANOSECONDS
       aggregation: AVG
       derivative: NONE
+    - name: changefeed.parallel_io_workers
+      exported_name: changefeed_parallel_io_workers
+      description: The number of workers in the ParallelIO
+      y_axis_label: Workers
+      type: GAUGE
+      unit: COUNT
+      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

pkg/ccl/changefeedccl/changefeed_test.go

@@ -11712,9 +11712,9 @@ func TestParallelIOMetrics(t *testing.T) {
 	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
 		registry := s.Server.JobRegistry().(*jobs.Registry)
 		metrics := registry.MetricsStruct().Changefeed.(*Metrics).AggMetrics
-
+		numWorkers := 1
 		db := sqlutils.MakeSQLRunner(s.DB)
-		db.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 1`)
+		db.Exec(t, fmt.Sprintf(`SET CLUSTER SETTING changefeed.sink_io_workers = %d`, numWorkers))
 		db.Exec(t, `
 		  CREATE TABLE foo (a INT PRIMARY KEY);
 		`)
@@ -11743,6 +11743,7 @@ func TestParallelIOMetrics(t *testing.T) {
 		// Set the frequency to 1s. The default frequency at the time of writing is
 		foo, err := f.Feed("CREATE CHANGEFEED FOR TABLE foo WITH pubsub_sink_config=" +
 			"'{\"Flush\": {\"Frequency\": \"100ms\"}}'")
+		defer closeFeed(t, foo)
 		require.NoError(t, err)
 
 		testutils.SucceedsSoon(t, func() error {
@@ -11775,6 +11776,19 @@ func TestParallelIOMetrics(t *testing.T) {
 			}
 			return nil
 		})
+
+		assert.Equal(t, int64(numWorkers), metrics.ParallelIOWorkers.Value())
+		jobFeed := foo.(cdctest.EnterpriseTestFeed)
+		require.NoError(t, jobFeed.Pause())
+		db.Exec(t, fmt.Sprintf(`SET CLUSTER SETTING changefeed.sink_io_workers = %d`, numWorkers+1))
+		require.NoError(t, jobFeed.Resume())
+		testutils.SucceedsSoon(t, func() error {
+			if metrics.ParallelIOWorkers.Value() != int64(numWorkers+1) {
+				return errors.Newf("waiting for workers: %d", metrics.ParallelIOWorkers.Value())
+			}
+			return nil
+		})
+
 		close(done)
 		require.NoError(t, g.Wait())
 		require.NoError(t, foo.Close())

pkg/ccl/changefeedccl/metrics.go

@@ -69,6 +69,7 @@ type AggMetrics struct {
 	ParallelIOPendingRows       *aggmetric.AggGauge
 	ParallelIOResultQueueNanos  *aggmetric.AggHistogram
 	ParallelIOInFlightKeys      *aggmetric.AggGauge
+	ParallelIOWorkers           *aggmetric.AggGauge
 	SinkIOInflight              *aggmetric.AggGauge
 	SinkBackpressureNanos       *aggmetric.AggHistogram
 	CommitLatency               *aggmetric.AggHistogram
@@ -128,6 +129,7 @@ type metricsRecorder interface {
 	recordSizeBasedFlush()
 	newParallelIOMetricsRecorder() parallelIOMetricsRecorder
 	recordSinkIOInflightChange(int64)
+	recordParallelIOWorkers(int64)
 	recordSinkBackpressure(time.Duration)
 	makeCloudstorageFileAllocCallback() func(delta int64)
 	getKafkaThrottlingMetrics(*cluster.Settings) metrics.Histogram
@@ -158,6 +160,7 @@ type sliMetrics struct {
 	ParallelIOPendingRows       *aggmetric.Gauge
 	ParallelIOResultQueueNanos  *aggmetric.Histogram
 	ParallelIOInFlightKeys      *aggmetric.Gauge
+	ParallelIOWorkers           *aggmetric.Gauge
 	SinkIOInflight              *aggmetric.Gauge
 	SinkBackpressureNanos       *aggmetric.Histogram
 	CommitLatency               *aggmetric.Histogram
@@ -626,6 +629,14 @@ func (m *sliMetrics) recordSinkIOInflightChange(delta int64) {
 	m.SinkIOInflight.Inc(delta)
 }
 
+func (m *sliMetrics) recordParallelIOWorkers(n int64) {
+	if m == nil {
+		return
+	}
+
+	m.ParallelIOWorkers.Update(n)
+}
+
 func (m *sliMetrics) recordSinkBackpressure(duration time.Duration) {
 	if m == nil {
 		return
@@ -712,6 +723,10 @@ func (w *wrappingCostController) recordSinkIOInflightChange(delta int64) {
 	w.inner.recordSinkIOInflightChange(delta)
 }
 
+func (w *wrappingCostController) recordParallelIOWorkers(n int64) {
+	w.inner.recordParallelIOWorkers(n)
+}
+
 func (w *wrappingCostController) recordSinkBackpressure(duration time.Duration) {
 	w.inner.recordSinkBackpressure(duration)
 }
@@ -992,6 +1007,12 @@ func newAggregateMetrics(histogramWindow time.Duration, lookup *cidr.Lookup) *Ag
 		Measurement: "Keys",
 		Unit:        metric.Unit_COUNT,
 	}
+	metaChangefeedParallelIOWorkers := metric.Metadata{
+		Name:        "changefeed.parallel_io_workers",
+		Help:        "The number of workers in the ParallelIO",
+		Measurement: "Workers",
+		Unit:        metric.Unit_COUNT,
+	}
 	metaChangefeedSinkIOInflight := metric.Metadata{
 		Name:        "changefeed.sink_io_inflight",
 		Help:        "The number of keys currently inflight as IO requests being sent to the sink",
@@ -1138,6 +1159,7 @@ func newAggregateMetrics(histogramWindow time.Duration, lookup *cidr.Lookup) *Ag
 			SigFigs:      2,
 			BucketConfig: metric.ChangefeedBatchLatencyBuckets,
 		}),
+		ParallelIOWorkers: b.Gauge(metaChangefeedParallelIOWorkers),
 		BatchHistNanos: b.Histogram(metric.HistogramOptions{
 			Metadata:     metaChangefeedBatchHistNanos,
 			Duration:     histogramWindow,
@@ -1245,6 +1267,7 @@ func (a *AggMetrics) getOrCreateScope(scope string) (*sliMetrics, error) {
 		ParallelIOPendingRows:       a.ParallelIOPendingRows.AddChild(scope),
 		ParallelIOResultQueueNanos:  a.ParallelIOResultQueueNanos.AddChild(scope),
 		ParallelIOInFlightKeys:      a.ParallelIOInFlightKeys.AddChild(scope),
+		ParallelIOWorkers:           a.ParallelIOWorkers.AddChild(scope),
 		SinkIOInflight:              a.SinkIOInflight.AddChild(scope),
 		SinkBackpressureNanos:       a.SinkBackpressureNanos.AddChild(scope),
 		CommitLatency:               a.CommitLatency.AddChild(scope),

pkg/ccl/changefeedccl/parallel_io.go

@@ -100,7 +100,7 @@ func NewParallelIO(
 	wg.GoCtx(func(ctx context.Context) error {
 		return io.processIO(ctx, numWorkers)
 	})
-
+	io.metrics.recordParallelIOWorkers(int64(numWorkers))
 	return io
 }