[beats receivers] Eliminate the need for the otelconsumer output and queue

[cmacknz]

This is a follow up from #44009 (comment) which identifies that we need more concurrent uses of the ConsumeLogs() function in the existing otelconsumer output.

A valid collector configuration today looks like and includes an output key that creates an instance of the otelconsumer output and requires the entire Beat pipeline including the queue to be duplicated inside the receiver.

  filebeatreceiver:
    filebeat:
      inputs:
        - type: benchmark
          enabled: true
          count: 1
          message: {{.Message}}
    output:
      otelconsumer:
    logging:
      level: info
      selectors:
        - '*'
    path.home: /tmp
exporters:
  file/no_rotation:
    path: /tmp/output
service:
  pipelines:
    logs:
      receivers: [filebeatreceiver]
      exporters: [file/no_rotation]

Evaluate if the otelconsumer output and the Beat internal queue can be removed. The largest impact is expected to be on the flow of acknowledgements through the Beat which need to remain correct after this change. The format conversions done by the otelconsumer output will need to remain in the Beat, perhaps as part of the pipeline publish client.

[elasticmachine]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[iamrajiv]

hey @cmacknz if its ok i can help on this issue ive been looking at the current otelconsumer implementation and i see it creates a full beat pipeline per receiver am i correct that the main issue is were duplicating queues unnecessarily the otel collector already has its own queuing mechanisms right?

[cmacknz]

is were duplicating queues unnecessarily the otel collector already has its own queuing mechanisms right?

Yes this is the problem. There is another piece to this though which is that if you remove the duplicated beat queue+output you will also remove the event acknowledgement mechanism inside the Beat event pipeline (a core part of the Filebeat at least once delivery guarantee and what prevents re-reading log files on startup).

We think reworking this is necessary but also that doing this correctly is going to be fairly complex which is why we haven't touched it yet. I suspect this one is probably too big to take on as a new contributor.

[cmacknz]

Recording here for future reference, when we do this we'll have to account for how we dealt with the per event GuaranteedSend flag in #45492

[swiatekm]

Maybe a simpler solution would be to just make the queue small and only use it for internal batching? If we're going to use the otel exporter batcher in synchronous mode, then sending one event per request is inefficient even if the semantics are correct. Something like batches of 100 events with a queue of size 300-500 would probably work fine?

Alternatively, we run the exporter with block_on_overflow disabled and add a shim which starts a goroutine for each event, acking when it's consumed, and does a blocking retry loop if it gets a queue full error. If both block_on_overlflow and wait_for_result are enabled, there's no way to know why we're blocking, unfortunately.

[cmacknz]

Shrinking the queue is a good idea that we can try and benchmark immediately. This would mitigate the additional memory overhead of having the beat receivers queue fill up when backpressured.