in_syslog: initialize parser timestamps for stream records

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);

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