[fbreceiver] - Fix batcher's configuration

[VihasMakwana]

Proposed commit message

As per batcher's documentation, there are two configurations related to flushing:

• min_size_items
  • Minimum number of log records/metric points in a request to immediately trigger a batcher flush.
  • For eg. if a plog.Log has more records than min_size_items then it is immediately forwarded to the exporter.
• max_size_items
  • Maximum log records/metric points per request.
  • For eg. if a plog.Log has more records than max_size_items then it is split into multiple requests.
  • This doesn't necessarily mean that it will be sent to exporter. That decision will be based on flush timeout or min_size_items.

This PR sets min_size_items to 0. This ensures both of the following conditions are satisfied:

1. No plog.Log contains more than bulk_max_size items.
2. Batcher flushes the data as soon it gets it

[elasticmachine]

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

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @VihasMakwana? 🙏.
For such, you'll need to label your PR with:

• The upcoming major version of the Elastic Stack
• The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit
• backport-active-all is the label that automatically backports to all active branches.
• backport-active-8 is the label that automatically backports to all active minor branches for the 8 major.
• backport-active-9 is the label that automatically backports to all active minor branches for the 9 major.

[khushijain21]

I think the correct parameter to map min_size_items on the exporter is flush.min_events.

They immediately trigger a flush based on minimum events/records set.

For more info https://www.elastic.co/guide/en/beats/filebeat/current/configuring-internal-queue.html#queue-mem-flush-min-events-option

[VihasMakwana]

flush.min_events

I initially did it, but as per the same document you linked:

flush.min_events is a legacy parameter, and new configurations should prefer to control batch size with bulk_max_size. 
As of 8.13, there is never a performance advantage to limiting batch size with flush.min_events instead of bulk_max_size.

Also, flush.min_events is used when we set presets. To avoid all this I would rather set this on batcher.

There is one more problem which is solved by this PR (I'll update description):

• default batcher min_size_items is 5000 and we set max_size_items to 1600 (same as filebeat bulk_max_size).
• This essentially means that no batch will have more than 1600 records and as a result, it will not be flushed until flush timeout is reached because minimum required size is 5000 (and flush timeout is 30s for batcher)

i hope that makes sense.

[mauri870]

@carsonip Would be great to get your opinion on this change.

[khushijain21]

This essentially means that no batch will have more than 1600 records and as a result, it will not be flushed until flush timeout is reached because minimum required size is 5000 (and flush timeout is 30s for batcher)

The batcher validates that max_size_items is greater than or equal to min_size_items so we never run into this situation.
https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterbatcher/config.go#L53-L55

Also, flush.min_events is used when we set presets. To avoid all this I would rather set this on batcher.

I agree with you. Since every event has to go through two queues before reaching ES. But setting min_size_items equal to max_size_items does not seem the correct thing to do. I'll wait for what others have to say.

[carsonip]

thanks for the ping and PR lgtm.
2 things to note, both are irrelevant to the PR

• one should NOT use es exporter batcher::max_size_items or batchprocessor's send_batch_max_size in metrics pipelines as it may break down metrics requests which would cause issues with TSDB. In this PR it is a log pipeline and it is fine.
• check out the es exporter flush::bytes config. It is respected regardless of batcher settings. If 1 batch is greater than flush::bytes (default ~5MB) it will be broken down into 2 bulk requests

[VihasMakwana]

This essentially means that no batch will have more than 1600 records and as a result, it will not be flushed until flush timeout is reached because minimum required size is 5000 (and flush timeout is 30s for batcher)

The batcher validates that max_size_items is greater than or equal to min_size_items so we never run into this situation. https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterbatcher/config.go#L53-L55

Also, flush.min_events is used when we set presets. To avoid all this I would rather set this on batcher.

I agree with you. Since every event has to go through two queues before reaching ES. But setting min_size_items equal to max_size_items does not seem the correct thing to do. I'll wait for what others have to say.

This is the bug in upstream exporter. It never validates the batcher configuration and we never run into any error. min_size_items is set to 5000 and we set max_size_items to 1600. We should run into this validation error, but we don't

I'm moving this PR to draft for now until I fix upstream.

[carsonip]

But setting min_size_items equal to max_size_items does not seem the correct thing to do. I'll wait for what others have to say.

if min_size_items == max_size_items, the (first) batch will always be of that size if it isn't triggered by batcher::flush_timeout.

[carsonip]

On a side note, setting min_size_items == max_size_items may be inefficient, as splitting a request and exporting them separately will incur some overhead (think about 2 bulk requests vs 1). e.g. if the current batched request has 1599 records, and the next request has 2 records, I assume that it will trigger a flush with a request of 1600 records and another request of 1 record, which will make it very inefficient. Would be good to have someone to confirm this.

[VihasMakwana]

BLOCKED ON open-telemetry/opentelemetry-collector-contrib#38072

[VihasMakwana]

On a side note, setting min_size_items == max_size_items may be inefficient, as splitting a request and exporting them separately will incur some overhead (think about 2 bulk requests vs 1). e.g. if the current batched request has 1599 records, and the next request has 2 records, I assume that it will trigger a flush with a request of 1600 records and another request of 1 record, which will make it very inefficient. Would be good to have someone to confirm this.

You're right. Can you please take a look at open-telemetry/opentelemetry-collector-contrib#38072? I'd like to get that merged before continuing on this one.

[carsonip]

can you test this out? I've always avoided setting otel configs to 0, because they will not override any non-0 default. In this case, I'm not sure if setting it explicitly as 0 will override the default 5000. With your PR in open-telemetry/opentelemetry-collector-contrib#38072 you should be able to tell it right away.

[VihasMakwana]

The performance testing is currently in progress. It looks good and EPS is similar to native filebeat. I'll manually test it as well.

[VihasMakwana]

@carsonip attaching a few screenshots for better transparency:

1. collector run with min_size_items: 0

2. collector run while skipping min_size_items (it defaults to 5000 in this case)

[pierrehilbert]

=== FAIL: libbeat/otelbeat/oteltranslate/outputs/elasticsearch TestToOtelConfig/check_preset_config_translation/config_translation_w/latency (0.00s)
--
  | config_otel_test.go:198:
  | Error Trace:	/opt/buildkite-agent/builds/bk-agent-prod-gcp-1740064094203004939/elastic/beats-libbeat/libbeat/otelbeat/oteltranslate/outputs/elasticsearch/config_otel_test.go:223
  | /opt/buildkite-agent/builds/bk-agent-prod-gcp-1740064094203004939/elastic/beats-libbeat/libbeat/otelbeat/oteltranslate/outputs/elasticsearch/config_otel_test.go:198
  | Error:      	Not equal:
  | expected: "api_key: \"\"\nbatcher:\n  enabled: true\n  max_size_items: 50\n  min_size_items: 50\nendpoints:\n- http://localhost:9200\nidle_conn_timeout: 1m0s\nlogs_index: some-index\nnum_workers: 1\npassword: changeme\nretry:\n  enabled: true\n  initial_interval: 1s\n  max_interval: 1m0s\n  max_retries: 3\ntimeout: 1m30s\nuser: elastic\n"
  | actual  : "api_key: \"\"\nbatcher:\n  enabled: true\n  max_size_items: 50\n  min_size_items: 0\nendpoints:\n- http://localhost:9200\nidle_conn_timeout: 1m0s\nlogs_index: some-index\nnum_workers: 1\npassword: changeme\nretry:\n  enabled: true\n  initial_interval: 1s\n  max_interval: 1m0s\n  max_retries: 3\ntimeout: 1m30s\nuser: elastic\n"

[VihasMakwana]

@khushijain21 @mauri870 can you take a look now? I've update PR description with new set of changes. We now set min_size_items: 0. This would immediately flush the data as soon as it gets it.

Let me know if you have any concerns. Happy to help you out.

[VihasMakwana]

/test