diff --git a/AGENTS.md b/AGENTS.md index 8797c9aa0e8..d75dc6c99b9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -101,30 +101,37 @@ Keep changes scoped: plugin logic in its plugin directory, shared behavior in `s - For touched components covered by `tests/integration`, agents must run the focused scenario(s) twice: - once normally to verify behavior; - - once with valgrind enabled to verify memory-safety behavior. + - once with the platform memory checker enabled to verify memory-safety + behavior: Valgrind on Linux or Leaks on macOS. - The default expectation for component verification is: `./tests/integration/setup-venv.sh` `cmake -S . -B build -DFLB_TESTS_RUNTIME=On -DFLB_TESTS_INTERNAL=On` `cmake --build build -j8` `tests/integration/.venv/bin/python -m pytest -q` + On Linux, run the memory-safety pass with: `VALGRIND=1 VALGRIND_STRICT=1 tests/integration/.venv/bin/python -m pytest -q` + On macOS, run the memory-safety pass with: + `LEAKS=1 LEAKS_STRICT=1 tests/integration/.venv/bin/python -m pytest -q` - On Windows, use the same `-DFLB_TESTS_RUNTIME=On` configuration and run - relevant focused runtime cases. Valgrind is normally unavailable on Windows; - report that exact blocker instead of conflating it with runtime-test support. + relevant focused runtime and functional integration cases. Valgrind and + macOS Leaks are normally unavailable on Windows; report that exact + memory-checker blocker instead of conflating it with test support. - Run broader test coverage when changing shared lifecycle, routing, storage, or accounting code. - Validate both success and failure paths (invalid payloads, boundary sizes, null/missing fields). - You can also run specific binaries from `build/bin` (e.g., `./bin/flb-it-opentelemetry`). - When changing code covered by `tests/integration`, agents must verify the - affected scenarios are valgrind-clean. Run the focused integration tests with - `tests/integration/run_tests.py --valgrind --valgrind-strict ...` and do not - stop at functional pass/fail if memory errors or leaks remain. -- If a focused integration or valgrind run cannot be executed, agents must not - silently skip it. They must report the exact blocker in the final response - (for example: missing binary, missing Python environment, unsupported - scenario, missing dependency, or infrastructure failure). + affected scenarios are clean under the platform memory checker. On Linux, + run `tests/integration/run_tests.py --valgrind --valgrind-strict ...`. On + macOS, run `tests/integration/run_tests.py --leaks --leaks-strict ...`. Do + not stop at functional pass/fail if memory errors or leaks remain. +- If a focused integration or platform memory-checker run cannot be executed, + agents must not silently skip it. They must report the exact blocker in the + final response (for example: missing binary, missing Python environment, + unsupported scenario, missing dependency, or infrastructure failure). - Final task close-outs must include proof of verification: - the exact focused integration command(s) run; - - whether valgrind was used; + - which platform memory checker was used (Valgrind on Linux or Leaks on + macOS), or `not run` with the exact blocker; - pass/fail status; - any concrete blocker if a required run could not be completed. - Keep generated integration artifacts out of git. Do not commit diff --git a/plugins/in_storage_backlog/sb.c b/plugins/in_storage_backlog/sb.c index 8e6aa70adbf..5cdf1ade202 100644 --- a/plugins/in_storage_backlog/sb.c +++ b/plugins/in_storage_backlog/sb.c @@ -312,8 +312,11 @@ static int sb_append_chunk_to_segregated_backlogs(struct cio_chunk *target_chun return -2; } - flb_routes_mask_set_by_tag(dummy_input_chunk.routes_mask, tag_buf, tag_len, - context->ins); + result = flb_routes_mask_set_by_tag(dummy_input_chunk.routes_mask, tag_buf, tag_len, + context->ins); + if (result == 0) { + return -4; + } mk_list_foreach_safe(head, tmp, &context->backlogs) { backlog = mk_list_entry(head, struct sb_out_queue, _head); @@ -412,6 +415,20 @@ int sb_segregate_chunks(struct flb_config *config) /* try to segregate a chunk */ ret = sb_append_chunk_to_segregated_backlogs(chunk, stream, context); if (ret) { + /* + * Leave chunks without a route on disk so a future configuration + * can reconsider them. Closing the ChunkIO handle removes them + * from this context's active chunk accounting without deleting + * the underlying files. + */ + if (ret == -4) { + flb_plg_info(context->ins, + "no matching route for %s/%s, keeping it on disk", + stream->name, chunk->name); + cio_chunk_close(chunk, CIO_FALSE); + continue; + } + /* * if the chunk could not be segregated, just remove it from the * queue, delete it and continue. diff --git a/tests/integration/scenarios/in_storage_backlog/config/reload_with_stale_route.yaml b/tests/integration/scenarios/in_storage_backlog/config/reload_with_stale_route.yaml new file mode 100644 index 00000000000..b547b02bb8d --- /dev/null +++ b/tests/integration/scenarios/in_storage_backlog/config/reload_with_stale_route.yaml @@ -0,0 +1,19 @@ +service: + flush: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + hot_reload: on + hot_reload.timeout: 10 + storage.path: ${STORAGE_BACKLOG_PATH} + +pipeline: + inputs: + - name: tail + tag: live + path: ${STORAGE_BACKLOG_MISSING_PATH} + storage.type: filesystem + + outputs: + - name: null + match: "*" diff --git a/tests/integration/scenarios/in_storage_backlog/config/reload_without_stale_route.yaml b/tests/integration/scenarios/in_storage_backlog/config/reload_without_stale_route.yaml new file mode 100644 index 00000000000..84c5e7112a9 --- /dev/null +++ b/tests/integration/scenarios/in_storage_backlog/config/reload_without_stale_route.yaml @@ -0,0 +1,19 @@ +service: + flush: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + hot_reload: on + hot_reload.timeout: 10 + storage.path: ${STORAGE_BACKLOG_PATH} + +pipeline: + inputs: + - name: tail + tag: live + path: ${STORAGE_BACKLOG_MISSING_PATH} + storage.type: filesystem + + outputs: + - name: null + match: live diff --git a/tests/integration/scenarios/in_storage_backlog/config/seed_unroutable_chunk.yaml b/tests/integration/scenarios/in_storage_backlog/config/seed_unroutable_chunk.yaml new file mode 100644 index 00000000000..43f6dfaf1c0 --- /dev/null +++ b/tests/integration/scenarios/in_storage_backlog/config/seed_unroutable_chunk.yaml @@ -0,0 +1,18 @@ +service: + flush: 1 + grace: 0 + storage.path: ${STORAGE_BACKLOG_PATH} + +pipeline: + inputs: + - name: dummy + tag: stale + samples: 1 + storage.type: filesystem + + outputs: + - name: http + match: stale + host: 127.0.0.1 + port: 9 + retry_limit: no_limits diff --git a/tests/integration/scenarios/in_storage_backlog/tests/test_in_storage_backlog_001.py b/tests/integration/scenarios/in_storage_backlog/tests/test_in_storage_backlog_001.py new file mode 100644 index 00000000000..608585d79ab --- /dev/null +++ b/tests/integration/scenarios/in_storage_backlog/tests/test_in_storage_backlog_001.py @@ -0,0 +1,174 @@ +import os +from pathlib import Path +import signal +import shutil +import subprocess +import time + +import pytest + +from utils.fluent_bit_manager import FluentBitManager + + +CONFIG_DIR = Path(__file__).resolve().parent.parent / "config" +SEED_CONFIG = CONFIG_DIR / "seed_unroutable_chunk.yaml" +RELOAD_WITHOUT_ROUTE = CONFIG_DIR / "reload_without_stale_route.yaml" +RELOAD_WITH_ROUTE = CONFIG_DIR / "reload_with_stale_route.yaml" +KEEP_ON_DISK_MESSAGE = "no matching route for dummy.0/" +REGISTER_MESSAGE = "register dummy.0/" +QUEUE_MESSAGE = "queueing dummy.0:" + + +def wait_for_persisted_chunk(storage_path, process, log_path, timeout=10): + deadline = time.time() + timeout + + while time.time() < deadline: + chunks = list(storage_path.glob("*/*.flb")) + if chunks and chunks[0].stat().st_size > 0: + return chunks[0] + + return_code = process.poll() + if return_code is not None: + if log_path.exists(): + log = log_path.read_text(encoding="utf-8", errors="replace") + else: + log = "seed log was not created" + raise RuntimeError( + f"seed Fluent Bit exited with code {return_code} before persisting a chunk:\n{log}" + ) + + time.sleep(0.1) + + raise TimeoutError("timed out waiting for the seed filesystem chunk") + + +def wait_for_log_occurrences(log_path, text, expected, timeout=10): + deadline = time.time() + timeout + + while time.time() < deadline: + if log_path.exists(): + log = log_path.read_text(encoding="utf-8", errors="replace") + if log.count(text) >= expected: + return + + time.sleep(0.1) + + raise TimeoutError(f"timed out waiting for {expected} occurrences of {text!r}") + + +def wait_for_path_removal(path, timeout=10): + deadline = time.time() + timeout + + while time.time() < deadline: + if not path.exists(): + return + + time.sleep(0.1) + + raise TimeoutError(f"timed out waiting for {path} to be removed") + + +def seed_persisted_chunk(storage_path, seed_log_path): + binary_path = FluentBitManager(str(SEED_CONFIG)).binary_absolute_path + seed_process = subprocess.Popen( + [binary_path, "-c", str(SEED_CONFIG), "-l", str(seed_log_path)], + stdout=subprocess.DEVNULL, + stderr=subprocess.STDOUT, + env=os.environ.copy(), + ) + + try: + return wait_for_persisted_chunk(storage_path, seed_process, seed_log_path) + finally: + if seed_process.poll() is None: + seed_process.kill() + seed_process.wait(timeout=10) + + +@pytest.fixture +def seeded_chunk(tmp_path, monkeypatch): + storage_path = tmp_path / "storage" + storage_path.mkdir() + seed_log_path = tmp_path / "seed.log" + missing_path = tmp_path / "does-not-exist.log" + + monkeypatch.setenv("STORAGE_BACKLOG_PATH", str(storage_path)) + monkeypatch.setenv("STORAGE_BACKLOG_MISSING_PATH", str(missing_path)) + + persisted_chunk = seed_persisted_chunk(storage_path, seed_log_path) + assert persisted_chunk.exists() + + return persisted_chunk + + +def test_routable_persisted_chunk_is_replayed_on_startup(seeded_chunk): + manager = FluentBitManager(str(RELOAD_WITH_ROUTE)) + + try: + manager.start() + wait_for_log_occurrences( + Path(manager.log_file), REGISTER_MESSAGE, expected=1 + ) + wait_for_log_occurrences( + Path(manager.log_file), QUEUE_MESSAGE, expected=1 + ) + wait_for_path_removal(seeded_chunk) + finally: + manager.stop() + + assert not seeded_chunk.exists() + + +@pytest.mark.skipif(not hasattr(signal, "SIGHUP"), reason="SIGHUP is unavailable") +def test_unroutable_chunk_stays_on_disk_across_hot_reload(seeded_chunk): + manager = FluentBitManager(str(RELOAD_WITHOUT_ROUTE)) + + try: + manager.start() + wait_for_log_occurrences( + Path(manager.log_file), KEEP_ON_DISK_MESSAGE, expected=1 + ) + assert seeded_chunk.exists() + + manager.send_sighup() + manager.wait_for_hot_reload_count(1, timeout=20) + wait_for_log_occurrences( + Path(manager.log_file), KEEP_ON_DISK_MESSAGE, expected=2, timeout=20 + ) + assert seeded_chunk.exists() + finally: + manager.stop() + + assert seeded_chunk.exists() + + +@pytest.mark.skipif(not hasattr(signal, "SIGHUP"), reason="SIGHUP is unavailable") +def test_restored_route_recovers_chunk_during_hot_reload(tmp_path, seeded_chunk): + runtime_config = tmp_path / "fluent-bit.yaml" + shutil.copyfile(RELOAD_WITHOUT_ROUTE, runtime_config) + manager = FluentBitManager(str(runtime_config)) + + try: + manager.start() + wait_for_log_occurrences( + Path(manager.log_file), KEEP_ON_DISK_MESSAGE, expected=1 + ) + assert seeded_chunk.exists() + + pending_config = tmp_path / "fluent-bit.yaml.tmp" + shutil.copyfile(RELOAD_WITH_ROUTE, pending_config) + os.replace(pending_config, runtime_config) + + manager.send_sighup() + manager.wait_for_hot_reload_count(1, timeout=20) + wait_for_log_occurrences( + Path(manager.log_file), REGISTER_MESSAGE, expected=1, timeout=20 + ) + wait_for_log_occurrences( + Path(manager.log_file), QUEUE_MESSAGE, expected=1, timeout=20 + ) + wait_for_path_removal(seeded_chunk, timeout=20) + finally: + manager.stop() + + assert not seeded_chunk.exists() diff --git a/tests/runtime/CMakeLists.txt b/tests/runtime/CMakeLists.txt index 0a3c10de616..34652aaa03a 100644 --- a/tests/runtime/CMakeLists.txt +++ b/tests/runtime/CMakeLists.txt @@ -113,6 +113,10 @@ if (FLB_CUSTOM_CALYPTIA) endforeach() endif() +if(FLB_IN_LIB AND FLB_OUT_HTTP AND FLB_OUT_NULL AND FLB_IN_STORAGE_BACKLOG) + FLB_RT_TEST(FLB_IN_STORAGE_BACKLOG "in_storage_backlog.c") +endif() + if(FLB_IN_EBPF) # Define common variables set(EBPF_TEST_INCLUDE_DIRS diff --git a/tests/runtime/in_storage_backlog.c b/tests/runtime/in_storage_backlog.c new file mode 100644 index 00000000000..16eafb512d3 --- /dev/null +++ b/tests/runtime/in_storage_backlog.c @@ -0,0 +1,212 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2026 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include +#include +#include +#include + +#include + +#include +#include + +#include "flb_tests_runtime.h" +#include "../include/flb_tests_tmpdir.h" + +#define TEST_RECORD "[0, {\"message\":\"persisted\"}]" + +static flb_ctx_t *start_context(const char *storage_path, + const char *input_tag, + const char *output_match, + const char *output_name, + int *input_id) +{ + int output_id; + int ret; + flb_ctx_t *ctx; + + ctx = flb_create(); + if (ctx == NULL) { + return NULL; + } + + if (flb_service_set(ctx, + "flush", "60", + "grace", "0", + "storage.path", storage_path, + NULL) != 0) { + flb_destroy(ctx); + return NULL; + } + + *input_id = flb_input(ctx, (char *) "lib", NULL); + if (*input_id < 0) { + flb_destroy(ctx); + return NULL; + } + + if (flb_input_set(ctx, *input_id, + "tag", input_tag, + "storage.type", "filesystem", + NULL) != 0) { + flb_destroy(ctx); + return NULL; + } + + output_id = flb_output(ctx, output_name, NULL); + if (output_id < 0) { + flb_destroy(ctx); + return NULL; + } + + if (strcmp(output_name, "http") == 0) { + ret = flb_output_set(ctx, output_id, + "match", output_match, + "host", "127.0.0.1", + "port", "9", + "retry_limit", "no_limits", + NULL); + } + else { + ret = flb_output_set(ctx, output_id, "match", output_match, NULL); + } + + if (ret != 0) { + flb_destroy(ctx); + return NULL; + } + + if (flb_start(ctx) != 0) { + flb_destroy(ctx); + return NULL; + } + + return ctx; +} + +static int wait_for_chunk_content(struct flb_input_instance *input_instance) +{ + int attempts; + struct flb_input_chunk *input_chunk; + + for (attempts = 0; attempts < 50; attempts++) { + if (mk_list_is_empty(&input_instance->chunks) != 0) { + input_chunk = mk_list_entry_first(&input_instance->chunks, + struct flb_input_chunk, + _head); + if (flb_input_chunk_get_size(input_chunk) > 0) { + return 0; + } + } + + flb_time_msleep(100); + } + + return -1; +} + +static int wait_for_filesystem_chunk_count(flb_ctx_t *ctx, int expected) +{ + int attempts; + int fs_chunks; + int mem_chunks; + + for (attempts = 0; attempts < 50; attempts++) { + flb_storage_chunk_count(ctx->config, &mem_chunks, &fs_chunks); + if (fs_chunks == expected) { + return 0; + } + + flb_time_msleep(100); + } + + return -1; +} + +static void test_unroutable_chunk_is_detached_and_preserved(void) +{ + int input_id; + int ret; + struct stat file_status; + struct flb_input_chunk *input_chunk; + struct flb_input_instance *input_instance; + struct flb_storage_input *storage_input; + flb_ctx_t *ctx; + char chunk_path[4096]; + char *storage_path; + + storage_path = flb_test_tmpdir_cat("/flb-storage-backlog-no-route-XXXXXX"); + TEST_ASSERT(storage_path != NULL); + TEST_ASSERT(mkdtemp(storage_path) != NULL); + + /* Seed one routed filesystem chunk whose output cannot deliver it. */ + ctx = start_context(storage_path, "stale", "stale", "http", &input_id); + TEST_ASSERT(ctx != NULL); + + ret = flb_lib_push(ctx, input_id, TEST_RECORD, sizeof(TEST_RECORD) - 1); + TEST_ASSERT(ret >= 0); + TEST_ASSERT(wait_for_filesystem_chunk_count(ctx, 1) == 0); + + input_instance = flb_input_get_instance(ctx->config, input_id); + TEST_ASSERT(input_instance != NULL); + TEST_ASSERT(wait_for_chunk_content(input_instance) == 0); + TEST_ASSERT(mk_list_is_empty(&input_instance->chunks) != 0); + + input_chunk = mk_list_entry_first(&input_instance->chunks, + struct flb_input_chunk, + _head); + storage_input = (struct flb_storage_input *) input_instance->storage; + TEST_ASSERT(storage_input != NULL); + + ret = snprintf(chunk_path, sizeof(chunk_path), "%s/%s/%s", + storage_path, storage_input->stream->name, + ((struct cio_chunk *) input_chunk->chunk)->name); + TEST_ASSERT(ret > 0 && (size_t) ret < sizeof(chunk_path)); + + flb_stop(ctx); + flb_destroy(ctx); + + TEST_ASSERT(stat(chunk_path, &file_status) == 0); + + /* Restore with a configuration whose only route does not match the chunk. */ + ctx = start_context(storage_path, "live", "live", "null", &input_id); + TEST_ASSERT(ctx != NULL); + + TEST_CHECK_(wait_for_filesystem_chunk_count(ctx, 0) == 0, + "unroutable restored chunk remained in active filesystem accounting"); + TEST_CHECK_(stat(chunk_path, &file_status) == 0, + "unroutable restored chunk was deleted"); + + flb_stop(ctx); + flb_destroy(ctx); + + TEST_CHECK_(stat(chunk_path, &file_status) == 0, + "unroutable restored chunk was not preserved across shutdown"); + + cio_utils_recursive_delete(storage_path); + flb_free(storage_path); +} + +TEST_LIST = { + {"unroutable_chunk_is_detached_and_preserved", + test_unroutable_chunk_is_detached_and_preserved}, + {NULL, NULL} +};