add rotation
This commit is contained in:
@@ -47,6 +47,13 @@ For UDP `514`, the script starts only the listener command with `sudo`:
|
|||||||
FGAI_SYSLOG_PORT=514 ./start.sh
|
FGAI_SYSLOG_PORT=514 ./start.sh
|
||||||
```
|
```
|
||||||
|
|
||||||
|
The syslog receiver rotates the active JSONL input at 25 MB by default. Rotated
|
||||||
|
files are gzip-compressed and 14 archives are retained. Override this when needed:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
FGAI_LOG_ROTATE_BYTES=$((100 * 1024 * 1024)) FGAI_LOG_ROTATE_COUNT=30 ./start.sh restart
|
||||||
|
```
|
||||||
|
|
||||||
Analyze local logs:
|
Analyze local logs:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
|||||||
@@ -181,7 +181,14 @@ def fetch_policies(args: argparse.Namespace) -> int:
|
|||||||
|
|
||||||
|
|
||||||
def listen_syslog(args: argparse.Namespace) -> int:
|
def listen_syslog(args: argparse.Namespace) -> int:
|
||||||
listen_udp_syslog(args.host, args.port, args.output, quiet=args.quiet)
|
listen_udp_syslog(
|
||||||
|
args.host,
|
||||||
|
args.port,
|
||||||
|
args.output,
|
||||||
|
rotate_bytes=args.rotate_bytes,
|
||||||
|
rotate_count=args.rotate_count,
|
||||||
|
quiet=args.quiet,
|
||||||
|
)
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
@@ -267,6 +274,8 @@ def build_parser() -> argparse.ArgumentParser:
|
|||||||
listener.add_argument("--host", default="0.0.0.0", help="Bind address")
|
listener.add_argument("--host", default="0.0.0.0", help="Bind address")
|
||||||
listener.add_argument("--port", type=int, default=5514, help="UDP port. Use 514 only with sudo/capability.")
|
listener.add_argument("--port", type=int, default=5514, help="UDP port. Use 514 only with sudo/capability.")
|
||||||
listener.add_argument("--output", default="logs/fg_syslog.jsonl", help="File to append received logs to")
|
listener.add_argument("--output", default="logs/fg_syslog.jsonl", help="File to append received logs to")
|
||||||
|
listener.add_argument("--rotate-bytes", type=int, default=25 * 1024 * 1024, help="Rotate active log at this size; 0 disables rotation")
|
||||||
|
listener.add_argument("--rotate-count", type=int, default=14, help="Number of compressed log archives to retain")
|
||||||
listener.add_argument("--quiet", action="store_true", help="Do not print each received syslog message")
|
listener.add_argument("--quiet", action="store_true", help="Do not print each received syslog message")
|
||||||
listener.set_defaults(func=listen_syslog)
|
listener.set_defaults(func=listen_syslog)
|
||||||
|
|
||||||
|
|||||||
@@ -1,24 +1,70 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import gzip
|
||||||
import socket
|
import socket
|
||||||
|
from datetime import datetime, timezone
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
def listen_udp_syslog(host: str, port: int, output: str, *, max_bytes: int = 65535, quiet: bool = False) -> None:
|
def rotate_log_file(output_path: Path, *, max_archives: int) -> Path | None:
|
||||||
|
"""Archive the active syslog file as gzip and retain only recent archives."""
|
||||||
|
if not output_path.exists() or output_path.stat().st_size == 0:
|
||||||
|
return None
|
||||||
|
|
||||||
|
timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
||||||
|
archive = output_path.with_name(f"{output_path.name}.{timestamp}.gz")
|
||||||
|
suffix = 1
|
||||||
|
while archive.exists():
|
||||||
|
archive = output_path.with_name(f"{output_path.name}.{timestamp}.{suffix}.gz")
|
||||||
|
suffix += 1
|
||||||
|
|
||||||
|
with output_path.open("rb") as source, gzip.open(archive, "wb") as destination:
|
||||||
|
while chunk := source.read(1024 * 1024):
|
||||||
|
destination.write(chunk)
|
||||||
|
output_path.unlink()
|
||||||
|
|
||||||
|
archives = sorted(output_path.parent.glob(f"{output_path.name}.*.gz"), key=lambda path: path.stat().st_mtime, reverse=True)
|
||||||
|
for stale_archive in archives[max(0, max_archives):]:
|
||||||
|
stale_archive.unlink()
|
||||||
|
return archive
|
||||||
|
|
||||||
|
|
||||||
|
def listen_udp_syslog(
|
||||||
|
host: str,
|
||||||
|
port: int,
|
||||||
|
output: str,
|
||||||
|
*,
|
||||||
|
max_bytes: int = 65535,
|
||||||
|
rotate_bytes: int = 25 * 1024 * 1024,
|
||||||
|
rotate_count: int = 14,
|
||||||
|
quiet: bool = False,
|
||||||
|
) -> None:
|
||||||
output_path = Path(output)
|
output_path = Path(output)
|
||||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
if rotate_bytes < 0 or rotate_count < 0:
|
||||||
|
raise ValueError("rotation settings must be zero or positive")
|
||||||
|
|
||||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||||
sock.bind((host, port))
|
sock.bind((host, port))
|
||||||
|
|
||||||
print(f"Listening for UDP syslog on {host}:{port}")
|
print(f"Listening for UDP syslog on {host}:{port}")
|
||||||
print(f"Writing logs to {output_path}")
|
print(f"Writing logs to {output_path}")
|
||||||
|
if rotate_bytes:
|
||||||
|
print(f"Rotation: {rotate_bytes} bytes, retaining {rotate_count} gzip archives")
|
||||||
with output_path.open("a", encoding="utf-8", buffering=1) as handle:
|
with output_path.open("a", encoding="utf-8", buffering=1) as handle:
|
||||||
while True:
|
while True:
|
||||||
data, address = sock.recvfrom(max_bytes)
|
data, address = sock.recvfrom(max_bytes)
|
||||||
message = data.decode("utf-8", errors="replace").strip()
|
message = data.decode("utf-8", errors="replace").strip()
|
||||||
if not message:
|
if not message:
|
||||||
continue
|
continue
|
||||||
handle.write(f"{message}\n")
|
line = f"{message}\n"
|
||||||
|
line_bytes = len(line.encode("utf-8"))
|
||||||
|
if rotate_bytes and output_path.stat().st_size + line_bytes > rotate_bytes:
|
||||||
|
handle.close()
|
||||||
|
archive = rotate_log_file(output_path, max_archives=rotate_count)
|
||||||
|
if archive and not quiet:
|
||||||
|
print(f"Rotated syslog to {archive}")
|
||||||
|
handle = output_path.open("a", encoding="utf-8", buffering=1)
|
||||||
|
handle.write(line)
|
||||||
if not quiet:
|
if not quiet:
|
||||||
print(f"{address[0]}:{address[1]} {message}")
|
print(f"{address[0]}:{address[1]} {message}")
|
||||||
|
|||||||
8
start.sh
8
start.sh
@@ -6,6 +6,8 @@ VENV_DIR="${FGAI_VENV_DIR:-$ROOT_DIR/.venv}"
|
|||||||
PORT="${FGAI_SYSLOG_PORT:-5514}"
|
PORT="${FGAI_SYSLOG_PORT:-5514}"
|
||||||
HOST="${FGAI_SYSLOG_HOST:-0.0.0.0}"
|
HOST="${FGAI_SYSLOG_HOST:-0.0.0.0}"
|
||||||
LOG_FILE="${FGAI_SYSLOG_FILE:-$ROOT_DIR/logs/fg_syslog.jsonl}"
|
LOG_FILE="${FGAI_SYSLOG_FILE:-$ROOT_DIR/logs/fg_syslog.jsonl}"
|
||||||
|
LOG_ROTATE_BYTES="${FGAI_LOG_ROTATE_BYTES:-26214400}"
|
||||||
|
LOG_ROTATE_COUNT="${FGAI_LOG_ROTATE_COUNT:-14}"
|
||||||
LISTENER_LOG="${FGAI_LISTENER_LOG:-$ROOT_DIR/logs/fgai-listener.log}"
|
LISTENER_LOG="${FGAI_LISTENER_LOG:-$ROOT_DIR/logs/fgai-listener.log}"
|
||||||
POLICY_FILE="${FGAI_POLICY_FILE:-$ROOT_DIR/exports/policies.json}"
|
POLICY_FILE="${FGAI_POLICY_FILE:-$ROOT_DIR/exports/policies.json}"
|
||||||
STATE_FILE="${FGAI_STATE_FILE:-$ROOT_DIR/state/fgai-status.json}"
|
STATE_FILE="${FGAI_STATE_FILE:-$ROOT_DIR/state/fgai-status.json}"
|
||||||
@@ -29,6 +31,8 @@ usage() {
|
|||||||
printf ' FGAI_SYSLOG_PORT=%s\n' "$PORT"
|
printf ' FGAI_SYSLOG_PORT=%s\n' "$PORT"
|
||||||
printf ' FGAI_SYSLOG_HOST=%s\n' "$HOST"
|
printf ' FGAI_SYSLOG_HOST=%s\n' "$HOST"
|
||||||
printf ' FGAI_SYSLOG_FILE=%s\n' "$LOG_FILE"
|
printf ' FGAI_SYSLOG_FILE=%s\n' "$LOG_FILE"
|
||||||
|
printf ' FGAI_LOG_ROTATE_BYTES=%s\n' "$LOG_ROTATE_BYTES"
|
||||||
|
printf ' FGAI_LOG_ROTATE_COUNT=%s\n' "$LOG_ROTATE_COUNT"
|
||||||
printf ' FGAI_LISTENER_LOG=%s\n' "$LISTENER_LOG"
|
printf ' FGAI_LISTENER_LOG=%s\n' "$LISTENER_LOG"
|
||||||
printf ' FGAI_DASHBOARD_PORT=%s\n' "$DASHBOARD_PORT"
|
printf ' FGAI_DASHBOARD_PORT=%s\n' "$DASHBOARD_PORT"
|
||||||
printf ' FGAI_LLM=%s\n' "$LLM_ENABLED"
|
printf ' FGAI_LLM=%s\n' "$LLM_ENABLED"
|
||||||
@@ -85,12 +89,16 @@ start_listener() {
|
|||||||
--host "$HOST" \
|
--host "$HOST" \
|
||||||
--port "$PORT" \
|
--port "$PORT" \
|
||||||
--output "$LOG_FILE" \
|
--output "$LOG_FILE" \
|
||||||
|
--rotate-bytes "$LOG_ROTATE_BYTES" \
|
||||||
|
--rotate-count "$LOG_ROTATE_COUNT" \
|
||||||
--quiet > "$LISTENER_LOG" 2>&1 &
|
--quiet > "$LISTENER_LOG" 2>&1 &
|
||||||
else
|
else
|
||||||
nohup "$VENV_DIR/bin/fgai" listen-syslog \
|
nohup "$VENV_DIR/bin/fgai" listen-syslog \
|
||||||
--host "$HOST" \
|
--host "$HOST" \
|
||||||
--port "$PORT" \
|
--port "$PORT" \
|
||||||
--output "$LOG_FILE" \
|
--output "$LOG_FILE" \
|
||||||
|
--rotate-bytes "$LOG_ROTATE_BYTES" \
|
||||||
|
--rotate-count "$LOG_ROTATE_COUNT" \
|
||||||
--quiet > "$LISTENER_LOG" 2>&1 &
|
--quiet > "$LISTENER_LOG" 2>&1 &
|
||||||
fi
|
fi
|
||||||
printf '%s\n' "$!" > "$LISTENER_PID_FILE"
|
printf '%s\n' "$!" > "$LISTENER_PID_FILE"
|
||||||
|
|||||||
28
tests/test_syslog_server.py
Normal file
28
tests/test_syslog_server.py
Normal file
@@ -0,0 +1,28 @@
|
|||||||
|
import gzip
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from fgai.syslog_server import rotate_log_file
|
||||||
|
|
||||||
|
|
||||||
|
class SyslogRotationTests(unittest.TestCase):
|
||||||
|
def test_rotation_compresses_and_prunes_archives(self):
|
||||||
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
||||||
|
output = Path(temporary_directory) / "fg_syslog.jsonl"
|
||||||
|
output.write_text("first log line\n", encoding="utf-8")
|
||||||
|
|
||||||
|
archive = rotate_log_file(output, max_archives=1)
|
||||||
|
|
||||||
|
self.assertIsNotNone(archive)
|
||||||
|
self.assertFalse(output.exists())
|
||||||
|
with gzip.open(archive, "rt", encoding="utf-8") as handle:
|
||||||
|
self.assertEqual(handle.read(), "first log line\n")
|
||||||
|
|
||||||
|
output.write_text("second log line\n", encoding="utf-8")
|
||||||
|
rotate_log_file(output, max_archives=1)
|
||||||
|
self.assertEqual(len(list(Path(temporary_directory).glob("fg_syslog.jsonl.*.gz"))), 1)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user