engine: expose internal logging call counts as internal metrics

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>
@@ -72,6 +74,7 @@ struct flb_log {
     pthread_t tid;             /* thread ID   */
     struct flb_worker *worker; /* non-real worker reference */
     struct mk_event_loop *evl;
+    struct flb_log_metrics *metrics;
 
     /* Initialization variables */
     int pth_init;
@@ -92,6 +95,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.

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>
@@ -438,6 +440,28 @@ int flb_log_get_level_str(char *str)
     return -1;
 }
 
+static inline const 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_set_file(struct flb_config *config, char *out)
 {
     struct flb_log *log = config->log;
@@ -454,6 +478,73 @@ int flb_log_set_file(struct flb_config *config, char *out)
     return 0;
 }
 
+/* Frees the metrics instance and its associated resources. */
+void flb_log_metrics_destroy(struct flb_log_metrics *metrics)
+{
+    if (metrics != NULL && metrics->cmt != NULL) {
+        cmt_destroy(metrics->cmt);
+    }
+    if (metrics != NULL) {
+        flb_free(metrics);
+    }
+}
+
+/*
+ * 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 *metrics;
+    int log_message_type;
+    const char *message_type_str;
+    uint64_t ts;
+    int ret;
+
+    metrics = flb_calloc(1, sizeof(struct flb_log_metrics));
+    if (metrics == NULL) {
+        flb_errno();
+        return NULL;
+    }
+
+    metrics->cmt = cmt_create();
+    if (metrics->cmt == NULL) {
+        flb_log_metrics_destroy(metrics);
+        return NULL;
+    }
+
+    metrics->logs_total_counter = cmt_counter_create(metrics->cmt,
+                                                     "fluentbit",
+                                                     "logger",
+                                                     "logs_total",
+                                                     "Total number of logs",
+                                                     1, (char *[]) {"message_type"});
+    if (metrics->logs_total_counter == NULL) {
+        flb_log_metrics_destroy(metrics);
+        return NULL;
+    }
+
+    /* Initialize counters for log message types to 0. */
+    ts = cfl_time_now();
+    for (log_message_type = FLB_LOG_ERROR; log_message_type <= FLB_LOG_TRACE; log_message_type++) {
+        message_type_str = flb_log_message_type_str(log_message_type);
+        if (!message_type_str) {
+            break;
+        }
+
+        ret = cmt_counter_set(metrics->logs_total_counter,
+                              ts,
+                              0,
+                              1, (char *[]) {message_type_str});
+        if (ret == -1) {
+            flb_log_metrics_destroy(metrics);
+            return NULL;
+        }
+    }
+
+    return metrics;
+}
+
 struct flb_log *flb_log_create(struct flb_config *config, int type,
                                int level, char *out)
 {
@@ -507,6 +598,16 @@ struct flb_log *flb_log_create(struct flb_config *config, int type,
         return NULL;
     }
 
+    /* Create metrics */
+    log->metrics = flb_log_metrics_create();
+    if (log->metrics == NULL) {
+        fprintf(stderr, "[log] could not create log metrics\n");
+        mk_event_loop_destroy(log->evl);
+        flb_free(log);
+        config->log = NULL;
+        return NULL;
+    }
+
     /*
      * Since the main process/thread might want to write log messages,
      * it will need a 'worker-like' context, here we create a fake worker
@@ -573,39 +674,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;
     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;
     }
@@ -630,6 +724,7 @@ int flb_log_construct(struct log_message *msg, int *ret_len,
         return -1;
     }
 
+    header_title = flb_log_message_type_str(type);
     len = snprintf(msg->msg, sizeof(msg->msg) - 1,
                    "%s[%s%i/%02i/%02i %02i:%02i:%02i%s]%s [%s%5s%s] ",
                    /*      time     */                    /* type */
@@ -704,8 +799,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;
+    const char *msg_type_str;
+    uint64_t ts;
 
     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 +815,25 @@ 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 != NULL && config->log != NULL) {
+            msg_type_str = flb_log_message_type_str(type);
+            if (msg_type_str == NULL) {
+                msg_type_str = "unknown";
+            }
+
+            ts = cfl_time_now();
+            ret = cmt_counter_inc(config->log->metrics->logs_total_counter,
+                                  ts,
+                                  1, (char *[]) {msg_type_str});
+            if (ret == -1) {
+                /* Not using flb_log_debug to avoid recursing into this same function. */
+                fprintf(stderr,
+                        "[log] failed to increment log total counter for message type '%s' (error=%d)\n",
+                        msg_type_str, ret);
+            }
+        }
+
         n = flb_pipe_write_all(w->log[1], &msg, sizeof(msg));
 
         if (n == -1) {
@@ -776,6 +893,7 @@ int flb_log_destroy(struct flb_log *log, struct flb_config *config)
     }
     flb_log_worker_destroy(log->worker);
     flb_free(log->worker);
+    flb_log_metrics_destroy(log->metrics);
     flb_free(log);
 
     return 0;

src/flb_metrics_exporter.c

@@ -299,6 +299,15 @@ struct cmt *flb_me_get_cmetrics(struct flb_config *ctx)
         }
     }
 
+    if (ctx->log != NULL) {
+        ret = cmt_cat(cmt, ctx->log->metrics->cmt);
+        if (ret != 0) {
+            flb_error("[metrics exporter] could not append global log metrics");
+            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")