Merge branch 'fluent:master' into enable-s390x

.github/actionlint.yml

@@ -1,4 +1,3 @@
 self-hosted-runner:
   labels:
-    - actuated
-    - actuated-aarch64
+    - actuated-arm64-8cpu-16gb

.github/workflows/call-build-linux-packages.yaml

@@ -106,7 +106,7 @@ jobs:
     name: ${{ matrix.distro }} package build and stage to S3
     environment: ${{ inputs.environment }}
     # Ensure for OSS Fluent Bit repo we enable usage of Actuated runners for ARM builds, for forks it should keep existing ubuntu-latest usage.
-    runs-on: ${{ (contains(matrix.distro, 'arm' ) && (github.repository == 'fluent/fluent-bit') && 'actuated-aarch64') || 'ubuntu-latest' }}
+    runs-on: ${{ (contains(matrix.distro, 'arm' ) && (github.repository == 'fluent/fluent-bit') && 'actuated-arm64-8cpu-16gb') || 'ubuntu-latest' }}
     permissions:
       contents: read
     strategy:
@@ -123,7 +123,7 @@ jobs:
       - name: Set up Actuated mirror
         if: contains(matrix.distro, 'arm' ) && (github.repository == 'fluent/fluent-bit')
         uses: self-actuated/hub-mirror@master
-  
+
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v3
 
@@ -137,7 +137,7 @@ jobs:
           echo "$INPUT --> $output"
           echo "replaced=$output" >> "$GITHUB_OUTPUT"
         shell: bash
-        env: 
+        env:
           INPUT: ${{ matrix.distro }}
 
       - name: fluent-bit - ${{ matrix.distro }} artifacts

include/fluent-bit/flb_config.h

@@ -261,6 +261,8 @@ struct flb_config {
     int enable_hot_reload;
     int ensure_thread_safety_on_hot_reloading;
     unsigned int hot_reloaded_count;
+    int shutdown_by_hot_reloading;
+    int hot_reloading;
 
     /* Co-routines */
     unsigned int coro_stack_size;

include/fluent-bit/flb_input.h

@@ -326,6 +326,9 @@ struct flb_input_instance {
     /* is the input instance overlimit ?: 1 or 0 */
     struct cmt_gauge   *cmt_storage_overlimit;
 
+    /* is the input instance paused or not ?: 1 or 0 */
+    struct cmt_gauge   *cmt_ingestion_paused;
+
     /* memory bytes used by chunks */
     struct cmt_gauge   *cmt_storage_memory_bytes;
 

include/fluent-bit/flb_output.h

@@ -1075,6 +1075,8 @@ void *flb_output_get_cmt_instance(struct flb_output_instance *ins);
 #endif
 void flb_output_net_default(const char *host, int port,
                             struct flb_output_instance *ins);
+int flb_output_enable_multi_threading(struct flb_output_instance *ins,
+                                      struct flb_config *config);
 const char *flb_output_name(struct flb_output_instance *ins);
 void flb_output_pre_run(struct flb_config *config);
 void flb_output_exit(struct flb_config *config);

include/fluent-bit/flb_plugin_proxy.h

@@ -71,10 +71,6 @@ struct flb_plugin_input_proxy_context {
 void *flb_plugin_proxy_symbol(struct flb_plugin_proxy *proxy,
                               const char *symbol);
 
-int flb_plugin_proxy_output_init(struct flb_plugin_proxy *proxy,
-                                 struct flb_output_instance *o_ins,
-                                 struct flb_config *config);
-
 int flb_plugin_proxy_register(struct flb_plugin_proxy *proxy,
                               struct flb_config *config);
 

include/fluent-bit/flb_version.h.in

@@ -80,7 +80,7 @@ static inline void flb_version_banner()
     fprintf(stderr,
             "%sFluent Bit v%s%s\n", bold_color, FLB_VERSION_STR, reset_color);
 #endif
-    fprintf(stderr, "* %sCopyright (C) 2015-2022 The Fluent Bit Authors%s\n",
+    fprintf(stderr, "* %sCopyright (C) 2015-2023 The Fluent Bit Authors%s\n",
             copyright_color, reset_color);
     fprintf(stderr, "* Fluent Bit is a CNCF sub-project under the "
             "umbrella of Fluentd\n");

lib/cmetrics/CMakeLists.txt

@@ -6,7 +6,7 @@ set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 # CMetrics Version
 set(CMT_VERSION_MAJOR  0)
 set(CMT_VERSION_MINOR  6)
-set(CMT_VERSION_PATCH  3)
+set(CMT_VERSION_PATCH  4)
 set(CMT_VERSION_STR "${CMT_VERSION_MAJOR}.${CMT_VERSION_MINOR}.${CMT_VERSION_PATCH}")
 
 # Include helpers

lib/cmetrics/src/cmt_decode_msgpack.c

@@ -977,8 +977,11 @@ static int unpack_basic_type_meta(mpack_reader_t *reader, size_t index, void *co
     result = cmt_mpack_unpack_map(reader, callbacks, context);
 
     if (CMT_DECODE_MSGPACK_SUCCESS == result) {
-        decode_context->map->label_count = cfl_list_size(&decode_context->map->label_keys);
+	if (decode_context->map == NULL || decode_context->map->parent == NULL) {
+            return CMT_DECODE_MSGPACK_INVALID_ARGUMENT_ERROR;
+	}
 
+        decode_context->map->label_count = cfl_list_size(&decode_context->map->label_keys);
         if (decode_context->map->type == CMT_HISTOGRAM) {
             histogram = (struct cmt_histogram *) decode_context->map->parent;
 

plugins/CMakeLists.txt

@@ -205,10 +205,13 @@ if(${CMAKE_SYSTEM_NAME} MATCHES "Linux")
   REGISTER_IN_PLUGIN("in_netif")
   REGISTER_IN_PLUGIN("in_docker")
   REGISTER_IN_PLUGIN("in_docker_events")
-  REGISTER_IN_PLUGIN("in_node_exporter_metrics")
   REGISTER_IN_PLUGIN("in_podman_metrics")
 endif()
 
+if(${CMAKE_SYSTEM_NAME} MATCHES "Linux" OR ${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
+  REGISTER_IN_PLUGIN("in_node_exporter_metrics")
+endif()
+
 REGISTER_IN_PLUGIN("in_kubernetes_events")
 REGISTER_IN_PLUGIN("in_kafka")
 REGISTER_IN_PLUGIN("in_fluentbit_metrics")

plugins/custom_calyptia/calyptia.c

@@ -37,6 +37,7 @@ struct calyptia {
     flb_sds_t cloud_host;
     flb_sds_t cloud_port;
     flb_sds_t machine_id;
+    int machine_id_auto_configured;
 
 /* used for reporting chunk trace records. */
 #ifdef FLB_HAVE_CHUNK_TRACE
@@ -426,6 +427,8 @@ static int cb_calyptia_init(struct flb_custom_instance *ins,
             flb_plg_error(ctx->ins, "unable to retrieve machine_id");
             return -1;
         }
+
+        ctx->machine_id_auto_configured = 1;
     }
 
     /* input collector */
@@ -517,6 +520,10 @@ static int cb_calyptia_exit(void *data, struct flb_config *config)
         return 0;
     }
 
+    if (ctx->machine_id && ctx->machine_id_auto_configured) {
+        flb_sds_destroy(ctx->machine_id);
+    }
+
     flb_free(ctx);
     return 0;
 }

plugins/in_forward/fw_prot.c

@@ -778,7 +778,8 @@ int fw_prot_process(struct flb_input_instance *ins, struct fw_conn *conn)
                         }
                         else if (event_type == FLB_EVENT_TYPE_METRICS) {
                             ret = cmt_decode_msgpack_create(&cmt, (char *) data, len, &off);
-                            if (ret == -1) {
+                            if (ret != CMT_DECODE_MSGPACK_SUCCESS) {
+                                flb_error("cmt_decode_msgpack_create failed. ret=%d", ret);
                                 msgpack_unpacked_destroy(&result);
                                 msgpack_unpacker_free(unp);
                                 flb_sds_destroy(out_tag);

plugins/in_node_exporter_metrics/CMakeLists.txt

@@ -11,6 +11,7 @@ set(src
   ne_loadavg.c
   ne_filefd_linux.c
   ne_textfile.c
+  ne_processes.c
   ne_utils.c
   ne_config.c
   ne.c
@@ -23,4 +24,8 @@ set(src
    )
 endif()
 
-FLB_PLUGIN(in_node_exporter_metrics "${src}" "")
+if (APPLE)
+  FLB_PLUGIN(in_node_exporter_metrics "${src}" "-framework Foundation -framework IOKit")
+else()
+  FLB_PLUGIN(in_node_exporter_metrics "${src}" "")
+endif()

plugins/in_node_exporter_metrics/ne.c

@@ -41,6 +41,7 @@
 #include "ne_netdev.h"
 #include "ne_textfile.h"
 #include "ne_systemd.h"
+#include "ne_processes.h"
 
 static int ne_timer_cpu_metrics_cb(struct flb_input_instance *ins,
                                    struct flb_config *config, void *in_context)
@@ -182,6 +183,15 @@ static int ne_timer_systemd_metrics_cb(struct flb_input_instance *ins,
     return 0;
 }
 
+static int ne_timer_processes_metrics_cb(struct flb_input_instance *ins,
+                                       struct flb_config *config, void *in_context)
+{
+    struct flb_ne *ctx = in_context;
+
+    ne_processes_update(ctx);
+
+    return 0;
+}
 struct flb_ne_callback {
     char *name;
     void (*func)(char *, void *, void *);
@@ -329,6 +339,13 @@ static void ne_systemd_update_cb(char *name, void *p1, void *p2)
     ne_systemd_update(ctx);
 }
 
+static void ne_processes_update_cb(char *name, void *p1, void *p2)
+{
+    struct flb_ne *ctx = p1;
+
+    ne_processes_update(ctx);
+}
+
 static int ne_update_cb(struct flb_ne *ctx, char *name)
 {
     int ret;
@@ -356,6 +373,7 @@ struct flb_ne_callback ne_callbacks[] = {
     { "filefd", ne_filefd_update_cb },
     { "textfile", ne_textfile_update_cb },
     { "systemd", ne_systemd_update_cb },
+    { "processes", ne_processes_update_cb },
     { 0 }
 };
 
@@ -392,6 +410,7 @@ static int in_ne_init(struct flb_input_instance *in,
     ctx->coll_filefd_fd = -1;
     ctx->coll_textfile_fd = -1;
     ctx->coll_systemd_fd = -1;
+    ctx->coll_processes_fd = -1;
 
     ctx->callback = flb_callback_create(in->name);
     if (!ctx->callback) {
@@ -422,7 +441,7 @@ static int in_ne_init(struct flb_input_instance *in,
 
             if (ret == FLB_FALSE) {
                 if (strncmp(entry->str, "cpufreq", 7) == 0) {
-                    if (ctx->cpu_scrape_interval == 0) {
+                    if (ctx->cpufreq_scrape_interval == 0) {
                         flb_plg_debug(ctx->ins, "enabled metrics %s", entry->str);
                         metric_idx = 0;
                     }
@@ -442,7 +461,7 @@ static int in_ne_init(struct flb_input_instance *in,
                     ne_cpufreq_init(ctx);
                 }
                 else if (strncmp(entry->str, "cpu", 3) == 0) {
-                    if (ctx->cpufreq_scrape_interval == 0) {
+                    if (ctx->cpu_scrape_interval == 0) {
                         flb_plg_debug(ctx->ins, "enabled metrics %s", entry->str);
                         metric_idx = 1;
                     }
@@ -701,6 +720,26 @@ static int in_ne_init(struct flb_input_instance *in,
                     }
                     ne_systemd_init(ctx);
                 }
+                else if (strncmp(entry->str, "processes", 9) == 0) {
+                    if (ctx->processes_scrape_interval == 0) {
+                        flb_plg_debug(ctx->ins, "enabled metrics %s", entry->str);
+                        metric_idx = 14;
+                    }
+                    else if (ctx->processes_scrape_interval > 0) {
+                        /* Create the filefd collector */
+                        ret = flb_input_set_collector_time(in,
+                                                           ne_timer_processes_metrics_cb,
+                                                           ctx->processes_scrape_interval, 0,
+                                                           config);
+                        if (ret == -1) {
+                            flb_plg_error(ctx->ins,
+                                          "could not set systemd collector for Node Exporter Metrics plugin");
+                            return -1;
+                        }
+                        ctx->coll_processes_fd = ret;
+                    }
+                    ne_processes_init(ctx);
+                }
                 else {
                     flb_plg_warn(ctx->ins, "Unknown metrics: %s", entry->str);
                     metric_idx = -1;
@@ -786,6 +825,9 @@ static int in_ne_exit(void *data, struct flb_config *config)
                 else if (strncmp(entry->str, "systemd", 7) == 0) {
                     ne_systemd_exit(ctx);
                 }
+                else if (strncmp(entry->str, "processes", 9) == 0) {
+                    ne_processes_exit(ctx);
+                }
                 else {
                     flb_plg_warn(ctx->ins, "Unknown metrics: %s", entry->str);
                 }
@@ -817,6 +859,9 @@ static int in_ne_exit(void *data, struct flb_config *config)
     if (ctx->coll_systemd_fd != -1) {
         ne_systemd_exit(ctx);
     }
+    if (ctx->coll_processes_fd != -1) {
+        ne_processes_exit(ctx);
+    }
 
     flb_ne_config_destroy(ctx);
 
@@ -870,6 +915,9 @@ static void in_ne_pause(void *data, struct flb_config *config)
     if (ctx->coll_systemd_fd != -1) {
         flb_input_collector_pause(ctx->coll_systemd_fd, ctx->ins);
     }
+    if (ctx->coll_processes_fd != -1) {
+        flb_input_collector_pause(ctx->coll_processes_fd, ctx->ins);
+    }
 }
 
 static void in_ne_resume(void *data, struct flb_config *config)
@@ -919,6 +967,9 @@ static void in_ne_resume(void *data, struct flb_config *config)
     if (ctx->coll_systemd_fd != -1) {
         flb_input_collector_resume(ctx->coll_systemd_fd, ctx->ins);
     }
+    if (ctx->coll_processes_fd != -1) {
+        flb_input_collector_resume(ctx->coll_processes_fd, ctx->ins);
+    }
 }
 
 /* Configuration properties map */
@@ -1013,9 +1064,15 @@ static struct flb_config_map config_map[] = {
      "scrape interval to collect systemd metrics from the node."
     },
 
+    {
+     FLB_CONFIG_MAP_TIME, "collector.processes.scrape_interval", "0",
+     0, FLB_TRUE, offsetof(struct flb_ne, processes_scrape_interval),
+     "scrape interval to collect processes metrics from the node."
+    },
+
     {
      FLB_CONFIG_MAP_CLIST, "metrics",
-     "cpu,cpufreq,meminfo,diskstats,filesystem,uname,stat,time,loadavg,vmstat,netdev,filefd,systemd",
+     NE_DEFAULT_ENABLED_METRICS,
      0, FLB_TRUE, offsetof(struct flb_ne, metrics),
      "Comma separated list of keys to enable metrics."
     },