engine: expose internal logging call counts as internal metrics

Internal logger calls increment a new v2 metric exposed by the HTTP server
Prometheus scrape endpoint. There is one time series per log message type.

Signed-off-by: Alec Holmes <[email protected]>

Author: alecholmes

include/fluent-bit/flb_config.h

@@ -164,6 +164,7 @@ struct flb_config {
     /* Logging */
     char *log_file;
     struct flb_log *log;
+    struct flb_log_metrics *log_metrics_ctx; /* Global metrics for logging calls */
 
     /* Parser Conf */
     char *parsers_file;

include/fluent-bit/flb_log.h

@@ -28,6 +28,8 @@
 #include <fluent-bit/flb_worker.h>
 #include <fluent-bit/flb_config.h>
 #include <fluent-bit/flb_sds.h>
+#include <cmetrics/cmetrics.h>
+#include <cmetrics/cmt_counter.h>
 #include <inttypes.h>
 #include <errno.h>
 #include <stdarg.h>
@@ -92,6 +94,14 @@ struct flb_log_cache {
     struct mk_list entries;         /* list for entries */
 };
 
+/* Global metrics for logging calls. */
+struct flb_log_metrics {
+    struct cmt *cmt;
+
+    /* cmetrics */
+    struct cmt_counter *logs_total_counter; /* total number of logs (by message type) */
+};
+
 /*
  * This function is used by plugins interface to check if an incoming log message
  * should be logged or not based in the log levels defined.
@@ -232,6 +242,8 @@ static inline int flb_log_suppress_check(int log_suppress_interval, const char *
 int flb_log_worker_init(struct flb_worker *worker);
 int flb_log_worker_destroy(struct flb_worker *worker);
 int flb_errno_print(int errnum, const char *file, int line);
+struct flb_log_metrics *flb_log_metrics_create();
+void flb_log_metrics_destroy(struct flb_log_metrics *ctx);
 #ifdef WIN32
 int flb_wsa_get_last_error_print(int errnum, const char *file, int line);
 #endif

src/flb_config.c

@@ -400,6 +400,14 @@ struct flb_config *flb_config_init()
     flb_regex_init();
 #endif
 
+    /* Create internal logger metrics */
+    config->log_metrics_ctx = flb_log_metrics_create();
+    if (!config->log_metrics_ctx) {
+        flb_error("[engine] could not create log metrics");
+        flb_config_exit(config);
+        return NULL;
+    }
+
     return config;
 }
 
@@ -570,6 +578,11 @@ void flb_config_exit(struct flb_config *config)
     flb_config_task_map_resize(config, 0);
     flb_routes_empty_mask_destroy(config);
 
+    /* Destroy internal logger metrics */
+    if (config->log_metrics_ctx) {
+        flb_log_metrics_destroy(config->log_metrics_ctx);
+    }
+
     flb_free(config);
 }
 

src/flb_log.c

@@ -32,6 +32,8 @@
 #include <fluent-bit/flb_config.h>
 #include <fluent-bit/flb_worker.h>
 #include <fluent-bit/flb_mem.h>
+#include <cmetrics/cmetrics.h>
+#include <cmetrics/cmt_counter.h>
 
 #ifdef WIN32
 #include <winsock.h>
@@ -564,6 +566,28 @@ struct flb_log *flb_log_create(struct flb_config *config, int type,
     return log;
 }
 
+static inline char *flb_log_message_type_str(int type)
+{
+    switch (type) {
+    case FLB_LOG_HELP:
+        return "help";
+    case FLB_LOG_INFO:
+        return "info";
+    case FLB_LOG_WARN:
+        return "warn";
+    case FLB_LOG_ERROR:
+        return "error";
+    case FLB_LOG_DEBUG:
+        return "debug";
+    case FLB_LOG_IDEBUG:
+        return "debug";
+    case FLB_LOG_TRACE:
+        return "trace";
+    default:
+        return NULL;
+    }
+}
+
 int flb_log_construct(struct log_message *msg, int *ret_len,
                      int type, const char *file, int line, const char *fmt, va_list *args)
 {
@@ -573,39 +597,32 @@ int flb_log_construct(struct log_message *msg, int *ret_len,
     int total;
     time_t now;
     const char *header_color = NULL;
-    const char *header_title = NULL;
+    const char *header_title = flb_log_message_type_str(type);
     const char *bold_color = ANSI_BOLD;
     const char *reset_color = ANSI_RESET;
     struct tm result;
     struct tm *current;
 
     switch (type) {
     case FLB_LOG_HELP:
-        header_title = "help";
         header_color = ANSI_CYAN;
         break;
     case FLB_LOG_INFO:
-        header_title = "info";
         header_color = ANSI_GREEN;
         break;
     case FLB_LOG_WARN:
-        header_title = "warn";
         header_color = ANSI_YELLOW;
         break;
     case FLB_LOG_ERROR:
-        header_title = "error";
         header_color = ANSI_RED;
         break;
     case FLB_LOG_DEBUG:
-        header_title = "debug";
         header_color = ANSI_YELLOW;
         break;
     case FLB_LOG_IDEBUG:
-        header_title = "debug";
         header_color = ANSI_CYAN;
         break;
     case FLB_LOG_TRACE:
-        header_title = "trace";
         header_color = ANSI_BLUE;
         break;
     }
@@ -704,8 +721,11 @@ void flb_log_print(int type, const char *file, int line, const char *fmt, ...)
     int ret;
     struct log_message msg = {0};
     va_list args;
+    char *msg_type_str = flb_log_message_type_str(type);
+    uint64_t ts = cfl_time_now();
 
     struct flb_worker *w;
+    struct flb_config *config;
 
     va_start(args, fmt);
     ret = flb_log_construct(&msg, &len, type, file, line, fmt, &args);
@@ -717,6 +737,11 @@ void flb_log_print(int type, const char *file, int line, const char *fmt, ...)
 
     w = flb_worker_get();
     if (w) {
+        config = w->config;
+        if (config && config->log_metrics_ctx) {
+            cmt_counter_inc(config->log_metrics_ctx->logs_total_counter, ts, 1, (char *[]) {msg_type_str}); // Ignoring inc error
+        }
+
         n = flb_pipe_write_all(w->log[1], &msg, sizeof(msg));
 
         if (n == -1) {
@@ -780,3 +805,75 @@ int flb_log_destroy(struct flb_log *log, struct flb_config *config)
 
     return 0;
 }
+
+/*
+ * Create and register cmetrics for the runtime logger.
+ * The caller must free the returned struct using flb_log_metrics_destroy.
+ */
+struct flb_log_metrics *flb_log_metrics_create()
+{
+    struct flb_log_metrics *lm;
+    int i;
+    char *message_type_str;
+    uint64_t ts;
+
+    lm = flb_calloc(1, sizeof(struct flb_log_metrics));
+    if (!lm) {
+        flb_errno();
+        goto error;
+    }
+
+    lm->cmt = cmt_create();
+    if (!lm->cmt) {
+        goto error;
+    }
+
+    lm->logs_total_counter = cmt_counter_create(lm->cmt,
+                                                "fluentbit", "logger", "logs_total",
+                                                "Total number of logs",
+                                                1, (char *[]) {"message_type"});
+    if (!lm->logs_total_counter) {
+        goto error;
+    }
+
+    /*
+     * Initialize counters for all log message types to 0.
+     * This assumes types are contiguous starting at 1 (FLB_LOG_ERROR).
+     */
+    i = 1;
+    ts = cfl_time_now();
+    for (i = 1; ; i++) {
+        message_type_str = flb_log_message_type_str(i);
+        if (!message_type_str) {
+            break;
+        }
+
+        if (cmt_counter_set(lm->logs_total_counter, ts, 0, 1, (char *[]) {message_type_str}) == -1) {
+            goto error;
+        }
+    }
+
+    return lm;
+
+error:
+    if (lm && lm->logs_total_counter) {
+        cmt_counter_destroy(lm->logs_total_counter);
+    }
+
+    if (lm && lm->cmt) {
+        cmt_destroy(lm->cmt);
+    }
+
+    if (lm) {
+        flb_free(lm);
+    }
+
+    return NULL;
+}
+
+void flb_log_metrics_destroy(struct flb_log_metrics *lm)
+{
+    cmt_counter_destroy(lm->logs_total_counter);
+    cmt_destroy(lm->cmt);
+    flb_free(lm);
+}

src/flb_metrics_exporter.c

@@ -299,6 +299,15 @@ struct cmt *flb_me_get_cmetrics(struct flb_config *ctx)
         }
     }
 
+    if (ctx->log_metrics_ctx) {
+        ret = cmt_cat(cmt, ctx->log_metrics_ctx->cmt);
+        if (ret == -1) {
+            flb_error("[metrics exporter] could not append global log_metrics_ctx");
+            cmt_destroy(cmt);
+            return NULL;
+        }
+    }
+
     /* Pipeline metrics: input, filters, outputs */
     mk_list_foreach(head, &ctx->inputs) {
         i = mk_list_entry(head, struct flb_input_instance, _head);

src/flb_sds.c

@@ -33,7 +33,7 @@
 #include <stdarg.h>
 #include <ctype.h>
 
-static flb_sds_t sds_alloc(size_t size)
+static flb_sds_t flb_sds_alloc_internal(size_t size)
 {
     void *buf;
     flb_sds_t s;
@@ -60,7 +60,7 @@ flb_sds_t flb_sds_create_len(const char *str, int len)
     flb_sds_t s;
     struct flb_sds *head;
 
-    s = sds_alloc(len);
+    s = flb_sds_alloc_internal(len);
     if (!s) {
         return NULL;
     }
@@ -91,7 +91,7 @@ flb_sds_t flb_sds_create(const char *str)
 
 flb_sds_t flb_sds_create_size(size_t size)
 {
-    return sds_alloc(size);
+    return flb_sds_alloc_internal(size);
 }
 
 /* Increase SDS buffer size 'len' bytes */

tests/runtime/CMakeLists.txt

@@ -25,6 +25,7 @@ endmacro()
 
 # Core
 FLB_RT_CORE_TEST(FLB_COROUTINE_TIMEOUT    "core-timeout.c")
+FLB_RT_CORE_TEST(FLB_INTERNAL_LOGGER      "core_internal_logger.c")
 
 FLB_RT_TEST(FLB_CHUNK_TRACE "core_chunk_trace.c")