diff --git a/README.md b/README.md index 9859578..31413df 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,93 @@ -# nginx-realtime-traffic-dashboard +# Nginx Live Traffic Monitor -Shows realtime traffic flows for an nginx reverse proxy per server. \ No newline at end of file +![Dashboard screenshot](docs/screenshot-placeholder.svg) + +A small Python/FastAPI dashboard for seeing live traffic through an Nginx reverse proxy. It follows one dedicated access log, aggregates requests into bounded one-second in-memory buckets, and streams telemetry to browsers over Server-Sent Events. There is no database or persistent application telemetry. + +## Requirements and installation + +Python 3.10+ is required. From a checkout: + +```sh +python3 -m venv .venv +. .venv/bin/activate +pip install -r requirements.txt +``` + +The dashboard uses Bulma and Chart.js from their public CDNs. For an offline deployment, vendor those two assets and update `app/static/index.html`. + +## Nginx log format + +Add the dedicated log (the application does not change Nginx configuration): + +```nginx +log_format traffic '$msec $host $request_method $status ' + '$request_length $bytes_sent ' + '$request_time $upstream_response_time'; + +access_log /var/log/nginx-traffic/traffic.log traffic; +``` + +The follower tolerates malformed lines, `-` upstream times, multiple upstream values, temporary disappearance, and rename/replacement rotation. It starts at the end of an existing file, like `tail -F`, so old log history is not mistaken for live traffic. The application keeps only configured rolling buckets (and bounded latency samples); totals are from application startup. + +## Running + +```sh +python -m app.main +``` + +Open . Configuration is available through flags or environment variables: + +| Setting | Flag | Environment variable | Default | +| --- | --- | --- | --- | +| Traffic log | `--log-path` | `TRAFFIC_LOG_PATH` | `/var/log/nginx-traffic/traffic.log` | +| Listen address | `--host` | `LISTEN_ADDRESS` | `127.0.0.1` | +| Listen port | `--port` | `LISTEN_PORT` | `8080` | +| Retention seconds | `--retention` | `TELEMETRY_RETENTION` | `3600` | + +Example: + +```sh +TRAFFIC_LOG_PATH=/var/log/nginx-traffic/traffic.log LISTEN_ADDRESS=0.0.0.0 LISTEN_PORT=8080 \ + python -m app.main --retention 3600 +``` + +The visible window can be set to 1 minute, 5 minutes, 15 minutes, or 1 hour. A new browser receives enough current history to fill its selected graph window. Host names are discovered automatically and can be selected for breakout graphs. Theme selection (system, light, dark) is stored in browser local storage. + +## tmpfs and permissions + +One possible tmpfs setup is: + +```sh +sudo install -d -o nginx -g nginx -m 0750 /var/log/nginx-traffic +sudo mount -t tmpfs -o size=64M,mode=0750,uid=nginx,gid=nginx tmpfs /var/log/nginx-traffic +``` + +The service account needs search/read permission on the directory and read permission on `traffic.log`. If Nginx writes as another group, grant the monitor account group access or use an ACL, for example: + +```sh +sudo setfacl -m u:nginx-monitor:rx /var/log/nginx-traffic +sudo setfacl -m u:nginx-monitor:r /var/log/nginx-traffic/traffic.log +``` + +## systemd + +Copy the included `nginx-traffic-monitor.service` to `/etc/systemd/system/`, adjust `WorkingDirectory`, virtualenv path, and the `User`/`Group` to match the installation, then: + +```sh +sudo systemctl daemon-reload +sudo systemctl enable --now nginx-traffic-monitor +sudo systemctl status nginx-traffic-monitor +``` + +The service should run as an unprivileged account. Put a reverse proxy or firewall in front of it if it must be accessed beyond localhost; this utility intentionally has no authentication. + +## Development + +Run the focused parser/aggregation tests with: + +```sh +pytest -q +``` + +The server task follows the path by inode, drains currently available data from a renamed file, and reopens a replacement. A missing path is retried without busy polling. The publisher sends one compact snapshot per second to each SSE subscriber, and disconnecting clients are removed automatically. diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..7d8631c --- /dev/null +++ b/app/__init__.py @@ -0,0 +1 @@ +"""Nginx live traffic monitor.""" diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..2b5f587 --- /dev/null +++ b/app/main.py @@ -0,0 +1,305 @@ +from __future__ import annotations + +import argparse +import asyncio +import json +import math +import os +import statistics +import threading +import time +from collections import deque +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, AsyncIterator + +from fastapi import FastAPI, Request +from fastapi.responses import FileResponse, StreamingResponse +from fastapi.staticfiles import StaticFiles + + +DEFAULT_LOG = "/var/log/nginx-traffic/traffic.log" +MAX_HOSTS = 1000 + + +@dataclass +class Bucket: + incoming: int = 0 + outgoing: int = 0 + requests: int = 0 + statuses: list[int] = field(default_factory=lambda: [0, 0, 0, 0]) + request_times: list[float] = field(default_factory=list) + upstream_times: list[float] = field(default_factory=list) + + +class Series: + def __init__(self, retention: int): + self.retention = retention + self.buckets: dict[int, Bucket] = {} + + def bucket(self, second: int) -> Bucket: + if second not in self.buckets: + self.buckets[second] = Bucket() + cutoff = second - self.retention + 1 + for old in [key for key in self.buckets if key < cutoff]: + del self.buckets[old] + return self.buckets[second] + + +def percentile(values: list[float], p: float) -> float | None: + if not values: + return None + values = sorted(values) + index = (len(values) - 1) * p + low, high = math.floor(index), math.ceil(index) + if low == high: + return values[low] + return values[low] + (values[high] - values[low]) * (index - low) + + +def parse_line(line: str) -> dict[str, Any] | None: + """Parse the configured, whitespace-delimited Nginx traffic format.""" + fields = line.strip().split() + if len(fields) < 8: + return None + try: + timestamp = float(fields[0]) + status = int(fields[3]) + request_length = int(fields[4]) + bytes_sent = int(fields[5]) + request_time = float(fields[6]) + upstream_raw = "".join(fields[7:]) + upstream_values = [float(value) for value in upstream_raw.split(",") if value not in ("", "-")] + if not all(math.isfinite(value) and value >= 0 for value in + (timestamp, request_length, bytes_sent, request_time, *upstream_values)): + return None + if status < 100 or status > 599: + return None + except (ValueError, OverflowError): + return None + return { + "timestamp": timestamp, + "host": fields[1] or "(unknown)", + "incoming": request_length, + "outgoing": bytes_sent, + "status": status, + "request_time": request_time, + "upstream_time": statistics.fmean(upstream_values) if upstream_values else None, + } + + +class Telemetry: + def __init__(self, retention: int): + self.retention = max(60, min(retention, 86400)) + self.global_series = Series(self.retention) + self.hosts: dict[str, Series] = {} + self.last_seen: dict[str, int] = {} + self.total_incoming = 0 + self.total_outgoing = 0 + self.lock = threading.RLock() + self.subscribers: set[asyncio.Queue[str]] = set() + + def ingest(self, event: dict[str, Any]) -> None: + now = int(time.time()) + second = int(event["timestamp"]) + # Log timestamps are normally current, but clamping protects bounded + # storage if a clock-jumped or otherwise surprising line appears. + if second < now - self.retention: + return + second = min(second, now) + with self.lock: + self._add(self.global_series.bucket(second), event) + host = event["host"] + if host not in self.hosts: + if len(self.hosts) >= MAX_HOSTS: + oldest = min(self.last_seen, key=self.last_seen.get) + self.hosts.pop(oldest, None) + self.last_seen.pop(oldest, None) + self.hosts[host] = Series(self.retention) + self._add(self.hosts[host].bucket(second), event) + self.last_seen[host] = second + self.total_incoming += event["incoming"] + self.total_outgoing += event["outgoing"] + + @staticmethod + def _add(bucket: Bucket, event: dict[str, Any]) -> None: + bucket.incoming += event["incoming"] + bucket.outgoing += event["outgoing"] + bucket.requests += 1 + status_class = event["status"] // 100 + if 2 <= status_class <= 5: + bucket.statuses[status_class - 2] += 1 + if len(bucket.request_times) < 256: + bucket.request_times.append(event["request_time"]) + if event["upstream_time"] is not None and len(bucket.upstream_times) < 256: + bucket.upstream_times.append(event["upstream_time"]) + + def subscribe(self) -> asyncio.Queue[str]: + queue: asyncio.Queue[str] = asyncio.Queue(maxsize=2) + with self.lock: + self.subscribers.add(queue) + return queue + + def unsubscribe(self, queue: asyncio.Queue[str]) -> None: + with self.lock: + self.subscribers.discard(queue) + + def snapshot(self, window: int) -> dict[str, Any]: + now = int(time.time()) + start = now - window + 1 + with self.lock: + points = [self._point(self.global_series.buckets.get(second), second) for second in range(start, now + 1)] + hosts = sorted(self.hosts, key=lambda host: self.last_seen.get(host, 0), reverse=True) + active = [host for host in hosts if self.last_seen.get(host, 0) >= now - 59] + host_points = { + host: [self._point(self.hosts[host].buckets.get(second), second) for second in range(start, now + 1)] + for host in hosts + } + current = self._summary(points[-60:]) + current["total_incoming"] = self.total_incoming + current["total_outgoing"] = self.total_outgoing + current["active_hosts"] = len(active) + current["busiest_host"] = max(active, key=lambda host: sum(p["incoming"] + p["outgoing"] for p in host_points[host][-60:]), default=None) + return {"type": "telemetry", "now": now, "points": points, "hosts": hosts, "host_points": host_points, "summary": current} + + @staticmethod + def _point(bucket: Bucket | None, second: int) -> dict[str, Any]: + if not bucket: + return {"time": second, "incoming": 0, "outgoing": 0, "requests": 0, "status": [0, 0, 0, 0], "latency": None, "p95": None, "upstream": None} + return {"time": second, "incoming": bucket.incoming, "outgoing": bucket.outgoing, "requests": bucket.requests, "status": bucket.statuses, "latency": statistics.fmean(bucket.request_times) if bucket.request_times else None, "p95": percentile(bucket.request_times, .95), "upstream": statistics.fmean(bucket.upstream_times) if bucket.upstream_times else None} + + @staticmethod + def _summary(points: list[dict[str, Any]]) -> dict[str, Any]: + seconds = max(1, len(points)) + total_requests = sum(point["requests"] for point in points) + statuses = [sum(point["status"][index] for point in points) for index in range(4)] + latencies = [point["latency"] for point in points if point["latency"] is not None] + upstream = [point["upstream"] for point in points if point["upstream"] is not None] + return {"incoming": sum(point["incoming"] for point in points) / seconds, "outgoing": sum(point["outgoing"] for point in points) / seconds, "requests": total_requests / seconds, "status": statuses, "latency": statistics.fmean(latencies) if latencies else None, "p95": percentile(latencies, .95), "upstream": statistics.fmean(upstream) if upstream else None} + + +async def follow_log(path: str, telemetry: Telemetry, stop: asyncio.Event) -> None: + handle = None + inode = None + pending = "" + while not stop.is_set(): + try: + stat = await asyncio.to_thread(os.stat, path) + if handle is None: + handle = await asyncio.to_thread(open, path, "r", encoding="utf-8", errors="replace") + await asyncio.to_thread(handle.seek, 0, os.SEEK_END) + inode = (stat.st_dev, stat.st_ino) + elif inode != (stat.st_dev, stat.st_ino): + # Drain what is currently available from the renamed file before switching. + for line in handle: + event = parse_line(line) + if event: + telemetry.ingest(event) + handle.close() + handle = await asyncio.to_thread(open, path, "r", encoding="utf-8", errors="replace") + inode = (stat.st_dev, stat.st_ino) + # Never join a partial line from the old inode to a new file. + pending = "" + data = await asyncio.to_thread(handle.read) + if data: + pending += data + lines = pending.split("\n") + pending = lines.pop() + for line in lines: + event = parse_line(line) + if event: + telemetry.ingest(event) + await asyncio.sleep(.15) + except FileNotFoundError: + if handle: + data = handle.read() + if data: + pending += data + await asyncio.sleep(.5) + except (OSError, ValueError): + if handle: + handle.close() + handle, inode = None, None + await asyncio.sleep(.5) + if handle: + handle.close() + + +def create_app(log_path: str = DEFAULT_LOG, retention: int = 3600) -> FastAPI: + telemetry = Telemetry(retention) + stop = asyncio.Event() + app = FastAPI(title="Nginx Live Traffic Monitor") + app.mount("/static", StaticFiles(directory=Path(__file__).parent / "static"), name="static") + + @app.on_event("startup") + async def startup() -> None: + app.state.task = asyncio.create_task(follow_log(log_path, telemetry, stop)) + app.state.publisher = asyncio.create_task(publish()) + + @app.on_event("shutdown") + async def shutdown() -> None: + stop.set() + for task in (getattr(app.state, "task", None), getattr(app.state, "publisher", None)): + if task: + task.cancel() + + async def publish() -> None: + while not stop.is_set(): + payload = json.dumps(telemetry.snapshot(3600), separators=(",", ":")) + with telemetry.lock: + subscribers = list(telemetry.subscribers) + for queue in subscribers: + try: + queue.put_nowait(payload) + except asyncio.QueueFull: + try: + queue.get_nowait() + except asyncio.QueueEmpty: + pass + queue.put_nowait(payload) + await asyncio.sleep(1) + + @app.get("/") + async def index() -> FileResponse: + return FileResponse(Path(__file__).parent / "static" / "index.html") + + @app.get("/api/telemetry") + async def telemetry_api(window: int = 300) -> dict[str, Any]: + return telemetry.snapshot(max(60, min(window, 3600))) + + @app.get("/api/events") + async def events(request: Request, window: int = 300) -> StreamingResponse: + queue = telemetry.subscribe() + initial = json.dumps(telemetry.snapshot(max(60, min(window, 3600))), separators=(",", ":")) + async def stream() -> AsyncIterator[str]: + try: + yield f"data: {initial}\n\n" + while True: + if await request.is_disconnected(): + break + try: + payload = await asyncio.wait_for(queue.get(), timeout=15) + yield f"data: {payload}\n\n" + except asyncio.TimeoutError: + yield ": keep-alive\n\n" + finally: + telemetry.unsubscribe(queue) + return StreamingResponse(stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) + + app.state.telemetry = telemetry + return app + + +def main() -> None: + parser = argparse.ArgumentParser(description="Live Nginx traffic dashboard") + parser.add_argument("--log-path", default=os.getenv("TRAFFIC_LOG_PATH", DEFAULT_LOG)) + parser.add_argument("--host", default=os.getenv("LISTEN_ADDRESS", "127.0.0.1")) + parser.add_argument("--port", type=int, default=int(os.getenv("LISTEN_PORT", "8080"))) + parser.add_argument("--retention", type=int, default=int(os.getenv("TELEMETRY_RETENTION", "3600"))) + args = parser.parse_args() + import uvicorn + uvicorn.run(create_app(args.log_path, args.retention), host=args.host, port=args.port) + + +if __name__ == "__main__": + main() diff --git a/app/static/index.html b/app/static/index.html new file mode 100644 index 0000000..969e679 --- /dev/null +++ b/app/static/index.html @@ -0,0 +1,54 @@ + + + + + Nginx Live Traffic Monitor + + + + + +
+

Traffic overview

Connecting to telemetry…

+

+
+
+

Throughput

Waiting for data
+

Host breakouts

Select hosts to compare their recent traffic.

+
+ + + diff --git a/docs/screenshot-placeholder.svg b/docs/screenshot-placeholder.svg new file mode 100644 index 0000000..e9c8dbd --- /dev/null +++ b/docs/screenshot-placeholder.svg @@ -0,0 +1,14 @@ + + + + Nginx Live Traffic + + INBOUND3.8 Mbit/s + OUTBOUND42.3 Mbit/s + REQUESTS183/sec + Throughput + + + + Screenshot placeholder — run the app to view live data. + diff --git a/nginx-traffic-monitor.service b/nginx-traffic-monitor.service new file mode 100644 index 0000000..3c74293 --- /dev/null +++ b/nginx-traffic-monitor.service @@ -0,0 +1,17 @@ +[Unit] +Description=Nginx live traffic monitor +After=network.target nginx.service + +[Service] +Type=simple +User=nginx-monitor +Group=nginx-monitor +WorkingDirectory=/opt/nginx-live-traffic-monitor +ExecStart=/opt/nginx-live-traffic-monitor/.venv/bin/nginx-traffic-monitor --host 127.0.0.1 --port 8080 +Restart=on-failure +RestartSec=2 +NoNewPrivileges=true +PrivateTmp=true + +[Install] +WantedBy=multi-user.target diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a75402f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,12 @@ +[project] +name = "nginx-live-traffic-monitor" +version = "0.1.0" +description = "Small in-memory live traffic dashboard for an Nginx reverse proxy" +requires-python = ">=3.10" +dependencies = ["fastapi>=0.115,<1", "uvicorn[standard]>=0.30,<1"] + +[project.optional-dependencies] +test = ["pytest>=8,<9"] + +[project.scripts] +nginx-traffic-monitor = "app.main:main" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a4c245d --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +fastapi>=0.115,<1 +uvicorn[standard]>=0.30,<1 diff --git a/tests/test_main.py b/tests/test_main.py new file mode 100644 index 0000000..1b769f9 --- /dev/null +++ b/tests/test_main.py @@ -0,0 +1,28 @@ +import time + +from app.main import Telemetry, parse_line + + +def test_parse_line_and_multiple_upstreams(): + event = parse_line("1720000000.123 example.test GET 200 120 240 0.150 0.100, 0.200") + assert event["host"] == "example.test" + assert event["incoming"] == 120 + assert event["upstream_time"] == 0.15 + + +def test_malformed_lines_are_ignored(): + assert parse_line("not enough fields") is None + assert parse_line("1 host GET nope 1 2 0.1 -") is None + assert parse_line("1 host GET 600 1 2 0.1 -") is None + + +def test_telemetry_is_aggregated_and_bounded(): + telemetry = Telemetry(60) + base = int(time.time()) - 30 + for second in range(70): + telemetry.ingest(parse_line(f"{base + second} host GET 200 10 20 .1 -")) + assert telemetry.total_incoming == 700 + assert len(telemetry.global_series.buckets) <= 60 + snapshot = telemetry.snapshot(60) + assert snapshot["summary"]["requests"] >= 1 + assert snapshot["hosts"] == ["host"]