config: Add dead letter queue related config

Signed-off-by: Hiroshi Hatake <[email protected]>

Author: cosmo0920

include/fluent-bit/flb_config.h

@@ -251,6 +251,10 @@ struct flb_config {
     char *storage_type;             /* global storage type */
     int   storage_inherit;          /* apply storage type to inputs */
 
+    /* DLQ for non-retriable output failures */
+    int   storage_keep_rejected;     /* 0/1 */
+    char *storage_rejected_path;     /* relative to storage_path, default "rejected" */
+
     /* Embedded SQL Database support (SQLite3) */
 #ifdef FLB_HAVE_SQLDB
     struct mk_list sqldb_list;
@@ -411,6 +415,9 @@ enum conf_type {
 #define FLB_CONF_STORAGE_TRIM_FILES    "storage.trim_files"
 #define FLB_CONF_STORAGE_TYPE          "storage.type"
 #define FLB_CONF_STORAGE_INHERIT       "storage.inherit"
+/* Storage DLQ */
+#define FLB_CONF_STORAGE_KEEP_REJECTED "storage.keep.rejected"
+#define FLB_CONF_STORAGE_REJECTED_PATH "storage.rejected.path"
 
 /* Coroutines */
 #define FLB_CONF_STR_CORO_STACK_SIZE "Coro_Stack_Size"

src/flb_config.c

@@ -163,6 +163,13 @@ struct flb_service_config service_configs[] = {
     {FLB_CONF_STORAGE_INHERIT,
      FLB_CONF_TYPE_BOOL,
      offsetof(struct flb_config, storage_inherit)},
+    /* Storage / DLQ */
+    {FLB_CONF_STORAGE_KEEP_REJECTED,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, storage_keep_rejected)},
+    {FLB_CONF_STORAGE_REJECTED_PATH,
+     FLB_CONF_TYPE_STR,
+     offsetof(struct flb_config, storage_rejected_path)},
 
     /* Coroutines */
     {FLB_CONF_STR_CORO_STACK_SIZE,
@@ -312,6 +319,7 @@ struct flb_config *flb_config_init()
     config->storage_type = NULL;
     config->storage_inherit = FLB_FALSE;
     config->storage_bl_flush_on_shutdown = FLB_FALSE;
+    config->storage_rejected_path = NULL;
     config->sched_cap  = FLB_SCHED_CAP;
     config->sched_base = FLB_SCHED_BASE;
     config->json_escape_unicode = FLB_TRUE;
@@ -573,6 +581,9 @@ void flb_config_exit(struct flb_config *config)
     if (config->storage_bl_mem_limit) {
         flb_free(config->storage_bl_mem_limit);
     }
+    if (config->storage_rejected_path) {
+        flb_free(config->storage_rejected_path);
+    }
 
 #ifdef FLB_HAVE_STREAM_PROCESSOR
     if (config->stream_processor_file) {