diff --git a/vector-vrl/src/vector_vrl/vector_test_utils.py b/vector-vrl/src/vector_vrl/vector_test_utils.py index cb8e7e64..373b8d61 100644 --- a/vector-vrl/src/vector_vrl/vector_test_utils.py +++ b/vector-vrl/src/vector_vrl/vector_test_utils.py @@ -99,9 +99,10 @@ def run_remap_chain( ) -> tuple[bool, list[dict], str]: """Run `input_logs` through `transforms`, in order, in a real Vector. - Vector reads the logs from stdin and exits once every event is written, - so the run ends when the work does. `timeout` bounds it: a run still - going then is killed and reported as a failure. + Vector reads the logs from stdin and writes the results to a file, then + exits once the file is flushed, so the run ends when the work does. + `timeout` bounds it: a run still going then is killed and reported as a + failure. Each transform keeps its own options (`drop_on_error`, `drop_on_abort` and so on), and only its `inputs` is rewired to the one before it. @@ -117,9 +118,10 @@ def run_remap_chain( try: data_dir = test_dir / "vector_data" data_dir.mkdir() + output_file = test_dir / "output.ndjson" config_file = test_dir / "config.json" config_file.write_text( - json.dumps(self._vector_config(transforms, data_dir)), + json.dumps(self._vector_config(transforms, data_dir, output_file)), encoding="utf-8", newline="\n", ) @@ -146,7 +148,13 @@ def run_remap_chain( results: list[dict] = [] undecodable: list[int] = [] - for number, line in enumerate(completed.stdout.splitlines(), start=1): + # The file sink creates the file on its first event, so none means no output. + output = ( + output_file.read_text(encoding="utf-8", errors="replace") + if output_file.exists() + else "" + ) + for number, line in enumerate(output.splitlines(), start=1): if not line.strip(): continue try: @@ -172,9 +180,16 @@ def run_remap_chain( @staticmethod def _vector_config( - transforms: list[tuple[str, dict[str, Any]]], data_dir: Path + transforms: list[tuple[str, dict[str, Any]]], + data_dir: Path, + output_file: Path, ) -> dict[str, Any]: - """A stdin -> transforms -> stdout Vector config, transforms chained in order.""" + """A stdin -> transforms -> file Vector config, transforms chained in order. + + The sink is a file because the console sink hands its last write to Tokio's + blocking pool without a flush, and Vector's exit drops that write when it + has not started: one event of a short run goes missing now and then. + """ chained: dict[str, Any] = {} upstream = "runner_stdin" for name, transform in transforms: @@ -185,8 +200,9 @@ def _vector_config( "sources": {"runner_stdin": {"type": "stdin"}}, "transforms": chained, "sinks": { - "runner_stdout": { - "type": "console", + "runner_output": { + "type": "file", + "path": str(output_file), "inputs": [upstream], "encoding": {"codec": "json"}, } diff --git a/vector-vrl/tests/integration/test_vector_test_runner.py b/vector-vrl/tests/integration/test_vector_test_runner.py index 694f5c12..1a1e9868 100644 --- a/vector-vrl/tests/integration/test_vector_test_runner.py +++ b/vector-vrl/tests/integration/test_vector_test_runner.py @@ -27,6 +27,23 @@ def test_runner_processes_logs_through_the_real_binary(vector_binary): assert all(result["flag"] is True for result in results) +def test_runner_reads_results_from_a_file_sink_not_the_console(tmp_path): + """The console sink's last write is lost when Vector exits first. + + About 1 run in 1,500 of a two-event input came back with one event, exit 0 and + empty stderr, so no repeat-run test catches it reliably. The file sink flushes + and syncs before it finishes. + """ + output_file = tmp_path / "out.ndjson" + config = VectorTestRunner._vector_config( # noqa: SLF001 + [("t", {"type": "remap", "source": "."})], tmp_path, output_file + ) + + (sink,) = config["sinks"].values() + assert sink["type"] == "file" + assert sink["path"] == str(output_file) + + def test_runner_reports_a_config_that_vector_rejects(vector_binary): runner = VectorTestRunner(vector_binary)