changefeedccl: Improve aggregator -> coordinator flush performance

[miretskiy]

See https://cockroachlabs.slack.com/archives/C05VAJ5H3QS/p1699528303796609 for discussion.

At very large scales (150k replicas/node), changefeed aggregators can still keep up (with mux rangefeed).
However, when it's time for aggregators to flush their progress to change frontier, they all tend
to flush at the same time:

This make sense as in the case when aggregators keep up, their frontier will advance to the same timestamp
at the same time, and they will want to flush at roughly the same time. We should add a bit of jitter to each aggregator.

In addition, when aggregator flushes its span frontier to the coordinator, the flush happens on the "main" processor goroutine
(tick() function). That's suboptimal for 2 reasons:

1. We fall ever slightly behind consuming from blocking buffer. The effective throughput drops while the flush is pending; and the flush might be pending for some time since the span frontier might be large (i.e. we might be sending multi MB message to the coordinator)
2. The aggregator processor is also single threaded -- thus when all aggregators flush at roughly same time, they will experience "head of queue" blocking. It is even possible (or even likely?) that flushes on the last few aggregators may be blocked for so long that the rangefeeds on those aggregators might disconnect with SLOW_CONSUMER error (i.e. we blocked long enough to fill up blocking buffer) -- and that causes more expensive and unnecessary catchup scans.

The solution on the aggregator side is quite simple: address (or at least mitigate) second issue by adding jitter on the flush logic;
The first issue should be addressed by trying to flush on a dedicated goroutine. The span frontier we use is thread safe, so that's okay. But, this might be made more complicated due to how distsql procs work.

Jira issue: CRDB-33363

[blathers-crl]

cc @cockroachdb/cdc

[jayshrivastava]

I think the spikes every 5 minutes are because the min_checkpoint_frequency is 5 mins.

Since we ran the changefeed with no initial scan, the initial catchup scans all finish very fast (say within 30 seconds at most) and each aggregator gets its first checkpoint in this 30 second window and emits it. The aggregators are tick()ing relatively fast (multiple times per second) and see new checkpoints every so often, but they wait for 5 mins to emit them and this cycle repeats. This causes a spike every 5 mins. The jitter can just be added to that first checkpoint and can be a function of min_checkpoint_frequency.

When running an initial scan, I would expect the initial scan portion to finish at drastically different times for each aggregator (in the order of minutes), so the jitter is kind of built in. Regardless, adding more jitter wouldn't hurt.

[jayshrivastava]

Note that, in the graph, the queue time is spiking on one node. The queue time is a metric only measured by aggregators. Just to be clear, it sounds like the working theory is that the spiking aggregator is on the same node as the coordinator and the coordinator is stealing CPU from the aggregator on that node.

[miretskiy]

It makes sense that they flush every 5 minutes -- that's how things were configured.

Note that, in the graph, the queue time is spiking on one node. The queue time is a metric only measured by aggregators. Just to be clear, it sounds like the working theory is that the spiking aggregator is on the same node as the coordinator and the coordinator is stealing CPU from the aggregator on that node.

That's possible (or even likely).
I think what we should be carefully measuring is how much the effective throughput drops when we begin flush.
I think the above description makes sense overall -- we can try to come up with better ways of observiing this (and we should!) -- and we will likely discover what I described above as the cause for these spikes (or drops in throughput).

[sean-]

Adding jitter made a significant improvement to the cluster's health by reducing the effects of the aggregator's distributed fan-in. Can we please get this merged and backported?

[williamkulju]

@miretskiy Out of curiosity, are you going to try to backport the fix to 23.2? If not, why not?