out_opentelemetry: add configurable log resource limits

Before this patch we had a hard-coded maximum of 100 log resources per chunk. The
change in question introduces a new configurable option called 'logs_max_resources'
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

@@ -955,7 +955,7 @@ static struct flb_config_map config_map[] = {
     },
 
     {
-      FLB_CONFIG_MAP_INT, "batch_size", DEFAULT_LOG_RECORD_BATCH_SIZE,
+     FLB_CONFIG_MAP_INT, "batch_size", DEFAULT_LOG_RECORD_BATCH_SIZE,
       0, FLB_TRUE, offsetof(struct opentelemetry_context, batch_size),
       "Set the maximum number of log records to be flushed at a time"
     },
@@ -964,10 +964,17 @@ static struct flb_config_map config_map[] = {
      0, FLB_FALSE, 0,
      "Set payload compression mechanism. Options available are 'gzip' and 'zstd'."
     },
+
     /*
      * Logs Properties
      * ---------------
      */
+    {
+     FLB_CONFIG_MAP_INT, "logs_max_resources", DEFAULT_MAX_RESOURCE_EXPORT,
+     0, FLB_TRUE, offsetof(struct opentelemetry_context, max_resources),
+     "Set the maximum number of OTLP log resources per export request (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

@@ -41,6 +41,7 @@
  * including the ones that succeeded. This is not ideal.
  */
 #define DEFAULT_LOG_RECORD_BATCH_SIZE "1000"
+#define DEFAULT_MAX_RESOURCE_EXPORT   "0"    /* no resource limits */
 
 struct opentelemetry_body_key {
     flb_sds_t key;
@@ -139,6 +140,9 @@ struct opentelemetry_context {
     /* Number of logs to flush at a time */
     int batch_size;
 
+    /* Maximum number of resources per OTLP export */
+    int max_resources;
+
     /* Log the response payload */
     int log_response_payload;
 

plugins/out_opentelemetry/opentelemetry_conf.c

@@ -279,6 +279,12 @@ struct opentelemetry_context *flb_opentelemetry_context_create(struct flb_output
         return NULL;
     }
 
+    if (ctx->max_resources < 0) {
+        flb_plg_error(ins, "max_resources 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

@@ -33,6 +33,8 @@
 #include "opentelemetry_conf.h"
 #include "opentelemetry_utils.h"
 
+#define RESOURCE_LOGS_INITIAL_CAPACITY 256
+
 static int hex_to_int(char ch)
 {
     if (ch >= '0' && ch <= '9') {
@@ -948,6 +950,9 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
     int max_scopes;
     int max_resources;
     int native_otel = FLB_FALSE;
+    size_t resource_logs_capacity;
+    size_t i;
+    size_t new_capacity;
     int64_t resource_id = -1;
     int64_t scope_id = -1;
     int64_t tmp_resource_id = -1;
@@ -958,6 +963,7 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
     struct flb_record_accessor *ra_match;
     Opentelemetry__Proto__Collector__Logs__V1__ExportLogsServiceRequest export_logs;
     Opentelemetry__Proto__Logs__V1__ResourceLogs **resource_logs = NULL;
+    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 *scope_log = NULL;
@@ -978,11 +984,19 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
     opentelemetry__proto__collector__logs__v1__export_logs_service_request__init(&export_logs);
 
     /* local limits */
-    max_resources = 100; /* maximim number of resources */
+    max_resources = ctx->max_resources; /* maximum number of resources */
     max_scopes = 100;    /* maximum number of scopes per resource */
 
-    /* allocate for 100 resource logs */
-    resource_logs = flb_calloc(max_resources, sizeof(Opentelemetry__Proto__Logs__V1__ResourceLogs *));
+    if (max_resources > 0) {
+        resource_logs_capacity = max_resources;
+    }
+    else {
+        resource_logs_capacity = RESOURCE_LOGS_INITIAL_CAPACITY; /* grow dynamically when unlimited */
+    }
+
+    /* allocate storage for the configured number of resource logs */
+    resource_logs = flb_calloc(resource_logs_capacity,
+                               sizeof(Opentelemetry__Proto__Logs__V1__ResourceLogs *));
     if (!resource_logs) {
         flb_errno();
         flb_log_event_decoder_destroy(decoder);
@@ -1021,10 +1035,42 @@ int otel_process_logs(struct flb_event_chunk *event_chunk,
 
             /* if we have a new resource_id, start a new resource context */
             if (resource_id != tmp_resource_id) {
-                if (export_logs.n_resource_logs >= max_resources) {
-                    flb_plg_error(ctx->ins, "max resources limit reached");
-                    ret = FLB_ERROR;
-                    break;
+                if (max_resources > 0) {
+                    if (export_logs.n_resource_logs >= max_resources) {
+                        /* respect the configured resource batching limit */
+                        flb_plg_error(ctx->ins, "max resources limit reached");
+                        ret = FLB_ERROR;
+                        break;
+                    }
+                }
+                else if (export_logs.n_resource_logs >= resource_logs_capacity) {
+                    new_capacity = resource_logs_capacity * 2;
+                    if (new_capacity <= resource_logs_capacity) {
+                        flb_plg_error(ctx->ins, "resource logs capacity overflow");
+                        ret = FLB_ERROR;
+                        break;
+                    }
+
+                    if (new_capacity < RESOURCE_LOGS_INITIAL_CAPACITY) {
+                        new_capacity = RESOURCE_LOGS_INITIAL_CAPACITY;
+                    }
+
+                    tmp_resource_logs = flb_realloc(resource_logs,
+                                                     new_capacity * sizeof(Opentelemetry__Proto__Logs__V1__ResourceLogs *));
+                    if (!tmp_resource_logs) {
+                        flb_errno();
+                        ret = FLB_RETRY;
+                        break;
+                    }
+
+                    resource_logs = tmp_resource_logs;
+
+                    for (i = resource_logs_capacity; i < new_capacity; i++) {
+                        resource_logs[i] = NULL;
+                    }
+
+                    resource_logs_capacity = new_capacity;
+                    export_logs.resource_logs = resource_logs;
                 }
 
 start_resource: