out_opentelemetry: add configurable log scope limits

Before this patch we had a hard-coded maximum of 100 log scopes per resource. The
change in question introduces a new configurable option called 'logs_max_scopes'
to control the maximum limit, the default value is: 0 (unlimited).

Signed-off-by: Eduardo Silva <[email protected]>

Author: edsiper

plugins/out_opentelemetry/opentelemetry.c

@@ -975,6 +975,12 @@ static struct flb_config_map config_map[] = {
      "Set the maximum number of OTLP log resources per export request (0 disables the limit; default: 0)"
     },
 
+    {
+     FLB_CONFIG_MAP_INT, "logs_max_scopes", DEFAULT_MAX_SCOPE_EXPORT,
+     0, FLB_TRUE, offsetof(struct opentelemetry_context, max_scopes),
+     "Set the maximum number of OTLP log scopes per resource (0 disables the limit; default: 0)"
+    },
+
     {
      FLB_CONFIG_MAP_STR, "logs_uri", "/v1/logs",
      0, FLB_TRUE, offsetof(struct opentelemetry_context, logs_uri),

plugins/out_opentelemetry/opentelemetry.h

@@ -42,6 +42,7 @@
  */
 #define DEFAULT_LOG_RECORD_BATCH_SIZE "1000"
 #define DEFAULT_MAX_RESOURCE_EXPORT   "0"    /* no resource limits */
+#define DEFAULT_MAX_SCOPE_EXPORT      "0"    /* no scope limits */
 
 struct opentelemetry_body_key {
     flb_sds_t key;
@@ -143,6 +144,9 @@ struct opentelemetry_context {
     /* Maximum number of resources per OTLP export */
     int max_resources;
 
+    /* Maximum number of scopes per OTLP resource */
+    int max_scopes;
+
     /* Log the response payload */
     int log_response_payload;
 

plugins/out_opentelemetry/opentelemetry_conf.c

@@ -285,6 +285,12 @@ struct opentelemetry_context *flb_opentelemetry_context_create(struct flb_output
         return NULL;
     }
 
+    if (ctx->max_scopes < 0) {
+        flb_plg_error(ins, "max_scopes must be greater than or equal to zero");
+        flb_opentelemetry_context_destroy(ctx);
+        return NULL;
+    }
+
     /* Parse 'add_label' */
     ret = config_add_labels(ins, ctx);
     if (ret == -1) {

plugins/out_opentelemetry/opentelemetry_logs.c

@@ -34,6 +34,7 @@
 #include "opentelemetry_utils.h"
 
 #define RESOURCE_LOGS_INITIAL_CAPACITY 256
+#define SCOPE_LOGS_INITIAL_CAPACITY    100
 
 static int hex_to_int(char ch)
 {
@@ -947,12 +948,15 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
     int ret;
     int record_type;
     int log_record_count;
-    int max_scopes;
+    int max_scopes_limit;
     int max_resources;
     int native_otel = FLB_FALSE;
     size_t resource_logs_capacity;
     size_t i;
     size_t new_capacity;
+    size_t resource_index = 0;
+    size_t scope_capacity = 0;
+    size_t new_scope_capacity = 0;
     int64_t resource_id = -1;
     int64_t scope_id = -1;
     int64_t tmp_resource_id = -1;
@@ -966,9 +970,12 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
     Opentelemetry__Proto__Logs__V1__ResourceLogs **tmp_resource_logs = NULL;
     Opentelemetry__Proto__Logs__V1__ResourceLogs *resource_log = NULL;
     Opentelemetry__Proto__Logs__V1__ScopeLogs **scope_logs = NULL;
+    Opentelemetry__Proto__Logs__V1__ScopeLogs **tmp_scope_logs = NULL;
     Opentelemetry__Proto__Logs__V1__ScopeLogs *scope_log = NULL;
     Opentelemetry__Proto__Logs__V1__LogRecord **log_records = NULL;
     Opentelemetry__Proto__Logs__V1__LogRecord  *log_record = NULL;
+    size_t *resource_scope_capacities = NULL;
+    size_t *tmp_scope_capacities = NULL;
 
     ctx = (struct opentelemetry_context *) out_context;
 
@@ -985,7 +992,7 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
 
     /* local limits */
     max_resources = ctx->max_resources; /* maximum number of resources */
-    max_scopes = 100;    /* maximum number of scopes per resource */
+    max_scopes_limit = ctx->max_scopes;    /* maximum number of scopes per resource */
 
     if (max_resources > 0) {
         resource_logs_capacity = max_resources;
@@ -1002,6 +1009,13 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
         flb_log_event_decoder_destroy(decoder);
         return -1;
     }
+    resource_scope_capacities = flb_calloc(resource_logs_capacity, sizeof(size_t));
+    if (!resource_scope_capacities) {
+        flb_errno();
+        flb_free(resource_logs);
+        flb_log_event_decoder_destroy(decoder);
+        return -1;
+    }
     export_logs.resource_logs = resource_logs;
     export_logs.n_resource_logs = 0;
 
@@ -1062,15 +1076,25 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
                         ret = FLB_RETRY;
                         break;
                     }
-
                     resource_logs = tmp_resource_logs;
+                    export_logs.resource_logs = resource_logs;
+
+                    tmp_scope_capacities = flb_realloc(resource_scope_capacities,
+                                                       new_capacity * sizeof(size_t));
+                    if (!tmp_scope_capacities) {
+                        flb_errno();
+                        ret = FLB_RETRY;
+                        break;
+                    }
+
+                    resource_scope_capacities = tmp_scope_capacities;
 
                     for (i = resource_logs_capacity; i < new_capacity; i++) {
                         resource_logs[i] = NULL;
+                        resource_scope_capacities[i] = 0;
                     }
 
                     resource_logs_capacity = new_capacity;
-                    export_logs.resource_logs = resource_logs;
                 }
 
 start_resource:
@@ -1090,6 +1114,8 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
                 resource_logs[export_logs.n_resource_logs] = resource_log;
                 export_logs.n_resource_logs++;
 
+                resource_index = export_logs.n_resource_logs - 1;
+
                 resource_log->resource = flb_calloc(1, sizeof(Opentelemetry__Proto__Resource__V1__Resource));
                 if (!resource_log->resource) {
                     flb_errno();
@@ -1107,7 +1133,17 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
 
                 /* prepare the scopes */
                 if (!resource_log->scope_logs) {
-                    scope_logs = flb_calloc(100, sizeof(Opentelemetry__Proto__Logs__V1__ScopeLogs *));
+                    if (max_scopes_limit > 0) {
+                        scope_capacity = (size_t) max_scopes_limit;
+                    }
+                    else {
+                        scope_capacity = resource_scope_capacities[resource_index];
+                        if (scope_capacity == 0) {
+                            scope_capacity = SCOPE_LOGS_INITIAL_CAPACITY;
+                        }
+                    }
+
+                    scope_logs = flb_calloc(scope_capacity, sizeof(Opentelemetry__Proto__Logs__V1__ScopeLogs *));
                     if (!scope_logs) {
                         flb_errno();
                         ret = FLB_RETRY;
@@ -1116,6 +1152,7 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
 
                     resource_log->scope_logs = scope_logs;
                     resource_log->n_scope_logs = 0;
+                    resource_scope_capacities[resource_index] = scope_capacity;
                 }
 
                 /* update the current resource_id and reset scope_id */
@@ -1124,11 +1161,46 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
             }
 
             if (scope_id != tmp_scope_id) {
+                resource_index = export_logs.n_resource_logs - 1;
+
                 /* check limits */
-                if (resource_log->n_scope_logs >= max_scopes) {
-                    flb_plg_error(ctx->ins, "max scopes limit reached");
-                    ret = FLB_ERROR;
-                    break;
+                if (max_scopes_limit > 0) {
+                    if (resource_log->n_scope_logs >= max_scopes_limit) {
+                        flb_plg_error(ctx->ins, "max scopes limit reached");
+                        ret = FLB_ERROR;
+                        break;
+                    }
+                }
+                else {
+                    if (resource_log->n_scope_logs >= resource_scope_capacities[resource_index]) {
+                        new_scope_capacity = resource_scope_capacities[resource_index] * 2;
+
+                        if (new_scope_capacity <= resource_scope_capacities[resource_index]) {
+                            flb_plg_error(ctx->ins, "scope logs capacity overflow");
+                            ret = FLB_ERROR;
+                            break;
+                        }
+
+                        if (new_scope_capacity < SCOPE_LOGS_INITIAL_CAPACITY) {
+                            new_scope_capacity = SCOPE_LOGS_INITIAL_CAPACITY;
+                        }
+
+                        tmp_scope_logs = flb_realloc(resource_log->scope_logs,
+                                                     new_scope_capacity * sizeof(Opentelemetry__Proto__Logs__V1__ScopeLogs *));
+                        if (!tmp_scope_logs) {
+                            flb_errno();
+                            ret = FLB_RETRY;
+                            break;
+                        }
+
+                        for (i = resource_scope_capacities[resource_index];
+                             i < new_scope_capacity; i++) {
+                            tmp_scope_logs[i] = NULL;
+                        }
+
+                        resource_log->scope_logs = tmp_scope_logs;
+                        resource_scope_capacities[resource_index] = new_scope_capacity;
+                    }
                 }
 
                 /* process the scope */
@@ -1155,7 +1227,8 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
                     flb_errno();
                     flb_free(scope_log->scope);
                     flb_free(scope_log);
-                    return -2;
+                    ret = FLB_RETRY;
+                    break;
                 }
                 log_record_count = 0;
 
@@ -1274,5 +1347,7 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
 
     /* release all protobuf resources */
     free_resource_logs(export_logs.resource_logs, export_logs.n_resource_logs);
+    flb_free(resource_scope_capacities);
+
     return ret;
 }