opentelemetry: in_opentelemetry: log_event_encoder: mitigate otel attributes quadratic encoding expansion

include/fluent-bit/flb_log_event_encoder.h

@@ -234,6 +234,8 @@ struct flb_log_event_encoder {
     int                                         dynamically_allocated;
     char                                       *output_buffer;
     size_t                                      output_length;
+    /* Maximum serialized output bytes; SIZE_MAX by default. */
+    size_t                                      output_limit;
     int                                         initialized;
 
     struct flb_time                             timestamp;

plugins/in_opentelemetry/opentelemetry.c

@@ -29,6 +29,8 @@ static int in_opentelemetry_init(struct flb_input_instance *ins,
                                  struct flb_config *config, void *data)
 {
     int                       ret;
+    int64_t                   logs_max_size_bytes;
+    const char               *logs_max_size;
     struct flb_opentelemetry *ctx;
     struct flb_http_server_options http_server_options;
 
@@ -49,6 +51,17 @@ static int in_opentelemetry_init(struct flb_input_instance *ins,
         return -1;
     }
 
+    logs_max_size = flb_input_get_property("logs_max_size", ins);
+    logs_max_size_bytes = ctx->logs_max_size;
+    if (logs_max_size != NULL) {
+        logs_max_size_bytes = flb_utils_size_to_bytes(logs_max_size);
+    }
+    if (logs_max_size_bytes <= 0 || (uint64_t) logs_max_size_bytes > SIZE_MAX) {
+        flb_plg_error(ctx->ins, "logs_max_size must be a valid positive size");
+        opentelemetry_config_destroy(ctx);
+        return -1;
+    }
+
     if (ctx->oauth2_cfg.validate) {
         if (!ctx->oauth2_cfg.issuer || !ctx->oauth2_cfg.jwks_url) {
             flb_plg_error(ctx->ins, "oauth2.issuer and oauth2.jwks_url are required when oauth2.validate is enabled");
@@ -205,6 +218,11 @@ static struct flb_config_map config_map[] = {
      "Forward traces without processing"
     },
 
+    {
+     FLB_CONFIG_MAP_SIZE, "logs_max_size", "100M",
+     0, FLB_TRUE, offsetof(struct flb_opentelemetry, logs_max_size),
+     "Maximum generated MessagePack bytes per logs request, including group metadata"
+    },
     {
      FLB_CONFIG_MAP_STR, "logs_metadata_key", "otlp",
      0, FLB_TRUE, offsetof(struct flb_opentelemetry, logs_metadata_key),

plugins/in_opentelemetry/opentelemetry.h

@@ -39,6 +39,7 @@ struct flb_opentelemetry {
     int  tag_from_uri;
     flb_sds_t logs_metadata_key;
     flb_sds_t logs_body_key;
+    size_t logs_max_size;
     int profile_support_enabled;
     int encode_profiles_as_log;
     struct flb_oauth2_jwt_cfg oauth2_cfg;

plugins/in_opentelemetry/opentelemetry_logs.c

@@ -30,6 +30,7 @@
 
 #include "opentelemetry_protobuf.h"
 #include "opentelemetry.h"
+#include "opentelemetry_logs.h"
 #include "opentelemetry_utils.h"
 
 /*
@@ -655,7 +656,10 @@ static int binary_payload_to_msgpack(struct flb_opentelemetry *ctx,
                 goto binary_payload_to_msgpack_end;
             }
 
-            flb_log_event_encoder_group_header_end(encoder);
+            ret = flb_log_event_encoder_group_header_end(encoder);
+            if (ret != FLB_EVENT_ENCODER_SUCCESS) {
+                goto binary_payload_to_msgpack_end;
+            }
 
             for (log_record_index=0; log_record_index < scope_log->n_log_records; log_record_index++) {
                 ret = flb_log_event_encoder_begin_record(encoder);
@@ -757,14 +761,20 @@ static int binary_payload_to_msgpack(struct flb_opentelemetry *ctx,
                     if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                         (*record_count)++;
                     }
+                    else {
+                        goto binary_payload_to_msgpack_end;
+                    }
                 }
                 else {
                     flb_plg_error(ctx->ins, "marshalling error");
                     goto binary_payload_to_msgpack_end;
                 }
             }
 
-            flb_log_event_encoder_group_end(encoder);
+            ret = flb_log_event_encoder_group_end(encoder);
+            if (ret != FLB_EVENT_ENCODER_SUCCESS) {
+                goto binary_payload_to_msgpack_end;
+            }
 
         }
     }
@@ -791,6 +801,18 @@ int opentelemetry_process_logs(struct flb_opentelemetry *ctx,
                                flb_sds_t tag,
                                size_t tag_len,
                                void *data, size_t size)
+{
+    size_t remaining = ctx->logs_max_size;
+
+    return opentelemetry_process_logs_with_budget(ctx, content_type, tag, tag_len,
+                                                  data, size, &remaining);
+}
+
+int opentelemetry_process_logs_with_budget(struct flb_opentelemetry *ctx,
+                                           flb_sds_t content_type,
+                                           flb_sds_t tag, size_t tag_len,
+                                           void *data, size_t size,
+                                           size_t *remaining)
 {
     int ret = -1;
     int is_proto = FLB_FALSE; /* default to JSON */
@@ -828,6 +850,7 @@ int opentelemetry_process_logs(struct flb_opentelemetry *ctx,
     if (encoder == NULL) {
         return -1;
     }
+    encoder->output_limit = *remaining;
 
     if (is_proto == FLB_TRUE) {
         ret = binary_payload_to_msgpack(ctx, encoder,
@@ -853,6 +876,7 @@ int opentelemetry_process_logs(struct flb_opentelemetry *ctx,
     }
 
     if (ret >= 0) {
+        *remaining -= encoder->output_length;
         if (opentelemetry_uses_worker_ingress_queue(ctx)) {
             size_t allocation_size;
             void *resized_buffer;

plugins/in_opentelemetry/opentelemetry_logs.h

@@ -28,4 +28,10 @@ int opentelemetry_process_logs(struct flb_opentelemetry *ctx,
                                size_t tag_len,
                                void *data, size_t size);
 
+int opentelemetry_process_logs_with_budget(struct flb_opentelemetry *ctx,
+                                           flb_sds_t content_type,
+                                           flb_sds_t tag, size_t tag_len,
+                                           void *data, size_t size,
+                                           size_t *remaining);
+
 #endif

plugins/in_opentelemetry/opentelemetry_prot.c

@@ -838,12 +838,14 @@ int opentelemetry_prot_handle_ng(struct flb_http_request *request,
     char payload_type;
     char *payload = NULL;
     size_t payload_size = 0;
+    size_t logs_remaining;
     struct flb_opentelemetry *context;
     struct cfl_list frames;
     struct otlp_grpc_frame *frame;
 
     cfl_list_init(&frames);
     context = (struct flb_opentelemetry *) response->stream->user_data;
+    logs_remaining = context->logs_max_size;
 
     if (request->path == NULL || request->path[0] != '/') {
         send_response_ng(response, 400, "error: invalid request\n");
@@ -1018,8 +1020,9 @@ int opentelemetry_prot_handle_ng(struct flb_http_request *request,
             goto cleanup;
         }
 
-        ret = opentelemetry_process_logs(context, request->content_type, tag, flb_sds_len(tag),
-                                         payload, payload_size);
+        ret = opentelemetry_process_logs_with_budget(context, request->content_type,
+                                                      tag, flb_sds_len(tag),
+                                                      payload, payload_size, &logs_remaining);
     }
     else if (context->profile_support_enabled &&
              is_profiles_export_path(request->path) == FLB_TRUE) {

src/flb_log_event_encoder.c

@@ -59,6 +59,21 @@ void flb_log_event_encoder_reset(struct flb_log_event_encoder *context)
     flb_log_event_encoder_update_internal_state(context);
 }
 
+static int encoder_buffer_write(void *data, const char *buffer, size_t length)
+{
+    struct flb_log_event_encoder *context;
+
+    context = (struct flb_log_event_encoder *)
+              ((char *) data - offsetof(struct flb_log_event_encoder, buffer));
+
+    if (context->buffer.size > context->output_limit ||
+        length > context->output_limit - context->buffer.size) {
+        return -1;
+    }
+
+    return msgpack_sbuffer_write(&context->buffer, buffer, length);
+}
+
 int flb_log_event_encoder_init(struct flb_log_event_encoder *context, int format)
 {
     if (context == NULL) {
@@ -75,11 +90,12 @@ int flb_log_event_encoder_init(struct flb_log_event_encoder *context, int format
     context->dynamically_allocated = FLB_FALSE;
     context->initialized = FLB_TRUE;
     context->format = format;
+    context->output_limit = SIZE_MAX;
 
     msgpack_sbuffer_init(&context->buffer);
     msgpack_packer_init(&context->packer,
                         &context->buffer,
-                        msgpack_sbuffer_write);
+                        encoder_buffer_write);
 
     flb_log_event_encoder_dynamic_field_init(&context->metadata,
                                              MSGPACK_OBJECT_MAP);
@@ -174,13 +190,15 @@ int flb_log_event_encoder_emit_record(struct flb_log_event_encoder *context)
     int      result;
     char     ts_buf[8];
     char     rec_header[5];
+    size_t   original_size;
     uint32_t sec;
     uint32_t nsec;
 
     if (context == NULL) {
         return FLB_EVENT_ENCODER_ERROR_INVALID_CONTEXT;
     }
 
+    original_size = context->buffer.size;
     result = FLB_EVENT_ENCODER_SUCCESS;
 
     if (context->root.size == 0) {
@@ -290,6 +308,10 @@ int flb_log_event_encoder_emit_record(struct flb_log_event_encoder *context)
         }
     }
 
+    if (result != FLB_EVENT_ENCODER_SUCCESS) {
+        context->buffer.size = original_size;
+    }
+
     flb_log_event_encoder_update_internal_state(context);
     flb_log_event_encoder_reset_record(context);
 
@@ -487,8 +509,8 @@ int flb_log_event_encoder_group_header_end(struct flb_log_event_encoder *context
     int ret;
 
     ret = flb_log_event_encoder_commit_record(context);
-    if (ret == -1) {
-        return -1;
+    if (ret != FLB_EVENT_ENCODER_SUCCESS) {
+        return ret;
     }
 
     return 0;
@@ -511,8 +533,8 @@ int flb_log_event_encoder_group_end(struct flb_log_event_encoder *context)
     }
 
     ret = flb_log_event_encoder_commit_record(context);
-    if (ret == -1) {
-        return -1;
+    if (ret != FLB_EVENT_ENCODER_SUCCESS) {
+        return ret;
     }
 
     return 0;

src/opentelemetry/flb_opentelemetry_logs.c

@@ -452,7 +452,7 @@ static int process_json_payload_log_records_entry(
     if (result == FLB_EVENT_ENCODER_SUCCESS) {
         result = flb_log_event_encoder_commit_record(encoder);
     }
-    else {
+    if (result != FLB_EVENT_ENCODER_SUCCESS) {
         if (error_status) {
             *error_status = FLB_OTEL_LOGS_ERR_ENCODER_FAILURE;
         }
@@ -516,8 +516,10 @@ static int process_json_payload_scope_logs_entry(
                     sizeof(FLB_OTEL_LOGS_METADATA_KEY) - 1,
                     logs_body_key,
                     &entry_status);
-        if (result < 0 && error_status) {
-            *error_status = entry_status;
+        if (result < 0) {
+            if (error_status) {
+                *error_status = entry_status;
+            }
             return result;
         }
     }
@@ -629,6 +631,22 @@ static int process_json_payload_resource_logs_entry (struct flb_log_event_encode
             return -FLB_OTEL_LOGS_ERR_UNEXPECTED_SCOPELOGS_TYPE;
         }
 
+        /* Empty scopes must not repeatedly serialize discarded resource metadata. */
+        result = flb_otel_utils_find_map_entry_by_key(&scope_logs->ptr[index].via.map,
+                                                     "logRecords", 0, FLB_TRUE);
+        if (result < 0) {
+            if (error_status) {
+                *error_status = FLB_OTEL_LOGS_ERR_EMPTY_PAYLOAD;
+            }
+            result = 0;
+            continue;
+        }
+        obj = &scope_logs->ptr[index].via.map.ptr[result].val;
+        if (obj->type == MSGPACK_OBJECT_ARRAY && obj->via.array.size == 0) {
+            result = 0;
+            continue;
+        }
+
         /*
          * we use a temporary encoder to hold the group information, if no record entries are added
          * we will discard it.
@@ -640,6 +658,7 @@ static int process_json_payload_resource_logs_entry (struct flb_log_event_encode
             }
             return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE;
         }
+        tmp_encoder->output_limit = encoder->output_limit - encoder->buffer.size;
         flb_log_event_encoder_group_init(tmp_encoder);
 
         /* pack internal schema */
@@ -794,7 +813,14 @@ static int process_json_payload_resource_logs_entry (struct flb_log_event_encode
             /* close scope map */
             flb_log_event_encoder_commit_map(tmp_encoder, FLB_LOG_EVENT_BODY);
         }
-        flb_log_event_encoder_group_header_end(tmp_encoder);
+        ret = flb_log_event_encoder_group_header_end(tmp_encoder);
+        if (ret != FLB_EVENT_ENCODER_SUCCESS) {
+            if (error_status) {
+                *error_status = FLB_OTEL_LOGS_ERR_ENCODER_FAILURE;
+            }
+            flb_log_event_encoder_destroy(tmp_encoder);
+            return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE;
+        }
 
         /* before processing the scope logs, grab the number of bytes written */
         size_before = tmp_encoder->buffer.size;
@@ -817,13 +843,26 @@ static int process_json_payload_resource_logs_entry (struct flb_log_event_encode
          * the main encoder
          */
         if (size_after > size_before) {
-            flb_log_event_encoder_group_end(tmp_encoder);
+            ret = flb_log_event_encoder_group_end(tmp_encoder);
+            if (ret != FLB_EVENT_ENCODER_SUCCESS) {
+                if (error_status) {
+                    *error_status = FLB_OTEL_LOGS_ERR_ENCODER_FAILURE;
+                }
+                flb_log_event_encoder_destroy(tmp_encoder);
+                return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE;
+            }
 
             /* Append the temporary encoder output to the main encoder */
-            msgpack_sbuffer_write(&encoder->buffer,
-                                  tmp_encoder->output_buffer,
-                                  tmp_encoder->output_length);
-
+            ret = msgpack_pack_str_body(&encoder->packer,
+                                        tmp_encoder->output_buffer,
+                                        tmp_encoder->output_length);
+            if (ret != FLB_EVENT_ENCODER_SUCCESS) {
+                if (error_status) {
+                    *error_status = FLB_OTEL_LOGS_ERR_ENCODER_FAILURE;
+                }
+                flb_log_event_encoder_destroy(tmp_encoder);
+                return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE;
+            }
             encoder->output_buffer = encoder->buffer.data;
             encoder->output_length = encoder->buffer.size;
         }
@@ -932,6 +971,9 @@ int flb_opentelemetry_logs_json_to_msgpack(struct flb_log_event_encoder *encoder
     if (encoder == NULL) {
         return -1;
     }
+    if (encoder->buffer.size > encoder->output_limit) {
+        return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE;
+    }
 
     /* Convert JSON to messagepack */
     result = flb_pack_json(body, len, &msgpack_body, &msgpack_body_length,
@@ -947,6 +989,7 @@ int flb_opentelemetry_logs_json_to_msgpack(struct flb_log_event_encoder *encoder
         flb_free(msgpack_body);
         return -1;
     }
+    local_log_encoder.output_limit = encoder->output_limit - encoder->buffer.size;
 
     offset = 0;
     msgpack_unpacked_init(&unpacked_root);
@@ -983,9 +1026,9 @@ int flb_opentelemetry_logs_json_to_msgpack(struct flb_log_event_encoder *encoder
 
     /* copy local buffer into caller encoder buffer */
     if (local_log_encoder.output_length > 0) {
-        msgpack_sbuffer_write(&encoder->buffer,
-                            local_log_encoder.output_buffer,
-                            local_log_encoder.output_length);
+        result = msgpack_pack_str_body(&encoder->packer,
+                                       local_log_encoder.output_buffer,
+                                       local_log_encoder.output_length);
         encoder->output_buffer = encoder->buffer.data;
         encoder->output_length = encoder->buffer.size;
     }