diff --git a/plugins/in_syslog/syslog_prot.c b/plugins/in_syslog/syslog_prot.c index 12316d70509..9c5aee2798e 100644 --- a/plugins/in_syslog/syslog_prot.c +++ b/plugins/in_syslog/syslog_prot.c @@ -281,6 +281,9 @@ int syslog_prot_process(struct syslog_conn *conn) continue; } + /* Parsers without a time key leave the output timestamp untouched. */ + flb_time_zero(&out_time); + /* Process the string */ ret = flb_parser_do(ctx->parser, p, len, &out_buf, &out_size, &out_time); diff --git a/tests/integration/scenarios/syslog_parser_time/tests/test_syslog_parser_time.py b/tests/integration/scenarios/syslog_parser_time/tests/test_syslog_parser_time.py new file mode 100644 index 00000000000..125c8055679 --- /dev/null +++ b/tests/integration/scenarios/syslog_parser_time/tests/test_syslog_parser_time.py @@ -0,0 +1,115 @@ +"""Bounded nesting and recovery checks for network ingestion.""" +import contextlib +import http.client +import os +from pathlib import Path +import signal +import socket +import struct +import subprocess +import time + +import pytest + + + +def wait_for(predicate, timeout=30): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(0.05) + assert predicate(), "Timed out waiting for Fluent Bit" + + +@contextlib.contextmanager +def daemon(tmp_path, mode): + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + port = listener.getsockname()[1] + plugin = mode.split("-")[0] + address = str(tmp_path / "input.sock") if plugin == "unix_socket" else ("127.0.0.1", port) + parser = tmp_path / "parsers.conf" + parser.write_text("[PARSER]\n Name json\n Format json\n") + command = [os.environ["FLUENT_BIT_BINARY"], "-f", "0.1", "-R", str(parser), "-i", plugin] + if plugin == "unix_socket": + command += ["-p", f"socket_path={address}"] + else: + command += ["-p", "listen=127.0.0.1", "-p", f"port={port}"] + if mode.endswith("-parser"): + command += ["-p", "format=none", "-p", "parser=json"] + if plugin == "syslog": + command += ["-p", "mode=tcp", "-p", "parser=json"] + command += ["-o", "stdout", "-m", "*", "-p", "format=json_lines"] + log = tmp_path / "fluent-bit.log" + memlog = tmp_path / "valgrind.log" + memory = os.environ.get("VALGRIND") == "1" + if memory: + command = ["valgrind", "--leak-check=full", "--show-leak-kinds=all", + "--errors-for-leak-kinds=definite,indirect", "--error-exitcode=99", + f"--log-file={memlog}"] + command + with log.open("w") as output: + process = subprocess.Popen(command, stdout=output, stderr=subprocess.STDOUT) + def ready(): + assert process.poll() is None, log.read_text() + return "[output:stdout:" in log.read_text() + try: + wait_for(ready) + yield address, process, log + finally: + if process.poll() is None: + process.send_signal(signal.SIGTERM) + try: + process.wait(timeout=30) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + pytest.fail("Fluent Bit did not shut down cleanly") + assert process.returncode == 0, log.read_text() + (memlog.read_text() if memory else "") + if memory: + assert "ERROR SUMMARY: 0 errors" in memlog.read_text(), memlog.read_text() + + +def send(mode, address, payload): + plugin = mode.split("-")[0] + if plugin == "http": + conn = http.client.HTTPConnection(*address, timeout=10) + try: + conn.request("POST", "/test", payload, {"Content-Type": "application/json"}) + response = conn.getresponse() + response.read() + finally: + conn.close() + return + if plugin == "udp": + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.sendto(payload + b"\n", address) + return + family = socket.AF_UNIX if plugin == "unix_socket" else socket.AF_INET + with socket.socket(family, socket.SOCK_STREAM) as sock: + sock.settimeout(5) + sock.connect(address) + if plugin == "mqtt": + # A regular MQTT CONNECT followed by a QoS 0 JSON publication. + sock.sendall(b"\x10\x10\x00\x04MQTT\x04\x02\x00\x0a\x00\x04test") + assert sock.recv(4)[0] == 0x20 + body = b"\x00\x01a" + payload + length = len(body) + encoded = bytearray() + while True: + digit = length % 128 + length //= 128 + encoded.append(digit | (0x80 if length else 0)) + if not length: + break + sock.sendall(b"\x30" + bytes(encoded) + body) + else: + sock.sendall(payload + b"\n") + + +def test_json_parser_without_timestamp(tmp_path): + with daemon(tmp_path, "syslog") as (address, process, log): + for marker in ("before", "after"): + send("syslog", address, ('{"marker":"' + marker + '"}').encode()) + wait_for(lambda: ('"marker":"' + marker + '"') in log.read_text()) + assert process.poll() is None