From 858ad131ebf9e88814335be23137f0b759fe5ed6 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 26 Oct 2026 13:34:32 -0600 Subject: [PATCH 1/8] input: avoid repeated initialization of private inputs Signed-off-by: Eduardo Silva --- include/fluent-bit/flb_input.h | 1 + src/flb_input.c | 6 ++++++ 2 files changed, 7 insertions(+) diff --git a/include/fluent-bit/flb_input.h b/include/fluent-bit/flb_input.h index 4824ea824af..bcf5af3e301 100644 --- a/include/fluent-bit/flb_input.h +++ b/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 diff --git a/src/flb_input.c b/src/flb_input.c index ddc2a46f297..db16ae43a25 100644 --- a/src/flb_input.c +++ b/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; } From e2032bd5967bfee1f52e3a9e42692da7678e0951 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 26 Oct 2026 13:34:32 -0600 Subject: [PATCH 2/8] tests: internal: verify private input initialization is idempotent Signed-off-by: Eduardo Silva --- tests/internal/processor.c | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/tests/internal/processor.c b/tests/internal/processor.c index 91824e7f3e3..f561f7c05b5 100644 --- a/tests/internal/processor.c +++ b/tests/internal/processor.c @@ -375,6 +375,8 @@ static void processor_grouped_filter_counters() static void processor_private_inputs_use_main_loop() { int ret; + int collector_count; + void *context; struct flb_config *config; struct flb_input_instance *ins; struct mk_event_loop *thread_evl; @@ -421,6 +423,14 @@ static void processor_private_inputs_use_main_loop() ret = flb_input_instance_init(ins, config); TEST_CHECK(ret == 0); + /* Initializing a private input again must not replace its owned state. */ + context = ins->context; + collector_count = mk_list_size(&ins->collectors); + ret = flb_input_instance_init(ins, config); + TEST_CHECK(ret == 0); + TEST_CHECK(ins->context == context); + TEST_CHECK(mk_list_size(&ins->collectors) == collector_count); + mk_list_foreach(head, &ins->collectors) { coll = mk_list_entry(head, struct flb_input_collector, _head); TEST_CHECK(coll->evl == config->evl); From 65a40ced48d5304079660f7fbae9f9a6fa03bbaf Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 26 Oct 2026 13:34:32 -0600 Subject: [PATCH 3/8] processor: free allocated empty filter results Signed-off-by: Eduardo Silva --- src/flb_processor.c | 1 + 1 file changed, 1 insertion(+) diff --git a/src/flb_processor.c b/src/flb_processor.c index 8e68a2353a1..a2d3faca185 100644 --- a/src/flb_processor.c +++ b/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; From c1147a19c648cb8653323155ceb98aa69a31fb96 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 26 Oct 2026 13:34:32 -0600 Subject: [PATCH 4/8] tests: internal: cover allocated empty processor results Signed-off-by: Eduardo Silva --- tests/internal/processor.c | 68 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 68 insertions(+) diff --git a/tests/internal/processor.c b/tests/internal/processor.c index f561f7c05b5..dfaa77775fc 100644 --- a/tests/internal/processor.c +++ b/tests/internal/processor.c @@ -1563,8 +1563,76 @@ static void processor_native_chain_counters(void) } } +/* A filter may return an owned allocation even when it drops every record. */ +static int filter_empty_allocated_result(const void *data, size_t bytes, + const char *tag, int tag_len, + void **out_buf, size_t *out_bytes, + struct flb_filter_instance *filter, + struct flb_input_instance *input, + void *context, struct flb_config *config) +{ + (void) data; + (void) bytes; + (void) tag; + (void) tag_len; + (void) filter; + (void) input; + (void) context; + (void) config; + + *out_buf = flb_malloc(65536); + *out_bytes = 0; + if (!TEST_CHECK(*out_buf != NULL)) { + return FLB_FILTER_NOTOUCH; + } + return FLB_FILTER_MODIFIED; +} + +static void processor_filter_empty_allocated_result(void) +{ + int ret; + char *input; + size_t input_size; + void *output = NULL; + size_t output_size = 0; + struct flb_config *config; + struct flb_processor *processor; + struct flb_processor_unit *unit; + struct flb_filter_instance *filter; + struct flb_filter_plugin plugin; + struct flb_filter_plugin *original_plugin; + + flb_init_env(); + config = flb_config_init(); + TEST_ASSERT(config != NULL); + processor = flb_processor_create(config, "empty_result", NULL, 0); + TEST_ASSERT(processor != NULL); + unit = flb_processor_unit_create(processor, FLB_PROCESSOR_LOGS, "stdout"); + TEST_ASSERT(unit != NULL); + TEST_ASSERT(flb_processor_init(processor) == 0); + TEST_ASSERT(create_msgpack_records(&input, &input_size) == 0); + + filter = unit->ctx; + original_plugin = filter->p; + plugin = *original_plugin; + plugin.cb_filter = filter_empty_allocated_result; + filter->p = &plugin; + ret = flb_processor_run(processor, 0, FLB_PROCESSOR_LOGS, + "test", 4, input, input_size, &output, &output_size); + filter->p = original_plugin; + TEST_CHECK(ret == 0); + TEST_CHECK(output == NULL); + TEST_CHECK(output_size == 0); + + /* Valgrind must also verify that the callback's allocation was released. */ + flb_free(input); + flb_processor_destroy(processor); + flb_config_exit(config); +} + TEST_LIST = { { "processor_native_chain_counters", processor_native_chain_counters }, + { "processor_filter_empty_allocated_result", processor_filter_empty_allocated_result }, {"processor_metrics_failure_clears_output", processor_metrics_failure_clears_output}, { "processor_private_inputs_use_main_loop", processor_private_inputs_use_main_loop }, { "processor", processor }, From 80826bf2ca64167d704a40aa5abc6276740985ec Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 26 Oct 2026 13:34:33 -0600 Subject: [PATCH 5/8] in_emitter: drain pending records during shutdown Signed-off-by: Eduardo Silva --- plugins/in_emitter/emitter.c | 34 +++++++++++++++++++--------------- 1 file changed, 19 insertions(+), 15 deletions(-) diff --git a/plugins/in_emitter/emitter.c b/plugins/in_emitter/emitter.c index 746e1ca2cf8..15b5ecee77a 100644 --- a/plugins/in_emitter/emitter.c +++ b/plugins/in_emitter/emitter.c @@ -59,6 +59,7 @@ 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 */ @@ -66,8 +67,7 @@ struct flb_emitter { 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 */ From d0bb6071fe4db4a91c201ca2ffe27a14bbeb21f7 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 26 Oct 2026 13:34:33 -0600 Subject: [PATCH 6/8] tests: integration: cover emitter shutdown and recovery Signed-off-by: Eduardo Silva --- .../tests/test_filter_rewrite_tag_paths.py | 165 ++++++++++++++++++ .../test_filter_rewrite_tag_regressions.py | 53 ++++++ 2 files changed, 218 insertions(+) create mode 100644 tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_paths.py create mode 100644 tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_regressions.py diff --git a/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_paths.py b/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_paths.py new file mode 100644 index 00000000000..fe5e19c1ce2 --- /dev/null +++ b/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_paths.py @@ -0,0 +1,165 @@ +import copy +import json +from collections import Counter +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +import threading +from types import SimpleNamespace + +import pytest +import requests +import yaml + +from utils.fluent_bit_manager import FluentBitStartupError +from utils.test_service import FluentBitTestService + + +def records(service): + result = [] + for line in Path(service.flb.log_file).read_text().splitlines(): + try: + value = json.loads(line) + except ValueError: + continue + if isinstance(value, dict) and "id" in value: + result.append(value) + return result + + +def config_for(tmp_path, messages, rewrite, placement="filter"): + input_config = {"name": "dummy", "tag": "source.app", "samples": 1, + "dummy": " ".join(json.dumps(message) for message in messages)} + config = { + "service": {"flush": 0.1, "grace": 1, "log_level": "info", + "scheduler.base": 1, "scheduler.cap": 1, + "storage.path": str(tmp_path / "storage"), + "http_server": "on", "http_port": "${FLUENT_BIT_HTTP_MONITORING_PORT}"}, + "pipeline": {"inputs": [input_config], "outputs": [ + {"name": "stdout", "match": "*", "format": "json_lines", "workers": 0}, + ]}, + } + if placement == "filter": + rewrite["match"] = "source.app" + config["pipeline"]["filters"] = [rewrite] + else: + input_config["processors"] = {"logs": [rewrite]} + for tag in ["source.app", "routed.error.blue.app", "routed", "wrong"]: + config["pipeline"].setdefault("filters", []).append( + {"name": "record_modifier", "match": tag, "record": f"route {tag}"} + ) + return config + + +def service_for(tmp_path, config, name="config.yaml"): + path = tmp_path / name + path.write_text(yaml.safe_dump(config)) + return FluentBitTestService(str(path)) + + +@pytest.fixture +def retry_sink(): + state = SimpleNamespace(allow=threading.Event(), attempted=threading.Event(), + lock=threading.Lock(), received=[]) + + class Sink(BaseHTTPRequestHandler): + def do_POST(self): + batch = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + state.attempted.set() + if state.allow.is_set(): + with state.lock: + state.received.extend(record["id"] for record in batch) + status = 200 + else: + status = 503 + self.send_response(status) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, *_args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Sink) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + state.port = server.server_port + try: + yield state + finally: + state.allow.set() + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def add_retry_output(config, sink): + config["pipeline"]["outputs"].append({ + "name": "http", "match": "routed", "host": "127.0.0.1", "port": sink.port, + "uri": "/records", "format": "json", "json_date_key": False, + "workers": 1, "retry_limit": False, + }) + + +def test_emitter_backpressure_keeps_records_and_resumes_input(tmp_path, retry_sink): + rewrite = {"name": "rewrite_tag", "emitter_name": "pressure_emitter", + "emitter_mem_buf_limit": "1K", "rule": "$kind ^emit$ routed false"} + config = config_for(tmp_path, [], rewrite) + config["pipeline"]["inputs"] = [{ + "name": "http", "listen": "127.0.0.1", "port": "${FLUENT_BIT_TEST_LISTENER_PORT}", + }] + add_retry_output(config, retry_sink) + service = service_for(tmp_path, config) + try: + service.start() + url = f"http://127.0.0.1:{service.flb_listener_port}/source.app" + messages = [{"id": i, "kind": "emit", "padding": "x" * 200} for i in range(24)] + response = requests.post(url, json=messages, timeout=15) + assert response.status_code == 201, response.text + service.wait_for_condition(lambda: len(records(service)) == 24, timeout=30) + service.wait_for_condition( + lambda: "pressure_emitter paused (mem buf overlimit" in Path(service.flb.log_file).read_text(), + timeout=30, description="emitter pause propagated under output retries", + ) + assert retry_sink.attempted.wait(30) + assert {r["route"] for r in records(service)} == {"source.app", "routed"} + retry_sink.allow.set() + service.wait_for_condition( + lambda: "pressure_emitter resume (mem buf overlimit" in Path(service.flb.log_file).read_text(), + timeout=30, description="emitter resumes after output recovery", + ) + response = requests.post(url, json={"id": 24, "kind": "emit"}, timeout=15) + assert response.status_code == 201, response.text + service.wait_for_condition(lambda: len(records(service)) == 25, timeout=30) + routed = Counter(r["id"] for r in records(service) if r["route"] == "routed") + service.wait_for_condition(lambda: Counter(retry_sink.received) == routed, timeout=30) + finally: + retry_sink.allow.set() + service.stop() + assert Counter(r["id"] for r in records(service)) == Counter(range(25)) + + +def test_filesystem_emitter_backlog_replays_with_live_records(tmp_path, retry_sink): + rewrite = {"name": "rewrite_tag", "emitter_name": "persistent_emitter", + "emitter_storage.type": "filesystem", "rule": "$kind ^emit$ routed false"} + config = config_for(tmp_path, [{"id": i, "kind": "emit"} for i in range(3)], rewrite) + add_retry_output(config, retry_sink) + first = service_for(tmp_path, config, "seed.yaml") + chunks = tmp_path / "storage" + try: + first.start() + assert retry_sink.attempted.wait(30) + first.wait_for_condition(lambda: list(chunks.glob("*/*.flb")), timeout=30) + finally: + first.stop() + assert list(chunks.glob("*/*.flb")), "undelivered emitter chunks must survive shutdown" + retry_sink.allow.set() + config["pipeline"]["inputs"][0]["dummy"] = " ".join( + json.dumps({"id": i, "kind": "emit"}) for i in range(3, 5) + ) + second = service_for(tmp_path, config, "replay.yaml") + try: + second.start() + second.wait_for_condition(lambda: Counter(retry_sink.received) == Counter(range(5)), timeout=30) + second.wait_for_condition(lambda: not list(chunks.glob("*/*.flb")), timeout=30) + finally: + second.stop() + assert Counter(retry_sink.received) == Counter(range(5)) diff --git a/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_regressions.py b/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_regressions.py new file mode 100644 index 00000000000..04fed98b38f --- /dev/null +++ b/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_regressions.py @@ -0,0 +1,53 @@ +import json +from collections import Counter +from pathlib import Path + +import pytest +import yaml + +from utils.fluent_bit_manager import FluentBitStartupError +from utils.test_service import FluentBitTestService + + +@pytest.mark.parametrize("chain", ["single", "forward", "reverse", "cycle-forward", "cycle-reverse"]) +@pytest.mark.parametrize("storage", ["memory", "filesystem"]) +def test_pending_emitter_records_drain_at_eof(tmp_path, chain, storage): + source = tmp_path / "source.log" + source.write_text("hello\n" * 1000) + filters = [{"name": "rewrite_tag", "match": "source", + "rule": f"$log ^hello$ {'routed' if chain == 'single' else 'middle'} false"}] + if chain != "single": + filters.append({"name": "rewrite_tag", "match": "middle", + "rule": f"$log ^hello$ {'source' if chain.startswith('cycle') else 'routed'} false"}) + if chain in ("reverse", "cycle-reverse"): + filters.reverse() + config = { + "service": {"flush": 0.2, "grace": 1, "log_level": "error", + "http_server": "on", "http_port": "${FLUENT_BIT_HTTP_MONITORING_PORT}"}, + "pipeline": { + "inputs": [{"name": "tail", "tag": "source", "path": str(source), + "read_from_head": True, "exit_on_eof": True}], + "filters": filters, + "outputs": [{"name": "counter", "match": "*" if chain.startswith("cycle") else "routed", + "workers": 0}], + }, + } + config["service"]["storage.path"] = str(tmp_path / "storage") + config["pipeline"]["inputs"][0]["storage.type"] = storage + for rewrite in filters: + rewrite["emitter_storage.type"] = storage + config_path = tmp_path / "eof.yaml" + config_path.write_text(yaml.safe_dump(config)) + service = FluentBitTestService(str(config_path)) + # EOF intentionally stops this finite input before the health check completes. + with pytest.raises(FluentBitStartupError, match="exited early with code 0"): + service.start() + delivered = 0 + for line in Path(service.flb.log_file).read_text().splitlines(): + try: + delivered += json.loads(line).get("log_records", 0) + except (ValueError, AttributeError): + pass + assert delivered == 1000 + if chain.startswith("cycle"): + assert "emitter cycle detected" in Path(service.flb.log_file).read_text() From 66d6eb98db19f1d1bd5df5fdff852c8380c2b93b Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 26 Oct 2026 13:34:33 -0600 Subject: [PATCH 7/8] filter_rewrite_tag: preserve records and emitter compatibility Signed-off-by: Eduardo Silva --- plugins/filter_rewrite_tag/rewrite_tag.c | 38 ++++++++++++++++++++---- 1 file changed, 33 insertions(+), 5 deletions(-) diff --git a/plugins/filter_rewrite_tag/rewrite_tag.c b/plugins/filter_rewrite_tag/rewrite_tag.c index 46d6bfbf134..7743795d4ce 100644 --- a/plugins/filter_rewrite_tag/rewrite_tag.c +++ b/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; } From a38d357da6f8a63bb586666f7a12ae21eafff9b1 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 26 Oct 2026 13:34:33 -0600 Subject: [PATCH 8/8] tests: integration: cover rewrite tag rules and processor placement Signed-off-by: Eduardo Silva --- .../tests/test_filter_rewrite_tag_groups.py | 17 +- .../tests/test_filter_rewrite_tag_paths.py | 154 ++++++++++++++++++ .../test_filter_rewrite_tag_regressions.py | 123 ++++++++++++++ 3 files changed, 293 insertions(+), 1 deletion(-) diff --git a/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_groups.py b/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_groups.py index 47fea50deff..03268863105 100644 --- a/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_groups.py +++ b/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)) diff --git a/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_paths.py b/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_paths.py index fe5e19c1ce2..e4d9667750f 100644 --- a/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_paths.py +++ b/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_paths.py @@ -56,6 +56,112 @@ def service_for(tmp_path, config, name="config.yaml"): return FluentBitTestService(str(path)) +@pytest.mark.parametrize("placement", ["filter", "processor"]) +def test_rule_precedence_and_accessors(tmp_path, placement): + messages = [ + {"id": 0, "meta": {"level": "error", "destination": "blue"}}, + {"id": 1, "meta": {"level": "info"}}, + {"id": 2}, + {"id": 3, "meta": {"level": 7}}, + ] + rewrite = {"name": "rewrite_tag", "rule": [ + "$meta['level'] ^(error)$ routed.$1.$meta['destination'].$TAG[1] false", + "$meta['level'] ^error$ wrong true", + ]} + service = service_for(tmp_path, config_for(tmp_path, messages, rewrite, placement)) + try: + service.start() + service.wait_for_condition(lambda: len(records(service)) >= 4, timeout=30) + finally: + service.stop() + assert Counter((r["id"], r["route"]) for r in records(service)) == Counter({ + (0, "routed.error.blue.app"): 1, (1, "source.app"): 1, + (2, "source.app"): 1, (3, "source.app"): 1, + }) + + +@pytest.mark.parametrize("placement", ["filter", "processor"]) +def test_invalid_first_expansion_does_not_try_later_rules(tmp_path, placement): + messages = [{"id": 0, "kind": "emit"}, + {"id": 1, "kind": "emit", "destination": "routed"}] + rewrite = {"name": "rewrite_tag", "rule": [ + "$kind ^emit$ $destination false", "$kind ^emit$ wrong false", + ]} + service = service_for(tmp_path, config_for(tmp_path, messages, rewrite, placement)) + try: + service.start() + service.wait_for_condition(lambda: len(records(service)) >= 2, timeout=30) + finally: + service.stop() + assert Counter((r["id"], r["route"]) for r in records(service)) == Counter({ + (0, "source.app"): 1, (1, "routed"): 1, + }) + + +@pytest.mark.parametrize("placement", ["filter", "processor"]) +def test_explicit_emitter_name_is_preserved(tmp_path, placement): + rewrite = {"name": "rewrite_tag", "emitter_name": "custom_emitter", + "rule": "$kind ^emit$ routed false"} + config = config_for(tmp_path, [{"id": 0, "kind": "emit"}], rewrite, placement) + service = service_for(tmp_path, config) + try: + service.start() + service.wait_for_condition(lambda: len(records(service)) == 1, timeout=30) + response = requests.get( + f"http://127.0.0.1:{service.flb.http_monitoring_port}/api/v1/metrics", timeout=5, + ) + response.raise_for_status() + assert "custom_emitter" in response.json()["input"] + finally: + service.stop() + + +@pytest.mark.parametrize("placement", ["filter", "processor"]) +@pytest.mark.parametrize("failure", ["duplicate-name", "invalid-regex", "invalid-storage", "missing-storage-path"]) +def test_invalid_configuration_is_rejected_cleanly(tmp_path, placement, failure): + rewrite = {"name": "rewrite_tag", "emitter_name": "shared_emitter", + "rule": "$kind ^emit$ routed false"} + if failure == "invalid-regex": + rewrite["rule"] = "$kind [ routed false" + elif failure == "invalid-storage": + rewrite["emitter_storage.type"] = "unsupported" + config = config_for(tmp_path, [{"id": 0, "kind": "emit"}], rewrite, placement) + if failure == "missing-storage-path": + rewrite["emitter_storage.type"] = "filesystem" + config["service"].pop("storage.path") + if failure == "duplicate-name": + if placement == "filter": + config["pipeline"]["filters"].insert(1, copy.deepcopy(rewrite)) + else: + config["pipeline"]["inputs"].append(copy.deepcopy(config["pipeline"]["inputs"][0])) + service = service_for(tmp_path, config) + with pytest.raises(FluentBitStartupError, match="exited early"): + service.start() + log = Path(service.flb.log_file).read_text() + if failure == "duplicate-name": + assert "emitter_name 'shared_emitter' already exists" in log + elif failure == "invalid-storage": + assert "invalid 'emitter_storage.type'" in log + elif failure == "invalid-regex": + assert "could not compile regex" in log + elif failure == "missing-storage-path": + assert "but no filesystem path was defined" in log + + +@pytest.mark.parametrize("placement", ["filter", "processor"]) +def test_no_rules_preserves_existing_passthrough_behavior(tmp_path, placement): + config = config_for(tmp_path, [{"id": 0}, {"id": 1}], {"name": "rewrite_tag"}, placement) + service = service_for(tmp_path, config) + try: + service.start() + service.wait_for_condition(lambda: len(records(service)) == 2, timeout=30) + finally: + service.stop() + assert Counter((r["id"], r["route"]) for r in records(service)) == Counter({ + (0, "source.app"): 1, (1, "source.app"): 1, + }) + + @pytest.fixture def retry_sink(): state = SimpleNamespace(allow=threading.Event(), attempted=threading.Event(), @@ -163,3 +269,51 @@ def test_filesystem_emitter_backlog_replays_with_live_records(tmp_path, retry_si finally: second.stop() assert Counter(retry_sink.received) == Counter(range(5)) + + +@pytest.mark.parametrize("threaded", [False, True]) +def test_multiple_rewrite_processor_stages_have_unique_emitters(tmp_path, threaded): + config = config_for(tmp_path, [{"id": 0, "kind": "emit"}], { + "name": "rewrite_tag", "rule": "$kind ^emit$ middle false", + }, "processor") + input_config = config["pipeline"]["inputs"][0] + input_config["threaded"] = threaded + input_config["processors"]["logs"].append({ + "name": "rewrite_tag", "rule": "$kind ^emit$ routed false", + }) + service = service_for(tmp_path, config) + try: + service.start() + service.wait_for_condition(lambda: len(records(service)) == 1, timeout=30) + response = requests.get( + f"http://127.0.0.1:{service.flb.http_monitoring_port}/api/v1/metrics", timeout=5, + ) + response.raise_for_status() + emitter_names = [name for name in response.json()["input"] if name.startswith("emitter_for_")] + assert len(emitter_names) == 2 + assert len(set(emitter_names)) == 2 + finally: + service.stop() + assert [(r["id"], r["route"]) for r in records(service)] == [(0, "routed")] + + +@pytest.mark.parametrize("placement", ["filter", "processor"]) +def test_existing_default_emitter_metric_name_is_preserved(tmp_path, placement): + config = config_for(tmp_path, [{"id": 0, "kind": "emit"}], { + "name": "rewrite_tag", "rule": "$kind ^emit$ routed false", + }, placement) + # No other global filters consume an instance ID before this processor. + if placement == "processor": + config["pipeline"].pop("filters") + service = service_for(tmp_path, config) + try: + service.start() + service.wait_for_condition(lambda: len(records(service)) == 1, timeout=30) + response = requests.get( + f"http://127.0.0.1:{service.flb.http_monitoring_port}/api/v1/metrics", timeout=5, + ) + response.raise_for_status() + emitter_names = [name for name in response.json()["input"] if name.startswith("emitter_for_")] + assert emitter_names == ["emitter_for_rewrite_tag.0"] + finally: + service.stop() diff --git a/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_regressions.py b/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_regressions.py index 04fed98b38f..a12d607d68d 100644 --- a/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_regressions.py +++ b/tests/integration/scenarios/filter_rewrite_tag/tests/test_filter_rewrite_tag_regressions.py @@ -9,6 +9,129 @@ from utils.test_service import FluentBitTestService +def read_records(service): + records = [] + for line in Path(service.flb.log_file).read_text().splitlines(): + try: + record = json.loads(line) + except ValueError: + continue + if isinstance(record, dict) and "id" in record: + records.append(record) + return records + + +@pytest.mark.parametrize("placement", ["filter", "processor"]) +@pytest.mark.parametrize("keep", [False, True]) +@pytest.mark.parametrize("storage", ["memory", "filesystem"]) +@pytest.mark.parametrize("threaded", [False, True]) +def test_routes_stages_and_invalid_tags(tmp_path, placement, keep, storage, threaded): + # Alternating tags exercise routing and per-tag order. Missing/empty tags + # must preserve the original even when other records emit successfully. + messages = [ + {"id": i, "kind": "emit", "destination": f"routed.{'a' if i % 2 == 0 else 'b'}"} + for i in range(6) + ] + messages.extend([{"id": 6, "kind": "emit", "destination": ""}, {"id": 7, "kind": "emit"}]) + script = tmp_path / "stages.lua" + script.write_text(''' +function before(tag, ts, record) + record.before = (record.before or 0) + 1 + return 1, ts, record +end +function after(tag, ts, record) + record.after = (record.after or 0) + 1 + record.after_tag = tag + return 1, ts, record +end +''') + config = { + "service": { + "flush": 0.2, "grace": 1, "log_level": "error", + "http_server": "on", "http_port": "${FLUENT_BIT_HTTP_MONITORING_PORT}", + }, + "pipeline": { + "inputs": [{ + "name": "dummy", "tag": "source", "samples": 1, + "dummy": " ".join(json.dumps(message) for message in messages), + "processors": {"logs": [ + {"name": "lua", "script": str(script), "call": "before"}, + {"name": "rewrite_tag", "rule": f"$kind ^emit$ $destination {str(keep).lower()}"}, + {"name": "lua", "script": str(script), "call": "after"}, + ]}, + }], + "filters": [ + {"name": "record_modifier", "match": tag, "record": f"route {tag}"} + for tag in ["source", "routed.a", "routed.b"] + ], + # Both outputs must receive all records independently. + "outputs": [ + {"name": "stdout", "match": "*", "format": "json_lines", "workers": 0}, + {"name": "stdout", "match": "*", "format": "json_lines", "workers": 0}, + ], + }, + } + config["service"]["storage.path"] = str(tmp_path / "storage") + input_config = config["pipeline"]["inputs"][0] + input_config["storage.type"] = storage + input_config["threaded"] = threaded + input_config["processors"]["logs"][1]["emitter_storage.type"] = storage + if placement == "filter": + stages = config["pipeline"]["inputs"][0]["processors"]["logs"] + rewrite = stages.pop(1) + rewrite["match"] = "source" + config["pipeline"]["filters"].insert(0, rewrite) + config_path = tmp_path / "routing.yaml" + config_path.write_text(yaml.safe_dump(config)) + service = FluentBitTestService(str(config_path)) + expected = Counter() + for message in messages: + if message.get("destination"): + expected[(message["id"], message["destination"])] += 2 + if keep: + expected[(message["id"], "source")] += 2 + else: + expected[(message["id"], "source")] += 2 + try: + service.start() + service.wait_for_condition( + lambda: len(read_records(service)) >= sum(expected.values()), + timeout=30, description="rewritten and retained records on both outputs", + ) + finally: + service.stop() + records = read_records(service) + assert Counter((record["id"], record["route"]) for record in records) == expected + for record in records: + assert record["before"] == 1 + assert record["after"] == 1 + assert record["after_tag"] == ("source" if placement == "filter" else record["route"]) + for route in ["routed.a", "routed.b"]: + # Two outputs each retain the source order within this destination. + ids = [record["id"] for record in records if record["route"] == route] + expected_ids = [message["id"] for message in messages if message.get("destination") == route] + assert ids == expected_ids + expected_ids + + +def test_output_processor_is_rejected(tmp_path): + config = { + "service": {"log_level": "error", "http_server": "on", + "http_port": "${FLUENT_BIT_HTTP_MONITORING_PORT}"}, + "pipeline": { + "inputs": [{"name": "dummy", "tag": "source", "samples": 1}], + "outputs": [{"name": "null", "match": "*", "processors": {"logs": [ + {"name": "rewrite_tag", "rule": "$log .* routed false"}, + ]}}], + }, + } + config_path = tmp_path / "output.yaml" + config_path.write_text(yaml.safe_dump(config)) + service = FluentBitTestService(str(config_path)) + with pytest.raises(FluentBitStartupError, match="exited early"): + service.start() + assert "rewrite_tag only supports input processors" in Path(service.flb.log_file).read_text() + + @pytest.mark.parametrize("chain", ["single", "forward", "reverse", "cycle-forward", "cycle-reverse"]) @pytest.mark.parametrize("storage", ["memory", "filesystem"]) def test_pending_emitter_records_drain_at_eof(tmp_path, chain, storage):