from __future__ import annotations import threading import time from datetime import datetime, timezone from typing import Any, Iterable from .live import RedisUnavailableError, TrafficHistory from .store import AlertStore SUMMARY_WINDOWS = (900, 3600, 18000, 21600, 86400) class AnalyticsSnapshotCache: """Disk-backed dashboard snapshots plus Redis-to-SQLite archive worker. 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__( self, store: AlertStore, 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: self.store = store self.history = history self.stop_event = stop_event self.interval_seconds = max(15, int(interval_seconds)) 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._archive_started = threading.Event() self._stopping = threading.Event() self._lock = threading.RLock() now = time.monotonic() # Standard UI ranges are kept warm. Arbitrary API ranges are added on demand. self._requested_at: dict[int, float] = {} self._last_refresh: dict[int, float] = {} self._errors = 0 self._refreshes = 0 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(120, self.interval_seconds * 3) def start(self) -> None: 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._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 cached = self.store.traffic_snapshot(window) cadence = self._refresh_interval(window) if cached is not None: cached = self._decorate(cached, "sqlite-snapshot") age = float(cached.get("snapshot_age_seconds") or 0) stale = age > cadence * 1.5 cached["snapshot_stale"] = stale cached["snapshot_refreshing"] = stale cached["snapshot_refresh_interval_seconds"] = cadence self._overlay_current_throughput(cached) if stale: self._snapshot_wake.set() return cached # 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)) return { "window_seconds": window, "events": 0, "bytes": 0, "alerts": 0, "blocked": 0, "timeline": [ { "ts_ms": now_ms - window * 1000 + idx * bin_ms, "events": 0, "bytes": 0, "alerts": 0, "bps": 0, "in_bps": 0, "out_bps": 0, "other_bps": 0, "pps": 0, } for idx in range(bins_count) ], "snapshot_source": "sqlite-background", "snapshot_age_seconds": 0, "snapshot_loading": True, "snapshot_refreshing": True, "snapshot_refresh_interval_seconds": cadence, } def refresh_all(self) -> None: self.refresh_windows(SUMMARY_WINDOWS) def refresh_windows(self, windows: Iterable[int]) -> None: normalized = sorted({self._normalise_window(window) for window in windows}) if not normalized: return try: snapshots = self.history.analytics_many(normalized) archive_backlog = not self._archive_ready.is_set() now_wall = time.time() for window in normalized: snapshots[window]["archive_backlog"] = archive_backlog if archive_backlog: snapshots[window]["analytics_complete"] = False self.store.save_traffic_snapshot(window, snapshots[window]) with self._lock: self._last_refresh[window] = now_wall self._refreshes += 1 except (RedisUnavailableError, KeyError, ValueError, OSError): with self._lock: self._errors += 1 except Exception: with self._lock: self._errors += 1 def status(self) -> dict[str, Any]: 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 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": "sqlite-archive", "archive_worker_running": self._archive_thread.is_alive(), "snapshot_worker_running": self._snapshot_thread.is_alive(), "archive_started": self._archive_started.is_set(), "archive_backlog": not self._archive_ready.is_set(), "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 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_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() self._archive_started.set() 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 >= 900: 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. Yield a # little CPU so EVE/NDR/web threads remain responsive. if self.stop_event.wait(0.02): break 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 # Do not race the very first archive pass. After that, snapshots may # use the committed SQLite prefix even while Redis still has a # backlog. This keeps historical charts moving instead of freezing # for hours; such snapshots are explicitly marked incomplete. if not self._archive_started.is_set(): continue try: due = self._due_windows() if due: self.refresh_windows(due) except Exception: with self._lock: self._errors += 1 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() due: list[int] = [] with self._lock: requests = dict(self._requested_at) last_refresh = dict(self._last_refresh) # When nobody requested analytics, do not touch SQLite just to keep # dashboard ranges warm. Current throughput is delivered independently # from the Rust telemetry path, so background chart work can stay idle. if not requests: return [] persisted = { int(row["window_seconds"]): row.get("generated_at") for row in self.store.traffic_snapshot_status().get("windows", []) } for window, requested_at in requests.items(): 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) if last <= 0 and window in persisted: last = self._generated_epoch(persisted[window]) if last <= 0 or now_wall - last >= cadence: due.append(window) return sorted(due) def _refresh_interval(self, window: int) -> int: base = self.interval_seconds if window <= 900: return base if window <= 3600: return max(base * 2, 120) if window <= 21600: return max(base * 5, 300) return max(base * 15, 900) def _overlay_current_throughput(self, payload: dict[str, Any]) -> None: try: sample = self.history.latest_throughput() except RedisUnavailableError: return if not sample: return now_ms = int(time.time() * 1000) ts_ms = int(sample.get("ts_ms") or 0) interval = max(float(sample.get("interval_ms") or 1000) / 1000.0, 0.001) age_ms = max(0, now_ms - ts_ms) if age_ms > max(3000, round(interval * 3000)): current = current_in = current_out = current_pps = 0 else: current = round(max(int(sample.get("bytes_total") or 0), 0) * 8 / interval) current_in = round(max(int(sample.get("bytes_in") or 0), 0) * 8 / interval) current_out = round(max(int(sample.get("bytes_out") or 0), 0) * 8 / interval) current_pps = round(max(int(sample.get("packets_total") or 0), 0) / interval, 2) payload["current_bps"] = current payload["current_in_bps"] = current_in payload["current_out_bps"] = current_out payload["current_other_bps"] = max(0, current - current_in - current_out) payload["current_pps"] = current_pps @staticmethod def _generated_epoch(value: Any) -> float: if not value: return 0.0 try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.timestamp() except (TypeError, ValueError): return 0.0 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]: result = dict(payload) generated = result.get("generated_at") age = 0.0 if generated: try: parsed = datetime.fromisoformat(str(generated).replace("Z", "+00:00")) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) age = max(0.0, (datetime.now(timezone.utc) - parsed.astimezone(timezone.utc)).total_seconds()) except ValueError: age = 0.0 result["snapshot_source"] = source result["snapshot_age_seconds"] = round(age, 1) return result