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); } 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()