admission,changefeedccl: integrate parallel workers w/ elastic cpu control

[irfansharif]

Is your feature request related to a problem? Please describe.

Here's what a CPU profile from a variant of admission-control/elastic-cdc (added in #89656) looks like. In this version we set up a few rangefeeds against 8vCPU nodes running TPC-C with 1000-warehouses, running with just initial_scan = 'only'.

The profile tells us that a lot of CPU hit is due to the parallel workers receiving events off of the KV feed. The tiny purple segment on the right is the scan portion down in KV. When we tried integrating just that bit with the elastic CPU limiter (a now lost revision of #89709), we were still ineffective in bounding effects on scheduling latencies since so much CPU is used up elsewhere. We ought to integrate this work with the elastic CPU limiter.

Epic CRDB-24463

[blathers-crl]

cc @cockroachdb/cdc

[amruss]

@irfansharif are you planning on picking this up?

[irfansharif]

No. That said this is relatively self-contained, and something I was discussing with @jayshrivastava+@shermanCRL whether it's something CDC has bandwidth to pick up. It's a good segue to/from another issue @miretskiy filed around changing what the default # of these parallel workers is, which I can't seem to find. For me I'm just planning to see #89709 through.

[shermanCRL]

Cool, assigning to @jayshrivastava for backlog.

[irfansharif]

To reproduce what this issue is describing, in this roachtest we can replace the following with the segment after:

cockroach/pkg/cmd/roachtest/tests/admission_control_elastic_cdc.go

Lines 121 to 124 in 42d08d8

	stmtWithCursor := fmt.Sprintf(`
	CREATE CHANGEFEED FOR tpcc.order_line, tpcc.stock, tpcc.customer
	INTO 'null://' WITH cursor = '-%ds'
	`, int64(float64(i+1)*padDuration.Seconds())) // scanning as far back as possible (~ when the workload started)

CREATE CHANGEFEED FOR tpcc.order_line, tpcc.stock, tpcc.customer
INTO 'null://' WITH initial_scan = 'only';

We can introduce this as a variant of the test, addressing this TODO:

cockroach/pkg/cmd/roachtest/tests/admission_control_elastic_cdc.go

Lines 130 to 134 in 42d08d8

	// TODO(irfansharif): Add a version of this test
	// with initial_scan = 'only' to demonstrate the
	// need+efficacy of using elastic CPU control in
	// changefeed workers. That too has a severe effect
	// on scheduling latencies.

[irfansharif]

Notes from a live-pairing session with @jayshrivastava, off of the branch on #89709.

diff --git i/pkg/ccl/changefeedccl/changefeed_processors.go w/pkg/ccl/changefeedccl/changefeed_processors.go
index 4256f69bb2..650e372026 100644
--- i/pkg/ccl/changefeedccl/changefeed_processors.go
+++ w/pkg/ccl/changefeedccl/changefeed_processors.go
@@ -476,6 +476,10 @@ func (ca *changeAggregator) Next() (rowenc.EncDatumRow, *execinfrapb.ProducerMet
 		}
 
 		if err := ca.tick(); err != nil {
+			// XXX: Is this a tight loop. Worth confirming. If so, ask for
+			// tokens before loop start, inform done-ness at loop end.
+			// XXX: There's another thing to consider: maybe the Pacer type from
+			// https://github.com/cockroachdb/cockroach/pull/89709 applies.
 			var e kvevent.ErrBufferClosed
 			if errors.As(err, &e) {
 				// ErrBufferClosed is a signal that our kvfeed has exited expectedly.
diff --git i/pkg/ccl/changefeedccl/event_processing.go w/pkg/ccl/changefeedccl/event_processing.go
index de387e1f02..14012e78e3 100644
--- i/pkg/ccl/changefeedccl/event_processing.go
+++ w/pkg/ccl/changefeedccl/event_processing.go
@@ -238,6 +238,11 @@ func (c *kvEventToRowConsumer) topicForEvent(eventMeta cdcevent.Metadata) (Topic
 
 // ConsumeEvent manages kv event lifetime: parsing, encoding and event being emitted to the sink.
 func (c *kvEventToRowConsumer) ConsumeEvent(ctx context.Context, ev kvevent.Event) error {
+	// XXX: One at a time.
+	// - this code is CPU dominant, that means it takes non-zero CPU time.
+	// - before running this code: ask the local (this node) elastic CPU granter
+	//   for some CPU time.
+	// - after you're done running, tell the CPU granter what you actually used.
 	if ev.Type() != kvevent.TypeKV {
 		return errors.AssertionFailedf("expected kv ev, got %v", ev.Type())
 	}
diff --git i/pkg/kv/kvserver/kvadmission/kvadmission.go w/pkg/kv/kvserver/kvadmission/kvadmission.go
index f4f3a0e893..34a7194068 100644
--- i/pkg/kv/kvserver/kvadmission/kvadmission.go
+++ w/pkg/kv/kvserver/kvadmission/kvadmission.go
@@ -258,6 +258,9 @@ func (n *controllerImpl) AdmitKVWork(
 				return Handle{}, err
 			}
 			ah.ElasticCPUWorkHandle = elasticWorkHandle
+			// XXX: Work typically happens between an Admit and
+			// AdmittedWorkDone. Not the one below which is just for error
+			// handling purposes, but there's another use at the caller.
 			defer func() {
 				if retErr != nil {
 					// No elastic work was done.
diff --git i/pkg/kv/kvserver/rangefeed/catchup_scan.go w/pkg/kv/kvserver/rangefeed/catchup_scan.go
index 06a2fd974c..e4e5693192 100644
--- i/pkg/kv/kvserver/rangefeed/catchup_scan.go
+++ w/pkg/kv/kvserver/rangefeed/catchup_scan.go
@@ -165,6 +165,8 @@ func (i *CatchUpIterator) CatchUpScan(
 			break
 		}
 
+		// XXX: Could use this library perhaps. At the top-most level of the
+		// changefeed iterator where we're reading off events.
 		if err := i.pacer.Pace(ctx); err != nil {
 			// We're unable to pace things automatically -- shout loudly
 			// semi-infrequently but don't fail the rangefeed itself.
diff --git i/pkg/server/node.go w/pkg/server/node.go
index d108838c90..87d377f8d0 100644
--- i/pkg/server/node.go
+++ w/pkg/server/node.go
@@ -218,6 +218,9 @@ func (nm nodeMetrics) callComplete(d time.Duration, pErr *roachpb.Error) {
 // its client.DB reference. Nodes use this to allocate node and store
 // IDs for bootstrapping the node itself or initializing new stores as
 // they're added on subsequent instantiations.
+//
+// XXX: Try connect dots from changefeed code to this guy,
+// which holds onto an elastic work queue (or can be made to)
 type Node struct {
 	stopper      *stop.Stopper
 	clusterID    *base.ClusterIDContainer // UUID for Cockroach cluster