Skip to content

input_metrics: discard cmetrics with no series #10700

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Aug 12, 2025
Merged

input_metrics: discard cmetrics with no series #10700

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
92d0cbe
macros: use stdbool for booleans
edsiper Aug 8, 2025
e65b7b9
metrics: add flb_metrics_is_empty() helper
edsiper Aug 8, 2025
943639d
input_metric: drop metrics with empty data series
edsiper Aug 8, 2025
1bbf67c
output: discard metrics with empty data series
edsiper Aug 8, 2025
c9ced09
output: safe cleanup on metricsprocessors failure
edsiper Aug 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions include/fluent-bit/flb_macros.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,11 @@
#ifndef FLB_MACROS_H
#define FLB_MACROS_H

#include <stdbool.h>
#include <monkey/mk_core.h>

#define FLB_FALSE 0
#define FLB_TRUE !FLB_FALSE
#define FLB_FALSE false
#define FLB_TRUE true

/* Return values */
#define FLB_ERROR 0
Expand Down
11 changes: 11 additions & 0 deletions include/fluent-bit/flb_metrics.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,13 @@
#include <cmetrics/cmt_decode_statsd.h>
#include <cmetrics/cmt_filter.h>

/*
* v1 HTTP endpoint metrics
* ------------------------
* These are functions that are not part of the CMetrics library, its' the old interface
* to ship internal metrics which is part of the v1 HTTP endpoint.
*/

/* Metrics IDs for general purpose (used by core and Plugins */
#define FLB_METRIC_N_RECORDS 0
#define FLB_METRIC_N_BYTES 1
Expand Down Expand Up @@ -85,6 +92,10 @@ int flb_metrics_dump_values(char **out_buf, size_t *out_size,
struct flb_metrics *me);
int flb_metrics_destroy(struct flb_metrics *metrics);
int flb_metrics_fluentbit_add(struct flb_config *ctx, struct cmt *cmt);
/* ! end of v1 HTTP endpoint metrics */

/* General metrics utilities */
bool flb_metrics_is_empty(struct cmt *cmt);

#endif
#endif /* FLB_HAVE_METRICS */
92 changes: 66 additions & 26 deletions include/fluent-bit/flb_output.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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) {
Expand Down Expand Up @@ -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 */
Expand Down Expand Up @@ -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;
}

Expand All @@ -803,6 +812,8 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
(char *) evc->data,
evc->size,
&chunk_offset)) == CMT_DECODE_MSGPACK_SUCCESS) {

cmt_out_context = NULL;
ret = flb_processor_run(o_ins->processor,
0,
FLB_PROCESSOR_METRICS,
Expand All @@ -814,6 +825,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 lacks 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,
Expand All @@ -822,7 +849,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,
Expand All @@ -836,23 +862,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;
}

Expand All @@ -868,29 +888,49 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,

cmt_encode_msgpack_destroy(serialized_context_buffer);
}
else {
cmt_destroy(metrics_context);
if (cmt_out_context != NULL && cmt_out_context != metrics_context) {
cmt_destroy(cmt_out_context);
}
flb_coro_destroy(coro);
flb_free(out_flush);
flb_free(p_buf);
return NULL;
}
}

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 {
p_size = serialization_buffer_offset;
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;
}
}
Expand Down
36 changes: 20 additions & 16 deletions src/flb_input_metric.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
#include <fluent-bit/flb_input_chunk.h>
#include <fluent-bit/flb_input_metric.h>
#include <fluent-bit/flb_input_plugin.h>
#include <cfl/cfl.h>

static int input_metrics_append(struct flb_input_instance *ins,
size_t processor_starting_stage,
Expand All @@ -33,6 +34,7 @@ static int input_metrics_append(struct flb_input_instance *ins,
size_t mt_size;
int processor_is_active;
struct cmt *out_context = NULL;
struct cmt *encode_context;

processor_is_active = flb_processor_is_active(ins->processor);
if (processor_is_active) {
Expand Down Expand Up @@ -60,29 +62,31 @@ static int input_metrics_append(struct flb_input_instance *ins,
}
}

if (out_context) {
encode_context = out_context;
}
else {
encode_context = cmt;
}

if (out_context != NULL) {
/* Convert metrics to msgpack */
ret = cmt_encode_msgpack_create(out_context, &mt_buf, &mt_size);

if (out_context != cmt) {
/* Drop the context if it contains no metrics */
if (encode_context == NULL || flb_metrics_is_empty(encode_context)) {
if (out_context && out_context != cmt) {
cmt_destroy(out_context);
}
return 0;
}

if (ret != 0) {
flb_plg_error(ins, "could not encode metrics");
/* Convert metrics to msgpack */
ret = cmt_encode_msgpack_create(encode_context, &mt_buf, &mt_size);

return -1;
}
if (out_context && out_context != cmt) {
cmt_destroy(out_context);
}
else {
/* Convert metrics to msgpack */
ret = cmt_encode_msgpack_create(cmt, &mt_buf, &mt_size);
if (ret != 0) {
flb_plg_error(ins, "could not encode metrics");
return -1;

}
if (ret != 0) {
flb_plg_error(ins, "could not encode metrics");
return -1;
}

/* Append packed metrics */
Expand Down
9 changes: 9 additions & 0 deletions src/flb_metrics.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -376,3 +376,12 @@ int flb_metrics_fluentbit_add(struct flb_config *ctx, struct cmt *cmt)

return 0;
}

bool flb_metrics_is_empty(struct cmt *cmt)
{
return cfl_list_is_empty(&cmt->counters) &&
cfl_list_is_empty(&cmt->gauges) &&
cfl_list_is_empty(&cmt->histograms) &&
cfl_list_is_empty(&cmt->summaries) &&
cfl_list_is_empty(&cmt->untypeds);
}
Comment on lines +380 to +387
Copy link

@coderabbitai coderabbitai bot Aug 8, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Guard against NULL input in flb_metrics_is_empty()

If a caller ever passes a NULL struct cmt *, the function will seg-fault.
A defensive early-return costs nothing and prevents a crash path that’s very hard to diagnose at runtime.

 bool flb_metrics_is_empty(struct cmt *cmt)
 {
-    return cfl_list_is_empty(&cmt->counters) &&
+    if (cmt == NULL) {
+        return true;
+    }
+
+    return cfl_list_is_empty(&cmt->counters) &&
            ...
 }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
bool flb_metrics_is_empty(struct cmt *cmt)
{
return cfl_list_is_empty(&cmt->counters) &&
cfl_list_is_empty(&cmt->gauges) &&
cfl_list_is_empty(&cmt->histograms) &&
cfl_list_is_empty(&cmt->summaries) &&
cfl_list_is_empty(&cmt->untypeds);
}
bool flb_metrics_is_empty(struct cmt *cmt)
{
if (cmt == NULL) {
return true;
}
return cfl_list_is_empty(&cmt->counters) &&
cfl_list_is_empty(&cmt->gauges) &&
cfl_list_is_empty(&cmt->histograms) &&
cfl_list_is_empty(&cmt->summaries) &&
cfl_list_is_empty(&cmt->untypeds);
}
🤖 Prompt for AI Agents 
In src/flb_metrics.c around lines 380 to 387, the function flb_metrics_is_empty
does not check if the input pointer cmt is NULL, which can cause a segmentation
fault. Add a defensive check at the start of the function to return true (or an
appropriate value) immediately if cmt is NULL to prevent crashes.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

use FLB_TRUE instead true

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Check the current definition of FLB_TRUE and FLB_FALSE
cat include/fluent-bit/flb_macros.h | grep -A2 -B2 "FLB_TRUE\|FLB_FALSE"

Length of output: 173

🏁 Script executed:

#!/bin/bash
# Check how other functions handle NULL checks with boolean returns
rg -A3 "if \(.*== NULL\)" src/flb_metrics.c | head -20

Length of output: 56

🏁 Script executed:

#!/bin/bash
# Look for other bool-returning functions that use FLB_TRUE
rg "bool.*\(" include/fluent-bit/ -A5 | grep -B3 "FLB_TRUE\|return true"

Length of output: 74

@edsiper, you're right! For consistency with the rest of the codebase, here's the updated suggestion using FLB_TRUE:

bool flb_metrics_is_empty(struct cmt *cmt)
{
+    if (cmt == NULL) {
+        return FLB_TRUE;
+    }
+
    return cfl_list_is_empty(&cmt->counters) &&
           cfl_list_is_empty(&cmt->gauges) &&
           cfl_list_is_empty(&cmt->histograms) &&
           cfl_list_is_empty(&cmt->summaries) &&
           cfl_list_is_empty(&cmt->untypeds);
}

This maintains consistency with the project's macro usage pattern, even though FLB_TRUE is now defined as the boolean literal true.

🧬 Code Graph Analysis Results

Relevant snippets from src/flb_metrics.c (lines 22-35)

static int id_exists(int id, struct flb_metrics *metrics)
{
    struct mk_list *head;
    struct flb_metric *metric;

    mk_list_foreach(head, &metrics->list) {
        metric = mk_list_entry(head, struct flb_metric, _head);
        if (metric->id == id) {
            return FLB_TRUE;
        }
    }

    return FLB_FALSE;
}
  • This function checks if a metric ID already exists in the metrics list.
  • It returns FLB_TRUE if the ID exists, otherwise FLB_FALSE.
  • The user comment about using FLB_TRUE instead of true is directly relevant here since this function returns FLB_TRUE and FLB_FALSE macros instead of standard C true/false.

Relevant snippets from src/flb_metrics.c (lines 37-51)

static int id_get(struct flb_metrics *metrics)
{
    int id;
    int ret = FLB_FALSE;

    /* Try to use 'count' as an id */
    id = metrics->count;

    while ((ret = id_exists(id, metrics)) == FLB_TRUE) {
        id++;
    }

    return id;
}
  • This function generates a new unique metric ID.
  • It uses FLB_TRUE and FLB_FALSE macros for boolean logic.
  • This is another place where the user comment about using FLB_TRUE instead of true is relevant.

Relevant snippet from include/fluent-bit/flb_mem.h (lines 84-96)

void *flb_calloc(size_t n, const size_t size) {
    if (size == 0) {
        return NULL;
    }
#ifdef FLB_HAVE_TESTS_OSSFUZZ
   // Add chance of failure. Used by fuzzing to test error-handling code.
   if (flb_fuzz_get_probability(1)) {
     return NULL;
   }
#endif

    return calloc(n, size);
}
  • This snippet shows the flb_calloc function used in flb_metrics_create for memory allocation.
  • It is relevant for understanding memory allocation and error handling in the metrics code.

Relevant snippet from src/flb_sds.c (lines 389-399)

void flb_sds_destroy(flb_sds_t s)
{
    struct flb_sds *head;

    if (!s) {
        return;
    }

    head = FLB_SDS_HEADER(s);
    flb_free(head);
}
  • This snippet shows how flb_sds_destroy frees string data structures used for metric titles.
  • Relevant for understanding resource cleanup in flb_metrics_destroy.

Relevant snippet from src/flb_sds.c (lines 58-76)

flb_sds_t flb_sds_create_len(const char *str, int len)
{
    flb_sds_t s;
    struct flb_sds *head;

    s = sds_alloc(len);
    if (!s) {
        return NULL;
    }

    if (str) {
        memcpy(s, str, len);
        s[len] = '\0';

        head = FLB_SDS_HEADER(s);
        head->len = len;
    }
    return s;
}
  • This snippet shows how metric titles are created with a length limit.
  • Relevant for understanding how titles are handled in flb_metrics_add and flb_metrics_title.

Relevant snippet from lib/cmetrics/src/cmt_counter.c (lines 26-81)

struct cmt_counter *cmt_counter_create(struct cmt *cmt,
                                       char *ns, char *subsystem,
                                       char *name, char *help,
                                       int label_count, char **label_keys)
{
    int ret;
    struct cmt_counter *counter;

    if (!ns) {
        cmt_log_error(cmt, "null ns not allowed");
        return NULL;
    }

    if (!subsystem) {
        cmt_log_error(cmt, "null subsystem not allowed");
        return NULL;
    }

    if (!name || strlen(name) == 0) {
        cmt_log_error(cmt, "undefined name");
        return NULL;
    }

    if (!help || strlen(help) == 0) {
        cmt_log_error(cmt, "undefined help");
        return NULL;
    }

    counter = calloc(1, sizeof(struct cmt_counter));
    if (!counter) {
        cmt_errno();
        return NULL;
    }
    cfl_list_add(&counter->_head, &cmt->counters);

    ret = cmt_opts_init(&counter->opts, ns, subsystem, name, help);
    if (ret == -1) {
        cmt_log_error(cmt, "unable to initialize options for counter");
        cmt_counter_destroy(counter);
        return NULL;
    }

    /* Create the map */
    counter->map = cmt_map_create(CMT_COUNTER, &counter->opts, label_count, label_keys,
                                  (void *) counter);
    if (!counter->map) {
        cmt_log_error(cmt, "unable to allocate map for counter");
        cmt_counter_destroy(counter);
        return NULL;
    }
    /* set default counter aggregation type to cumulative */
    counter->aggregation_type = CMT_AGGREGATION_TYPE_CUMULATIVE;

    counter->cmt = cmt;
    return counter;
}
  • This snippet shows how counters are created in the cmetrics library, which is used in flb_metrics.c for attaching metrics.
  • Relevant for understanding the creation of counters in functions like attach_uptime.

Relevant snippet from lib/cmetrics/src/cmt_gauge.c (lines 27-81)

struct cmt_gauge *cmt_gauge_create(struct cmt *cmt,
                                   char *ns, char *subsystem, char *name,
                                   char *help, int label_count, char **label_keys)
{
    int ret;
    struct cmt_gauge *gauge;

    if (!ns) {
        cmt_log_error(cmt, "null ns not allowed");
        return NULL;
    }

    if (!subsystem) {
        cmt_log_error(cmt, "null subsystem not allowed");
        return NULL;
    }

    if (!name || strlen(name) == 0) {
        cmt_log_error(cmt, "undefined name");
        return NULL;
    }

    if (!help || strlen(help) == 0) {
        cmt_log_error(cmt, "undefined help");
        return NULL;
    }

    gauge = calloc(1, sizeof(struct cmt_gauge));
    if (!gauge) {
        cmt_errno();
        return NULL;
    }
    cfl_list_add(&gauge->_head, &cmt->gauges);

    /* Initialize options */
    ret = cmt_opts_init(&gauge->opts, ns, subsystem, name, help);
    if (ret == -1) {
        cmt_log_error(cmt, "unable to initialize options for gauge");
        cmt_gauge_destroy(gauge);
        return NULL;
    }

    /* Create the map */
    gauge->map = cmt_map_create(CMT_GAUGE, &gauge->opts, label_count, label_keys,
                                (void *) gauge);
    if (!gauge->map) {
        cmt_log_error(cmt, "unable to allocate map for gauge");
        cmt_gauge_destroy(gauge);
        return NULL;
    }

    gauge->cmt = cmt;

    return gauge;
}
  • This snippet shows how gauges are created in the cmetrics library, used in flb_metrics.c for attaching metrics like process start time and build info.
  • Relevant for understanding the metric attachment functions.

Relevant snippet from lib/cmetrics/src/cmt_gauge.c (lines 94-109)

int cmt_gauge_set(struct cmt_gauge *gauge, uint64_t timestamp, double val,
                  int labels_count, char **label_vals)
{
    struct cmt_metric *metric;

    metric = cmt_map_metric_get(&gauge->opts, gauge->map, labels_count, label_vals,
                                CMT_TRUE);
    if (!metric) {
        cmt_log_error(gauge->cmt, "unable to retrieve metric for gauge %s_%s_%s",
                      gauge->opts.ns, gauge->opts.subsystem,
                      gauge->opts.name);
        return -1;
    }
    cmt_metric_set(metric, timestamp, val);
    return 0;
}
  • This snippet shows how gauge values are set, using CMT_TRUE macro for boolean.
  • Relevant for understanding the boolean macro usage in metric setting functions.

[End of relevant snippets.]

Loading