[beats receivers] The interactions between otelconsumer and the ES exporter lose data on partial failure of _bulk

[cmacknz]

• Relates Deprecate Batcher and configure QueueBatch with Partitioner open-telemetry/opentelemetry-collector-contrib#41373

The interactions between the otelconsumer and the ES exporter likely lose data in cases where there is a document level failure in a batch request and the collector restarts. It is likely that most of these problems are actually bugs in the ES exporter, but we should add detailed failure mode testing here to ensure we preserve Filebeat's at least once delivery guarantees.

For reference, when you make a _bulk request to Elasticsearch the response will give you document level HTTP status codes. This lets you detect partial failure along with which specific documents failed. The Beats Elasticsearch output uses this to drop permanent failures (e.g. mapping conflict on a single document) and retry individual documents (e.g. document level 429s). This happens here:

beats/libbeat/outputs/elasticsearch/client.go

Lines 263 to 271 in d73c6c9

	eventsToRetry, stats := client.bulkCollectPublishFails(bulkResult)
	stats.reportToObserver(client.observer)
	if len(eventsToRetry) > 0 {
	span.Context.SetLabel("events_failed", len(eventsToRetry))
	batch.RetryEvents(eventsToRetry)
	} else {
	batch.ACK()
	}

The bulk indexer used in the ES exporter has similar logic here.

The otelconsumer only accepts an error which is for the whole batch of logs it tried to ingest:

beats/x-pack/libbeat/outputs/otelconsumer/otelconsumer.go

Lines 153 to 170 in d73c6c9

	err := out.logsConsumer.ConsumeLogs(ctx, pLogs)
	if err != nil {
	// Permanent errors shouldn't be retried. This tipically means
	// the data cannot be serialized by the exporter that is attached
	// to the pipeline or when the destination refuses the data because
	// it cannot decode it. Retrying in this case is useless.
	//
	// See https://github.com/open-telemetry/opentelemetry-collector/blob/1c47d89/receiver/doc.go#L23-L40
	if consumererror.IsPermanent(err) {
	st.PermanentErrors(len(events))
	batch.Drop()
	} else {
	st.RetryableErrors(len(events))
	batch.Retry()
	}
	return fmt.Errorf("failed to send batch events to otel collector: %w", err)
	}

It can detect permanent failures but I think this is only going to be for the entire batch.

The bulk indexer can return nil on flush if there were document level errors. Problem one is that our otelconsumer can’t see this and will ack the batch when this happens but the document in this case hasn't been persisted in Elasticsearch yet.

The bulk indexer logic linked earlier to retry failed individual documents covers for this by re-adding failed documents to the indexer’s writer, but this also has the effect that those logs will be get Flushed on a different pushDataLogs call (ES exporters ConsumeLogs function) in the future decoupling any error from the original call in our otelconsumer. This is problem two.

To observe this causing problems you’d need to cause partial _bulk failures that exhaust any configured retries and/or restart the collector so that the effect of our ack in otelconsumer has caused us to move past the uningested data in Filebeats registry file when ingesting a log file.

We need to do some targeted testing here where we force partial _bulk failures (e.g. permanent mapping conflict on only one doc and long lived but recoverable 429 response on only one doc).

I think the fundamental problem is that otelconsumer can’t see that only some documents may have succeeded, we are relying on retries in the ES exporter that can be stopped via a collector restart. We could consider partial failures as total failures and that would avoid data loss but cause us to re-index a lot more documents than we do today.

[elasticmachine]

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

[cmacknz]

We are still generating configurations that use the Batcher configuration

beats/libbeat/otelbeat/oteltranslate/outputs/elasticsearch/config_otel.go

Lines 135 to 140 in 9b4b114

	// Batcher is experimental
	"batcher": map[string]any{
	"enabled": true,
	"max_size": escfg.BulkMaxSize, // bulk_max_size
	"min_size": 0, // 0 means immediately trigger a flush
	},

This was deprecated in open-telemetry/opentelemetry-collector-contrib#41373 I believe to address some of the data loss concerns that seem to be primarily caused by the ES exporter using it's own post-queue in-memory buffer. @carsonip @lahsivjar it'd be great if you can confirm this is the path to fixing the data loss concerns in the ES exporter, Filebeat has a strict at least once delivery guarantee that we don't want to regress on when we switch to running Beat inputs in the collector.

[carsonip]

This was deprecated in open-telemetry/opentelemetry-collector-contrib#41373 I believe to address some of the data loss concerns that seem to be primarily caused by the ES exporter using it's own post-queue in-memory buffer.

There are only 2 types of bulk indexers in es exporter: sync and async.

• async: the old type where ES exporter has "its own post-queue in-memory buffer" and has reliability issues
• sync: the new type, where callers e.g. receivers and processors are blocked until the entire request finishes, including request level and doc level retries.

From beats/libbeat/otelbeat/oteltranslate/outputs/elasticsearch/config_otel.go I see that you're already using batcher, which uses the sync bulk indexer. I don't expect any "data loss" on partial failure.

This was deprecated in open-telemetry/opentelemetry-collector-contrib#41373

The deprecation changed the way how batcher is configured. Instead of being a standalone root level config, it is now under sending_queue. It also has some implications around batching without sending queue.

@lahsivjar apparently beats is using batcher without sending queue. Can you confirm the appropriate config after the deprecation, e.g. wait_for_result? I still think it isn't super straightforward why batching requires sending queue

[lahsivjar]

apparently beats is using batcher without sending queue. Can you confirm the appropriate config after the deprecation, e.g. wait_for_result? I still think it isn't super straightforward why batching requires sending queue

Yes, the following configuration should work:

        sending_queue:
          enabled: true
          num_consumers: <num_consumers>
          wait_for_result: true
          block_on_overflow: true 
          sizer: requests
          queue_size: <queue_length>
          batch:
             flush_timeout: {{ .flushTimeout }}
             sizer: items # items or bytes
             min_size: <min_size>
             max_size: <max_size>

[mauri870]

I started migrating the deprecated batcher config to sending_queue, but @VihasMakwana is already doing that as part of #46111, we'll end up implementing this there.

I added tests for otelconsumer delivery guarantees in TestConsumeContract, but they only cover ConsumeLogs-level behavior. I also wrote a mock-es test to simulate _bulk level failures, but that still does not give us per-document coverage for the scenarios mentioned. I'm starting with more in-depth testing of elasticsearchexporter for these cases.

[mauri870]

After investigating this for a bit and diving into the internals of the exporter and docappender, here is what I found regarding document-level failures and retries. This only applies to the synchronous bulk indexer mode with batcher/sending_queue + retryoptions. Please correct me if I misunderstood anything.

The bulk indexer can return nil on flush if there were document level errors. Problem one is that our otelconsumer can’t see this and will ack the batch when this happens but the document in this case hasn't been persisted in Elasticsearch yet.

While the otelconsumer does acknowledge the whole batch, the responsibility for retrying failed documents at a more granular level is delegated to the Elasticsearch exporter. Each ConsumeLogs call creates a new synchronous bulk indexer session. Within that session, the exporter automatically re-queues documents that failed with retryable errors and keeps flushing until either all retryable documents are ingested or the maximum retry attempts are reached. I added a test to go-docappender to confirm this holds true, I'm also going to add similar tests to elasticsearchexporter as well. This ensures that by the time ConsumeLogs returns, everything that could be successfully ingested has been ingested.

The bulk indexer logic linked earlier to retry failed individual documents covers for this by re-adding failed documents to the indexer’s writer, but this also has the effect that those logs will be get Flushed on a different pushDataLogs call (ES exporters ConsumeLogs function) in the future decoupling any error from the original call in our otelconsumer. This is problem two.

It’s true that retried documents are decoupled from their original batch in the lowest level, but the retry logic is still bounded to the scope of the same synchronous bulk indexer session. Since the session is unique to each call to ConsumeLogs, failures cannot "escape" into future sessions. That means tracking remains consistent: once ConsumeLogs completes, we have a clear and definitive outcome: either all retryable documents were accepted, or retries were exhausted. Only then otelconsumer will ack the batch.

So the current design guarantees that all documents that could be ingested are either successfully indexed or have reached their retry limit by the time the batch is acknowledged.

One problem I found while working on a document-level failures test for fbreceiver is that otelconsumer currently skews output metrics. It has no visibility on what the exporter does internally, some documents might have been retried and accepted, dropped due to max retries or permanent failures, but otelconsumer will always ACK the whole batch because it receives no error from the exporter and ConsumeLogs only works at the batch level. The only way I could find to access document-level results is to somehow peek into the ES exporter metrics.

[cmacknz]

Thanks all, I clearly did not fully parse the way the session ties back to the consume logs call when I went through the ES exporter code.

Each ConsumeLogs call creates a new synchronous bulk indexer session. Within that session, the exporter automatically re-queues documents that failed with retryable errors and keeps flushing until either all retryable documents are ingested or the maximum retry attempts are reached.

Does this mean that the _bulk request size shrinks as document level failures are retried or am I misinterpreting? Beats sends document level failures back into the queue to be retried as part of a different batch, to keep the batch size efficiently large. This works because our acknowledgements use a per event callback mechanism.

[carsonip]

Does this mean that the _bulk request size shrinks as document level failures are retried or am I misinterpreting?

Correct. There is no mixing between batches, to maintain the tracking that when a Consume* returns, all docs associated with this request are either all successful or some of them failed after retrying until max. It was a deliberate design to avoid data loss when used with e.g. persistent queue. But we can discuss whether this is desirable.

In contrary to this, apm-server does something like beats where per-doc retries are done in async and will be batched up more optimally, but without using the per-event callback in es bulk indexer client. We implemented it in go-docappender: https://github.com/elastic/go-docappender/blob/c58b6b2659d50131d6b4a3bcfb66145234c9e5dc/bulk_indexer.go#L676

[cmacknz]

Thanks for now prioritizing no data loss is more important. I think this means we are going to need to do some performance tests when ES is experiencing 429s, though honestly my suspicion is that mOTLP is going to flush this out on our behalf and we can just benefit from improvements driven from that side of things.