From 074d17be894976f9af7d075f61749a5e15425a92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mateusz=20Gruszczy=C5=84ski?= Date: Sun, 16 Aug 2026 22:43:06 +0200 Subject: [PATCH] fix memory usage redis --- README.md | 51 ++- VERSION | 2 +- app/analytics_cache.py | 192 +++++--- app/config.py | 22 +- app/eve.py | 2 +- app/live.py | 811 +++++++++++++++++++++++----------- app/main.py | 10 +- app/redis_service.py | 31 +- app/static/js/app.js | 4 +- app/store.py | 369 +++++++++++++++- app/templates/index.html | 6 +- app/webui.py | 15 +- deploy-routeros.env.example | 11 +- docs/routeros-tzsp-capture.md | 6 +- rust/tzsp-receiver/Cargo.toml | 2 +- scripts/deploy-routeros.sh | 15 +- tests/test_analytics_cache.py | 47 +- tests/test_config.py | 10 + tests/test_deploy_script.py | 4 +- tests/test_live.py | 96 ++++ tests/test_redis_service.py | 18 +- tests/test_store.py | 35 +- 22 files changed, 1377 insertions(+), 382 deletions(-) diff --git a/README.md b/README.md index 8da16a1..5d55385 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,17 @@ # MikroSuricata -Project version: `0.11.2` +Project version: `0.11.3` + + +## What changed in 0.11.3 + +- Reworked traffic retention so **Redis is only a short-lived ingestion buffer**. A dedicated archive worker commits normalized EVE events and throughput samples to SQLite and removes them from Redis only after the SQLite transaction succeeds. +- Dashboard analytics and chart snapshots are now calculated from the **SQLite traffic archive** and persisted in SQLite. Multi-hour views no longer scan large Redis sorted sets. +- Added an exact **5 hour (`18000` seconds)** dashboard range. Requested ranges are no longer rounded to the nearest predefined analytics window. +- Added bounded-memory analytics for long windows: endpoint/application cardinality is aggregated inside SQLite and a separate snapshot worker processes one requested window at a time instead of retaining large Python flow/IP sets. Archiving continues while a long 24h snapshot is being calculated. +- Managed Redis now defaults to a **128 MiB hard memory limit**, `noeviction`, no RDB snapshots and no AOF. The defaults prevent Redis from growing until the RouterOS container is killed by OOM; durable traffic history lives in `/data/ids.db`. +- Added `TRAFFIC_ARCHIVE_INTERVAL_SECONDS`, `TRAFFIC_ARCHIVE_LAG_SECONDS` and `TRAFFIC_ARCHIVE_BATCH_SIZE` controls. Defaults move committed data out of Redis every 5 seconds with a 10 second live-buffer lag. +- Upgrades with a large legacy Redis backlog drain it in bounded batches before publishing fresh SQLite snapshots, preventing partially migrated history from appearing in charts. ## What changed in 0.11.2 @@ -116,10 +127,15 @@ NDR_AUTO_BLOCK_RISK=92 ROUTEROS_INVENTORY_INTERVAL_SECONDS=300 NOTIFY_WEBHOOK_URL= NOTIFY_MIN_RISK=80 -REDIS_AOF=true +REDIS_MAXMEMORY_MB=128 +REDIS_SNAPSHOT_SECONDS=0 +REDIS_AOF=false +TRAFFIC_ARCHIVE_INTERVAL_SECONDS=5 +TRAFFIC_ARCHIVE_LAG_SECONDS=10 +TRAFFIC_ARCHIVE_BATCH_SIZE=1000 ``` -All NDR state, IOC data, Redis persistence, Suricata logs/rules and forensic PCAP rotation remain below the single persistent `/data` mount. +All durable NDR state, IOC data, traffic history/chart snapshots, Suricata logs/rules and forensic PCAP rotation remain below the single persistent `/data` mount. Redis is a bounded transient ingest buffer by default; SQLite holds the durable traffic archive. ## Architecture @@ -134,10 +150,11 @@ single RouterOS container + TAP suritap0 + Suricata IDS + Python control plane / EVE JSON watcher - + SQLite alerts / assets / NDR incidents / sessions + + SQLite alerts / assets / NDR incidents / sessions / traffic archive + MikroSuricata behavior + correlation engine + local IOC datasets (IP/domain/SHA256/JA3/JA4/HASSH) - + Redis traffic history + + bounded Redis ingest buffer + + SQLite dashboard snapshots + Rust -> Python 1 Hz Unix telemetry + WebSocket live throughput / event stream + Web UI :8080 @@ -152,7 +169,7 @@ RouterOS TZSP UDP v Rust receiver -- recvmmsg() --> TZSP decode --> TAP write --> Suricata | - +-- 1 Hz counters only --> Unix datagram --> Python --> Redis / WebSocket / Prometheus + +-- 1 Hz counters only --> Unix datagram --> Python --> short Redis buffer / SQLite / WebSocket / Prometheus ``` If Redis, the browser or an analytics request is slow, it cannot block UDP receive/TAP injection. @@ -424,7 +441,9 @@ Administrative actions require a dashboard session. Configure `ADMIN_USERNAME` a Authenticated maintenance includes clearing incident/history data, SQLite `VACUUM`, runtime counter reset, RouterOS block-list actions, validated custom-rule/threshold edits, live Suricata rule reloads and managed signature-feed updates. Keep port `8080` on a trusted management network or place the dashboard behind HTTPS. -SQLite also stores the four rolling chart summaries (`900`, `3600`, `21600`, `86400` seconds). They refresh every `ANALYTICS_SNAPSHOT_INTERVAL_SECONDS` and are served immediately after UI entry/restart when newer live history is temporarily unavailable. +SQLite also stores the normalized traffic archive and rolling dashboard summaries. Standard warm ranges are `900`, `3600`, `18000`, `21600` and `86400` seconds (15m / 1h / 5h / 6h / 24h), while arbitrary requested ranges are materialized on demand. The archive worker first commits old Redis entries to SQLite, then deletes those exact Redis members; chart calculations therefore do not require multi-hour Redis history. + +Redis is deliberately transient in 0.11.3. New deployments use `REDIS_MAXMEMORY_MB=128`, `REDIS_SNAPSHOT_SECONDS=0` and `REDIS_AOF=false`. If an existing RouterOS `IDS_ENV` from an older release still explicitly sets `REDIS_AOF=true` or a non-zero `REDIS_SNAPSHOT_SECONDS`, change those values to the new defaults when you want the fully transient Redis model. --- @@ -626,15 +645,15 @@ Then deploy by giving the **RouterOS-side TAR path** directly: ./scripts/deploy-routeros.sh routeros-suricata-tzsp-arm64.tar ``` -The deployer no longer builds, detects image architecture, renames, or re-uploads the image. For project version `0.11.2` it creates: +The deployer no longer builds, detects image architecture, renames, or re-uploads the image. For project version `0.11.3` it creates: ```text -name=suricata_0.11.2 +name=suricata_0.11.3 file=routeros-suricata-tzsp-arm64.tar -root-dir=/containers/suricata_0.11.2/root +root-dir=/containers/suricata_0.11.3/root ``` -The remaining deployment work is unchanged: private container bridge/VETH/NAT, environment, persistent mounts, optional RouterOS REST user/firewall integration, hybrid TZSP capture configuration, image extraction wait, container start, and final status. Existing containers are not removed. Re-running deployment for the same version stops with `Container suricata_0.11.2 already exists`. +The remaining deployment work is unchanged: private container bridge/VETH/NAT, environment, persistent mounts, optional RouterOS REST user/firewall integration, hybrid TZSP capture configuration, image extraction wait, container start, and final status. Existing containers are not removed. Re-running deployment for the same version stops with `Container suricata_0.11.3 already exists`. For SSH key authentication set: @@ -715,8 +734,8 @@ The deployment uses one persistent directory outside the image root: Inside it the application keeps: ```text -/data/ids.db SQLite, sessions and analytics snapshots -/data/redis/ Redis persistence +/data/ids.db SQLite, sessions, traffic archive and analytics snapshots +/data/redis/ Redis runtime directory (persistence disabled by default) /data/logs/suricata/ EVE/raw Suricata logs /data/lib/suricata/ suricata-update feeds, cache and vendor rules /data/suricata/ custom rules, thresholds and update filters @@ -972,12 +991,12 @@ After the first deployment, when the VETH/private bridge/NAT, hybrid TZSP captur ./scripts/upgrade-routeros-container.sh routeros-suricata-tzsp-arm64.tar ``` -For version `0.11.2` the second command creates: +For version `0.11.3` the second command creates: ```text -name=suricata_0.11.2 +name=suricata_0.11.3 file=routeros-suricata-tzsp-arm64.tar -root-dir=/containers/suricata_0.11.2/root +root-dir=/containers/suricata_0.11.3/root interface=veth-ids envlist=IDS_ENV mountlists=IDS_MOUNTS diff --git a/VERSION b/VERSION index bc859cb..1a96df1 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.11.2 +0.11.3 diff --git a/app/analytics_cache.py b/app/analytics_cache.py index 9a00c81..5a4de1d 100644 --- a/app/analytics_cache.py +++ b/app/analytics_cache.py @@ -9,16 +9,16 @@ from .live import RedisUnavailableError, TrafficHistory from .store import AlertStore -SUMMARY_WINDOWS = (900, 3600, 21600, 86400) +SUMMARY_WINDOWS = (900, 3600, 18000, 21600, 86400) class AnalyticsSnapshotCache: - """Redis-backed dashboard snapshots refreshed only for windows actually in use. + """Disk-backed dashboard snapshots plus Redis-to-SQLite archive worker. - Older builds rebuilt all four windows every minute, which meant scanning and - decoding the complete 24-hour Redis history even when the browser displayed - only 15 minutes. This cache keeps persisted snapshots, but refreshes only - recently requested windows and uses a slower cadence for wider ranges. + Redis is intentionally only a short-lived ingestion buffer. This worker moves + normalized EVE records and TZSP rate samples to SQLite, deletes committed + entries from Redis, and calculates dashboard snapshots from SQLite instead of + scanning multi-hour Redis sorted sets. """ def __init__( @@ -27,45 +27,76 @@ class AnalyticsSnapshotCache: history: TrafficHistory, stop_event: threading.Event, interval_seconds: int = 60, + *, + archive_interval_seconds: int = 5, + archive_lag_seconds: int = 10, + archive_batch_size: int = 1000, ) -> None: - # AlertStore stays in the signature for backwards compatibility with the - # application wiring, but traffic analytics are Redis-only. self.store = store self.history = history self.stop_event = stop_event self.interval_seconds = max(15, int(interval_seconds)) - self._thread = threading.Thread(target=self._run, name="analytics-snapshots", daemon=True) - self._wake = threading.Event() + self.archive_interval_seconds = max(1, int(archive_interval_seconds)) + self.archive_lag_seconds = max(2, int(archive_lag_seconds)) + self.archive_batch_size = max(100, min(int(archive_batch_size), 5000)) + # Archiving and analytics are intentionally separate. A long 24h SQL + # aggregation must never pause Redis draining and allow the live buffer + # to grow back toward its memory ceiling. + self._archive_thread = threading.Thread( + target=self._run_archive, + name="traffic-archive", + daemon=True, + ) + self._snapshot_thread = threading.Thread( + target=self._run_snapshots, + name="analytics-snapshot", + daemon=True, + ) + self._archive_wake = threading.Event() + self._snapshot_wake = threading.Event() + self._archive_ready = threading.Event() + self._stopping = threading.Event() self._lock = threading.RLock() - self._requested_at: dict[int, float] = {} + now = time.monotonic() + # Standard UI ranges are kept warm. Arbitrary API ranges are added on demand. + self._requested_at: dict[int, float] = {window: now for window in SUMMARY_WINDOWS} self._last_refresh: dict[int, float] = {} self._errors = 0 self._refreshes = 0 - # If a browser has not used a window for this long, stop rebuilding it. + self._archive_runs = 0 + self._archive_moved_events = 0 + self._archive_moved_throughput = 0 + self._last_archive_run = 0.0 + self._last_purge = 0.0 self._active_ttl_seconds = max(300, self.interval_seconds * 10) def start(self) -> None: - if not self._thread.is_alive(): - self._thread.start() + self._stopping.clear() + if not self._archive_thread.is_alive(): + self._archive_thread.start() + if not self._snapshot_thread.is_alive(): + self._snapshot_thread.start() + self._archive_wake.set() + self._snapshot_wake.set() def stop(self, timeout: float = 2.0) -> None: - self._wake.set() - if self._thread.is_alive(): - self._thread.join(timeout=timeout) + self._stopping.set() + self._archive_wake.set() + self._snapshot_wake.set() + if self._archive_thread.is_alive(): + self._archive_thread.join(timeout=timeout) + if self._snapshot_thread.is_alive(): + self._snapshot_thread.join(timeout=timeout) def get(self, window_seconds: int) -> dict[str, Any]: window = self._normalise_window(window_seconds) now = time.monotonic() with self._lock: self._requested_at[window] = now - try: - cached = self.history.snapshot(window) - except RedisUnavailableError: - raise - + cached = self.store.traffic_snapshot(window) cadence = self._refresh_interval(window) if cached is not None: - cached = self._decorate(cached, "redis-cache") + cached = self._decorate(cached, "sqlite-snapshot") age = float(cached.get("snapshot_age_seconds") or 0) stale = age > cadence * 1.5 cached["snapshot_stale"] = stale @@ -73,12 +104,11 @@ class AnalyticsSnapshotCache: cached["snapshot_refresh_interval_seconds"] = cadence self._overlay_current_throughput(cached) if stale: - self._wake.set() + self._snapshot_wake.set() return cached - # First request after an empty Redis volume returns a shell immediately; - # only this requested range is built in the background. - self._wake.set() + # The first request for an arbitrary range is calculated by the worker. + self._snapshot_wake.set() now_ms = int(time.time() * 1000) bins_count = 60 bin_ms = max(1000, int(window * 1000 / bins_count)) @@ -102,7 +132,7 @@ class AnalyticsSnapshotCache: } for idx in range(bins_count) ], - "snapshot_source": "redis-background", + "snapshot_source": "sqlite-background", "snapshot_age_seconds": 0, "snapshot_loading": True, "snapshot_refreshing": True, @@ -110,7 +140,6 @@ class AnalyticsSnapshotCache: } def refresh_all(self) -> None: - """Explicit maintenance/test operation; normal background work is demand-driven.""" self.refresh_windows(SUMMARY_WINDOWS) def refresh_windows(self, windows: Iterable[int]) -> None: @@ -118,11 +147,10 @@ class AnalyticsSnapshotCache: if not normalized: return try: - # analytics_many scans only the widest requested window once. snapshots = self.history.analytics_many(normalized) now_wall = time.time() for window in normalized: - self.history.save_snapshot(window, snapshots[window]) + self.store.save_traffic_snapshot(window, snapshots[window]) with self._lock: self._last_refresh[window] = now_wall self._refreshes += 1 @@ -134,38 +162,105 @@ class AnalyticsSnapshotCache: self._errors += 1 def status(self) -> dict[str, Any]: - persisted = self.history.snapshot_status(SUMMARY_WINDOWS) + persisted = self.store.traffic_snapshot_status().get("windows", []) now = time.monotonic() with self._lock: active = [ window for window, requested in self._requested_at.items() - if now - requested <= self._active_ttl_seconds + if window in SUMMARY_WINDOWS or now - requested <= self._active_ttl_seconds ] refreshes = self._refreshes errors = self._errors + archive_runs = self._archive_runs + moved_events = self._archive_moved_events + moved_throughput = self._archive_moved_throughput + last_archive_run = self._last_archive_run return { - "backend": "redis", + "backend": "sqlite-archive", + "archive_worker_running": self._archive_thread.is_alive(), + "snapshot_worker_running": self._snapshot_thread.is_alive(), "interval_seconds": self.interval_seconds, + "archive_interval_seconds": self.archive_interval_seconds, + "archive_lag_seconds": self.archive_lag_seconds, "windows": list(SUMMARY_WINDOWS), "persisted": persisted, "active_windows": sorted(active), - "cadence_seconds": {str(window): self._refresh_interval(window) for window in SUMMARY_WINDOWS}, + "cadence_seconds": {str(window): self._refresh_interval(window) for window in sorted(set(active) | set(SUMMARY_WINDOWS))}, "refreshes": refreshes, "errors": errors, + "archive_runs": archive_runs, + "archive_moved_events": moved_events, + "archive_moved_throughput": moved_throughput, + "last_archive_run": last_archive_run, } - def _run(self) -> None: - # Do not scan 24h on process startup. The first browser/API request marks - # its selected range active and wakes this worker. - while not self.stop_event.is_set(): - self._wake.wait(timeout=min(5.0, float(self.interval_seconds))) - self._wake.clear() - if self.stop_event.is_set(): + def _run_archive(self) -> None: + while not self.stop_event.is_set() and not self._stopping.is_set(): + self._archive_wake.wait(timeout=float(self.archive_interval_seconds)) + self._archive_wake.clear() + if self.stop_event.is_set() or self._stopping.is_set(): break + backlog = self._archive_once() + if backlog: + self._archive_ready.clear() + elif not self._archive_ready.is_set(): + self._archive_ready.set() + self._snapshot_wake.set() + + now = time.monotonic() + if now - self._last_purge >= 60: + try: + self.history.purge_archive() + except Exception: + with self._lock: + self._errors += 1 + self._last_purge = now + + if backlog: + # Continue draining old installations immediately instead of + # sleeping while Redis still holds a large backlog. Do not + # publish a partial SQLite snapshot while migration is pending. + self._archive_wake.set() + + def _run_snapshots(self) -> None: + poll_seconds = float(min(max(self.interval_seconds, 5), 15)) + while not self.stop_event.is_set() and not self._stopping.is_set(): + self._snapshot_wake.wait(timeout=poll_seconds) + self._snapshot_wake.clear() + if self.stop_event.is_set() or self._stopping.is_set(): + break + # On startup/upgrades wait until the bounded archive worker has + # drained any legacy Redis backlog. Existing persisted snapshots + # remain available to the UI while this happens. + if not self._archive_ready.is_set(): + continue due = self._due_windows() if due: self.refresh_windows(due) + def _archive_once(self) -> bool: + cutoff_ms = int((time.time() - self.archive_lag_seconds) * 1000) + try: + result = self.history.archive_redis_to_store( + cutoff_ms, + batch_size=self.archive_batch_size, + max_batches=100, + ) + except RedisUnavailableError: + with self._lock: + self._errors += 1 + return False + except Exception: + with self._lock: + self._errors += 1 + return False + with self._lock: + self._archive_runs += 1 + self._archive_moved_events += int(result.get("events") or 0) + self._archive_moved_throughput += int(result.get("throughput_samples") or 0) + self._last_archive_run = time.time() + return int(result.get("batches") or 0) >= 100 + def _due_windows(self) -> list[int]: now_mono = time.monotonic() now_wall = time.time() @@ -175,10 +270,10 @@ class AnalyticsSnapshotCache: last_refresh = dict(self._last_refresh) persisted = { int(row["window_seconds"]): row.get("generated_at") - for row in self.history.snapshot_status(requests.keys()) + for row in self.store.traffic_snapshot_status().get("windows", []) } for window, requested_at in requests.items(): - if now_mono - requested_at > self._active_ttl_seconds: + if window not in SUMMARY_WINDOWS and now_mono - requested_at > self._active_ttl_seconds: continue cadence = self._refresh_interval(window) last = last_refresh.get(window, 0.0) @@ -199,7 +294,6 @@ class AnalyticsSnapshotCache: return max(base * 15, 900) def _overlay_current_throughput(self, payload: dict[str, Any]) -> None: - """Keep the 'now' rate fresh without rescanning the selected history window.""" try: sample = self.history.latest_throughput() except RedisUnavailableError: @@ -235,12 +329,8 @@ class AnalyticsSnapshotCache: except (TypeError, ValueError): return 0.0 - @staticmethod - def _normalise_window(value: int) -> int: - value = int(value) - if value in SUMMARY_WINDOWS: - return value - return min(SUMMARY_WINDOWS, key=lambda item: abs(item - value)) + def _normalise_window(self, value: int) -> int: + return min(max(int(value), 60), self.history.retention_hours * 3600) @staticmethod def _decorate(payload: dict[str, Any], source: str) -> dict[str, Any]: diff --git a/app/config.py b/app/config.py index c572f9f..8b5f05e 100644 --- a/app/config.py +++ b/app/config.py @@ -88,6 +88,9 @@ class Config: session_hours: int session_cookie_secure: bool analytics_snapshot_interval_seconds: int + traffic_archive_interval_seconds: int + traffic_archive_lag_seconds: int + traffic_archive_batch_size: int redis_url: str redis_managed: bool redis_data_dir: str @@ -194,15 +197,20 @@ class Config: analytics_snapshot_interval_seconds=max( 15, _int("ANALYTICS_SNAPSHOT_INTERVAL_SECONDS", 60) ), + traffic_archive_interval_seconds=max(1, _int("TRAFFIC_ARCHIVE_INTERVAL_SECONDS", 5)), + traffic_archive_lag_seconds=max(2, _int("TRAFFIC_ARCHIVE_LAG_SECONDS", 10)), + traffic_archive_batch_size=max(100, min(5000, _int("TRAFFIC_ARCHIVE_BATCH_SIZE", 1000))), redis_url=os.getenv("REDIS_URL", "redis://127.0.0.1:6379/0"), redis_managed=_bool("REDIS_MANAGED", True), redis_data_dir=os.getenv("REDIS_DATA_DIR", "/data/redis"), redis_port=_int("REDIS_PORT", 6379), - # Managed Redis is the sole traffic-history store. Do not evict by - # count/memory; retention time is the authoritative bound. - redis_maxmemory_mb=0, - redis_snapshot_seconds=_int("REDIS_SNAPSHOT_SECONDS", 1800), - redis_aof=_bool("REDIS_AOF", True), + # Redis is only a short-lived ingestion buffer. Keep a hard memory + # ceiling so a stalled archive worker cannot trigger host OOMK. + redis_maxmemory_mb=max(32, _int("REDIS_MAXMEMORY_MB", 128) or 128), + # Durable history lives in SQLite, so Redis persistence is optional + # and disabled by default to avoid RDB/AOF fork memory spikes. + redis_snapshot_seconds=max(0, _int("REDIS_SNAPSHOT_SECONDS", 0)), + redis_aof=_bool("REDIS_AOF", False), traffic_retention_hours=_int("TRAFFIC_RETENTION_HOURS", 24), traffic_max_events=0, traffic_memory_events=0, @@ -257,8 +265,12 @@ class Config: "admin_username": self.admin_username, "session_hours": self.session_hours, "analytics_snapshot_interval_seconds": self.analytics_snapshot_interval_seconds, + "traffic_archive_interval_seconds": self.traffic_archive_interval_seconds, + "traffic_archive_lag_seconds": self.traffic_archive_lag_seconds, + "traffic_archive_batch_size": self.traffic_archive_batch_size, "traffic_retention_hours": self.traffic_retention_hours, "redis_managed": self.redis_managed, + "redis_maxmemory_mb": self.redis_maxmemory_mb, "redis_snapshot_seconds": self.redis_snapshot_seconds, "redis_aof": self.redis_aof, "traffic_max_events": self.traffic_max_events, diff --git a/app/eve.py b/app/eve.py index 7ffb2d9..7fbeba9 100644 --- a/app/eve.py +++ b/app/eve.py @@ -113,7 +113,7 @@ class EVEWatcher(threading.Thread): key = f"alerts_filtered_{tuning.reason}" self.stats.inc(key) # A filtered alert is deliberately excluded from the dashboard and - # Redis traffic history. The original EVE record stays on disk. + # traffic ingest/archive path. The original EVE record stays on disk. return duplicate_id = self.store.find_recent_duplicate(event, self.dedup_window_seconds) diff --git a/app/live.py b/app/live.py index 677d2a9..7f23eb7 100644 --- a/app/live.py +++ b/app/live.py @@ -460,10 +460,11 @@ class RedisConnection: class TrafficHistory: - """Persistent traffic history backed by Redis. + """Traffic history with Redis ingest buffering and optional SQLite archive. - Production can require Redis and disable RAM fallback entirely. The optional - memory mode is retained only for development/unit tests. + Production requires Redis for the live ingest queue and drains committed + history to SQLite. The optional memory mode is retained only for + development/unit tests. """ REDIS_KEY = "suricata:traffic:v2" @@ -481,11 +482,18 @@ class TrafficHistory: *, require_redis: bool = False, allow_memory_fallback: bool = True, + archive_store: Any | None = None, ) -> None: self.retention_hours = max(1, int(retention_hours)) # 0 means no count cap. Time retention is the authoritative bound. self.max_events = max(0, int(max_events)) self.allow_memory_fallback = bool(allow_memory_fallback) + self.archive_store = archive_store + self._archived_events = 0 + self._archived_throughput = 0 + self._archive_batches = 0 + self._archive_errors = 0 + self._last_archive_at = 0.0 memory_capacity = max(0, int(memory_events)) if self.allow_memory_fallback else 0 self._memory: collections.deque[dict[str, Any]] = collections.deque(maxlen=memory_capacity) self._throughput_memory: collections.deque[dict[str, Any]] = collections.deque( @@ -508,7 +516,7 @@ class TrafficHistory: self._redis = None self._redis_error = str(exc) if require_redis and self._redis is None: - raise RedisUnavailableError(f"Redis traffic history is required: {self._redis_error}") + raise RedisUnavailableError(f"Redis traffic ingest buffer is required: {self._redis_error}") def add(self, event: dict[str, Any]) -> None: self.add_many([event]) @@ -588,19 +596,67 @@ class TrafficHistory: "app_proto": app_proto.strip().lower(), "direction": direction.strip().lower(), } + + combined: list[dict[str, Any]] = [] + seen: set[str] = set() + if self.archive_store is not None: + offset = 0 + page_size = min(2000, max(limit * 3, 250)) + while len(combined) < limit: + page = self.archive_store.traffic_event_page( + since_ms, + until_ms, + offset=offset, + limit=page_size, + event_type=filters["event_type"], + proto=filters["proto"], + app_proto=filters["app_proto"], + direction=filters["direction"], + text=filters["text"], + ) + if not page: + break + for item in page: + archive_key = str(item.pop("_archive_key", "")) + if not _matches_search(item, filters): + continue + key = _event_identity(item) + if key in seen: + continue + seen.add(key) + combined.append(item) + offset += len(page) + if len(page) < page_size: + break + redis = self._redis_or_retry() if redis is not None: remote = self._redis_search(redis, since_ms, until_ms, limit, filters) if remote is not None: - return remote - if not self.allow_memory_fallback: + for item in remote: + key = _event_identity(item) + if key in seen: + continue + seen.add(key) + combined.append(item) + elif self.archive_store is None and not self.allow_memory_fallback: raise RedisUnavailableError(self._redis_error or "Redis is unavailable") - with self._lock: - candidates = [ - item for item in reversed(self._memory) - if since_ms <= int(item.get("ts_ms") or 0) <= until_ms - ] - return [item for item in candidates if _matches_search(item, filters)][:limit] + + if self.allow_memory_fallback and self.archive_store is None: + with self._lock: + candidates = [ + item for item in reversed(self._memory) + if since_ms <= int(item.get("ts_ms") or 0) <= until_ms + ] + for item in candidates: + if _matches_search(item, filters): + key = _event_identity(item) + if key not in seen: + seen.add(key) + combined.append(item) + + combined.sort(key=lambda item: int(item.get("ts_ms") or 0), reverse=True) + return combined[:limit] def latest_throughput(self) -> dict[str, Any] | None: """Return only the newest persisted TZSP rate sample (constant-cost Redis read).""" @@ -623,8 +679,13 @@ class TrafficHistory: return None def throughput_analytics(self, window_seconds: int = 3600) -> dict[str, Any]: - """Build the speed/volume chart from the compact 1 Hz TZSP series only.""" + """Build the speed/volume chart without scanning long-lived Redis history.""" window_seconds = min(max(int(window_seconds), 60), self.retention_hours * 3600) + if self.archive_store is not None: + payload = self._archive_analytics_many((window_seconds,), include_events=False)[window_seconds] + payload["throughput_only"] = True + return payload + now_ms = int(time.time() * 1000) since_ms = now_ms - window_seconds * 1000 throughput = self._redis_throughput_candidates(since_ms, now_ms + 1000) @@ -644,6 +705,9 @@ class TrafficHistory: def analytics(self, window_seconds: int = 3600, sample_limit: int | None = None) -> dict[str, Any]: window_seconds = min(max(int(window_seconds), 60), self.retention_hours * 3600) + if self.archive_store is not None: + return self._archive_analytics_many((window_seconds,))[window_seconds] + now_ms = int(time.time() * 1000) since_ms = now_ms - window_seconds * 1000 limit = None if sample_limit is None else max(int(sample_limit), 1) @@ -674,6 +738,9 @@ class TrafficHistory: }) if not normalized: return {} + if self.archive_store is not None: + return self._archive_analytics_many(normalized) + now_ms = int(time.time() * 1000) max_window = max(normalized) oldest_ms = now_ms - max_window * 1000 @@ -706,6 +773,94 @@ class TrafficHistory: result[window] = payload return result + def _archive_analytics_many( + self, + windows: Iterable[int], + *, + include_events: bool = True, + ) -> dict[int, dict[str, Any]]: + if self.archive_store is None: + return {} + normalized = sorted({ + min(max(int(window), 60), self.retention_hours * 3600) for window in windows + }) + if not normalized: + return {} + now_ms = int(time.time() * 1000) + archive_status = self.archive_store.traffic_archive_status() + newest_archived_ms = max( + int(archive_status.get("newest_event_ms") or 0), + int(archive_status.get("newest_throughput_ms") or 0), + ) + # Freeze the upper bound for this calculation. The archive worker can + # keep appending newer rows through WAL without shifting OFFSET-based + # pages underneath the snapshot worker. + read_until_ms = min(now_ms + 1000, newest_archived_ms) if newest_archived_ms else now_ms + 1000 + result: dict[int, dict[str, Any]] = {} + # Deliberately build one window at a time. The worker therefore has a + # bounded Python memory footprint even when the SQLite archive contains + # millions of rows. High-cardinality endpoint/application aggregations + # are delegated to SQL below instead of retaining large sets/counters. + for window in normalized: + since_ms = now_ms - window * 1000 + accumulator = _AnalyticsAccumulator( + since_ms, + now_ms, + window, + track_high_cardinality=False, + ) + event_count = 0 + throughput_count = 0 + + if include_events: + offset = 0 + page_size = 2000 + while True: + page = self.archive_store.traffic_event_page( + since_ms, read_until_ms, offset=offset, limit=page_size + ) + if not page: + break + for item in page: + item.pop("_archive_key", None) + accumulator.add_event(item) + event_count += 1 + offset += len(page) + if len(page) < page_size: + break + + offset = 0 + page_size = 5000 + while True: + page = self.archive_store.traffic_throughput_page( + since_ms, read_until_ms, offset=offset, limit=page_size + ) + if not page: + break + for sample in page: + sample.pop("_archive_key", None) + accumulator.add_throughput(sample) + throughput_count += 1 + offset += len(page) + if len(page) < page_size: + break + + payload = accumulator.finish() + if include_events: + payload.update( + self.archive_store.traffic_dimension_summary( + since_ms, + read_until_ms, + limit=10, + ) + ) + payload["analytics_source"] = "sqlite-archive" + payload["analytics_complete"] = True + payload["retained_events_scanned"] = event_count + payload["throughput_samples_scanned"] = throughput_count + result[window] = payload + return result + def save_snapshot(self, window_seconds: int, payload: dict[str, Any]) -> None: window = int(window_seconds) stored = dict(payload) @@ -779,32 +934,129 @@ class TrafficHistory: raise RedisUnavailableError(str(exc)) from exc return local_count + def archive_redis_to_store( + self, + cutoff_ms: int, + *, + batch_size: int = 1000, + max_batches: int = 0, + ) -> dict[str, int]: + """Move old Redis queue entries into the disk-backed SQLite archive. + + Redis is only the ingestion buffer. A batch is removed from Redis only + after SQLite commits it, so retries after a crash are safe through the + archive tables' stable primary keys. + """ + if self.archive_store is None: + return {"events": 0, "throughput_samples": 0, "batches": 0} + redis = self._redis_or_retry() + if redis is None: + raise RedisUnavailableError(self._redis_error or "Redis is unavailable") + batch_size = max(50, min(int(batch_size), 5000)) + max_batches = max(0, int(max_batches)) + moved_events = 0 + moved_throughput = 0 + batches = 0 + + try: + streams = [ + [self.REDIS_KEY, _decode_redis_member, self.archive_store.archive_traffic_events, True], + [self.THROUGHPUT_KEY, _decode_throughput_member, self.archive_store.archive_traffic_throughput, True], + ] + # Alternate event and throughput batches so a large legacy EVE + # backlog cannot starve rate samples in Redis during migration. + while any(bool(stream[3]) for stream in streams) and (max_batches == 0 or batches < max_batches): + for stream in streams: + if not stream[3] or (max_batches and batches >= max_batches): + continue + key, decoder, writer, _active = stream + raw = redis.execute( + "ZRANGEBYSCORE", key, "-inf", int(cutoff_ms), + "LIMIT", 0, batch_size, + ) + if not raw: + stream[3] = False + continue + records: list[tuple[str, dict[str, Any]]] = [] + for member in raw: + payload = decoder(member) + if payload is None: + continue + digest = hashlib.blake2s(bytes(member), digest_size=16).hexdigest() + records.append((digest, payload)) + writer(records) + redis.execute("ZREM", key, *raw) + if key == self.REDIS_KEY: + moved_events += len(raw) + else: + moved_throughput += len(raw) + batches += 1 + if len(raw) < batch_size: + stream[3] = False + self._redis_error = "" + with self._lock: + self._archived_events += moved_events + self._archived_throughput += moved_throughput + self._archive_batches += batches + self._last_archive_at = time.time() + return { + "events": moved_events, + "throughput_samples": moved_throughput, + "batches": batches, + } + except RedisUnavailableError: + raise + except Exception as exc: + with self._lock: + self._archive_errors += 1 + if isinstance(exc, (RedisProtocolError, OSError, ConnectionError)): + self._mark_redis_down(exc) + raise RedisUnavailableError(str(exc)) from exc + raise + + def purge_archive(self) -> dict[str, int]: + if self.archive_store is None: + return {"events": 0, "throughput_samples": 0} + cutoff_ms = int((time.time() - self.retention_hours * 3600) * 1000) + return self.archive_store.purge_traffic_archive_before(cutoff_ms) + def clear(self) -> int: with self._lock: count = len(self._memory) self._memory.clear() self._throughput_memory.clear() self._snapshot_memory.clear() + archived_events = 0 + if self.archive_store is not None: + archived = self.archive_store.clear_traffic_archive() + archived_events = int(archived.get("events") or 0) redis = self._redis_or_retry() if redis is None: - if self.allow_memory_fallback: - return count + if self.allow_memory_fallback or self.archive_store is not None: + return count + archived_events raise RedisUnavailableError(self._redis_error or "Redis is unavailable") try: remote = _safe_int(redis.execute("ZCARD", self.REDIS_KEY)) keys = [self.REDIS_KEY, self.LEGACY_REDIS_KEY, self.THROUGHPUT_KEY] - keys.extend(f"{self.SNAPSHOT_PREFIX}{window}" for window in (900, 3600, 21600, 86400)) + keys.extend(f"{self.SNAPSHOT_PREFIX}{window}" for window in (900, 3600, 18000, 21600, 86400)) redis.execute("DEL", *keys) - return max(count, remote) + return count + archived_events + remote except Exception as exc: self._mark_redis_down(exc) - if not self.allow_memory_fallback: + if not self.allow_memory_fallback and self.archive_store is None: raise RedisUnavailableError(str(exc)) from exc - return count + return count + archived_events def status(self) -> dict[str, Any]: with self._lock: memory_count = len(self._memory) + archive_stats = { + "archived_events_total": self._archived_events, + "archived_throughput_total": self._archived_throughput, + "archive_batches": self._archive_batches, + "archive_errors": self._archive_errors, + "last_archive_at": self._last_archive_at, + } redis = self._redis_or_retry() remote_count = None throughput_count = None @@ -815,7 +1067,14 @@ class TrafficHistory: self._redis_error = "" except Exception as exc: self._mark_redis_down(exc) - if self._redis_url and not self.allow_memory_fallback: + if self.archive_store is not None: + try: + archive_stats.update(self.archive_store.traffic_archive_status()) + except Exception: + archive_stats["archive_errors"] = int(archive_stats.get("archive_errors") or 0) + 1 + if self._redis_url and self.archive_store is not None: + backend = "redis-buffer+sqlite" + elif self._redis_url and not self.allow_memory_fallback: backend = "redis" elif self._redis_url: backend = "redis+memory-dev" @@ -833,6 +1092,7 @@ class TrafficHistory: "memory_capacity": self._memory.maxlen if self.allow_memory_fallback else 0, "retention_hours": self.retention_hours, "max_events": self.max_events, + "archive": archive_stats if self.archive_store is not None else None, } def _redis_search( @@ -978,10 +1238,14 @@ class TrafficHistory: old_count = _safe_int(redis.execute("ZCARD", self.LEGACY_REDIS_KEY)) if new_count == 0 and old_count > 0: redis.execute("RENAME", self.LEGACY_REDIS_KEY, self.REDIS_KEY) - # Analytics semantics changed in 0.9.5 (TZSP volume + noise/app - # filtering). Remove cached v2 calculations so stale inflated values - # cannot survive an image upgrade. Raw Redis event history is kept. - redis.execute("DEL", *(f"{self.LEGACY_SNAPSHOT_PREFIX}{window}" for window in (900, 3600, 21600, 86400))) + # Dashboard snapshots are SQLite-backed now. Remove both generations + # of obsolete Redis snapshot keys during upgrade; the raw queue is + # preserved here and drained transactionally by the archive worker. + keys = [ + *(f"{self.LEGACY_SNAPSHOT_PREFIX}{window}" for window in (900, 3600, 21600, 86400)), + *(f"{self.SNAPSHOT_PREFIX}{window}" for window in (900, 3600, 18000, 21600, 86400)), + ] + redis.execute("DEL", *keys) except Exception: # Migration is best-effort; a missing legacy key is normal. pass @@ -995,6 +1259,14 @@ def _decode_redis_member(member: bytes) -> dict[str, Any] | None: return None +def _event_identity(item: dict[str, Any]) -> str: + event_id = _text(item.get("id"), 128) + if event_id: + return event_id + raw = json.dumps(item, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.blake2s(raw, digest_size=16).hexdigest() + + def _decode_throughput_member(member: bytes) -> dict[str, Any] | None: try: parts = member.split(b"|", 2) @@ -1078,114 +1350,130 @@ def _search_blob(item: dict[str, Any]) -> str: return " ".join(_text(item.get(name)).lower() for name in fields) -def _analytics( - events: Iterable[dict[str, Any]], - since_ms: int, - now_ms: int, - window_seconds: int, - throughput_samples: Iterable[dict[str, Any]] | None = None, -) -> dict[str, Any]: - event_list = list(events) - throughput_list = list(throughput_samples or []) - bins_count = 60 - bin_ms = max(1000, int(window_seconds * 1000 / bins_count)) - bins = [ - { - "ts_ms": since_ms + idx * bin_ms, - "events": 0, - "bytes": 0, - "bytes_in": 0, - "bytes_out": 0, - "packets": 0, - "alerts": 0, - } - for idx in range(bins_count) - ] - apps: collections.Counter[str] = collections.Counter() - protocols: collections.Counter[str] = collections.Counter() - sources: collections.Counter[str] = collections.Counter() - destinations: collections.Counter[str] = collections.Counter() - local_clients: collections.Counter[str] = collections.Counter() - remote_peers: collections.Counter[str] = collections.Counter() - local_client_bytes: collections.Counter[str] = collections.Counter() - remote_peer_bytes: collections.Counter[str] = collections.Counter() - app_bytes: collections.Counter[str] = collections.Counter() - directions: collections.Counter[str] = collections.Counter() - types: collections.Counter[str] = collections.Counter() - signatures: collections.Counter[str] = collections.Counter() - severities: collections.Counter[str] = collections.Counter() - fingerprints: collections.Counter[str] = collections.Counter() - assets: collections.Counter[str] = collections.Counter() - file_activity: collections.Counter[str] = collections.Counter() - included_events = 0 - eve_flow_bytes = 0 - app_flow_seen: set[tuple[str, str]] = set() - alerts = 0 - blocked = 0 - anomalies = 0 - dns_nxdomain = 0 - files = 0 - encrypted = 0 - cleartext = 0 +class _AnalyticsAccumulator: + """Streaming analytics builder used for both Redis/dev and SQLite archive reads.""" - for item in event_list: + def __init__( + self, + since_ms: int, + now_ms: int, + window_seconds: int, + *, + track_high_cardinality: bool = True, + ) -> None: + self.since_ms = int(since_ms) + self.now_ms = int(now_ms) + self.window_seconds = int(window_seconds) + self.track_high_cardinality = bool(track_high_cardinality) + self.bins_count = 60 + self.bin_ms = max(1000, int(self.window_seconds * 1000 / self.bins_count)) + self.bins = [ + { + "ts_ms": self.since_ms + idx * self.bin_ms, + "events": 0, + "bytes": 0, + "bytes_in": 0, + "bytes_out": 0, + "packets": 0, + "alerts": 0, + "rate_bytes": 0, + "rate_bytes_in": 0, + "rate_bytes_out": 0, + "rate_packets": 0, + } + for idx in range(self.bins_count) + ] + self.apps: collections.Counter[str] = collections.Counter() + self.protocols: collections.Counter[str] = collections.Counter() + self.sources: collections.Counter[str] = collections.Counter() + self.destinations: collections.Counter[str] = collections.Counter() + self.local_clients: collections.Counter[str] = collections.Counter() + self.remote_peers: collections.Counter[str] = collections.Counter() + self.local_client_bytes: collections.Counter[str] = collections.Counter() + self.remote_peer_bytes: collections.Counter[str] = collections.Counter() + self.app_bytes: collections.Counter[str] = collections.Counter() + self.directions: collections.Counter[str] = collections.Counter() + self.types: collections.Counter[str] = collections.Counter() + self.signatures: collections.Counter[str] = collections.Counter() + self.severities: collections.Counter[str] = collections.Counter() + self.fingerprints: collections.Counter[str] = collections.Counter() + self.assets: collections.Counter[str] = collections.Counter() + self.file_activity: collections.Counter[str] = collections.Counter() + self.app_flow_seen: set[tuple[str, str]] = set() + self.included_events = 0 + self.eve_flow_bytes = 0 + self.alerts = 0 + self.blocked = 0 + self.anomalies = 0 + self.dns_nxdomain = 0 + self.files = 0 + self.encrypted = 0 + self.cleartext = 0 + self.throughput_bytes = 0 + self.throughput_classified_bytes = 0 + self.throughput_packets = 0 + self.throughput_samples = 0 + self.latest_sample: dict[str, Any] | None = None + + def add_event(self, item: dict[str, Any]) -> None: ts = _safe_int(item.get("ts_ms")) - if ts < since_ms or ts > now_ms + 1000 or is_dashboard_noise(item): - continue - included_events += 1 - idx = min(max((ts - since_ms) // bin_ms, 0), bins_count - 1) + if ts < self.since_ms or ts > self.now_ms + 1000 or is_dashboard_noise(item): + return + self.included_events += 1 + idx = min(max((ts - self.since_ms) // self.bin_ms, 0), self.bins_count - 1) is_flow = _text(item.get("type"), 32).lower() == "flow" size = max(_safe_int(item.get("bytes")), 0) if is_flow else 0 bytes_in = max(_safe_int(item.get("bytes_in")), 0) if is_flow else 0 bytes_out = max(_safe_int(item.get("bytes_out")), 0) if is_flow else 0 packets = max(_safe_int(item.get("packets")), 0) if is_flow else 0 - bins[idx]["events"] += 1 - bins[idx]["bytes"] += size - bins[idx]["bytes_in"] += bytes_in - bins[idx]["bytes_out"] += bytes_out - bins[idx]["packets"] += packets + bucket = self.bins[idx] + bucket["events"] += 1 + bucket["bytes"] += size + bucket["bytes_in"] += bytes_in + bucket["bytes_out"] += bytes_out + bucket["packets"] += packets if item.get("type") == "alert": - bins[idx]["alerts"] += 1 - alerts += 1 + bucket["alerts"] += 1 + self.alerts += 1 signature = _text(item.get("signature"), 160) if signature: - signatures[signature] += 1 + self.signatures[signature] += 1 severity = item.get("severity") if severity not in (None, ""): - severities[f"S{severity}"] += 1 + self.severities[f"S{severity}"] += 1 if item.get("blocked"): - blocked += 1 + self.blocked += 1 if item.get("type") == "anomaly": - anomalies += 1 + self.anomalies += 1 if item.get("type") == "dns" and _text(item.get("dns_rcode"), 32).upper() == "NXDOMAIN": - dns_nxdomain += 1 + self.dns_nxdomain += 1 if item.get("type") == "fileinfo": - files += 1 + self.files += 1 filename = _text(item.get("filename"), 180) or "unnamed file" digest = _text(item.get("file_sha256") or item.get("file_sha1") or item.get("file_md5"), 32) - file_activity[f"{filename}{' · ' + digest if digest else ''}"] += 1 + self.file_activity[f"{filename}{' · ' + digest if digest else ''}"] += 1 direction = _text(item.get("direction"), 24) or "unknown" src_ip = _text(item.get("src_ip"), 64) dest_ip = _text(item.get("dest_ip"), 64) ether_src = _text(item.get("ether_src"), 32) ether_dest = _text(item.get("ether_dest"), 32) if direction in {"outbound", "internal"} and src_ip and ether_src: - assets[f"{src_ip} · {ether_src}"] += 1 + self.assets[f"{src_ip} · {ether_src}"] += 1 if direction in {"inbound", "internal"} and dest_ip and ether_dest: - assets[f"{dest_ip} · {ether_dest}"] += 1 + self.assets[f"{dest_ip} · {ether_dest}"] += 1 if item.get("type") == "dhcp": asset_ip = _text(item.get("dhcp_assigned_ip") or item.get("src_ip"), 64) identity = _text(item.get("dhcp_hostname") or item.get("dhcp_client_mac"), 160) if asset_ip or identity: - assets[f"{asset_ip}{' · ' if asset_ip and identity else ''}{identity}"] += 1 + self.assets[f"{asset_ip}{' · ' if asset_ip and identity else ''}{identity}"] += 1 elif item.get("type") == "arp": asset_ip = _text(item.get("arp_src_ip") or item.get("src_ip"), 64) mac = _text(item.get("arp_src_mac"), 32) if asset_ip or mac: - assets[f"{asset_ip}{' · ' if asset_ip and mac else ''}{mac}"] += 1 + self.assets[f"{asset_ip}{' · ' if asset_ip and mac else ''}{mac}"] += 1 app_proto = _valid_app_proto(item.get("app_proto")) if item.get("type") in {"tls", "quic", "ssh"} or app_proto in {"tls", "quic", "ssh"}: - encrypted += 1 + self.encrypted += 1 for label, key in ( ("JA4", "tls_ja4"), ("JA3", "tls_ja3"), @@ -1196,179 +1484,182 @@ def _analytics( ): value = _text(item.get(key), 160) if value: - fingerprints[f"{label} {value}"] += 1 + self.fingerprints[f"{label} {value}"] += 1 if item.get("type") in {"http", "ftp", "smtp"} or app_proto in {"http", "ftp", "smtp", "telnet"}: - cleartext += 1 + self.cleartext += 1 if is_flow: - eve_flow_bytes += size - if app_proto: + self.eve_flow_bytes += size + if app_proto and self.track_high_cardinality: flow_identity = _text(item.get("flow_id") or item.get("community_id") or item.get("id"), 128) app_key = (app_proto, flow_identity) - if app_key not in app_flow_seen: - app_flow_seen.add(app_key) - apps[app_proto] += 1 + if app_key not in self.app_flow_seen: + self.app_flow_seen.add(app_key) + self.apps[app_proto] += 1 if is_flow: - app_bytes[app_proto] += size + self.app_bytes[app_proto] += size if item.get("proto"): - protocols[_text(item.get("proto"), 24)] += 1 - if item.get("src_ip"): - sources[_text(item.get("src_ip"), 64)] += 1 - if item.get("dest_ip"): - destinations[_text(item.get("dest_ip"), 64)] += 1 + self.protocols[_text(item.get("proto"), 24)] += 1 + if self.track_high_cardinality: + if item.get("src_ip"): + self.sources[_text(item.get("src_ip"), 64)] += 1 + if item.get("dest_ip"): + self.destinations[_text(item.get("dest_ip"), 64)] += 1 + if direction == "outbound": + if src_ip: + self.local_clients[src_ip] += 1 + self.local_client_bytes[src_ip] += size + if dest_ip: + self.remote_peers[dest_ip] += 1 + self.remote_peer_bytes[dest_ip] += size + elif direction == "inbound": + if dest_ip: + self.local_clients[dest_ip] += 1 + self.local_client_bytes[dest_ip] += size + if src_ip: + self.remote_peers[src_ip] += 1 + self.remote_peer_bytes[src_ip] += size + elif direction == "internal": + if src_ip: + self.local_clients[src_ip] += 1 + self.local_client_bytes[src_ip] += size + if dest_ip and dest_ip != src_ip: + self.local_clients[dest_ip] += 1 + self.local_client_bytes[dest_ip] += size + else: + if src_ip: + self.remote_peers[src_ip] += 1 + self.remote_peer_bytes[src_ip] += size + if dest_ip and dest_ip != src_ip: + self.remote_peers[dest_ip] += 1 + self.remote_peer_bytes[dest_ip] += size + self.directions[direction] += 1 + self.types[_text(item.get("type"), 32)] += 1 - if direction == "outbound": - if src_ip: - local_clients[src_ip] += 1 - local_client_bytes[src_ip] += size - if dest_ip: - remote_peers[dest_ip] += 1 - remote_peer_bytes[dest_ip] += size - elif direction == "inbound": - if dest_ip: - local_clients[dest_ip] += 1 - local_client_bytes[dest_ip] += size - if src_ip: - remote_peers[src_ip] += 1 - remote_peer_bytes[src_ip] += size - elif direction == "internal": - if src_ip: - local_clients[src_ip] += 1 - local_client_bytes[src_ip] += size - if dest_ip and dest_ip != src_ip: - local_clients[dest_ip] += 1 - local_client_bytes[dest_ip] += size + def add_throughput(self, sample: dict[str, Any]) -> None: + ts = _safe_int(sample.get("ts_ms")) + if ts < self.since_ms or ts > self.now_ms + 1000: + return + idx = min(max((ts - self.since_ms) // self.bin_ms, 0), self.bins_count - 1) + bytes_in = max(_safe_int(sample.get("bytes_in")), 0) + bytes_out = max(_safe_int(sample.get("bytes_out")), 0) + bytes_total = max( + _safe_int(sample.get("bytes_total")), + bytes_in + bytes_out + max(_safe_int(sample.get("bytes_internal")), 0) + max(_safe_int(sample.get("bytes_external")), 0), + ) + packets_total = max(_safe_int(sample.get("packets_total")), 0) + bucket = self.bins[idx] + bucket["rate_bytes"] += bytes_total + bucket["rate_bytes_in"] += bytes_in + bucket["rate_bytes_out"] += bytes_out + bucket["rate_packets"] += packets_total + self.throughput_bytes += bytes_total + self.throughput_classified_bytes += bytes_in + bytes_out + self.throughput_packets += packets_total + self.throughput_samples += 1 + if self.latest_sample is None or ts > _safe_int(self.latest_sample.get("ts_ms")): + self.latest_sample = sample + + def finish(self) -> dict[str, Any]: + bucket_seconds = max(self.window_seconds / self.bins_count, 1) + has_throughput = self.throughput_samples > 0 + for bucket in self.bins: + if has_throughput: + bucket["bps"] = round(bucket.pop("rate_bytes") * 8 / bucket_seconds) + bucket["in_bps"] = round(bucket.pop("rate_bytes_in") * 8 / bucket_seconds) + bucket["out_bps"] = round(bucket.pop("rate_bytes_out") * 8 / bucket_seconds) + bucket["pps"] = round(bucket.pop("rate_packets") / bucket_seconds, 2) + else: + bucket.pop("rate_bytes", None) + bucket.pop("rate_bytes_in", None) + bucket.pop("rate_bytes_out", None) + bucket.pop("rate_packets", None) + bucket["bps"] = round(bucket["bytes"] * 8 / bucket_seconds) + bucket["in_bps"] = round(bucket["bytes_in"] * 8 / bucket_seconds) + bucket["out_bps"] = round(bucket["bytes_out"] * 8 / bucket_seconds) + bucket["pps"] = round(bucket["packets"] / bucket_seconds, 2) + bucket["other_bps"] = max(0, bucket["bps"] - bucket["in_bps"] - bucket["out_bps"]) + + latest_sample = self.latest_sample + if latest_sample is not None: + interval = max(float(latest_sample.get("interval_ms") or 1000) / 1000.0, 0.001) + sample_age_ms = max(0, self.now_ms - _safe_int(latest_sample.get("ts_ms"))) + if sample_age_ms > max(3000, round(interval * 3000)): + current_bps = current_in_bps = current_out_bps = current_pps = 0 + else: + current_bps = round(max(_safe_int(latest_sample.get("bytes_total")), 0) * 8 / interval) + current_in_bps = round(max(_safe_int(latest_sample.get("bytes_in")), 0) * 8 / interval) + current_out_bps = round(max(_safe_int(latest_sample.get("bytes_out")), 0) * 8 / interval) + current_pps = round(max(_safe_int(latest_sample.get("packets_total")), 0) / interval, 2) else: - if src_ip: - remote_peers[src_ip] += 1 - remote_peer_bytes[src_ip] += size - if dest_ip and dest_ip != src_ip: - remote_peers[dest_ip] += 1 - remote_peer_bytes[dest_ip] += size - directions[direction] += 1 - types[_text(item.get("type"), 32)] += 1 + current_bps = self.bins[-1]["bps"] if self.bins else 0 + current_in_bps = self.bins[-1]["in_bps"] if self.bins else 0 + current_out_bps = self.bins[-1]["out_bps"] if self.bins else 0 + current_pps = self.bins[-1]["pps"] if self.bins else 0 - # Raw TZSP throughput samples are the authoritative speed source. EVE flow - # bytes remain useful for traffic volume/application accounting, but their - # timestamps describe flow lifecycle events and are not an instantaneous rate. - throughput_bytes = 0 - throughput_classified_bytes = 0 - throughput_packets = 0 - latest_sample: dict[str, Any] | None = None - if throughput_list: - for bucket in bins: - bucket["rate_bytes"] = 0 - bucket["rate_bytes_in"] = 0 - bucket["rate_bytes_out"] = 0 - bucket["rate_packets"] = 0 - for sample in throughput_list: - ts = _safe_int(sample.get("ts_ms")) - if ts < since_ms or ts > now_ms + 1000: - continue - idx = min(max((ts - since_ms) // bin_ms, 0), bins_count - 1) - bytes_in = max(_safe_int(sample.get("bytes_in")), 0) - bytes_out = max(_safe_int(sample.get("bytes_out")), 0) - bytes_total = max( - _safe_int(sample.get("bytes_total")), - bytes_in + bytes_out + max(_safe_int(sample.get("bytes_internal")), 0) + max(_safe_int(sample.get("bytes_external")), 0), - ) - packets_total = max(_safe_int(sample.get("packets_total")), 0) - bins[idx]["rate_bytes"] += bytes_total - bins[idx]["rate_bytes_in"] += bytes_in - bins[idx]["rate_bytes_out"] += bytes_out - bins[idx]["rate_packets"] += packets_total - throughput_bytes += bytes_total - throughput_classified_bytes += bytes_in + bytes_out - throughput_packets += packets_total - if latest_sample is None or ts > _safe_int(latest_sample.get("ts_ms")): - latest_sample = sample + current_other_bps = max(0, current_bps - current_in_bps - current_out_bps) + direction_coverage_pct = round( + (self.throughput_classified_bytes / self.throughput_bytes) * 100.0, 1 + ) if self.throughput_bytes else 0.0 + observed_bytes = self.throughput_bytes if has_throughput else self.eve_flow_bytes + return { + "window_seconds": self.window_seconds, + "events": self.included_events, + "bytes": observed_bytes, + "eve_flow_bytes": self.eve_flow_bytes, + "throughput_bytes": self.throughput_bytes, + "throughput_packets": self.throughput_packets, + "current_bps": current_bps, + "current_in_bps": current_in_bps, + "current_out_bps": current_out_bps, + "current_other_bps": current_other_bps, + "throughput_direction_coverage_pct": direction_coverage_pct, + "current_pps": current_pps, + "avg_bps": round(observed_bytes * 8 / max(self.window_seconds, 1)), + "peak_bps": max((bucket["bps"] for bucket in self.bins), default=0), + "peak_in_bps": max((bucket["in_bps"] for bucket in self.bins), default=0), + "peak_out_bps": max((bucket["out_bps"] for bucket in self.bins), default=0), + "alerts": self.alerts, + "blocked": self.blocked, + "anomalies": self.anomalies, + "dns_nxdomain": self.dns_nxdomain, + "files": self.files, + "encrypted_sessions": self.encrypted, + "cleartext_sessions": self.cleartext, + "unique_local_clients": len(self.local_clients), + "unique_remote_peers": len(self.remote_peers), + "timeline": self.bins, + "top_apps": _counter_rows(self.apps), + "protocols": _counter_rows(self.protocols), + "top_sources": _counter_rows(self.sources), + "top_destinations": _counter_rows(self.destinations), + "top_local_clients": _counter_rows(self.local_clients), + "top_remote_peers": _counter_rows(self.remote_peers), + "top_local_clients_by_bytes": _counter_rows_metric(self.local_client_bytes, "bytes"), + "top_remote_peers_by_bytes": _counter_rows_metric(self.remote_peer_bytes, "bytes"), + "top_apps_by_bytes": _counter_rows_metric(self.app_bytes, "bytes"), + "directions": _counter_rows(self.directions), + "event_types": _counter_rows(self.types), + "top_signatures": _counter_rows(self.signatures), + "severities": _counter_rows(self.severities), + "top_fingerprints": _counter_rows(self.fingerprints), + "top_assets": _counter_rows(self.assets), + "top_files": _counter_rows(self.file_activity), + } - bucket_seconds = max(window_seconds / bins_count, 1) - for bucket in bins: - if throughput_list: - bucket["bps"] = round(bucket.pop("rate_bytes") * 8 / bucket_seconds) - bucket["in_bps"] = round(bucket.pop("rate_bytes_in") * 8 / bucket_seconds) - bucket["out_bps"] = round(bucket.pop("rate_bytes_out") * 8 / bucket_seconds) - bucket["pps"] = round(bucket.pop("rate_packets") / bucket_seconds, 2) - else: - bucket["bps"] = round(bucket["bytes"] * 8 / bucket_seconds) - bucket["in_bps"] = round(bucket["bytes_in"] * 8 / bucket_seconds) - bucket["out_bps"] = round(bucket["bytes_out"] * 8 / bucket_seconds) - bucket["pps"] = round(bucket["packets"] / bucket_seconds, 2) - bucket["other_bps"] = max(0, bucket["bps"] - bucket["in_bps"] - bucket["out_bps"]) - if latest_sample is not None: - interval = max(float(latest_sample.get("interval_ms") or 1000) / 1000.0, 0.001) - sample_age_ms = max(0, now_ms - _safe_int(latest_sample.get("ts_ms"))) - # Do not display a stale non-zero "current" rate after traffic stops. - # Three sample intervals (minimum 3 s) gives the writer enough jitter - # tolerance while still returning the live metric to zero quickly. - if sample_age_ms > max(3000, round(interval * 3000)): - current_bps = 0 - current_in_bps = 0 - current_out_bps = 0 - current_pps = 0 - else: - current_bps = round(max(_safe_int(latest_sample.get("bytes_total")), 0) * 8 / interval) - current_in_bps = round(max(_safe_int(latest_sample.get("bytes_in")), 0) * 8 / interval) - current_out_bps = round(max(_safe_int(latest_sample.get("bytes_out")), 0) * 8 / interval) - current_pps = round(max(_safe_int(latest_sample.get("packets_total")), 0) / interval, 2) - else: - current_bps = bins[-1]["bps"] if bins else 0 - current_in_bps = bins[-1]["in_bps"] if bins else 0 - current_out_bps = bins[-1]["out_bps"] if bins else 0 - current_pps = bins[-1]["pps"] if bins else 0 - - current_other_bps = max(0, current_bps - current_in_bps - current_out_bps) - direction_coverage_pct = round( - (throughput_classified_bytes / throughput_bytes) * 100.0, 1 - ) if throughput_bytes else 0.0 - - observed_bytes = throughput_bytes if throughput_list else eve_flow_bytes - - return { - "window_seconds": window_seconds, - "events": included_events, - "bytes": observed_bytes, - "eve_flow_bytes": eve_flow_bytes, - "throughput_bytes": throughput_bytes, - "throughput_packets": throughput_packets, - "current_bps": current_bps, - "current_in_bps": current_in_bps, - "current_out_bps": current_out_bps, - "current_other_bps": current_other_bps, - "throughput_direction_coverage_pct": direction_coverage_pct, - "current_pps": current_pps, - "avg_bps": round(observed_bytes * 8 / max(window_seconds, 1)), - "peak_bps": max((bucket["bps"] for bucket in bins), default=0), - "peak_in_bps": max((bucket["in_bps"] for bucket in bins), default=0), - "peak_out_bps": max((bucket["out_bps"] for bucket in bins), default=0), - "alerts": alerts, - "blocked": blocked, - "anomalies": anomalies, - "dns_nxdomain": dns_nxdomain, - "files": files, - "encrypted_sessions": encrypted, - "cleartext_sessions": cleartext, - "unique_local_clients": len(local_clients), - "unique_remote_peers": len(remote_peers), - "timeline": bins, - "top_apps": _counter_rows(apps), - "protocols": _counter_rows(protocols), - "top_sources": _counter_rows(sources), - "top_destinations": _counter_rows(destinations), - "top_local_clients": _counter_rows(local_clients), - "top_remote_peers": _counter_rows(remote_peers), - "top_local_clients_by_bytes": _counter_rows_metric(local_client_bytes, "bytes"), - "top_remote_peers_by_bytes": _counter_rows_metric(remote_peer_bytes, "bytes"), - "top_apps_by_bytes": _counter_rows_metric(app_bytes, "bytes"), - "directions": _counter_rows(directions), - "event_types": _counter_rows(types), - "top_signatures": _counter_rows(signatures), - "severities": _counter_rows(severities), - "top_fingerprints": _counter_rows(fingerprints), - "top_assets": _counter_rows(assets), - "top_files": _counter_rows(file_activity), - } +def _analytics( + events: Iterable[dict[str, Any]], + since_ms: int, + now_ms: int, + window_seconds: int, + throughput_samples: Iterable[dict[str, Any]] | None = None, +) -> dict[str, Any]: + accumulator = _AnalyticsAccumulator(since_ms, now_ms, window_seconds) + for item in events: + accumulator.add_event(item) + for sample in throughput_samples or (): + accumulator.add_throughput(sample) + return accumulator.finish() def _counter_rows(counter: collections.Counter[str], limit: int = 10) -> list[dict[str, Any]]: @@ -1381,7 +1672,7 @@ def _counter_rows_metric( return [{"name": name, key: value} for name, value in counter.most_common(limit)] class LiveEventPipeline: - """Immediate WebSocket fan-out plus asynchronous Redis persistence.""" + """Immediate WebSocket fan-out plus asynchronous Redis-buffer persistence.""" def __init__(self, bus: EventBus, history: TrafficHistory, queue_size: int = 10000) -> None: self.bus = bus diff --git a/app/main.py b/app/main.py index d06c1bc..6f67eb0 100644 --- a/app/main.py +++ b/app/main.py @@ -244,12 +244,16 @@ def main() -> int: 0, require_redis=True, allow_memory_fallback=False, + archive_store=store, ) analytics_cache = AnalyticsSnapshotCache( store, traffic_history, stop_event, cfg.analytics_snapshot_interval_seconds, + archive_interval_seconds=cfg.traffic_archive_interval_seconds, + archive_lag_seconds=cfg.traffic_archive_lag_seconds, + archive_batch_size=cfg.traffic_archive_batch_size, ) live_pipeline = LiveEventPipeline(event_bus, traffic_history) normalizer = TrafficNormalizer(cfg.monitored_networks) @@ -384,12 +388,12 @@ def main() -> int: "traffic_history": { "name": "Live traffic history", "status": "up" if traffic_history.status().get("redis_ok") else "degraded", - "details": f"Redis-only persistent history; retention={cfg.traffic_retention_hours}h; no event-count cap", + "details": f"Redis ingest buffer -> SQLite archive; retention={cfg.traffic_retention_hours}h", }, "analytics_cache": { "name": "Persistent dashboard summaries", "status": "up", - "details": f"Redis snapshots for 15m/1h/6h/24h every {cfg.analytics_snapshot_interval_seconds}s", + "details": f"SQLite snapshots for 15m/1h/5h/6h/24h; Redis is not scanned for charts", }, "ndr": { "name": "MikroSuricata NDR correlation", @@ -409,7 +413,7 @@ def main() -> int: else "degraded" ), "details": ( - f"{cfg.redis_data_dir}; maxmemory=unlimited; persistence={redis_status.get('persistence')}" + f"{cfg.redis_data_dir}; maxmemory={cfg.redis_maxmemory_mb} MB; persistence={redis_status.get('persistence')}" if cfg.redis_managed else "Managed Redis disabled; REDIS_URL may point to an external server" ), diff --git a/app/redis_service.py b/app/redis_service.py index 878f0f9..766bcbd 100644 --- a/app/redis_service.py +++ b/app/redis_service.py @@ -12,24 +12,25 @@ from typing import Any class RedisSupervisor: - """Run the persistent Redis history service inside the IDS container.""" + """Run the bounded Redis ingestion buffer inside the IDS container.""" def __init__( self, enabled: bool, data_dir: str, port: int = 6379, - maxmemory_mb: int = 0, - snapshot_seconds: int = 1800, - aof: bool = True, + maxmemory_mb: int = 128, + snapshot_seconds: int = 0, + aof: bool = False, ) -> None: self.enabled = bool(enabled) self.data_dir = data_dir self.port = int(port) - # 0 means unlimited. Traffic retention is time-based; Redis must not evict - # arbitrary history just because an old deployment exported a memory cap. + # Redis is an ingestion buffer. A hard ceiling prevents host OOMK if the + # SQLite archive worker stalls; noeviction makes overload explicit instead + # of silently discarding arbitrary history. self.maxmemory_mb = max(0, int(maxmemory_mb)) - self.snapshot_seconds = max(300, int(snapshot_seconds)) + self.snapshot_seconds = max(0, int(snapshot_seconds)) self.aof = bool(aof) self.executable = shutil.which("redis-server") self._lock = threading.RLock() @@ -108,7 +109,12 @@ class RedisSupervisor: "maxmemory_mb": self.maxmemory_mb, "snapshot_seconds": self.snapshot_seconds, "aof": self.aof, - "persistence": "AOF everysec + RDB" if self.aof else "RDB", + "persistence": ( + "AOF everysec + RDB" if self.aof and self.snapshot_seconds > 0 + else "AOF everysec" if self.aof + else "RDB" if self.snapshot_seconds > 0 + else "disabled (SQLite archive is durable)" + ), "last_error": self._last_error, } @@ -146,7 +152,12 @@ class RedisSupervisor: "--bind", "127.0.0.1", "--protected-mode", "yes", "--port", str(self.port), - "--save", str(self.snapshot_seconds), "100", + ] + if self.snapshot_seconds > 0: + cmd.extend(["--save", str(self.snapshot_seconds), "100"]) + else: + cmd.extend(["--save", ""]) + cmd.extend([ "--appendonly", "yes" if self.aof else "no", "--appendfsync", "everysec", "--aof-use-rdb-preamble", "yes", @@ -154,7 +165,7 @@ class RedisSupervisor: "--dbfilename", "traffic.rdb", "--maxmemory-policy", "noeviction", "--loglevel", "warning", - ] + ]) if self.maxmemory_mb > 0: cmd.extend(["--maxmemory", f"{self.maxmemory_mb}mb"]) else: diff --git a/app/static/js/app.js b/app/static/js/app.js index bcdc8a4..2a48dc6 100644 --- a/app/static/js/app.js +++ b/app/static/js/app.js @@ -8,7 +8,7 @@ reports:'/reports', feeds:'/feeds', rules:'/rules', system:'/system' }; const PATH_VIEWS = Object.fromEntries(Object.entries(VIEW_PATHS).map(([view,path])=>[path,view])); - const WINDOW_LABELS = {900:'Last 15 minutes',3600:'Last 1 hour',21600:'Last 6 hours',86400:'Last 24 hours'}; + const WINDOW_LABELS = {900:'Last 15 minutes',3600:'Last 1 hour',18000:'Last 5 hours',21600:'Last 6 hours',86400:'Last 24 hours'}; const state = { view: 'overview', ws: null, reconnectTimer: null, reconnectDelay: 1000, liveEnabled: false, paused: false, live: [], liveById: new Map(), liveSequence: 0, @@ -467,7 +467,7 @@ $('coverageStatus').innerHTML=coverage.map(([k,v])=>`
${esc(k)}${Number(v).toLocaleString()}
`).join(''); const age=Number(a.snapshot_age_seconds||0), source=a.snapshot_source||'live', stale=Boolean(a.snapshot_stale), refreshing=Boolean(a.snapshot_refreshing); const ageText=age<60?Math.round(age)+'s':age<3600?Math.round(age/60)+'m':Math.round(age/3600)+'h'; const completeness=a.analytics_complete===false?'fallback':`all ${Number(a.retained_events_scanned??a.events??0).toLocaleString()} retained`; - $('snapshotMeta').textContent=source==='redis-cache'?`${refreshing?'refreshing':'Redis cached'} · ${ageText} · ${completeness}`:`Redis · ${completeness}`; + $('snapshotMeta').textContent=source==='sqlite-snapshot'?`${refreshing?'refreshing':'SQLite cached'} · ${ageText} · ${completeness}`:`${source} · ${completeness}`; $('snapshotMeta').className=`status-chip ${a.analytics_complete===false||stale?'warn':'ok'}`; updateReportWindowState(a); scheduleChartRender(); diff --git a/app/store.py b/app/store.py index e2a49a2..23a9cf5 100644 --- a/app/store.py +++ b/app/store.py @@ -12,15 +12,30 @@ from .mitre import classify as classify_mitre, merge as merge_mitre class AlertStore: - SCHEMA_VERSION = 11 + SCHEMA_VERSION = 12 def __init__(self, path: str) -> None: self.path = path os.makedirs(os.path.dirname(path) or ".", exist_ok=True) self._lock = threading.RLock() + self._analytics_lock = threading.RLock() self._conn = sqlite3.connect(path, check_same_thread=False) self._conn.row_factory = sqlite3.Row self._init_schema() + # WAL permits a dedicated read connection to run long dashboard + # aggregates while the archive worker keeps committing new batches. + # This avoids coupling Redis draining to a 24h GROUP BY query. + self._analytics_conn: sqlite3.Connection | None = None + if path != ":memory:": + self._analytics_conn = sqlite3.connect(path, check_same_thread=False) + self._analytics_conn.row_factory = sqlite3.Row + self._analytics_conn.executescript( + """ + PRAGMA query_only=ON; + PRAGMA temp_store=FILE; + PRAGMA busy_timeout=5000; + """ + ) def _init_schema(self) -> None: with self._lock: @@ -30,6 +45,8 @@ class AlertStore: PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA foreign_keys=ON; + PRAGMA temp_store=FILE; + PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, @@ -59,6 +76,24 @@ class AlertStore: generated_at TEXT NOT NULL, payload_json TEXT NOT NULL ); + CREATE TABLE IF NOT EXISTS traffic_events ( + event_key TEXT PRIMARY KEY, + ts_ms INTEGER NOT NULL, + event_type TEXT NOT NULL DEFAULT '', + proto TEXT NOT NULL DEFAULT '', + app_proto TEXT NOT NULL DEFAULT '', + direction TEXT NOT NULL DEFAULT '', + flow_id TEXT NOT NULL DEFAULT '', + src_ip TEXT NOT NULL DEFAULT '', + dest_ip TEXT NOT NULL DEFAULT '', + flow_bytes INTEGER NOT NULL DEFAULT 0, + payload_json TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS traffic_throughput ( + sample_key TEXT PRIMARY KEY, + ts_ms INTEGER NOT NULL, + payload_json TEXT NOT NULL + ); CREATE TABLE IF NOT EXISTS web_sessions ( token_hash TEXT PRIMARY KEY, username TEXT NOT NULL, @@ -164,9 +199,19 @@ class AlertStore: # Existing 0.3.x databases do not have last_seen/hit_count. Add # columns before creating indexes that reference the new schema. self._migrate_columns() + self._migrate_traffic_archive_columns() self._conn.executescript( """ CREATE INDEX IF NOT EXISTS idx_alerts_timestamp ON alerts(timestamp DESC); + CREATE INDEX IF NOT EXISTS idx_traffic_events_ts ON traffic_events(ts_ms DESC); + CREATE INDEX IF NOT EXISTS idx_traffic_events_type_ts ON traffic_events(event_type, ts_ms DESC); + CREATE INDEX IF NOT EXISTS idx_traffic_events_proto_ts ON traffic_events(proto, ts_ms DESC); + CREATE INDEX IF NOT EXISTS idx_traffic_events_app_ts ON traffic_events(app_proto, ts_ms DESC); + CREATE INDEX IF NOT EXISTS idx_traffic_events_direction_ts ON traffic_events(direction, ts_ms DESC); + CREATE INDEX IF NOT EXISTS idx_traffic_events_src_ts ON traffic_events(src_ip, ts_ms DESC); + CREATE INDEX IF NOT EXISTS idx_traffic_events_dest_ts ON traffic_events(dest_ip, ts_ms DESC); + CREATE INDEX IF NOT EXISTS idx_traffic_events_flow_ts ON traffic_events(flow_id, ts_ms DESC); + CREATE INDEX IF NOT EXISTS idx_traffic_throughput_ts ON traffic_throughput(ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_alerts_last_seen ON alerts(last_seen DESC); CREATE INDEX IF NOT EXISTS idx_alerts_signature_id ON alerts(signature_id); CREATE INDEX IF NOT EXISTS idx_alerts_blocked ON alerts(blocked); @@ -198,6 +243,21 @@ class AlertStore: self._conn.execute(f"PRAGMA user_version={self.SCHEMA_VERSION}") self._conn.commit() + def _migrate_traffic_archive_columns(self) -> None: + columns = { + str(row["name"]) + for row in self._conn.execute("PRAGMA table_info(traffic_events)").fetchall() + } + additions = { + "flow_id": "TEXT NOT NULL DEFAULT ''", + "src_ip": "TEXT NOT NULL DEFAULT ''", + "dest_ip": "TEXT NOT NULL DEFAULT ''", + "flow_bytes": "INTEGER NOT NULL DEFAULT 0", + } + for name, ddl in additions.items(): + if name not in columns: + self._conn.execute(f"ALTER TABLE traffic_events ADD COLUMN {name} {ddl}") + def save_traffic_snapshot(self, window_seconds: int, payload: dict[str, Any]) -> None: window_seconds = int(window_seconds) if window_seconds <= 0: @@ -257,6 +317,309 @@ class AlertStore: self._conn.commit() return count + def archive_traffic_events(self, records: list[tuple[str, dict[str, Any]]]) -> int: + if not records: + return 0 + inserted = 0 + with self._lock: + for event_key, payload in records: + if ( + str(payload.get("type") or "").strip().lower() == "alert" + and str(payload.get("signature") or "").strip().casefold() + in {"suricata ipv4 truncated packet", "suricata ipv6 truncated packet"} + ): + # These decoder alerts have always been excluded from the + # dashboard. Do not spend disk or analytics work on them. + continue + raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + cursor = self._conn.execute( + """ + INSERT OR IGNORE INTO traffic_events( + event_key, ts_ms, event_type, proto, app_proto, direction, + flow_id, src_ip, dest_ip, flow_bytes, payload_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + str(event_key), + int(payload.get("ts_ms") or 0), + str(payload.get("type") or "")[:32].lower(), + str(payload.get("proto") or "")[:24].lower(), + str(payload.get("app_proto") or "")[:48].lower(), + str(payload.get("direction") or "")[:24].lower(), + str(payload.get("flow_id") or payload.get("community_id") or "")[:128], + str(payload.get("src_ip") or "")[:64], + str(payload.get("dest_ip") or "")[:64], + max(0, int(payload.get("bytes") or 0)) if str(payload.get("type") or "").lower() == "flow" else 0, + raw, + ), + ) + inserted += max(0, int(cursor.rowcount or 0)) + self._conn.commit() + return inserted + + def archive_traffic_throughput(self, records: list[tuple[str, dict[str, Any]]]) -> int: + if not records: + return 0 + inserted = 0 + with self._lock: + for sample_key, payload in records: + raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + cursor = self._conn.execute( + "INSERT OR IGNORE INTO traffic_throughput(sample_key, ts_ms, payload_json) VALUES (?, ?, ?)", + (str(sample_key), int(payload.get("ts_ms") or 0), raw), + ) + inserted += max(0, int(cursor.rowcount or 0)) + self._conn.commit() + return inserted + + def traffic_event_page( + self, + since_ms: int, + until_ms: int, + *, + offset: int = 0, + limit: int = 2000, + event_type: str = "", + proto: str = "", + app_proto: str = "", + direction: str = "", + text: str = "", + ) -> list[dict[str, Any]]: + clauses = ["ts_ms>=?", "ts_ms<=?"] + params: list[Any] = [int(since_ms), int(until_ms)] + for column, value in ( + ("event_type", event_type), + ("proto", proto), + ("app_proto", app_proto), + ("direction", direction), + ): + value = str(value or "").strip().lower() + if value: + clauses.append(f"{column}=?") + params.append(value) + text = str(text or "").strip().lower() + if text: + clauses.append("LOWER(payload_json) LIKE ?") + params.append(f"%{text}%") + params.extend((max(1, min(int(limit), 5000)), max(0, int(offset)))) + sql = ( + "SELECT event_key, payload_json FROM traffic_events WHERE " + + " AND ".join(clauses) + + " ORDER BY ts_ms DESC, event_key DESC LIMIT ? OFFSET ?" + ) + conn = self._analytics_conn or self._conn + lock = self._analytics_lock if self._analytics_conn is not None else self._lock + with lock: + rows = conn.execute(sql, params).fetchall() + result: list[dict[str, Any]] = [] + for row in rows: + try: + payload = json.loads(str(row["payload_json"])) + except (TypeError, ValueError, json.JSONDecodeError): + continue + if isinstance(payload, dict): + payload["_archive_key"] = str(row["event_key"]) + result.append(payload) + return result + + def traffic_throughput_page( + self, + since_ms: int, + until_ms: int, + *, + offset: int = 0, + limit: int = 5000, + ) -> list[dict[str, Any]]: + conn = self._analytics_conn or self._conn + lock = self._analytics_lock if self._analytics_conn is not None else self._lock + with lock: + rows = conn.execute( + """ + SELECT sample_key, payload_json FROM traffic_throughput + WHERE ts_ms>=? AND ts_ms<=? + ORDER BY ts_ms DESC, sample_key DESC LIMIT ? OFFSET ? + """, + (int(since_ms), int(until_ms), max(1, min(int(limit), 10000)), max(0, int(offset))), + ).fetchall() + result: list[dict[str, Any]] = [] + for row in rows: + try: + payload = json.loads(str(row["payload_json"])) + except (TypeError, ValueError, json.JSONDecodeError): + continue + if isinstance(payload, dict): + payload["_archive_key"] = str(row["sample_key"]) + result.append(payload) + return result + + def traffic_dimension_summary(self, since_ms: int, until_ms: int, limit: int = 10) -> dict[str, Any]: + """High-cardinality dashboard dimensions calculated inside SQLite. + + Keeping endpoint/flow DISTINCT work in SQLite prevents the Python worker + from retaining millions of flow IDs or remote peers in RAM for 24h views. + """ + since_ms = int(since_ms) + until_ms = int(until_ms) + limit = max(1, min(int(limit), 25)) + base = (since_ms, until_ms) + ignored_apps = ("", "failed", "unknown", "none", "null", "notset") + conn = self._analytics_conn or self._conn + lock = self._analytics_lock if self._analytics_conn is not None else self._lock + + def rows(sql: str, params: tuple[Any, ...]) -> list[dict[str, Any]]: + with lock: + return [dict(row) for row in conn.execute(sql, params).fetchall()] + + def scalar(sql: str, params: tuple[Any, ...]) -> int: + with lock: + row = conn.execute(sql, params).fetchone() + return int(row[0] or 0) if row is not None else 0 + + local_cte = """ + WITH endpoints AS ( + SELECT src_ip AS name, flow_bytes AS bytes + FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND direction IN ('outbound','internal') AND src_ip<>'' + UNION ALL + SELECT dest_ip AS name, flow_bytes AS bytes + FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND direction IN ('inbound','internal') AND dest_ip<>'' + AND (direction<>'internal' OR dest_ip<>src_ip) + ) + """ + remote_cte = """ + WITH endpoints AS ( + SELECT dest_ip AS name, flow_bytes AS bytes + FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND direction='outbound' AND dest_ip<>'' + UNION ALL + SELECT src_ip AS name, flow_bytes AS bytes + FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND direction='inbound' AND src_ip<>'' + UNION ALL + SELECT src_ip AS name, flow_bytes AS bytes + FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND direction NOT IN ('outbound','inbound','internal') AND src_ip<>'' + UNION ALL + SELECT dest_ip AS name, flow_bytes AS bytes + FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND direction NOT IN ('outbound','inbound','internal') AND dest_ip<>'' AND dest_ip<>src_ip + ) + """ + top_sources = rows( + """ + SELECT src_ip AS name, COUNT(*) AS count FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND src_ip<>'' + GROUP BY src_ip ORDER BY count DESC, name LIMIT ? + """, + base + (limit,), + ) + top_destinations = rows( + """ + SELECT dest_ip AS name, COUNT(*) AS count FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND dest_ip<>'' + GROUP BY dest_ip ORDER BY count DESC, name LIMIT ? + """, + base + (limit,), + ) + app_params = base + ignored_apps + (limit,) + top_apps = rows( + """ + SELECT app_proto AS name, + COUNT(DISTINCT CASE WHEN flow_id<>'' THEN flow_id ELSE event_key END) AS count + FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND app_proto NOT IN (?, ?, ?, ?, ?, ?) + GROUP BY app_proto ORDER BY count DESC, name LIMIT ? + """, + app_params, + ) + top_apps_by_bytes = rows( + """ + SELECT app_proto AS name, COALESCE(SUM(flow_bytes),0) AS bytes + FROM traffic_events + WHERE ts_ms>=? AND ts_ms<=? AND app_proto NOT IN (?, ?, ?, ?, ?, ?) + GROUP BY app_proto ORDER BY bytes DESC, name LIMIT ? + """, + app_params, + ) + + local_params = base + base + top_local = rows( + local_cte + " SELECT name, COUNT(*) AS count FROM endpoints GROUP BY name ORDER BY count DESC, name LIMIT ?", + local_params + (limit,), + ) + local_bytes = rows( + local_cte + " SELECT name, COALESCE(SUM(bytes),0) AS bytes FROM endpoints GROUP BY name ORDER BY bytes DESC, name LIMIT ?", + local_params + (limit,), + ) + unique_local = scalar( + local_cte + " SELECT COUNT(DISTINCT name) FROM endpoints", + local_params, + ) + + remote_params = base + base + base + base + top_remote = rows( + remote_cte + " SELECT name, COUNT(*) AS count FROM endpoints GROUP BY name ORDER BY count DESC, name LIMIT ?", + remote_params + (limit,), + ) + remote_bytes = rows( + remote_cte + " SELECT name, COALESCE(SUM(bytes),0) AS bytes FROM endpoints GROUP BY name ORDER BY bytes DESC, name LIMIT ?", + remote_params + (limit,), + ) + unique_remote = scalar( + remote_cte + " SELECT COUNT(DISTINCT name) FROM endpoints", + remote_params, + ) + + return { + "top_apps": top_apps, + "top_apps_by_bytes": top_apps_by_bytes, + "top_sources": top_sources, + "top_destinations": top_destinations, + "top_local_clients": top_local, + "top_remote_peers": top_remote, + "top_local_clients_by_bytes": local_bytes, + "top_remote_peers_by_bytes": remote_bytes, + "unique_local_clients": unique_local, + "unique_remote_peers": unique_remote, + } + + def traffic_archive_status(self) -> dict[str, Any]: + with self._lock: + events = self._conn.execute( + "SELECT COUNT(*) AS count, MIN(ts_ms) AS oldest, MAX(ts_ms) AS newest FROM traffic_events" + ).fetchone() + throughput = self._conn.execute( + "SELECT COUNT(*) AS count, MIN(ts_ms) AS oldest, MAX(ts_ms) AS newest FROM traffic_throughput" + ).fetchone() + return { + "events": int(events["count"] or 0), + "throughput_samples": int(throughput["count"] or 0), + "oldest_event_ms": int(events["oldest"] or 0), + "newest_event_ms": int(events["newest"] or 0), + "oldest_throughput_ms": int(throughput["oldest"] or 0), + "newest_throughput_ms": int(throughput["newest"] or 0), + } + + def purge_traffic_archive_before(self, cutoff_ms: int) -> dict[str, int]: + with self._lock: + events = self._conn.execute("DELETE FROM traffic_events WHERE ts_ms dict[str, int]: + with self._lock: + events = int(self._conn.execute("SELECT COUNT(*) FROM traffic_events").fetchone()[0]) + throughput = int(self._conn.execute("SELECT COUNT(*) FROM traffic_throughput").fetchone()[0]) + self._conn.execute("DELETE FROM traffic_events") + self._conn.execute("DELETE FROM traffic_throughput") + self._conn.commit() + return {"events": events, "throughput_samples": throughput} + def create_web_session( self, token_hash: str, @@ -1078,6 +1441,10 @@ class AlertStore: self._conn.execute("VACUUM") def close(self) -> None: + if self._analytics_conn is not None: + with self._analytics_lock: + self._analytics_conn.close() + self._analytics_conn = None with self._lock: self._conn.close() diff --git a/app/templates/index.html b/app/templates/index.html index dae04d0..8a9aa52 100644 --- a/app/templates/index.html +++ b/app/templates/index.html @@ -39,7 +39,7 @@
Offline @@ -60,7 +60,7 @@

Traffic throughput

Total, inbound and outbound speed from the Rust TZSP data-plane; 1 s samples are retained in Redis.

loading
TotalInboundOutbound

Traffic direction

Inbound / outbound / internal

-

Events & alerts

Complete retained event history for the selected time range.

EventsAlerts
+

Events & alerts

Complete disk-backed event history for the selected time range.

EventsAlerts

Event mix

Flow, DNS, TLS, HTTP and alerts

Top applications

Unique detected flows; failed/unknown classifications are excluded.

Top local clients

Traffic volume by monitored endpoint

@@ -78,7 +78,7 @@
-

Live Sessions

Continuous streaming is off by default. Capture and Redis history continue independently.

Live off
+

Live Sessions

Continuous streaming is off by default. Capture and disk-backed traffic history continue independently.

Live off
The browser receives coalesced batches instead of every packet/update. Filters are applied server-side while live mode is active.
diff --git a/app/webui.py b/app/webui.py index e136e12..78d7c5b 100644 --- a/app/webui.py +++ b/app/webui.py @@ -308,7 +308,7 @@ class WebServer: direction=self._query_text(query, "direction", 24), ) except RedisUnavailableError as exc: - self._json({"error": f"Redis traffic history unavailable: {exc}"}, status=503) + self._json({"error": f"Traffic history unavailable: {exc}"}, status=503) return self._json({"events": events, "history": traffic_history.status()}) return @@ -322,7 +322,7 @@ class WebServer: payload = traffic_history.throughput_analytics(window) self._json(outer._overlay_current_throughput(payload, window)) except RedisUnavailableError as exc: - self._json({"error": f"Redis throughput history unavailable: {exc}"}, status=503) + self._json({"error": f"Traffic throughput unavailable: {exc}"}, status=503) return if parsed.path == "/api/traffic/analytics": if traffic_history is None: @@ -333,7 +333,7 @@ class WebServer: try: payload = outer._analytics_payload(window) except RedisUnavailableError as exc: - self._json({"error": f"Redis analytics unavailable: {exc}"}, status=503) + self._json({"error": f"Traffic analytics unavailable: {exc}"}, status=503) return self._json(payload) return @@ -548,12 +548,11 @@ class WebServer: try: count = traffic_history.clear() if traffic_history is not None else 0 except RedisUnavailableError as exc: - self._json({"error": f"Redis traffic history unavailable: {exc}"}, status=503) + self._json({"error": f"Traffic history unavailable: {exc}"}, status=503) return - # Remove snapshots from older builds; current traffic snapshots live in Redis. - legacy_snapshots = store.clear_traffic_snapshots() - self._audit("traffic.clear", details={"events": count, "legacy_snapshots": legacy_snapshots}) - self._json({"ok": True, "message": f"Cleared {count} Redis traffic events and cached chart snapshots"}) + snapshots = store.clear_traffic_snapshots() + self._audit("traffic.clear", details={"events": count, "snapshots": snapshots}) + self._json({"ok": True, "message": f"Cleared {count} archived/live traffic events and {snapshots} chart snapshots"}) return if parsed.path == "/api/admin/ndr/incidents/status": try: diff --git a/deploy-routeros.env.example b/deploy-routeros.env.example index 92b975e..20268bb 100644 --- a/deploy-routeros.env.example +++ b/deploy-routeros.env.example @@ -57,14 +57,17 @@ SURICATA_PERSIST_LIB_DIR=/data/lib/suricata RULE_UPDATE_INTERVAL_HOURS=24 ALERT_RETENTION_DAYS=14 -# Bounded live-traffic history. Managed Redis runs inside the same RouterOS container. +# Bounded live-traffic ingest buffer. Durable traffic history is stored in SQLite. REDIS_MANAGED=true REDIS_DATA_DIR=/data/redis REDIS_PORT=6379 -REDIS_MAXMEMORY_MB=0 -REDIS_SNAPSHOT_SECONDS=1800 -REDIS_AOF=true +REDIS_MAXMEMORY_MB=128 +REDIS_SNAPSHOT_SECONDS=0 +REDIS_AOF=false TRAFFIC_RETENTION_HOURS=24 +TRAFFIC_ARCHIVE_INTERVAL_SECONDS=5 +TRAFFIC_ARCHIVE_LAG_SECONDS=10 +TRAFFIC_ARCHIVE_BATCH_SIZE=1000 TRAFFIC_MAX_EVENTS=0 TRAFFIC_MEMORY_EVENTS=0 WEBSOCKET_QUEUE_SIZE=512 diff --git a/docs/routeros-tzsp-capture.md b/docs/routeros-tzsp-capture.md index e69116d..abe3907 100644 --- a/docs/routeros-tzsp-capture.md +++ b/docs/routeros-tzsp-capture.md @@ -1,6 +1,6 @@ # RouterOS TZSP capture architecture -MikroSuricata 0.11.2 uses a **hybrid RouterOS capture path**. The goal is to keep high-volume routed IPv4 off `/tool/sniffer` while still preserving visibility into ARP and other non-IPv4 Ethernet traffic. +MikroSuricata 0.11.3 uses a **hybrid RouterOS capture path**. The goal is to keep high-volume routed IPv4 off `/tool/sniffer` while still preserving visibility into ARP and other non-IPv4 Ethernet traffic. ## Default topology @@ -100,7 +100,7 @@ With the default filter the stream can include, among other protocols: ### IPv6 performance note -IPv6 is deliberately left in the Packet Sniffer complement in 0.11.2 because the requested design is "mangle for IPv4, stream everything except IPv4". If the network later carries sustained high-volume IPv6, the Packet Sniffer performance ceiling can reappear for that traffic. At that point IPv6 should be split into its own high-throughput capture path and the L2 sniffer narrowed to explicit non-IP EtherTypes. +IPv6 is deliberately left in the Packet Sniffer complement in 0.11.3 because the requested design is "mangle for IPv4, stream everything except IPv4". If the network later carries sustained high-volume IPv6, the Packet Sniffer performance ceiling can reappear for that traffic. At that point IPv6 should be split into its own high-throughput capture path and the L2 sniffer narrowed to explicit non-IP EtherTypes. ## No bridge or interface name is assumed @@ -264,7 +264,7 @@ If `TZSP RX` is close to the expected traffic and capture remains close to 100%, - Packet Sniffer cannot necessarily see traffic switched entirely in hardware by a hardware-offloaded bridge. Broadcast/multicast behavior can differ by platform. - The default mangle rule covers routed IPv4 only, not bridge-only IPv4 switching. -- IPv6 remains on Packet Sniffer in 0.11.2 and can therefore inherit Packet Sniffer throughput limits under sustained high-rate IPv6 traffic. +- IPv6 remains on Packet Sniffer in 0.11.3 and can therefore inherit Packet Sniffer throughput limits under sustained high-rate IPv6 traffic. - `filter-interface=all` can expose the same low-level broadcast/L2 frame on more than one logical/physical observation point on some topologies. If this is noisy, set `TZSP_L2_INTERFACE` explicitly. - The global Packet Sniffer has no per-consumer instance. MikroSuricata therefore cannot preserve a second independent sniffer configuration while also owning the non-IPv4 TZSP stream. - Do not run an additional full-traffic Packet Sniffer stream to the same TZSP destination at the same time; duplicate packets will inflate traffic and Suricata processing. diff --git a/rust/tzsp-receiver/Cargo.toml b/rust/tzsp-receiver/Cargo.toml index a558ba1..4f74ed9 100644 --- a/rust/tzsp-receiver/Cargo.toml +++ b/rust/tzsp-receiver/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mikrosuricata-tzsp" -version = "0.11.2" +version = "0.11.3" edition = "2021" publish = false diff --git a/scripts/deploy-routeros.sh b/scripts/deploy-routeros.sh index 9081c8d..4c0d3da 100755 --- a/scripts/deploy-routeros.sh +++ b/scripts/deploy-routeros.sh @@ -70,10 +70,13 @@ TZSP_L2_FILTER_INTERFACE="${TZSP_L2_INTERFACE:-all}" : "${REDIS_MANAGED:=true}" : "${REDIS_DATA_DIR:=/data/redis}" : "${REDIS_PORT:=6379}" -: "${REDIS_MAXMEMORY_MB:=0}" -: "${REDIS_SNAPSHOT_SECONDS:=1800}" -: "${REDIS_AOF:=true}" +: "${REDIS_MAXMEMORY_MB:=128}" +: "${REDIS_SNAPSHOT_SECONDS:=0}" +: "${REDIS_AOF:=false}" : "${TRAFFIC_RETENTION_HOURS:=24}" +: "${TRAFFIC_ARCHIVE_INTERVAL_SECONDS:=5}" +: "${TRAFFIC_ARCHIVE_LAG_SECONDS:=10}" +: "${TRAFFIC_ARCHIVE_BATCH_SIZE:=1000}" : "${TRAFFIC_MAX_EVENTS:=0}" : "${TRAFFIC_MEMORY_EVENTS:=0}" : "${WEBSOCKET_QUEUE_SIZE:=512}" @@ -179,6 +182,9 @@ for numeric_pair in \ "REDIS_MAXMEMORY_MB=$REDIS_MAXMEMORY_MB" \ "REDIS_SNAPSHOT_SECONDS=$REDIS_SNAPSHOT_SECONDS" \ "TRAFFIC_RETENTION_HOURS=$TRAFFIC_RETENTION_HOURS" \ + "TRAFFIC_ARCHIVE_INTERVAL_SECONDS=$TRAFFIC_ARCHIVE_INTERVAL_SECONDS" \ + "TRAFFIC_ARCHIVE_LAG_SECONDS=$TRAFFIC_ARCHIVE_LAG_SECONDS" \ + "TRAFFIC_ARCHIVE_BATCH_SIZE=$TRAFFIC_ARCHIVE_BATCH_SIZE" \ "TRAFFIC_MAX_EVENTS=$TRAFFIC_MAX_EVENTS" \ "TRAFFIC_MEMORY_EVENTS=$TRAFFIC_MEMORY_EVENTS" \ "WEBSOCKET_QUEUE_SIZE=$WEBSOCKET_QUEUE_SIZE" \ @@ -370,6 +376,9 @@ cat > "$LOCAL_RSC" <