openetelemetry: Implement cmetrics max datapoints and its splitting

include/fluent-bit/flb_opentelemetry.h

@@ -95,6 +95,16 @@ struct flb_otel_error_map {
 
 struct cmt;
 struct ctrace;
+
+struct flb_opentelemetry_metrics_proto_batch {
+    flb_sds_t payload;
+    size_t data_point_count;
+};
+
+struct flb_opentelemetry_metrics_proto_batches {
+    size_t count;
+    struct flb_opentelemetry_metrics_proto_batch *entries;
+};
 struct flb_log_event;
 
 enum flb_opentelemetry_otlp_json_result {
@@ -263,6 +273,31 @@ flb_sds_t flb_opentelemetry_metrics_msgpack_to_otlp_proto(const void *data,
                                                           size_t size,
                                                           int *result);
 
+/*
+ * Split an encoded OTLP ExportMetricsServiceRequest without changing metric,
+ * resource, or scope metadata. A zero limit returns the original request as a
+ * single batch. The returned payloads are owned by the batch collection.
+ */
+struct flb_opentelemetry_metrics_proto_batches *
+flb_opentelemetry_metrics_proto_batches_create(const void *payload,
+                                               size_t payload_size,
+                                               size_t max_data_points,
+                                               int *result);
+
+struct flb_opentelemetry_metrics_proto_batches *
+flb_opentelemetry_metrics_to_otlp_proto_batches(struct cmt *context,
+                                                size_t max_data_points,
+                                                int *result);
+
+struct flb_opentelemetry_metrics_proto_batches *
+flb_opentelemetry_metrics_msgpack_to_otlp_proto_batches(const void *data,
+                                                        size_t size,
+                                                        size_t max_data_points,
+                                                        int *result);
+
+void flb_opentelemetry_metrics_proto_batches_destroy(
+    struct flb_opentelemetry_metrics_proto_batches *batches);
+
 flb_sds_t flb_opentelemetry_logs_to_otlp_proto(const void *event_chunk_data,
                                                size_t event_chunk_size,
                                                struct flb_opentelemetry_otlp_logs_options *options,

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>
@@ -730,6 +731,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 encode_result;
+    size_t index;
+    struct flb_opentelemetry_metrics_proto_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 = flb_opentelemetry_metrics_proto_batches_create(
+                  payload,
+                  flb_sds_len(payload),
+                  (size_t) ctx->metrics_max_datapoints,
+                  &encode_result);
+    if (batches == NULL) {
+        flb_plg_error(ctx->ins,
+                      "could not split metric payload into batches: %i",
+                      encode_result);
+        if (encode_result == FLB_OPENTELEMETRY_OTLP_PROTO_NOT_SUPPORTED) {
+            return FLB_RETRY;
+        }
+        return FLB_ERROR;
+    }
+
+    result = FLB_OK;
+    for (index = 0; index < batches->count; index++) {
+        result = opentelemetry_post(ctx,
+                                    batches->entries[index].payload,
+                                    flb_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;
+        }
+    }
+
+    flb_opentelemetry_metrics_proto_batches_destroy(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 +858,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 +1078,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 +1181,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/CMakeLists.txt

@@ -122,6 +122,7 @@ set(src
   ${src}
   opentelemetry/flb_opentelemetry_logs.c
   opentelemetry/flb_opentelemetry_metrics.c
+  opentelemetry/flb_opentelemetry_metrics_batch.c
   opentelemetry/flb_opentelemetry_otlp_json.c
   opentelemetry/flb_opentelemetry_otlp_proto.c
   opentelemetry/flb_opentelemetry_traces.c

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);