core: signal blob implementation

cmake/plugins_options.cmake

@@ -11,6 +11,7 @@ option(FLB_MINIMAL "Enable minimal build configuration" No)
 
 # Inputs (sources, data collectors)
 # =================================
+DEFINE_OPTION(FLB_IN_BLOB                     "Enable Blob input plugin"                     ON)
 DEFINE_OPTION(FLB_IN_CALYPTIA_FLEET           "Enable Calyptia Fleet input plugin"           ON)
 DEFINE_OPTION(FLB_IN_COLLECTD                 "Enable Collectd input plugin"                 ON)
 DEFINE_OPTION(FLB_IN_CPU                      "Enable CPU input plugin"                      ON)

include/fluent-bit/flb_config.h

@@ -48,11 +48,11 @@ struct flb_config {
     int is_running;           /* service running ?              */
     double flush;             /* Flush timeout                  */
 
-    /* 
-     * Maximum grace time on shutdown. If set to -1, the engine will 
+    /*
+     * Maximum grace time on shutdown. If set to -1, the engine will
      * shutdown when all remaining tasks are flushed
      */
-    int grace; 
+    int grace;
     int grace_count;          /* Count of grace shutdown tries  */
     flb_pipefd_t flush_fd;    /* Timer FD associated to flush   */
     int convert_nan_to_null;  /* convert null to nan ?          */
@@ -71,6 +71,10 @@ struct flb_config {
 
     flb_pipefd_t ch_self_events[2]; /* channel to recieve thread tasks        */
 
+    int notification_channels_initialized;
+    flb_pipefd_t notification_channels[2];
+    struct mk_event notification_event;
+
     /* Channel event loop (just for ch_notif) */
     struct mk_event_loop *ch_evl;
 

include/fluent-bit/flb_engine_macros.h

@@ -26,16 +26,20 @@
 /* Types of events handled by the Server engine */
 #define FLB_ENGINE_EV_CORE          MK_EVENT_NOTIFICATION
 #define FLB_ENGINE_EV_CUSTOM        MK_EVENT_CUSTOM
-#define FLB_ENGINE_EV_THREAD        1024
-#define FLB_ENGINE_EV_SCHED         2048
-#define FLB_ENGINE_EV_SCHED_FRAME   (FLB_ENGINE_EV_SCHED + 4096)
 
-#define FLB_ENGINE_EV_INPUT          8192
-#define FLB_ENGINE_EV_THREAD_INPUT  16384  /* reserved, not used yet */
+#define FLB_ENGINE_EV_THREAD        (1 << 10)                          /*  1024 */
+#define FLB_ENGINE_EV_SCHED         (1 << 11)                          /*  2048 */
+#define FLB_ENGINE_EV_SCHED_FRAME   (FLB_ENGINE_EV_SCHED | (1 << 12))  /*  2048 | 4096 = 6144  */
+#define FLB_ENGINE_EV_SCHED_CORO    (1 << 13)                          /*  8192 */
 
-#define FLB_ENGINE_EV_OUTPUT        32768
-#define FLB_ENGINE_EV_THREAD_OUTPUT 65536
-#define FLB_ENGINE_EV_THREAD_ENGINE 131072
+#define FLB_ENGINE_EV_INPUT         (1 << 14)                          /*  16384 */
+#define FLB_ENGINE_EV_THREAD_INPUT  (1 << 15)                          /*  32768 */
+
+#define FLB_ENGINE_EV_OUTPUT        (1 << 16)                          /*  65536 */
+#define FLB_ENGINE_EV_THREAD_OUTPUT (1 << 17)                          /* 131072 */
+#define FLB_ENGINE_EV_THREAD_ENGINE (1 << 18)                          /* 262144 */
+
+#define FLB_ENGINE_EV_NOTIFICATION  (1 << 19)                          /* 524288 */
 
 /* Engine events: all engine events set the left 32 bits to '1' */
 #define FLB_ENGINE_EV_STARTED   FLB_BITS_U64_SET(1, 1) /* Engine started    */

include/fluent-bit/flb_event.h

@@ -28,6 +28,7 @@
 #define FLB_EVENT_TYPE_LOGS       FLB_INPUT_CHUNK_TYPE_LOGS
 #define FLB_EVENT_TYPE_METRICS    FLB_INPUT_CHUNK_TYPE_METRICS
 #define FLB_EVENT_TYPE_TRACES     FLB_INPUT_CHUNK_TYPE_TRACES
+#define FLB_EVENT_TYPE_BLOBS      FLB_INPUT_CHUNK_TYPE_BLOBS
 
 #define FLB_EVENT_TYPE_HAS_TRACE FLB_INPUT_CHUNK_HAS_TRACE
 

include/fluent-bit/flb_filter.h

@@ -74,6 +74,9 @@ struct flb_filter_plugin {
                       void *, struct flb_config *);
     int (*cb_exit) (void *, struct flb_config *);
 
+    /* Notification: this callback will be invoked anytime a notification is received*/
+    int (*cb_notification) (struct flb_filter_instance *, struct flb_config *, void *);
+
     struct mk_list _head;  /* Link to parent list (config->filters) */
 };
 
@@ -110,6 +113,7 @@ struct flb_filter_instance {
 #ifdef FLB_HAVE_METRICS
     struct flb_metrics *metrics;   /* metrics                  */
 #endif
+    flb_pipefd_t notification_channel;
 
     /* Keep a reference to the original context this instance belongs to */
     struct flb_config *config;

include/fluent-bit/flb_http_client.h

@@ -36,6 +36,7 @@
 #define FLB_HTTP_HEAD        3
 #define FLB_HTTP_CONNECT     4
 #define FLB_HTTP_PATCH       5
+#define FLB_HTTP_DELETE      6
 
 /* HTTP Flags */
 #define FLB_HTTP_10          1

include/fluent-bit/flb_input.h

@@ -102,6 +102,9 @@ struct flb_input_plugin {
     /* Collect: every certain amount of time, Fluent Bit trigger this callback */
     int (*cb_collect) (struct flb_input_instance *, struct flb_config *, void *);
 
+    /* Notification: this callback will be invoked anytime a notification is received*/
+    int (*cb_notification) (struct flb_input_instance *, struct flb_config *, void *);
+
     /*
      * Flush: each plugin during a collection, it does some buffering,
      * when the Flush timer takes place on the Engine, it will trigger
@@ -381,6 +384,8 @@ struct flb_input_instance {
     struct mk_list *net_config_map;
     struct mk_list net_properties;
 
+    flb_pipefd_t notification_channel;
+
     /* Keep a reference to the original context this instance belongs to */
     struct flb_config *config;
 };
@@ -577,7 +582,7 @@ static FLB_INLINE int flb_input_is_threaded(struct flb_input_instance *ins)
  * number of retries, if it has exceeded the 'retry_limit' option, an FLB_ERROR
  * will be returned instead.
  */
-static inline void flb_input_return(struct flb_coro *coro) {
+static FLB_INLINE void flb_input_return(struct flb_coro *coro) {
     int n;
     uint64_t val;
     struct flb_input_coro *input_coro;
@@ -599,7 +604,7 @@ static inline void flb_input_return(struct flb_coro *coro) {
     flb_input_coro_prepare_destroy(input_coro);
 }
 
-static inline void flb_input_return_do(int ret) {
+static FLB_INLINE void flb_input_return_do(int ret) {
     struct flb_coro *coro = flb_coro_get();
 
     flb_input_return(coro);
@@ -733,6 +738,7 @@ int flb_input_upstream_set(struct flb_upstream *u, struct flb_input_instance *in
 int flb_input_downstream_set(struct flb_downstream *stream,
                              struct flb_input_instance *ins);
 
+
 /* processors */
 int flb_input_instance_processors_load(struct flb_input_instance *ins, struct flb_cf_group *processors);
 

include/fluent-bit/flb_input_blob.h

@@ -0,0 +1,47 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2024 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_INPUT_BLOB_H
+#define FLB_INPUT_BLOB_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_input.h>
+#include <fluent-bit/flb_notification.h>
+#include <fluent-bit/flb_log_event_encoder.h>
+
+struct flb_blob_delivery_notification {
+    struct flb_notification base;
+    cfl_sds_t path;
+    int success;
+};
+
+
+struct flb_blob_file {
+    cfl_sds_t path;
+};
+
+void flb_input_blob_delivery_notification_destroy(void *instance);
+
+int flb_input_blob_file_get_info(msgpack_object map, cfl_sds_t *source,
+                                 cfl_sds_t *file_path, size_t *size);
+int flb_input_blob_file_register(struct flb_input_instance *ins,
+                                 struct flb_log_event_encoder *encoder,
+                                 const char *tag, size_t tag_len,
+                                 char *file_path, size_t size);
+#endif

include/fluent-bit/flb_input_chunk.h

@@ -50,6 +50,7 @@
 #define FLB_INPUT_CHUNK_TYPE_LOGS      0
 #define FLB_INPUT_CHUNK_TYPE_METRICS   1
 #define FLB_INPUT_CHUNK_TYPE_TRACES    2
+#define FLB_INPUT_CHUNK_TYPE_BLOBS     3
 
 #ifdef FLB_HAVE_CHUNK_TRACE
 #define FLB_INPUT_CHUNK_HAS_TRACE     1 << 31

include/fluent-bit/flb_input_event.h

@@ -20,10 +20,10 @@
 #ifndef FLB_INPUT_EVENT_H
 #define FLB_INPUT_EVENT_H
 
-/* support event types by input plugins*/
-
+/* support event types by input plugins */
 #define FLB_INPUT_LOGS        0
 #define FLB_INPUT_METRICS     1
 #define FLB_INPUT_TRACES      2
+#define FLB_INPUT_BLOBS       3
 
 #endif

include/fluent-bit/flb_input_thread.h

@@ -64,6 +64,9 @@ struct flb_input_thread_instance {
     struct mk_event_loop *evl;           /* thread event loop context */
     flb_pipefd_t ch_parent_events[2];    /* communication between parent and thread */
     flb_pipefd_t ch_thread_events[2];    /* local messages in the thread event loop */
+    int notification_channels_initialized;
+    flb_pipefd_t notification_channels[2];
+    struct mk_event notification_event;
     struct flb_input_instance *ins;      /* output plugin instance */
     struct flb_tp *tp;
     struct flb_tp_thread *th;

include/fluent-bit/flb_notification.h

@@ -0,0 +1,54 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2024 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_NOTIFICATION_H
+#define FLB_NOTIFICATION_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_socket.h>
+#include <fluent-bit/flb_config.h>
+#include <cfl/cfl_sds.h>
+
+
+#define FLB_NOTIFICATION_TYPE_BLOB_DELIVERY 1
+
+
+struct flb_notification {
+    int   plugin_type;
+    void *plugin_instance;
+    int   notification_type;
+    int   dynamically_allocated;
+
+    void (*destructor)(void *);
+};
+
+int flb_notification_enqueue(int plugin_type,
+                             char *instance_name,
+                             struct flb_notification *notification,
+                             struct flb_config *config);
+
+int flb_notification_receive(flb_pipefd_t channel,
+                             struct flb_notification **notification);
+
+int flb_notification_deliver(struct flb_notification *notification);
+
+void flb_notification_cleanup(struct flb_notification *notification);
+
+#endif

include/fluent-bit/flb_output.h

@@ -89,6 +89,7 @@ int flb_chunk_trace_output(struct flb_chunk_trace *trace, struct flb_output_inst
 #define FLB_OUTPUT_LOGS        1
 #define FLB_OUTPUT_METRICS     2
 #define FLB_OUTPUT_TRACES      4
+#define FLB_OUTPUT_BLOBS       8
 
 #define FLB_OUTPUT_FLUSH_COMPAT_OLD_18()                 \
     const void *data   = event_chunk->data;              \
@@ -232,6 +233,12 @@ struct flb_output_plugin {
     /* Default number of worker threads */
     int workers;
 
+    int (*cb_worker_init) (void *, struct flb_config *);
+    int (*cb_worker_exit) (void *, struct flb_config *);
+
+    /* Notification: this callback will be invoked anytime a notification is received*/
+    int (*cb_notification) (struct flb_output_instance *, struct flb_config *, void *);
+
     /* Tests */
     struct flb_test_out_formatter test_formatter;
 
@@ -445,6 +452,8 @@ struct flb_output_instance {
     struct mk_list flush_list;
     struct mk_list flush_list_destroy;
 
+    flb_pipefd_t notification_channel;
+
     /* Keep a reference to the original context this instance belongs to */
     struct flb_config *config;
 };

include/fluent-bit/flb_output_thread.h

@@ -63,6 +63,9 @@ struct flb_out_thread_instance {
     struct flb_bucket_queue *evl_bktq;    /* bucket queue for evl track event priority */
     flb_pipefd_t ch_parent_events[2];    /* channel to receive parent notifications */
     flb_pipefd_t ch_thread_events[2];    /* channel to send messages local event loop */
+    int notification_channels_initialized;
+    flb_pipefd_t notification_channels[2];
+    struct mk_event notification_event;
     struct flb_output_instance *ins;     /* output plugin instance */
     struct flb_config *config;
     struct flb_tp_thread *th;

include/fluent-bit/flb_processor.h

@@ -119,6 +119,8 @@ struct flb_processor {
     void *data;
     int source_plugin_type;
 
+    flb_pipefd_t notification_channel;
+
     /* Fluent Bit context */
     struct flb_config *config;
 };
@@ -155,6 +157,9 @@ struct flb_processor_plugin {
 
     int (*cb_exit) (struct flb_processor_instance *, void *);
 
+    /* Notification: this callback will be invoked anytime a notification is received*/
+    int (*cb_notification) (struct flb_processor_instance *, struct flb_config *, void *);
+
     struct mk_list _head;  /* Link to parent list (config->filters) */
 };
 
@@ -179,6 +184,8 @@ struct flb_processor_instance {
      */
     struct cmt *cmt;                      /* parent context               */
 
+    flb_pipefd_t notification_channel;
+
     /* Keep a reference to the original context this instance belongs to */
     struct flb_config *config;
 };

include/fluent-bit/flb_router.h

@@ -44,6 +44,10 @@ static inline int flb_router_match_type(int in_event_type,
              !(o_ins->event_type & FLB_OUTPUT_TRACES)) {
         return FLB_FALSE;
     }
+    else if (in_event_type == FLB_INPUT_BLOBS &&
+             !(o_ins->event_type & FLB_OUTPUT_BLOBS)) {
+        return FLB_FALSE;
+    }
 
     return FLB_TRUE;
 }