opentelemetry: support event_name in logs

lib/fluent-otel-proto/proto_c/opentelemetry/proto/logs/v1/logs.pb-c.c

@@ -353,7 +353,7 @@ const ProtobufCMessageDescriptor opentelemetry__proto__logs__v1__scope_logs__des
   (ProtobufCMessageInit) opentelemetry__proto__logs__v1__scope_logs__init,
   NULL,NULL,NULL    /* reserved[123] */
 };
-static const ProtobufCFieldDescriptor opentelemetry__proto__logs__v1__log_record__field_descriptors[10] =
+static const ProtobufCFieldDescriptor opentelemetry__proto__logs__v1__log_record__field_descriptors[11] =
 {
   {
     "time_unix_nano",
@@ -475,11 +475,24 @@ static const ProtobufCFieldDescriptor opentelemetry__proto__logs__v1__log_record
     0,             /* flags */
     0,NULL,NULL    /* reserved1,reserved2, etc */
   },
+  {
+    "event_name",
+    12,
+    PROTOBUF_C_LABEL_NONE,
+    PROTOBUF_C_TYPE_STRING,
+    0,   /* quantifier_offset */
+    offsetof(Opentelemetry__Proto__Logs__V1__LogRecord, event_name),
+    NULL,
+    &protobuf_c_empty_string,
+    0,             /* flags */
+    0,NULL,NULL    /* reserved1,reserved2, etc */
+  },
 };
 static const unsigned opentelemetry__proto__logs__v1__log_record__field_indices_by_name[] = {
   4,   /* field[4] = attributes */
   3,   /* field[3] = body */
   5,   /* field[5] = dropped_attributes_count */
+  10,   /* field[10] = event_name */
   6,   /* field[6] = flags */
   9,   /* field[9] = observed_time_unix_nano */
   1,   /* field[1] = severity_number */
@@ -492,7 +505,7 @@ static const ProtobufCIntRange opentelemetry__proto__logs__v1__log_record__numbe
 {
   { 1, 0 },
   { 5, 3 },
-  { 0, 10 }
+  { 0, 11 }
 };
 const ProtobufCMessageDescriptor opentelemetry__proto__logs__v1__log_record__descriptor =
 {
@@ -502,7 +515,7 @@ const ProtobufCMessageDescriptor opentelemetry__proto__logs__v1__log_record__des
   "Opentelemetry__Proto__Logs__V1__LogRecord",
   "opentelemetry.proto.logs.v1",
   sizeof(Opentelemetry__Proto__Logs__V1__LogRecord),
-  10,
+  11,
   opentelemetry__proto__logs__v1__log_record__field_descriptors,
   opentelemetry__proto__logs__v1__log_record__field_indices_by_name,
   2,  opentelemetry__proto__logs__v1__log_record__number_ranges,

plugins/in_opentelemetry/opentelemetry_logs.c

@@ -281,6 +281,14 @@ static int otel_pack_v1_metadata(struct flb_opentelemetry *ctx,
         }
     }
 
+    if (log_record->event_name != NULL && strlen(log_record->event_name) > 0) {
+        flb_mp_map_header_append(&mh);
+        msgpack_pack_str(mp_pck, 10);
+        msgpack_pack_str_body(mp_pck, "event_name", 10);
+        msgpack_pack_str(mp_pck, strlen(log_record->event_name));
+        msgpack_pack_str_body(mp_pck, log_record->event_name, strlen(log_record->event_name));
+    }
+
     flb_mp_map_header_append(&mh);
     msgpack_pack_str(mp_pck, 11);
     msgpack_pack_str_body(mp_pck, "trace_flags", 11);

plugins/out_opentelemetry/opentelemetry.c

@@ -1096,6 +1096,16 @@ static struct flb_config_map config_map[] = {
      0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_severity_number_message_key),
      "Specify a Severity Number key"
     },
+    {
+     FLB_CONFIG_MAP_STR, "logs_event_name_metadata_key", "$EventName",
+     0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_event_name_metadata_key),
+     "Specify an Event Name key for metadata"
+    },
+    {
+     FLB_CONFIG_MAP_STR, "logs_event_name_message_key", "$EventName",
+     0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_event_name_message_key),
+     "Specify an Event Name key for message body"
+    },
 
 
     /* EOF */

plugins/out_opentelemetry/opentelemetry.h

@@ -136,6 +136,12 @@ struct opentelemetry_context {
     flb_sds_t logs_severity_number_message_key;
     struct flb_record_accessor *ra_severity_number_message;
 
+    flb_sds_t logs_event_name_metadata_key;
+    struct flb_record_accessor *ra_event_name_metadata;
+
+    flb_sds_t logs_event_name_message_key;
+    struct flb_record_accessor *ra_event_name_message;
+
     /* Number of logs to flush at a time */
     int batch_size;
 
@@ -199,6 +205,7 @@ struct opentelemetry_context {
     struct flb_record_accessor *ra_log_meta_otlp_trace_id;
     struct flb_record_accessor *ra_log_meta_otlp_span_id;
     struct flb_record_accessor *ra_log_meta_otlp_trace_flags;
+    struct flb_record_accessor *ra_log_meta_otlp_event_name;
 };
 
 int opentelemetry_post(struct opentelemetry_context *ctx,

plugins/out_opentelemetry/opentelemetry_conf.c

@@ -517,6 +517,18 @@ struct opentelemetry_context *flb_opentelemetry_context_create(struct flb_output
         flb_plg_error(ins, "failed to create ra for message severity number");
     }
 
+    ctx->ra_event_name_metadata = flb_ra_create((char*)ctx->logs_event_name_metadata_key,
+                                               FLB_FALSE);
+    if (ctx->ra_event_name_metadata == NULL) {
+        flb_plg_error(ins, "failed to create ra for metadata event name");
+    }
+
+    ctx->ra_event_name_message = flb_ra_create((char*)ctx->logs_event_name_message_key,
+                                              FLB_FALSE);
+    if (ctx->ra_event_name_message == NULL) {
+        flb_plg_error(ins, "failed to create ra for message event name");
+    }
+
     /* record accessor: group metadata */
     ctx->ra_meta_schema = flb_ra_create("$schema", FLB_FALSE);
     if (ctx->ra_meta_schema == NULL) {
@@ -606,6 +618,11 @@ struct opentelemetry_context *flb_opentelemetry_context_create(struct flb_output
         flb_plg_error(ins, "failed to create record accessor for otlp trace flags");
     }
 
+    ctx->ra_log_meta_otlp_event_name = flb_ra_create("$otlp['event_name']", FLB_FALSE);
+    if (ctx->ra_log_meta_otlp_event_name == NULL) {
+        flb_plg_error(ins, "failed to create record accessor for otlp event name");
+    }
+
     http_client_flags = FLB_HTTP_CLIENT_FLAG_AUTO_DEFLATE |
                         FLB_HTTP_CLIENT_FLAG_AUTO_INFLATE;
 
@@ -795,6 +812,18 @@ void flb_opentelemetry_context_destroy(struct opentelemetry_context *ctx)
         flb_ra_destroy(ctx->ra_log_meta_otlp_trace_flags);
     }
 
+    if (ctx->ra_log_meta_otlp_event_name) {
+        flb_ra_destroy(ctx->ra_log_meta_otlp_event_name);
+    }
+
+    if (ctx->ra_event_name_metadata) {
+        flb_ra_destroy(ctx->ra_event_name_metadata);
+    }
+
+    if (ctx->ra_event_name_message) {
+        flb_ra_destroy(ctx->ra_event_name_message);
+    }
+
 #ifdef FLB_HAVE_SIGNV4
 #ifdef FLB_HAVE_AWS
     if (ctx->aws_provider) {

plugins/out_opentelemetry/opentelemetry_logs.c

@@ -392,6 +392,7 @@ static int append_v1_logs_metadata_and_fields(struct opentelemetry_context *ctx,
     int severity_text_set = FLB_FALSE;
     int severity_number_set = FLB_FALSE;
     int trace_flags_set = FLB_FALSE;
+    int event_name_set = FLB_FALSE;
     size_t attr_count = 0;
     struct flb_ra_value *ra_val;
     Opentelemetry__Proto__Common__V1__KeyValue **attrs = NULL;
@@ -648,6 +649,52 @@ static int append_v1_logs_metadata_and_fields(struct opentelemetry_context *ctx,
         }
     }
 
+    /* EventName */
+    ra_val = flb_ra_get_value_object(ctx->ra_log_meta_otlp_event_name, *event->metadata);
+    if (ra_val != NULL) {
+        if (ra_val->o.type == MSGPACK_OBJECT_STR) {
+            log_record->event_name = flb_calloc(1, ra_val->o.via.str.size + 1);
+            if (log_record->event_name) {
+                strncpy(log_record->event_name, ra_val->o.via.str.ptr, ra_val->o.via.str.size);
+                event_name_set = FLB_TRUE;
+            }
+        }
+        flb_ra_key_value_destroy(ra_val);
+    }
+
+    if (!event_name_set && ctx->ra_event_name_metadata) {
+        ra_val = flb_ra_get_value_object(ctx->ra_event_name_metadata, *event->metadata);
+        if (ra_val != NULL) {
+            if (ra_val->o.type == MSGPACK_OBJECT_STR) {
+                log_record->event_name = flb_calloc(1, ra_val->o.via.str.size + 1);
+                if (log_record->event_name) {
+                    strncpy(log_record->event_name, ra_val->o.via.str.ptr, ra_val->o.via.str.size);
+                    event_name_set = FLB_TRUE;
+                }
+            }
+            flb_ra_key_value_destroy(ra_val);
+        }
+    }
+
+    if (!event_name_set && ctx->ra_event_name_message) {
+        ra_val = flb_ra_get_value_object(ctx->ra_event_name_message, *event->body);
+        if (ra_val != NULL) {
+            if (ra_val->o.type == MSGPACK_OBJECT_STR) {
+                log_record->event_name = flb_calloc(1, ra_val->o.via.str.size + 1);
+                if (log_record->event_name) {
+                    strncpy(log_record->event_name, ra_val->o.via.str.ptr, ra_val->o.via.str.size);
+                    event_name_set = FLB_TRUE;
+                }
+            }
+            flb_ra_key_value_destroy(ra_val);
+        }
+    }
+
+    if (!event_name_set) {
+        /* To prevent invalid free */
+        log_record->event_name = NULL;
+    }
+
     return 0;
 }
 
@@ -675,6 +722,9 @@ static void free_log_records(Opentelemetry__Proto__Logs__V1__LogRecord **logs, s
         if (log->severity_text != NULL && log->severity_text != protobuf_c_empty_string) {
             flb_free(log->severity_text);
         }
+        if (log->event_name != NULL && log->event_name != protobuf_c_empty_string) {
+            flb_free(log->event_name);
+        }
         if (log->span_id.data != NULL) {
             flb_free(log->span_id.data);
         }

tests/internal/data/opentelemetry/test_cases.json

@@ -228,6 +228,26 @@
     }
   },
 
+  "valid_log_with_event_name": {
+    "input": {
+      "resourceLogs": [{
+        "scopeLogs": [{
+          "logRecords": [{
+            "timeUnixNano": "1640995200000000000",
+            "eventName": "user.login",
+            "body": {"stringValue": "test log with event name"}
+          }]
+        }]
+      }]
+    },
+    "expected": {
+      "group_metadata": {"schema":"otlp","resource_id":0,"scope_id":0},
+      "group_body": {"resource":{}},
+      "log_metadata": {"otlp":{"event_name":"user.login"}},
+      "log_body": {"log": "test log with event name"}
+    }
+  },
+
   "valid_log_with_attributes": {
     "input": {
       "resourceLogs": [{