changefeed,kv: Excessive memory usage when using rangefeeds.

[miretskiy]

This is technically a rangefeed issue as the problem clearly lies in the rangefeed subsystem.

However, it is only applicable to the changefeed system at this point (but will almost certainly impact replication streaming as well). Other users of rangefeeds use them on small system tables and are unlikely to experience this issue.

The crux of the problem is that running rangefeed against substantially sized table (thousands of ranges per node) can result in OOMs.

When establishing rangefeed streaming rpc, each range is treated as a separate stream. Each stream is configured to use
2MB initial window size (pkg/rpc/context.go). The "initial" name is a bit of a misnomer since using any window size above gRPC default of 65K turns off dynamic window sizing -- so, in effect, this is an upper bound on the amount of memory which can be used by an RPC stream (lots of historical context can be found here: #35161).

Critically though the memory used by gRPC stream is not tracked by any memory monitors.
This memory is "inflight" and the consumer of RPC stream is expected to consume this data quickly. However, no matter how fast the consumer is, it is possible that in a sufficiently sized system, just the mere fact of establishing a rangefeed could result in an OOM. For example, consider a 31 node cluster where a single node wants to talk to all other nodes, each containing 500 ranges. This would result in 30->1 fan-in scenario, where if those 500 ranges per node had enough data to send, we could wind up attempting to ingest 30GB of data very rapidly. Pre-reserving 30GB of memory on any memory monitor is excessive -- and probably incorrect.

The second issue, applicable to changefeeds (and soon to replication streaming) is the fact that these system emit data downstream to external system -- possibly over high latency links. Changefeed has an effective pushback mechanism (which uses memory monitor) to deal with slow sinks. Unfortunately, as described above, if the consumer of the RPC stream is not consuming, then there will be 2MB worth of data buffered by the stream itself. From the dist_sender_rangefeed.go:

for {
			event, err := stream.Recv()
			...
			select {
			case eventCh <- event:
			...
			}
		}

When emission to events channel blocks -- for any reason, the stream still keeps receiving the data, up to 2MB.

The above observations were verified by writing up a unit test that emitted 4MB of data into each of the 100 ranges, and capturing heap profiles.

As expected, 200 MB (2MB per stream) are allocated by http2 client transport.

Lowering initial window size to 64K+1 (plus 1 byte to ensure hard limit) results in 6MB:

A full solution to this problem is expected not to be backported. So, we'll start off with a partial solution that
will be backportable.

We need to add a client side connection class specifically for rangefeed. This connection class will use much lower initial window size. Perhaps 196K (hopefully making OOMs 10x less likely); or even 65K+1 -- a 30x reduction.

A real solution is to ensure that each client rangefeed call (which can span many ranges) establishes 1 stream per node and multiplex all of feeds on that single stream so that per-stream pushback mechanism is effective. That is, each changefeed will establish 1 stream per node instead of 1 stream per range.

Solutions that probably don't solve anything:

• Let's have changefeeds disconnect when they enter pushback. This doesn't solve anything because resume w/ catchup scan could hit us with more data even more rapidly. Changefeed disconnect is probably too late anyway.
• Let's have connection level limits and pushback: GRPC explicitly removed connection level limits. All pushback is per stream.

Jira issue: CRDB-11977
Epic CRDB-23738

[blathers-crl]

cc @cockroachdb/cdc

[mwang1026]

@miretskiy @amruss removing kv since you seem to be handling this

[miretskiy]

Mux rangefeed merged; it is somewhat undertested, but that's not part of this issue.
Closing this issue.