output: discard metrics with empty data series

When a metrics context is about to be flushed, it may pass through one or more
processors before reaching the output plugin. Processing can result in a valid
metrics context with no time series data, which is useless for output plugins.
This change detects that case, skips the flush, and returns FLB_OK to the engine.

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

Author: edsiper

include/fluent-bit/flb_output.h

@@ -49,6 +49,7 @@
 #include <fluent-bit/flb_event.h>
 #include <fluent-bit/flb_processor.h>
 
+#include <cfl/cfl.h>
 #include <cmetrics/cmetrics.h>
 #include <cmetrics/cmt_gauge.h>
 #include <cmetrics/cmt_counter.h>
@@ -669,6 +670,13 @@ static FLB_INLINE void output_pre_cb_flush(void)
         flb_debug("[output] skipping flush for event chunk with zero records.");
         FLB_OUTPUT_RETURN(FLB_OK);
     }
+    /* Skip flush if processed event chunk has no data (empty after processing) */
+    else if (persisted_params.event_chunk &&
+             persisted_params.event_chunk->type == FLB_EVENT_TYPE_METRICS &&
+             persisted_params.event_chunk->size == 0) {
+        flb_debug("[output] skipping flush for event chunk with no data after processing.");
+        FLB_OUTPUT_RETURN(FLB_OK);
+    }
 
     /* Continue, we will resume later */
     out_p = persisted_params.out_plugin;
@@ -708,7 +716,7 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
 {
     int ret;
     size_t records;
-    void *p_buf;
+    void *p_buf = NULL;
     size_t p_size;
     size_t stack_size;
     struct flb_coro *coro;
@@ -725,8 +733,10 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
     struct ctrace *trace_context;
     struct cprof *profile_context;
     size_t chunk_offset;
+    struct cmt *encode_context = NULL;
     struct cmt *cmt_out_context = NULL;
 
+
     /* Custom output coroutine info */
     out_flush = (struct flb_output_flush *) flb_calloc(1, sizeof(struct flb_output_flush));
     if (!out_flush) {
@@ -756,6 +766,7 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
 
     /* Logs processor */
     evc = task->event_chunk;
+
     if (flb_processor_is_active(o_ins->processor)) {
         if (evc->type == FLB_EVENT_TYPE_LOGS) {
             /* run the processor */
@@ -786,10 +797,8 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
 
             if (p_buf == NULL) {
                 flb_errno();
-
                 flb_coro_destroy(coro);
                 flb_free(out_flush);
-
                 return NULL;
             }
 
@@ -803,6 +812,7 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
                             (char *) evc->data,
                             evc->size,
                             &chunk_offset)) == CMT_DECODE_MSGPACK_SUCCESS) {
+
                 ret = flb_processor_run(o_ins->processor,
                                         0,
                                         FLB_PROCESSOR_METRICS,
@@ -814,6 +824,22 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
                                         NULL);
 
                 if (ret == 0) {
+                    if (cmt_out_context) {
+                        encode_context = cmt_out_context;
+                    }
+                    else {
+                        encode_context = metrics_context;
+                    }
+
+                    /* if the cmetrics context lack of time series just skip it */
+                    if (flb_metrics_is_empty(encode_context)) {
+                        if (encode_context != metrics_context) {
+                            cmt_destroy(encode_context);
+                        }
+                        cmt_destroy(metrics_context);
+                        continue;
+                    }
+
                     if (cmt_out_context != NULL) {
                         ret = cmt_encode_msgpack_create(cmt_out_context,
                                                         &serialized_context_buffer,
@@ -822,7 +848,6 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
                         if (cmt_out_context != metrics_context) {
                             cmt_destroy(cmt_out_context);
                         }
-
                     }
                     else {
                         ret = cmt_encode_msgpack_create(metrics_context,
@@ -836,23 +861,17 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
                         flb_coro_destroy(coro);
                         flb_free(out_flush);
                         flb_free(p_buf);
-
                         return NULL;
                     }
 
-                    if ((serialization_buffer_offset +
-                         serialized_context_size) > p_size) {
-                        resized_serialization_buffer = \
-                            flb_realloc(p_buf, p_size + serialized_context_size);
-
+                    if ((serialization_buffer_offset + serialized_context_size) > p_size) {
+                        resized_serialization_buffer = flb_realloc(p_buf, p_size + serialized_context_size);
                         if (resized_serialization_buffer == NULL) {
                             flb_errno();
-
                             cmt_encode_msgpack_destroy(serialized_context_buffer);
                             flb_coro_destroy(coro);
                             flb_free(out_flush);
                             flb_free(p_buf);
-
                             return NULL;
                         }
 
@@ -871,26 +890,35 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
             }
 
             if (serialization_buffer_offset == 0) {
-                flb_coro_destroy(coro);
-                flb_free(out_flush);
+                flb_debug("[output] skipping flush for metrics event chunk with zero metrics after processing.");
                 flb_free(p_buf);
-
-                return NULL;
+                p_buf = NULL; /* Mark as freed to avoid double-free */
+
+                /* Create an empty processed event chunk to signal success */
+                out_flush->processed_event_chunk = flb_event_chunk_create(
+                                                    evc->type,
+                                                    0,
+                                                    evc->tag,
+                                                    flb_sds_len(evc->tag),
+                                                    NULL,
+                                                    0);
+            }
+            else {
+                out_flush->processed_event_chunk = flb_event_chunk_create(
+                                                    evc->type,
+                                                    0,
+                                                    evc->tag,
+                                                    flb_sds_len(evc->tag),
+                                                    p_buf,
+                                                    p_size);
             }
-
-            out_flush->processed_event_chunk = flb_event_chunk_create(
-                                                evc->type,
-                                                0,
-                                                evc->tag,
-                                                flb_sds_len(evc->tag),
-                                                p_buf,
-                                                p_size);
 
             if (out_flush->processed_event_chunk == NULL) {
                 flb_coro_destroy(coro);
                 flb_free(out_flush);
-                flb_free(p_buf);
-
+                if (p_buf != NULL) {
+                    flb_free(p_buf);
+                }
                 return NULL;
             }
         }