out_opentelemetry: support metrics max datapoints

include/fluent-bit/flb_input_chunk.h

@@ -179,6 +179,10 @@ int flb_input_chunk_set_up_down(struct flb_input_chunk *ic);
 int flb_input_chunk_set_up(struct flb_input_chunk *ic);
 int flb_input_chunk_down(struct flb_input_chunk *ic);
 int flb_input_chunk_is_up(struct flb_input_chunk *ic);
+int flb_input_chunk_find_space_new_data(struct flb_input_chunk *ic,
+                                        size_t chunk_size);
+int flb_input_chunk_has_overlimit_routes(struct flb_input_chunk *ic,
+                                         size_t chunk_size);
 void flb_input_chunk_update_output_instances(struct flb_input_chunk *ic,
                                              ssize_t chunk_size);
 

plugins/out_opentelemetry/opentelemetry.c

@@ -25,6 +25,7 @@
 #include <fluent-bit/flb_kv.h>
 #include <fluent-bit/flb_pack.h>
 #include <fluent-bit/flb_log_event_decoder.h>
+#include <fluent-bit/flb_opentelemetry.h>
 #include <fluent-bit/flb_ra_key.h>
 
 #include <cfl/cfl.h>
@@ -49,9 +50,6 @@
 #endif
 #endif
 
-extern cfl_sds_t cmt_encode_opentelemetry_create(struct cmt *cmt);
-extern void cmt_encode_opentelemetry_destroy(cfl_sds_t text);
-
 #include "opentelemetry.h"
 #include "opentelemetry_conf.h"
 #include "opentelemetry_utils.h"
@@ -730,6 +728,67 @@ static int opentelemetry_format_test(struct flb_config *config,
     return 0;
 }
 
+static int post_metrics_payload(struct opentelemetry_context *ctx,
+                                struct flb_event_chunk *event_chunk,
+                                flb_sds_t payload)
+{
+    int result;
+    int split_result;
+    size_t index;
+    struct cmt_opentelemetry_batches *batches;
+
+    if (ctx->metrics_max_datapoints == 0) {
+        return opentelemetry_post(ctx,
+                                  payload,
+                                  flb_sds_len(payload),
+                                  event_chunk->tag,
+                                  flb_sds_len(event_chunk->tag),
+                                  ctx->metrics_uri_sanitized,
+                                  ctx->grpc_metrics_uri);
+    }
+
+    batches = cmt_encode_opentelemetry_split_payload(
+                  payload,
+                  flb_sds_len(payload),
+                  (size_t) ctx->metrics_max_datapoints,
+                  &split_result);
+    if (batches == NULL) {
+        flb_plg_error(ctx->ins,
+                      "could not split metric payload into batches: %i",
+                      split_result);
+        if (split_result == CMT_ENCODE_OPENTELEMETRY_ALLOCATION_ERROR) {
+            return FLB_RETRY;
+        }
+        return FLB_ERROR;
+    }
+
+    result = FLB_OK;
+    for (index = 0; index < batches->count; index++) {
+        result = opentelemetry_post(ctx,
+                                    batches->entries[index].payload,
+                                    cfl_sds_len(batches->entries[index].payload),
+                                    event_chunk->tag,
+                                    flb_sds_len(event_chunk->tag),
+                                    ctx->metrics_uri_sanitized,
+                                    ctx->grpc_metrics_uri);
+        if (result != FLB_OK) {
+            if (result == FLB_RETRY && index > 0) {
+                flb_plg_warn(ctx->ins,
+                             "metric payload partially succeeded (%zu/%zu batches); "
+                             "skipping retry to avoid resending accepted data",
+                             index,
+                             batches->count);
+                result = FLB_OK;
+            }
+            break;
+        }
+    }
+
+    cmt_encode_opentelemetry_destroy_batches(batches);
+
+    return result;
+}
+
 static int process_metrics(struct flb_event_chunk *event_chunk,
                            struct flb_output_flush *out1_flush,
                            struct flb_input_instance *ins, void *out_context,
@@ -796,11 +855,7 @@ static int process_metrics(struct flb_event_chunk *event_chunk,
         flb_plg_debug(ctx->ins, "final payload size: %lu", flb_sds_len(buf));
         if (buf && flb_sds_len(buf) > 0) {
             /* Send HTTP request */
-            result = opentelemetry_post(ctx, buf, flb_sds_len(buf),
-                                        event_chunk->tag,
-                                        flb_sds_len(event_chunk->tag),
-                                        ctx->metrics_uri_sanitized,
-                                        ctx->grpc_metrics_uri);
+            result = post_metrics_payload(ctx, event_chunk, buf);
 
             /* Debug http_post() result statuses */
             if (result == FLB_OK) {
@@ -1020,6 +1075,12 @@ static int cb_opentelemetry_init(struct flb_output_instance *ins,
         ctx->batch_size = atoi(DEFAULT_LOG_RECORD_BATCH_SIZE);
     }
 
+    if (ctx->metrics_max_datapoints < 0) {
+        flb_plg_error(ins, "metrics_max_datapoints must be zero or greater");
+        flb_opentelemetry_context_destroy(ctx);
+        return -1;
+    }
+
     flb_output_set_context(ins, ctx);
 
     /*
@@ -1117,6 +1178,12 @@ static struct flb_config_map config_map[] = {
      0, FLB_TRUE, offsetof(struct opentelemetry_context, grpc_metrics_uri),
      "Specify an optional gRPC URI for the target OTel endpoint."
     },
+    {
+     FLB_CONFIG_MAP_INT, "metrics_max_datapoints", DEFAULT_METRICS_MAX_DATAPOINTS,
+     0, FLB_TRUE, offsetof(struct opentelemetry_context, metrics_max_datapoints),
+     "Set the maximum number of metric data points per OTLP export request "
+     "(0 disables the limit; default: 0)"
+    },
 
     {
      FLB_CONFIG_MAP_INT, "batch_size", DEFAULT_LOG_RECORD_BATCH_SIZE,

plugins/out_opentelemetry/opentelemetry.h

@@ -44,6 +44,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 */
+#define DEFAULT_METRICS_MAX_DATAPOINTS "0"    /* no data point limit */
 
 struct opentelemetry_body_key {
     flb_sds_t key;
@@ -145,6 +146,9 @@ struct opentelemetry_context {
     /* Number of logs to flush at a time */
     int batch_size;
 
+    /* Maximum number of metric data points per OTLP export request */
+    int metrics_max_datapoints;
+
     /* Maximum number of resources per OTLP export */
     int max_resources;
 

src/flb_http_client.c

@@ -2560,6 +2560,7 @@ struct flb_http_client_session *flb_http_client_session_begin(struct flb_http_cl
 
     if (protocol_version == HTTP_PROTOCOL_VERSION_20) {
         flb_stream_disable_keepalive(&upstream->base);
+        flb_upstream_conn_recycle(connection, FLB_FALSE);
     }
 
     session = flb_http_client_session_create(client, protocol_version, connection);

src/flb_input_chunk.c

@@ -1170,6 +1170,35 @@ int flb_input_chunk_release_space_compound(
     return 0;
 }
 
+static int flb_input_chunk_output_would_exceed_limit(
+                struct flb_output_instance *o_ins,
+                size_t chunk_size,
+                size_t *available_space)
+{
+    size_t remaining_space;
+
+    remaining_space = 0;
+
+    if (o_ins->fs_chunks_size <= o_ins->total_limit_size) {
+        remaining_space = o_ins->total_limit_size - o_ins->fs_chunks_size;
+        if (o_ins->fs_backlog_chunks_size <= remaining_space) {
+            remaining_space -= o_ins->fs_backlog_chunks_size;
+
+            if (available_space != NULL) {
+                *available_space = remaining_space;
+            }
+
+            return chunk_size > remaining_space;
+        }
+    }
+
+    if (available_space != NULL) {
+        *available_space = 0;
+    }
+
+    return FLB_TRUE;
+}
+
 /*
  * Find a slot in the output instance to append the new data with size chunk_size, it
  * will drop the the oldest chunks when the limitation on local disk is reached.
@@ -1198,9 +1227,8 @@ int flb_input_chunk_find_space_new_data(struct flb_input_chunk *ic,
             (flb_routes_mask_get_bit(ic->routes_mask,
                                      o_ins->id,
                                      o_ins->config->router) == 0) ||
-            (o_ins->fs_chunks_size +
-             o_ins->fs_backlog_chunks_size +
-             chunk_size) <= o_ins->total_limit_size) {
+            flb_input_chunk_output_would_exceed_limit(o_ins, chunk_size,
+                                                      NULL) == FLB_FALSE) {
             continue;
         }
 
@@ -1234,6 +1262,8 @@ int flb_input_chunk_has_overlimit_routes(struct flb_input_chunk *ic,
                                          size_t chunk_size)
 {
     int overlimit = 0;
+    int route_overlimit;
+    size_t available_space;
     struct mk_list *head;
     struct flb_output_instance *o_ins;
 
@@ -1248,16 +1278,14 @@ int flb_input_chunk_has_overlimit_routes(struct flb_input_chunk *ic,
         }
 
         FS_CHUNK_SIZE_DEBUG(o_ins);
+        route_overlimit = flb_input_chunk_output_would_exceed_limit(
+                              o_ins, chunk_size, &available_space);
         flb_trace("[input chunk] chunk %s required %ld bytes and %ld bytes left "
                   "in plugin %s", flb_input_chunk_get_name(ic), chunk_size,
-                  o_ins->total_limit_size -
-                  o_ins->fs_backlog_chunks_size -
-                  o_ins->fs_chunks_size,
+                  available_space,
                   o_ins->name);
 
-        if ((o_ins->fs_chunks_size +
-             o_ins->fs_backlog_chunks_size +
-             chunk_size) > o_ins->total_limit_size) {
+        if (route_overlimit == FLB_TRUE) {
             overlimit = FLB_TRUE;
         }
     }

tests/integration/README.md

@@ -490,6 +490,7 @@ Covers:
 - metadata and message key mapping
 - `add_label`
 - `batch_size`
+- `metrics_max_datapoints`
 - `logs_max_resources`
 - `logs_max_scopes`
 

tests/integration/scenarios/out_opentelemetry/config/out_otel_grpc_metrics_max_datapoints.conf

@@ -0,0 +1,19 @@
+[SERVICE]
+    flush        1
+    log_level    info
+    http_server  on
+    http_port    ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+[INPUT]
+    name  opentelemetry
+    port  ${FLUENT_BIT_TEST_LISTENER_PORT}
+
+[OUTPUT]
+    name                    opentelemetry
+    match                   *
+    host                    127.0.0.1
+    port                    ${TEST_SUITE_HTTP_PORT}
+    http2                   on
+    grpc                    on
+    grpc_metrics_uri        /batched.metrics.v1.Metrics/Export
+    metrics_max_datapoints  4

tests/integration/scenarios/out_opentelemetry/config/out_otel_http_metrics_max_datapoints.conf

@@ -0,0 +1,17 @@
+[SERVICE]
+    flush        1
+    log_level    info
+    http_server  on
+    http_port    ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+[INPUT]
+    name  opentelemetry
+    port  ${FLUENT_BIT_TEST_LISTENER_PORT}
+
+[OUTPUT]
+    name                    opentelemetry
+    match                   *
+    host                    127.0.0.1
+    port                    ${TEST_SUITE_HTTP_PORT}
+    metrics_uri             /batched/metrics
+    metrics_max_datapoints  4