Limit data production from receivers to avoid backpressure scenarios

[wbh1]

Component(s)

receiver/filelog, receiver/journald

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

This feature request is based on the FluentBit feature surrounding buffering described here: https://docs.fluentbit.io/manual/administration/backpressure

In our environment (and I'm sure many others), the destinations we send telemetry data to apply some known rate limits to the rate at which we are able to transmit data. For example, we operate a Loki cluster that applies a per-stream rate limit of XMB/s. However, there is currently no way (that I know of, at least) in otelcol to apply any sort of throttling to the amount of data produced [without losing data like through sampling].

In our specific use case, we're using the filelog receiver. Under heavy load, some systems may exceed the aforementioned rate limit in the amount of data they're producing. However, there is no good way to communicate this back to the data producer. This is because we observe the rate limits on our gateways and there's currently no way to limit the data coming into the gateway, although #6908 probably solves this.

Regardless, that rate limiter will just shift the problem of getting a clogged-up queue from the gateway to the agent. Instead, we'd rather be able to limit the speed at which data is produced to prevent either the agent or the gateway from getting backed up with retries.

Describe the solution you'd like

For receivers where it makes sense (e.g. filelog, journald, basically anything that is reading from an external data source instead of having data pushed to it), there could be a way to specify a maximum ingestion rate. I think this should be configured per-receiver, but I'm sure there's an argument to be made that it should be a processor.

Measuring by size is preferable, but you could also limit by a count of data points (e.g. log lines). When this limit is reached, ingestion is paused on the receiver until the next interval. At the next interval, the receiver will resume ingestion at the same offset where it left off so there is no data loss.

The receiver is expected to be allowed to fall behind, but this is an acceptable tradeoff in order to ensure that -- eventually -- all data will be present. This is in contrast to the current system in which backpressure scenarios can result in loss of data due to exhausting retries.

Describe alternatives you've considered

The alternatives I've considered are outlined above, but all result in different types of backpressure scenarios just shifting where it occurs to different places in a pipeline. To my knowledge, the only way to prevent backpressure is to either scale up your exporters' destination(s) to accept data faster (not always in control of users) or to throttle the production of data.

Additional context

Some receivers like journald don't currently have a system to track their progress in receiving data the way that filelog does with offsets. Instead, the journald receiver constantly receives data by tailing the journal. In such a case, pausing ingestion would still cause memory to balloon. This would then trigger the memory_limiter, if configured, and data would still be dropped. Probably not desirable, but that's a separate issue for implementing that sort of logic in that receiver.

[github-actions]

Pinging code owners:

• receiver/filelog: @djaglowski
• receiver/journald: @sumo-drosiek @djaglowski

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[djaglowski]

My understanding is that we expect these receivers to block when downstream components are not ready to consume data. @dmitryax, isn't this what was achieved with #20864?

[wbh1]

I would view this feature request as supplementary to the linked PR.

It is still useful to explicitly limit the production of data without needing to encounter backpressure and trigger retries. It would help to avoid potential loss of data due to retries being exhausted in a situation where (data_produced_in_interval / upstream_rate_limit) > max_possible_retries, and reduce the load on the receiving system when you already know the limits of it in advance.

Additionally, relying on the exporter to trigger the slowdown is not feasible in all architectures. For example, if you're sending from collector -> otel gateway -> destination and the destination is the one rate-limiting data ingestion, the collector will remain unaware of errors since -- from its perspective -- the otel gateway successfully received the data.

[djaglowski]

I think you have some good points here, that backpressure should not necessarily come from exporters only.

I'm thinking it would make the most sense as a per-receiver config, along the lines of consumerretry, which can be embedded into any receiver config. I'd be curious to here what @dmitryax thinks of this.

[github-actions]

This issue has been inactive for 60 days. It will be closed in 60 days if there is no activity. To ping code owners by adding a component label, see Adding Labels via Comments, or if you are unsure of which component this issue relates to, please ping @open-telemetry/collector-contrib-triagers. If this issue is still relevant, please ping the code owners or leave a comment explaining why it is still relevant. Otherwise, please close it.

Pinging code owners:

• receiver/journald: @sumo-drosiek @djaglowski
• receiver/filelog: @djaglowski

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[sumo-drosiek]

For journald I wonder if there is a way to rely on max system PIPE size, so if we don;t read from stdout fast enough, journalctl will wait until we start again.
If not, this change may require reading from journald manually 🤔

[wbh1]

Yeah... I don't think that we'd be able to keep using the -f flag to journalctl. Instead, I think we'd need to keep track of the __CURSOR value from the last read log line and then resume reading from that cursor position on the next read.

[atoulme]

Please look at the new exporterhelper batch sender here: https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/batch_sender.go it implements the ability to apply backpressure back to receivers.