From 5d348a62387fc2b7b55f243d203d463f53cf5ac4 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Sat, 17 Oct 2026 07:17:07 -0600 Subject: [PATCH 1/2] out_opentelemetry: send every pending log record once The final export of a chunk only checked the record count of the last scope, which restarts at zero for every scope. A chunk that ended with a group start and no records yet, such as a group whose records arrive in a later append, dropped all its earlier records while the flush reported success. Count the records pending since the last export instead. A batch export also released only the records of the current scope, so records of earlier scopes in the same request were sent again with the next batch. Release the records of every scope after an export. Signed-off-by: Eduardo Silva --- .../out_opentelemetry/opentelemetry_logs.c | 42 +++++++++++++++++-- 1 file changed, 39 insertions(+), 3 deletions(-) diff --git a/plugins/out_opentelemetry/opentelemetry_logs.c b/plugins/out_opentelemetry/opentelemetry_logs.c index e1366a995bb..8e8f8d11af7 100644 --- a/plugins/out_opentelemetry/opentelemetry_logs.c +++ b/plugins/out_opentelemetry/opentelemetry_logs.c @@ -862,6 +862,36 @@ static void free_log_records(Opentelemetry__Proto__Logs__V1__LogRecord **logs, s } } +/* + * Release the log records of a request that was already sent, keeping its + * resources and scopes, so a later request of the same chunk does not repeat + * them. + */ +static void release_sent_log_records(Opentelemetry__Proto__Collector__Logs__V1__ExportLogsServiceRequest *export_logs) +{ + size_t i; + size_t j; + Opentelemetry__Proto__Logs__V1__ResourceLogs *resource_log; + Opentelemetry__Proto__Logs__V1__ScopeLogs *scope_log; + + for (i = 0; i < export_logs->n_resource_logs; i++) { + resource_log = export_logs->resource_logs[i]; + if (resource_log == NULL) { + continue; + } + + for (j = 0; j < resource_log->n_scope_logs; j++) { + scope_log = resource_log->scope_logs[j]; + if (scope_log == NULL || scope_log->log_records == NULL) { + continue; + } + + free_log_records(scope_log->log_records, scope_log->n_log_records); + scope_log->n_log_records = 0; + } + } +} + static void free_resource_logs(Opentelemetry__Proto__Logs__V1__ResourceLogs **resource_logs, size_t resource_count) { int i; @@ -1132,6 +1162,7 @@ int otel_process_logs(struct flb_event_chunk *event_chunk, int ret; int record_type; int log_record_count; + int pending_records = 0; int max_scopes_limit; int max_resources; int native_otel = FLB_FALSE; @@ -1582,13 +1613,14 @@ int otel_process_logs(struct flb_event_chunk *event_chunk, ret = FLB_OK; log_record_count++; + pending_records++; scope_log->n_log_records = log_record_count; if (log_record_count >= ctx->batch_size) { ret = logs_flush_to_otel(ctx, event_chunk, &export_logs); - free_log_records(log_records, log_record_count); + release_sent_log_records(&export_logs); log_record_count = 0; - scope_log->n_log_records = 0; + pending_records = 0; /* Do not let a later batch overwrite this failure. */ if (ret != FLB_OK) { @@ -1599,7 +1631,11 @@ int otel_process_logs(struct flb_event_chunk *event_chunk, flb_log_event_decoder_destroy(decoder); - if (log_record_count > 0 && ret == FLB_OK) { + /* + * Records of earlier scopes are pending too: a chunk can end with a group + * that has no records yet, for example one opened by the previous append. + */ + if (pending_records > 0 && ret == FLB_OK) { ret = logs_flush_to_otel(ctx, event_chunk, &export_logs); } From 784b18312e5f59144e3e8642c07c6ac6b4e50b5b Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Sat, 17 Oct 2026 07:17:07 -0600 Subject: [PATCH 2/2] tests: integration: out_opentelemetry: cover empty trailing groups and batches Signed-off-by: Eduardo Silva --- ...out_otel_http_logs_forward_batch_size.yaml | 18 ++++ .../out_otel_http_logs_forward_groups.yaml | 17 ++++ .../tests/test_out_opentelemetry_001.py | 89 +++++++++++++++++++ 3 files changed, 124 insertions(+) create mode 100644 tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_forward_batch_size.yaml create mode 100644 tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_forward_groups.yaml diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_forward_batch_size.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_forward_batch_size.yaml new file mode 100644 index 00000000000..0fdad8c9533 --- /dev/null +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_forward_batch_size.yaml @@ -0,0 +1,18 @@ +service: + flush: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: forward + listen: 127.0.0.1 + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + + outputs: + - name: opentelemetry + match: "*" + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + batch_size: 3 diff --git a/tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_forward_groups.yaml b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_forward_groups.yaml new file mode 100644 index 00000000000..2d327081491 --- /dev/null +++ b/tests/integration/scenarios/out_opentelemetry/config/out_otel_http_logs_forward_groups.yaml @@ -0,0 +1,17 @@ +service: + flush: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: forward + listen: 127.0.0.1 + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + + outputs: + - name: opentelemetry + match: "*" + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} diff --git a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py index 0481f64b65c..647131c6720 100644 --- a/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py +++ b/tests/integration/scenarios/out_opentelemetry/tests/test_out_opentelemetry_001.py @@ -6,6 +6,7 @@ import shutil import socket import threading +import time from collections import Counter import requests @@ -1454,3 +1455,91 @@ def test_out_opentelemetry_logs_max_scopes_enforcement(): output = json.loads(json_format.MessageToJson(logs_seen[0])) assert len(output["resourceLogs"]) == 4 assert all(len(resource_log["scopeLogs"]) == 1 for resource_log in output["resourceLogs"]) + + +def _pack(value): + """Encode forward fixtures without an optional MessagePack dependency.""" + if isinstance(value, bytes): + return value + if isinstance(value, int): + return b"\xcf" + value.to_bytes(8, "big") + if isinstance(value, str): + data = value.encode("utf-8") + return b"\xdb" + len(data).to_bytes(4, "big") + data + if isinstance(value, list): + return b"\xdd" + len(value).to_bytes(4, "big") + b"".join(_pack(item) for item in value) + if isinstance(value, dict): + return b"\xdf" + len(value).to_bytes(4, "big") + b"".join( + _pack(key) + _pack(item) for key, item in value.items()) + raise TypeError(type(value)) + + +def _event_time(seconds): + return b"\xd7\x00" + seconds.to_bytes(4, "big") + bytes(4) + + +def _group_start(resource_id, name): + return [[_event_time(0xFFFFFFFF), {"schema": "otlp", "resource_id": resource_id, "scope_id": 0}], + {"resource": {"attributes": {"service.name": name}}, "scope": {"name": "scope"}}] + + +def _group_end(): + return [[_event_time(0xFFFFFFFE), {}], {}] + + +def _log_record(record_id): + return [[_event_time(1700000000), {"otlp": {}}], {"id": record_id}] + + +def _send_forward(service, entries): + with socket.create_connection(("127.0.0.1", service.flb_listener_port), timeout=10) as connection: + connection.sendall(_pack(["test", entries])) + + +def _exported_record_ids(): + ids = [] + for export_request in data_storage["logs"]: + output = json.loads(json_format.MessageToJson(export_request)) + for record, _, _ in iter_log_records(output): + for item in record["body"]["kvlistValue"]["values"]: + if item["key"] == "id": + ids.append(int(item["value"]["intValue"])) + return sorted(ids) + + +def test_out_opentelemetry_chunk_ending_with_empty_group_exports_its_records(): + """A chunk that ends with a group start and no records still exports the earlier groups.""" + service = Service("out_otel_http_logs_forward_groups.yaml") + service.start() + try: + # the next group starts in this append and receives its records later + _send_forward(service, [_group_start(0, "first"), _log_record(0), _log_record(1), + _group_end(), _group_start(1, "second")]) + service.service.wait_for_condition( + lambda: _exported_record_ids() == [0, 1] or None, + timeout=15, interval=0.5, description="records of the first group") + + _send_forward(service, [_log_record(2), _group_end()]) + service.service.wait_for_condition( + lambda: _exported_record_ids() == [0, 1, 2] or None, + timeout=15, interval=0.5, description="record appended to the open group") + finally: + service.stop() + + +def test_out_opentelemetry_batch_size_exports_each_record_once(): + """Records of earlier scopes are not sent again by a later batch.""" + service = Service("out_otel_http_logs_forward_batch_size.yaml") + service.start() + try: + entries = [_group_start(0, "first")] + [_log_record(i) for i in range(2)] + [_group_end()] + entries += [_group_start(1, "second")] + [_log_record(i) for i in range(2, 7)] + [_group_end()] + _send_forward(service, entries) + service.service.wait_for_condition( + lambda: len(_exported_record_ids()) >= 7 or None, + timeout=15, interval=0.5, description="seven exported records") + # give a duplicated export time to arrive + time.sleep(2) + assert _exported_record_ids() == list(range(7)) + finally: + service.stop()