bvandeusenandClaude Opus 5.5
51892a02c0
- plugins/syslog: RFC 5424 and RFC 3164 parsing, RFC 6587 TCP framing. - Lines are queued as they arrive and stored every 5 s through a Holding, so a storage stall delays them rather than losing them. - A new sender is an offered source, counted and not kept, until an admin adds it on the Syslog page (name, host); it can be ignored instead. - The listener follows the plugin's settings (port, UDP, TCP), rebinding within seconds of a save; a port it can't open is said on the page. - Hourly ring per source: age window and byte cap from Settings. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
97 lines
3.5 KiB
Python
97 lines
3.5 KiB
Python
"""The syslog listener (#5999): lines queued as they arrive, bounded, read in
|
|
rounds; the listener bound as the settings say, and a port it can't open
|
|
said, not raised."""
|
|
import asyncio
|
|
import socket
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from plugins.syslog import receiver
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fresh(monkeypatch):
|
|
monkeypatch.setattr(receiver, "LISTENER", receiver.Listener())
|
|
monkeypatch.setattr(receiver, "QUEUE", receiver.deque())
|
|
|
|
|
|
def test_the_queue_is_bounded_and_says_what_it_lost(monkeypatch):
|
|
monkeypatch.setattr(receiver, "MAX_QUEUED", 3)
|
|
for i in range(5):
|
|
receiver.take("10.0.0.5", f"<13>line {i}".encode())
|
|
assert receiver.LISTENER.received == 5 and receiver.LISTENER.dropped == 2
|
|
assert [r.message.text for r in receiver.read_round()] == ["line 2", "line 3", "line 4"]
|
|
|
|
|
|
def test_a_round_takes_at_most_its_share_and_leaves_the_rest():
|
|
for i in range(5):
|
|
receiver.take("10.0.0.5", f"<13>line {i}".encode())
|
|
first = receiver.read_round(limit=2)
|
|
assert [r.message.text for r in first] == ["line 0", "line 1"] and len(receiver.QUEUE) == 3
|
|
assert first[0].address == "10.0.0.5"
|
|
|
|
|
|
def _app(**cfg):
|
|
return SimpleNamespace(config={"PLUGINS": {"syslog": cfg}})
|
|
|
|
|
|
def _free_port() -> int:
|
|
with socket.socket() as s:
|
|
s.bind(("0.0.0.0", 0))
|
|
return s.getsockname()[1]
|
|
|
|
|
|
def test_it_listens_on_udp_and_tcp_and_takes_what_arrives():
|
|
async def go():
|
|
port = _free_port()
|
|
await receiver.ensure_listening(_app(port=port, udp=True, tcp=True))
|
|
assert receiver.LISTENER.listening and receiver.LISTENER.error is None
|
|
udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
udp.sendto(b"<13>over udp", ("127.0.0.1", port))
|
|
udp.close()
|
|
reader, writer = await asyncio.wait_for(asyncio.open_connection("127.0.0.1", port), 5)
|
|
writer.write(b"12 <13>over tcp<14>and a line\n")
|
|
await writer.drain()
|
|
writer.close()
|
|
for _ in range(50):
|
|
if len(receiver.QUEUE) >= 3:
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
texts = sorted(r.message.text for r in receiver.read_round())
|
|
await receiver.close()
|
|
return texts
|
|
|
|
assert asyncio.run(go()) == ["and a line", "over tcp", "over udp"]
|
|
|
|
|
|
def test_a_port_it_cannot_open_is_said_and_tried_again_later():
|
|
async def go():
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as held:
|
|
held.bind(("0.0.0.0", 0))
|
|
port = held.getsockname()[1]
|
|
now = datetime(2026, 10, 19, 3, 0, tzinfo=timezone.utc)
|
|
await receiver.ensure_listening(_app(port=port, udp=True, tcp=False), now)
|
|
said = receiver.LISTENER.error
|
|
tried = receiver.LISTENER.tried_at
|
|
await receiver.ensure_listening(_app(port=port, udp=True, tcp=False), now + timedelta(seconds=10))
|
|
assert receiver.LISTENER.tried_at == tried # not retried within the minute
|
|
return said
|
|
|
|
said = asyncio.run(go())
|
|
assert said and said.startswith("Port ") and "UDP" in said
|
|
|
|
|
|
def test_a_change_of_port_moves_the_listener():
|
|
async def go():
|
|
a, b = _free_port(), _free_port()
|
|
await receiver.ensure_listening(_app(port=a, udp=True, tcp=False))
|
|
await receiver.ensure_listening(_app(port=b, udp=True, tcp=False))
|
|
port = receiver.LISTENER._udp_transport.get_extra_info("sockname")[1]
|
|
await receiver.close()
|
|
return port, b
|
|
|
|
port, b = asyncio.run(go())
|
|
assert port == b
|