processor_opentelemetry_envelope: new processor to package non OTel Logs content as OTel

cmake/plugins_options.cmake

@@ -67,7 +67,7 @@ DEFINE_OPTION(FLB_PROCESSOR_CONTENT_MODIFIER  "Enable content modifier processor
 DEFINE_OPTION(FLB_PROCESSOR_LABELS            "Enable metrics label manipulation processor"  ON)
 DEFINE_OPTION(FLB_PROCESSOR_METRICS_SELECTOR  "Enable metrics selector processor"            ON)
 DEFINE_OPTION(FLB_PROCESSOR_SQL               "Enable SQL processor"                         ON)
-
+DEFINE_OPTION(FLB_PROCESSOR_OPENTELEMETRY_ENVELOPE "Enable OpenTelemetry envelope processor"  ON)
 
 # Filters
 # =======

plugins/CMakeLists.txt

@@ -286,6 +286,7 @@ REGISTER_PROCESSOR_PLUGIN("processor_content_modifier")
 REGISTER_PROCESSOR_PLUGIN("processor_labels")
 REGISTER_PROCESSOR_PLUGIN("processor_metrics_selector")
 REGISTER_PROCESSOR_PLUGIN("processor_sql")
+REGISTER_PROCESSOR_PLUGIN("processor_opentelemetry_envelope")
 
 # OUTPUTS
 # =======

plugins/out_opentelemetry/opentelemetry_logs.c

@@ -528,8 +528,8 @@ static void free_resource_logs(Opentelemetry__Proto__Logs__V1__ResourceLogs **re
 
         if (resource_log->resource->attributes != NULL) {
             otlp_kvarray_destroy(resource_log->resource->attributes, resource_log->resource->n_attributes);
-            flb_free(resource_log->resource);
         }
+        flb_free(resource_log->resource);
 
         /* iterate scoipe logs */
         if (resource_log->n_scope_logs > 0) {

plugins/processor_opentelemetry_envelope/CMakeLists.txt

@@ -0,0 +1,4 @@
+set(src
+    otel_envelope.c)
+
+FLB_PLUGIN(processor_opentelemetry_envelope "${src}" "")

plugins/processor_opentelemetry_envelope/otel_envelope.c

@@ -0,0 +1,224 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2024 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+
+#include <fluent-bit/flb_processor_plugin.h>
+#include <cfl/cfl.h>
+
+/* Processor initialization */
+static int cb_init(struct flb_processor_instance *ins,
+                   void *source_plugin_instance,
+                   int source_plugin_type,
+                   struct flb_config *config)
+{
+    return FLB_PROCESSOR_SUCCESS;
+}
+
+/* Processor exit */
+static int cb_exit(struct flb_processor_instance *ins, void *data)
+{
+    return FLB_PROCESSOR_SUCCESS;
+}
+
+/* Create an group start with OTLP style-signature */
+static struct flb_mp_chunk_record *envelop_init(struct cfl_list *list, struct flb_mp_chunk_record *active_record)
+{
+    int ret;
+    struct cfl_kvlist *kvlist_meta = NULL;
+    struct cfl_kvlist *kvlist_record = NULL;
+    struct cfl_kvlist *kvlist_resource = NULL;
+    struct cfl_kvlist *kvlist_scope = NULL;
+    struct cfl_object *cobj_meta = NULL;
+    struct cfl_object *cobj_record = NULL;
+    struct flb_mp_chunk_record *record = NULL;
+    struct flb_time tm;
+
+    /* metadata */
+    kvlist_meta = cfl_kvlist_create();
+    if (!kvlist_meta) {
+        return NULL;
+    }
+
+    cfl_kvlist_insert_string(kvlist_meta, "schema", "otlp");
+    cfl_kvlist_insert_int64(kvlist_meta, "resource_id", 0);
+    cfl_kvlist_insert_int64(kvlist_meta, "scope_id", 0);
+
+    /* empty content */
+    kvlist_record = cfl_kvlist_create();
+    if (!kvlist_record) {
+        goto failure;
+    }
+
+    kvlist_resource = cfl_kvlist_create();
+    if (!kvlist_resource) {
+        goto failure;
+    }
+
+    kvlist_scope = cfl_kvlist_create();
+    if (!kvlist_scope) {
+        goto failure;
+    }
+
+    cfl_kvlist_insert_kvlist(kvlist_record, "resource", kvlist_resource);
+    cfl_kvlist_insert_kvlist(kvlist_record, "scope", kvlist_scope);
+
+    record = flb_mp_chunk_record_create(NULL);
+    if (!record) {
+        goto failure;
+    }
+
+    cobj_meta = cfl_object_create();
+    if (!cobj_meta) {
+        goto failure;
+    }
+    ret = cfl_object_set(cobj_meta, CFL_OBJECT_KVLIST, kvlist_meta);
+    if (ret != 0) {
+        goto failure;
+    }
+
+    cobj_record = cfl_object_create();
+    if (!cobj_record) {
+        goto failure;
+    }
+    ret = cfl_object_set(cobj_record, CFL_OBJECT_KVLIST, kvlist_record);
+    if (ret != 0) {
+        goto failure;
+    }
+
+    /* set the group flag in the timestamp field */
+    flb_time_set(&tm, FLB_LOG_EVENT_GROUP_START, 0);
+    flb_time_copy(&record->event.timestamp, &tm);
+
+    record->modified  = FLB_TRUE;
+    record->cobj_metadata = cobj_meta;
+    record->cobj_record = cobj_record;
+
+    /* add the envelop before the active record */
+    cfl_list_add_before(&record->_head, &active_record->_head, list);
+
+    return record;
+
+failure:
+    if (kvlist_meta) {
+        cfl_kvlist_destroy(kvlist_meta);
+    }
+    if (kvlist_record) {
+        cfl_kvlist_destroy(kvlist_record);
+    }
+    if (kvlist_resource) {
+        cfl_kvlist_destroy(kvlist_resource);
+    }
+    if (kvlist_scope) {
+        cfl_kvlist_destroy(kvlist_scope);
+    }
+    if (cobj_meta) {
+        cfl_object_destroy(cobj_meta);
+    }
+    if (cobj_record) {
+        cfl_object_destroy(cobj_record);
+    }
+    if (record) {
+        flb_mp_chunk_cobj_record_destroy(NULL, record);
+    }
+
+    return NULL;
+}
+
+/* Create an group end */
+static void envelop_end(struct cfl_list *list, struct flb_mp_chunk_record *active_record)
+{
+    struct flb_time tm;
+    struct flb_mp_chunk_record *record;
+
+    /* set the group flag in the timestamp field */
+    record = flb_mp_chunk_record_create(NULL);
+    if (!record) {
+        return;
+    }
+
+    flb_time_set(&tm, FLB_LOG_EVENT_GROUP_END, 0);
+    flb_time_copy(&record->event.timestamp, &tm);
+
+    record->modified  = FLB_TRUE;
+    record->cobj_metadata = NULL;
+    record->cobj_record = NULL;
+
+    /* add the envelop before the active record */
+    cfl_list_add_after(&record->_head, &active_record->_head, list);
+}
+
+
+#include <fluent-bit/flb_pack.h>
+
+/* Logs callback */
+static int cb_process_logs(struct flb_processor_instance *ins,
+                           void *chunk_data, const char *tag, int tag_len)
+{
+    int ret;
+    int record_type;
+    int grouped = FLB_FALSE;
+    struct flb_mp_chunk_record *prev_record;
+    struct flb_mp_chunk_record *record;
+    struct flb_mp_chunk_cobj *chunk_cobj = (struct flb_mp_chunk_cobj *) chunk_data;
+
+
+    /* Iterate records */
+    while (flb_mp_chunk_cobj_record_next(chunk_cobj, &record) == FLB_MP_CHUNK_RECORD_OK) {
+        prev_record = record;
+
+        /* get record type */
+        ret = flb_log_event_decoder_get_record_type(&record->event, &record_type);
+        if (ret != 0) {
+            flb_plg_error(ins, "record has invalid event type");
+            continue;
+        }
+
+        if (record_type == FLB_LOG_EVENT_NORMAL && grouped == FLB_FALSE) {
+            envelop_init(&chunk_cobj->records, record);
+            grouped = FLB_TRUE;
+        }
+        else if (record_type == FLB_LOG_EVENT_GROUP_START && grouped == FLB_TRUE) {
+            envelop_end(&chunk_cobj->records, record);
+            grouped = FLB_FALSE;
+        }
+    }
+
+    if (grouped == FLB_TRUE) {
+        envelop_end(&chunk_cobj->records, prev_record);
+    }
+
+    return FLB_PROCESSOR_SUCCESS;
+}
+
+static struct flb_config_map config_map[] = {
+    /* EOF */
+    {0}
+};
+
+struct flb_processor_plugin processor_opentelemetry_envelope_plugin = {
+    .name               = "opentelemetry_envelope",
+    .description        = "Package log records inside an OpenTelemetry Logs schema",
+    .cb_init            = cb_init,
+    .cb_process_logs    = cb_process_logs,
+    .cb_process_metrics = NULL,
+    .cb_process_traces  = NULL,
+    .cb_exit            = cb_exit,
+    .config_map         = config_map,
+    .flags              = 0,
+};

src/flb_mp.c

@@ -1055,6 +1055,23 @@ struct flb_mp_chunk_cobj *flb_mp_chunk_cobj_create(struct flb_log_event_encoder
     return chunk_cobj;
 }
 
+static int generate_empty_msgpack_map(char **out_buf, size_t *out_size)
+{
+    msgpack_sbuffer mp_sbuf;
+    msgpack_packer mp_pck;
+
+    /* initialize msgpack buffer */
+    msgpack_sbuffer_init(&mp_sbuf);
+    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+    msgpack_pack_map(&mp_pck, 0);
+
+    *out_buf = mp_sbuf.data;
+    *out_size = mp_sbuf.size;
+
+    return 0;
+}
+
 int flb_mp_chunk_cobj_encode(struct flb_mp_chunk_cobj *chunk_cobj, char **out_buf, size_t *out_size)
 {
     int ret;
@@ -1070,9 +1087,6 @@ int flb_mp_chunk_cobj_encode(struct flb_mp_chunk_cobj *chunk_cobj, char **out_bu
     /* Iterate all records */
     cfl_list_foreach(head, &chunk_cobj->records) {
         record = cfl_list_entry(head, struct flb_mp_chunk_record, _head);
-        if (record->modified == FLB_TRUE) {
-            continue;
-        }
 
         ret = flb_log_event_encoder_begin_record(chunk_cobj->log_encoder);
         if (ret == -1) {
@@ -1089,19 +1103,33 @@ int flb_mp_chunk_cobj_encode(struct flb_mp_chunk_cobj *chunk_cobj, char **out_bu
             if (ret == -1) {
                 return -1;
             }
-
-            ret = flb_log_event_encoder_set_metadata_from_raw_msgpack(chunk_cobj->log_encoder, mp_buf, mp_size);
-            if (ret != FLB_EVENT_ENCODER_SUCCESS) {
-                flb_free(mp_buf);
+        }
+        else {
+            ret = generate_empty_msgpack_map(&mp_buf, &mp_size);
+            if (ret == -1) {
                 return -1;
             }
-            flb_free(mp_buf);
         }
 
-        ret = flb_mp_cfl_to_msgpack(record->cobj_record, &mp_buf, &mp_size);
-        if (ret == -1) {
+        ret = flb_log_event_encoder_set_metadata_from_raw_msgpack(chunk_cobj->log_encoder, mp_buf, mp_size);
+        if (ret != FLB_EVENT_ENCODER_SUCCESS) {
+            flb_free(mp_buf);
             return -1;
         }
+        flb_free(mp_buf);
+
+        if (record->cobj_record) {
+            ret = flb_mp_cfl_to_msgpack(record->cobj_record, &mp_buf, &mp_size);
+            if (ret == -1) {
+                return -1;
+            }
+        }
+        else {
+            ret = generate_empty_msgpack_map(&mp_buf, &mp_size);
+            if (ret == -1) {
+                return -1;
+            }
+        }
 
         ret = flb_log_event_encoder_set_body_from_raw_msgpack(chunk_cobj->log_encoder, mp_buf, mp_size);
         if (ret != FLB_EVENT_ENCODER_SUCCESS) {
@@ -1229,8 +1257,7 @@ int flb_mp_chunk_cobj_record_destroy(struct flb_mp_chunk_cobj *chunk_cobj,
         return -1;
     }
 
-
-    if (chunk_cobj->record_pos) {
+    if (chunk_cobj && chunk_cobj->record_pos) {
         first = cfl_list_entry_first(&chunk_cobj->records, struct flb_mp_chunk_record, _head);
         last = cfl_list_entry_last(&chunk_cobj->records, struct flb_mp_chunk_record, _head);
 

src/flb_pack.c

@@ -476,8 +476,7 @@ static int pack_print_fluent_record(size_t cnt, msgpack_unpacked result)
     flb_time_pop_from_msgpack(&tms, &result, &obj);
     flb_metadata_pop_from_msgpack(&metadata, &result, &obj);
 
-    fprintf(stdout, "[%zd] [%"PRIu32".%09lu, ", cnt,
-            (uint32_t) tms.tm.tv_sec, tms.tm.tv_nsec);
+    fprintf(stdout, "[%zd] [%"PRId32".%09lu, ", cnt, (int32_t) tms.tm.tv_sec, tms.tm.tv_nsec);
 
     msgpack_object_print(stdout, *metadata);