filter_rewrite_tag: fix emitter lifecycle and record preservation

include/fluent-bit/flb_input.h

@@ -257,6 +257,7 @@ struct flb_input_instance {
      * as the tags will be composed used the file name being watched.
      */
     int flags;
+    int initialized;                     /* successful instance initialization */
 
     int id;                              /* instance id                  */
 #ifdef FLB_HAVE_CHUNK_TRACE

plugins/filter_rewrite_tag/rewrite_tag.c

@@ -34,6 +34,8 @@
 
 #include "rewrite_tag.h"
 
+static void destroy_rules(struct flb_rewrite_tag *ctx);
+
 /* Create an emitter input instance */
 static int emitter_create(struct flb_rewrite_tag *ctx)
 {
@@ -216,8 +218,19 @@ static int cb_rewrite_tag_init(struct flb_filter_instance *ins,
     flb_sds_t tmp;
     flb_sds_t emitter_name = NULL;
     struct flb_rewrite_tag *ctx;
+    struct flb_processor_unit *processor_unit;
+    struct flb_processor *processor;
     (void) data;
 
+    if (ins->parent_processor != NULL) {
+        processor_unit = ins->parent_processor;
+        processor = processor_unit->parent;
+        if (processor->source_plugin_type != FLB_PLUGIN_INPUT) {
+            flb_plg_error(ins, "rewrite_tag only supports input processors");
+            return -1;
+        }
+    }
+
     /* Create context */
     ctx = flb_calloc(1, sizeof(struct flb_rewrite_tag));
     if (!ctx) {
@@ -250,8 +263,15 @@ static int cb_rewrite_tag_init(struct flb_filter_instance *ins,
             return -1;
         }
 
-        tmp = flb_sds_printf(&emitter_name, "emitter_for_%s",
-                             flb_filter_name(ins));
+        tmp = flb_sds_printf(&emitter_name, "emitter_for_%s", flb_filter_name(ins));
+        if (tmp != NULL && ins->parent_processor != NULL &&
+            flb_input_name_exists(tmp, config) == FLB_TRUE) {
+            /* Preserve existing metric names unless a processor name collides. */
+            flb_sds_len_set(emitter_name, 0);
+            tmp = flb_sds_printf(&emitter_name, "emitter_for_%s_%s_%zu",
+                                 ((struct flb_input_instance *) processor->data)->name,
+                                 flb_filter_name(ins), processor_unit->stage);
+        }
         if (!tmp) {
             flb_error("[filter rewrite_tag] cannot compose emitter_name");
             flb_sds_destroy(emitter_name);
@@ -292,13 +312,13 @@ static int cb_rewrite_tag_init(struct flb_filter_instance *ins,
     /* Process the configuration */
     ret = process_config(ctx);
     if (ret == -1) {
-        return -1;
+        goto error;
     }
 
     /* Create the emitter context */
     ret = emitter_create(ctx);
     if (ret == -1) {
-        return -1;
+        goto error;
     }
 
     /* Register a metric to count the number of emitted records */
@@ -314,6 +334,12 @@ static int cb_rewrite_tag_init(struct flb_filter_instance *ins,
 #endif
 
     return 0;
+
+error:
+    destroy_rules(ctx);
+    flb_filter_set_context(ins, NULL);
+    flb_free(ctx);
+    return -1;
 }
 
 static int ingest_inline(struct flb_rewrite_tag *ctx,
@@ -396,7 +422,9 @@ static int process_record(const char *tag, int tag_len, msgpack_object map,
     flb_regex_results_release(&result);
 
     /* Validate new outgoing tag */
-    if (!out_tag) {
+    if (out_tag == NULL || flb_sds_len(out_tag) == 0) {
+        flb_sds_destroy(out_tag);
+        *keep = FLB_TRUE;
         return FLB_FALSE;
     }
 

plugins/in_emitter/emitter.c

@@ -59,15 +59,15 @@ struct flb_emitter {
     struct mk_list i_ins_list;          /* instance list of linked/sending inputs */
     int cycle_reported;                 /* self-emission cycle already logged ? */
     int propagating;                    /* pause/resume propagation in progress */
+    int ingesting;                      /* currently publishing a pending chunk */
     int owner_flush_enabled;             /* owner supports lossless shutdown flush */
     int shutdown_flush_done;            /* owner shutdown hook already invoked */
     int shutdown_flush_active;          /* owner shutdown hook is emitting */
 };
 
 static inline int shutdown_flush_enabled(struct flb_emitter *ctx)
 {
-    if (ctx->ins->config->is_shutting_down == FLB_TRUE &&
-        ctx->owner_flush_enabled == FLB_TRUE) {
+    if (ctx->ins->config->is_shutting_down == FLB_TRUE) {
         return FLB_TRUE;
     }
 
@@ -163,10 +163,12 @@ int static do_in_emitter_add_record(struct em_chunk *ec,
     }
 
     /* Associate this backlog chunk to this instance into the engine */
+    ctx->ingesting = FLB_TRUE;
     ret = flb_input_log_append(in,
                                ec->tag, flb_sds_len(ec->tag),
                                ec->mp_sbuf.data,
                                ec->mp_sbuf.size);
+    ctx->ingesting = FLB_FALSE;
     if (ret == -1) {
         flb_plg_error(ctx->ins, "error registering chunk with tag: %s", ec->tag);
         /* Release the echunk */
@@ -206,7 +208,7 @@ int in_emitter_add_record(const char *tag, int tag_len,
      * same instance until the stack is exhausted, and the emission itself is an
      * endless tag rewrite cycle.
      */
-    if (i_ins == ctx->ins) {
+    if (i_ins == ctx->ins || ctx->ingesting == FLB_TRUE) {
         if (ctx->cycle_reported == FLB_FALSE) {
             ctx->cycle_reported = FLB_TRUE;
             flb_plg_error(ctx->ins,
@@ -233,7 +235,7 @@ int in_emitter_add_record(const char *tag, int tag_len,
         i_ref = flb_malloc(sizeof(struct input_ref));
         if (!i_ref) {
             flb_errno();
-            return FLB_FILTER_NOTOUCH;
+            return -1;
         }
         i_ref->i_ins = i_ins;
         mk_list_add(&i_ref->_head, &ctx->i_ins_list);
@@ -325,6 +327,10 @@ int in_emitter_add_record(const char *tag, int tag_len,
     }
 
     ctx->pending_bytes += buf_size;
+    /* A downstream emitter may already have received its shutdown pause. */
+    if (shutdown_flush_enabled(ctx) == FLB_TRUE && ctx->shutdown_flush_done == FLB_TRUE) {
+        return do_in_emitter_add_record(ec, in);
+    }
     return 0;
 }
 
@@ -428,9 +434,10 @@ static int cb_emitter_init(struct flb_input_instance *in,
     mk_list_init(&ctx->i_ins_list);
 
     callbacks = (struct flb_emitter_callbacks *) in->data;
+    /* Pending records must be published before ingestion stops at shutdown. */
+    in->flags |= FLB_INPUT_SHUTDOWN_FLUSH;
     if (callbacks != NULL && callbacks->pause != NULL) {
         ctx->owner_flush_enabled = FLB_TRUE;
-        in->flags |= FLB_INPUT_SHUTDOWN_FLUSH;
     }
 
 
@@ -507,24 +514,21 @@ static void cb_emitter_pause(void *data, struct flb_config *config)
     ctx->propagating = FLB_TRUE;
 
     if (shutdown_flush_enabled(ctx) == FLB_TRUE &&
-        ctx->shutdown_flush_done == FLB_FALSE &&
-        ctx->ins->data != NULL) {
+        ctx->shutdown_flush_done == FLB_FALSE) {
         callbacks = (struct flb_emitter_callbacks *) ctx->ins->data;
         ctx->shutdown_flush_done = FLB_TRUE;
 
-        if (callbacks->pause != NULL) {
-            /* Publish any records already waiting in a threaded emitter. */
-            if (ctx->msgs != NULL) {
-                in_emitter_ingest_ring_buffer(ctx->ins, config, ctx);
-            }
+        /* Publish queued records even when this emitter has no owner hook. */
+        if (ctx->msgs != NULL) {
+            in_emitter_ingest_ring_buffer(ctx->ins, config, ctx);
+        }
 
+        if (callbacks != NULL && callbacks->pause != NULL) {
             ctx->shutdown_flush_active = FLB_TRUE;
             callbacks->pause(callbacks->data, config);
             ctx->shutdown_flush_active = FLB_FALSE;
-
-            /* Convert owner-emitted records into input chunks before pausing. */
-            cb_queue_chunks(ctx->ins, config, ctx);
         }
+        cb_queue_chunks(ctx->ins, config, ctx);
     }
 
     /* Pause all known senders */

src/flb_input.c

@@ -1576,6 +1576,11 @@ int flb_input_instance_init(struct flb_input_instance *ins,
     struct flb_input_plugin *p = ins->p;
     int tls_session_mode;
 
+    /* Private inputs can be initialized while another input initializes. */
+    if (ins->initialized == FLB_TRUE) {
+        return 0;
+    }
+
     if (ins->log_level == -1 && config->log != NULL) {
         ins->log_level = config->log->level;
     }
@@ -2049,6 +2054,7 @@ int flb_input_instance_init(struct flb_input_instance *ins,
         }
     }
 
+    ins->initialized = FLB_TRUE;
     return 0;
 }
 

src/flb_processor.c

@@ -2026,6 +2026,7 @@ static int processor_run(struct flb_processor *proc, struct chunk_codec *codec,
                  * no more data exists to be processed.
                  */
                 if (tmp_size == 0) {
+                    flb_free(tmp_buf);
                     *out_buf = NULL;
                     *out_size = 0;
                     unit_out_items = 0;

tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_groups.py

@@ -28,7 +28,9 @@ def exported_records(log_file):
 
 @pytest.mark.parametrize("reverse", [False, True])
 @pytest.mark.parametrize("mode", ["split", "same", "keep", "unmatched"])
-def test_rewrite_tag_preserves_group_context(tmp_path, reverse, mode):
+@pytest.mark.parametrize("placement", ["filter", "processor"])
+@pytest.mark.parametrize("profile", ["memory", "filesystem", "threaded-filesystem"])
+def test_rewrite_tag_preserves_group_context(tmp_path, reverse, mode, placement, profile):
     messages = [
         {"level": "info", "message": "order accepted"},
         {"level": "error", "message": "payment authorization failed"},
@@ -86,6 +88,19 @@ def test_rewrite_tag_preserves_group_context(tmp_path, reverse, mode):
             "dummy": json.dumps({"level": "info", "message": "ungrouped event"}),
         },
     ])
+    config["service"]["storage.path"] = str(tmp_path / "storage")
+    storage = "memory" if profile == "memory" else "filesystem"
+    config["pipeline"]["filters"][0]["emitter_storage.type"] = storage
+    for input_config in config["pipeline"]["inputs"]:
+        input_config["storage.type"] = storage
+        input_config["threaded"] = profile == "threaded-filesystem"
+    if placement != "filter":
+        rewrite = config["pipeline"].pop("filters")[0]
+        rewrite.pop("match")
+        for input_config in config["pipeline"]["inputs"]:
+            input_config.setdefault("processors", {}).setdefault("logs", []).append(
+                copy.deepcopy(rewrite)
+            )
     config_path = tmp_path / "rewrite_tag.yaml"
     config_path.write_text(yaml.safe_dump(config))
     service = FluentBitTestService(str(config_path))