engine: Add a capability to handle dead letter queue for preserving invalid chunks for later verifications

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

Author: cosmo0920

src/flb_engine.c

@@ -231,6 +231,50 @@ static inline double calculate_chunk_capacity_percent(struct flb_output_instance
                   ((double)ins->total_limit_size));
 }
 
+static void handle_dlq_if_available(struct flb_config *config,
+                                    struct flb_task *task,
+                                    struct flb_output_instance *ins,
+                                    int status_code /* pass 0 if unknown */)
+{
+    const char *tag_buf = NULL;
+    int         tag_len = 0;
+    flb_sds_t   tag_sds = NULL;
+    const char *tag     = NULL;
+    const char *out     = NULL;
+    struct flb_input_chunk *ic;
+    struct cio_chunk *cio_ch;
+
+    if (!config || !config->storage_keep_rejected || !task || !task->ic || !ins) {
+        return;
+    }
+
+    ic = (struct flb_input_chunk *) task->ic;
+
+    if (!ic || !ic->chunk) {
+        return;
+    }
+
+    /* Obtain tag from the input chunk API (no direct field available) */
+    if (flb_input_chunk_get_tag(ic, &tag_buf, &tag_len) == 0 && tag_buf && tag_len > 0) {
+        tag_sds = flb_sds_create_len(tag_buf, tag_len);  /* make it NUL-terminated */
+        tag     = tag_sds;
+    }
+    else {
+        /* Fallback: use input instance name */
+        tag = flb_input_name(task->i_ins);
+    }
+
+    out    = flb_output_name(ins);
+    cio_ch = (struct cio_chunk *) ic->chunk;  /* ic->chunk is a cio_chunk* under the hood */
+
+    /* Copy bytes into DLQ stream (filesystem) */
+    (void) flb_storage_quarantine_chunk(config, cio_ch, tag, status_code, out);
+
+    if (tag_sds) {
+        flb_sds_destroy(tag_sds);
+    }
+}
+
 static inline int handle_output_event(uint64_t ts,
                                       struct flb_config *config,
                                       uint64_t val)
@@ -353,6 +397,8 @@ static inline int handle_output_event(uint64_t ts,
     }
     else if (ret == FLB_RETRY) {
         if (ins->retry_limit == FLB_OUT_RETRY_NONE) {
+            handle_dlq_if_available(config, task, ins, 0);
+
             /* cmetrics: output_dropped_records_total */
             cmt_counter_add(ins->cmt_dropped_records, ts, task->records,
                             1, (char *[]) {name});
@@ -388,6 +434,8 @@ static inline int handle_output_event(uint64_t ts,
              * - It reached the maximum number of re-tries
              */
 
+            handle_dlq_if_available(config, task, ins, 0);
+
             /* cmetrics */
             cmt_counter_inc(ins->cmt_retries_failed, ts, 1, (char *[]) {name});
             cmt_counter_add(ins->cmt_dropped_records, ts, task->records,
@@ -429,6 +477,8 @@ static inline int handle_output_event(uint64_t ts,
          * memory available or we ran out of file descriptors.
          */
         if (retry_seconds == -1) {
+            handle_dlq_if_available(config, task, ins, 0);
+
             flb_warn("[engine] retry for chunk '%s' could not be scheduled: "
                      "input=%s > output=%s",
                      flb_input_chunk_get_name(task->ic),
@@ -465,6 +515,7 @@ static inline int handle_output_event(uint64_t ts,
         }
     }
     else if (ret == FLB_ERROR) {
+        handle_dlq_if_available(config, task, ins, 0);
         /* cmetrics */
         cmt_counter_inc(ins->cmt_errors, ts, 1, (char *[]) {name});
         cmt_counter_add(ins->cmt_dropped_records, ts, task->records,