371 lines
15 KiB
Python
371 lines
15 KiB
Python
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
|