Merge branch 'master' into fix-storage-inherit-test

Signed-off-by: Jorge Niedbalski <[email protected]>

Author: niedbalski

include/fluent-bit/flb_api.h

@@ -22,6 +22,7 @@
 
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_output.h>
+#include <fluent-bit/flb_custom.h>
 
 struct flb_api {
     const char *(*output_get_property) (const char *, struct flb_output_instance *);
@@ -33,6 +34,11 @@ struct flb_api {
     void (*log_print) (int, const char*, int, const char*, ...);
     int (*input_log_check) (struct flb_input_instance *, int);
     int (*output_log_check) (struct flb_output_instance *, int);
+
+    /* To preserve ABI, we need to add these APIs after the
+     * input/output definitions. */
+    const char *(*custom_get_property) (const char *, struct flb_custom_instance *);
+    int (*custom_log_check) (struct flb_custom_instance *, int);
 };
 
 #ifdef FLB_CORE

include/fluent-bit/flb_config.h

@@ -67,9 +67,10 @@ struct flb_config {
      * shutdown when all remaining tasks are flushed
      */
     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 ?          */
+    int grace_count;          /* Count of grace shutdown tries              */
+    int grace_input;          /* Shutdown grace to keep inputs ingesting    */
+    flb_pipefd_t flush_fd;    /* Timer FD associated to flush               */
+    int convert_nan_to_null;  /* Convert null to nan ?                      */
 
     int daemon;               /* Run as a daemon ?              */
     flb_pipefd_t shutdown_fd; /* Shutdown FD, 5 seconds         */
@@ -243,6 +244,7 @@ struct flb_config {
     int   storage_max_chunks_up;    /* max number of chunks 'up' in memory */
     int   storage_del_bad_chunks;   /* delete irrecoverable chunks */
     char *storage_bl_mem_limit;     /* storage backlog memory limit */
+    int   storage_bl_flush_on_shutdown; /* enable/disable backlog chunks flush on shutdown */
     struct flb_storage_metrics *storage_metrics_ctx; /* storage metrics context */
     int   storage_trim_files;       /* enable/disable file trimming */
     char *storage_type;             /* global storage type */
@@ -394,6 +396,8 @@ enum conf_type {
 #define FLB_CONF_STORAGE_METRICS       "storage.metrics"
 #define FLB_CONF_STORAGE_CHECKSUM      "storage.checksum"
 #define FLB_CONF_STORAGE_BL_MEM_LIMIT  "storage.backlog.mem_limit"
+#define FLB_CONF_STORAGE_BL_FLUSH_ON_SHUTDOWN \
+                                       "storage.backlog.flush_on_shutdown" 
 #define FLB_CONF_STORAGE_MAX_CHUNKS_UP "storage.max_chunks_up"
 #define FLB_CONF_STORAGE_DELETE_IRRECOVERABLE_CHUNKS \
                                        "storage.delete_irrecoverable_chunks"

include/fluent-bit/flb_custom.h

@@ -31,9 +31,20 @@
 #define FLB_CUSTOM_NET_CLIENT   1   /* custom may use upstream net.* properties  */
 #define FLB_CUSTOM_NET_SERVER   2   /* custom may use downstream net.* properties  */
 
+/* Custom plugin types */
+#define FLB_CUSTOM_PLUGIN_CORE   0
+#define FLB_CUSTOM_PLUGIN_PROXY  1
+
 struct flb_custom_instance;
 
 struct flb_custom_plugin {
+    /*
+     * The type defines if this is a core-based plugin or it's handled by
+     * some specific proxy.
+     */
+    int type;
+    void *proxy;
+
     int flags;             /* Flags (not available at the moment */
     char *name;            /* Custom plugin short name           */
     char *description;     /* Description                        */
@@ -49,6 +60,9 @@ struct flb_custom_plugin {
                    void *, struct flb_config *);
     int (*cb_exit) (void *, struct flb_config *);
 
+    /* Destroy */
+    void (*cb_destroy) (struct flb_custom_plugin *);
+
     struct mk_list _head;  /* Link to parent list (config->custom) */
 };
 
@@ -96,5 +110,6 @@ int flb_custom_plugin_property_check(struct flb_custom_instance *ins,
 int flb_custom_init_all(struct flb_config *config);
 void flb_custom_set_context(struct flb_custom_instance *ins, void *context);
 void flb_custom_instance_destroy(struct flb_custom_instance *ins);
+int flb_custom_log_check(struct flb_custom_instance *ins, int l);
 
 #endif

include/fluent-bit/flb_engine.h

@@ -37,6 +37,7 @@ int flb_engine_exit_status(struct flb_config *config, int status);
 int flb_engine_shutdown(struct flb_config *config);
 int flb_engine_destroy_tasks(struct mk_list *tasks);
 void flb_engine_reschedule_retries(struct flb_config *config);
+void flb_engine_stop_ingestion(struct flb_config *config);
 
 /* Engine event loop */
 void flb_engine_evl_init();

include/fluent-bit/flb_plugin_proxy.h

@@ -28,6 +28,7 @@
 /* Plugin Types */
 #define FLB_PROXY_INPUT_PLUGIN     1
 #define FLB_PROXY_OUTPUT_PLUGIN    2
+#define FLB_PROXY_CUSTOM_PLUGIN    3
 
 /* Proxies available */
 #define FLB_PROXY_GOLANG          11

include/fluent-bit/flb_plugins.h.in

@@ -64,6 +64,9 @@ void flb_plugins_unregister(struct flb_config *config)
 
     mk_list_foreach_safe(head, tmp, &config->custom_plugins) {
         custom = mk_list_entry(head, struct flb_custom_plugin, _head);
+        if(custom->cb_destroy) {
+            custom->cb_destroy(custom);
+        }
         mk_list_del(&custom->_head);
         flb_free(custom);
     }

include/fluent-bit/flb_storage.h

@@ -83,4 +83,6 @@ struct flb_storage_metrics *flb_storage_metrics_create(struct flb_config *ctx);
 /* cmetrics */
 int flb_storage_metrics_update(struct flb_config *config, struct flb_storage_metrics *sm);
 
+void flb_storage_chunk_count(struct flb_config *ctx, int *mem_chunks, int *fs_chunks);
+
 #endif

lib/miniz/CMakeLists.txt

@@ -7,7 +7,7 @@ if(DEFINED PROJECT_NAME)
 endif()
 
 if(CMAKE_MINOR_VERSION LESS 12)
-  project(miniz)
+  project(miniz C)
   # see issue https://gitlab.kitware.com/cmake/cmake/merge_requests/1799
 else()
   project(miniz C)

plugins/in_forward/fw.c

@@ -358,7 +358,9 @@ static int in_fw_init(struct flb_input_instance *ins,
 
 static void in_fw_pause(void *data, struct flb_config *config)
 {
+    int ret;
     struct flb_in_fw_config *ctx = data;
+
     if (config->is_running == FLB_TRUE) {
         /*
          * This is the case when we are not in a shutdown phase, but
@@ -367,11 +369,20 @@ static void in_fw_pause(void *data, struct flb_config *config)
          * and wait for the ingestion to resume.
          */
         flb_input_collector_pause(ctx->coll_fd, ctx->ins);
-        if (pthread_mutex_lock(&ctx->conn_mutex)) {
-            fw_conn_del_all(ctx);
-            ctx->is_paused = FLB_TRUE;
+
+        ret = pthread_mutex_lock(&ctx->conn_mutex);
+        if (ret != 0) {
+            flb_plg_error(ctx->ins, "cannot lock collector mutex");
+            return;
+        }
+
+        fw_conn_del_all(ctx);
+        ctx->is_paused = FLB_TRUE;
+        ret = pthread_mutex_unlock(&ctx->conn_mutex);
+        if (ret != 0) {
+            flb_plg_error(ctx->ins, "cannot unlock collector mutex");
+            return;
         }
-        pthread_mutex_unlock(&ctx->conn_mutex);
     }
 
     /*
@@ -388,13 +399,24 @@ static void in_fw_pause(void *data, struct flb_config *config)
 }
 
 static void in_fw_resume(void *data, struct flb_config *config) {
+    int ret;
     struct flb_in_fw_config *ctx = data;
+
     if (config->is_running == FLB_TRUE) {
         flb_input_collector_resume(ctx->coll_fd, ctx->ins);
-        if (pthread_mutex_lock(&ctx->conn_mutex)) {
-            ctx->is_paused = FLB_FALSE;
+
+        ret = pthread_mutex_lock(&ctx->conn_mutex);
+        if (ret != 0) {
+            flb_plg_error(ctx->ins, "cannot lock collector mutex");
+            return;
+        }
+
+        ctx->is_paused = FLB_FALSE;
+        ret = pthread_mutex_unlock(&ctx->conn_mutex);
+        if (ret != 0) {
+            flb_plg_error(ctx->ins, "cannot unlock collector mutex");
+            return;
         }
-        pthread_mutex_unlock(&ctx->conn_mutex);
     }
 }
 

plugins/out_azure_blob/CMakeLists.txt

@@ -6,6 +6,7 @@ set(src
   azure_blob_db.c
   azure_blob_appendblob.c
   azure_blob_blockblob.c
+  azure_blob_store.c
   )
 
 FLB_PLUGIN(out_azure_blob "${src}" "")