out_opentelemetry: send every pending log record once

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

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

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}

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