"""数据统计热缓存:后台按窗口并行刷新 summary/curve;浏览器只读缓存 + SSE.""" from __future__ import annotations import json import os import queue import threading import time from collections.abc import Iterator from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any from lib.stats_agg_lib import build_stats_curve, build_stats_summary, normalize_window STATS_WINDOWS = ("today", "7d", "30d", "all") def _poll_interval_sec() -> float: try: return max(5.0, float(os.getenv("HUB_STATS_POLL_INTERVAL_SEC") or os.getenv("HUB_PULL_INTERVAL_SEC") or 15)) except (TypeError, ValueError): return 15.0 def _sse_heartbeat_sec() -> float: try: return max(10.0, float(os.getenv("HUB_SSE_HEARTBEAT_SEC") or 25)) except (TypeError, ValueError): return 25.0 class StatsStore: def __init__(self) -> None: self._lock = threading.RLock() self.version = 0 self.summaries: dict[str, dict[str, Any]] = {} self.curves: dict[str, dict[str, Any]] = {} self.refreshing = True self.last_error: str | None = None self.updated_at: str = "" self._ready = False self._subscribers: list[queue.Queue[str | None]] = [] self._stop = threading.Event() self._refresh = threading.Event() self._thread: threading.Thread | None = None def start(self) -> None: if self._thread and self._thread.is_alive(): return self._stop.clear() with self._lock: self.refreshing = True self._ready = False self._thread = threading.Thread( target=self._loop, daemon=True, name="hub-stats-pull", ) self._thread.start() def stop(self) -> None: self._stop.set() self._refresh.set() self._broadcast(close=True) def request_refresh(self) -> None: self._refresh.set() def summary_dict(self, window: str = "7d") -> dict[str, Any]: w = normalize_window(window) with self._lock: body = dict(self.summaries.get(w) or {}) ver = self.version refreshing = self.refreshing err = self.last_error updated_at = self.updated_at ready = self._ready if not body: return { "ok": False, "window": w, "msg": "统计缓存尚未就绪" if not ready or refreshing else "暂无统计数据", "stats_version": ver, "refreshing": refreshing, "updated_at": updated_at, "error": err, "from_cache": True, "poll_interval_sec": _poll_interval_sec(), } body = dict(body) body["stats_version"] = ver body["refreshing"] = refreshing body["updated_at"] = updated_at body["error"] = err body["from_cache"] = True body["poll_interval_sec"] = _poll_interval_sec() body.setdefault("ok", True) return body def curve_dict(self, range_key: str = "7d") -> dict[str, Any]: w = normalize_window(range_key) with self._lock: body = dict(self.curves.get(w) or {}) ver = self.version refreshing = self.refreshing err = self.last_error updated_at = self.updated_at if not body: return { "ok": True, "range": w, "total": [], "series": [], "stats_version": ver, "refreshing": refreshing, "updated_at": updated_at, "error": err, "from_cache": True, "poll_interval_sec": _poll_interval_sec(), } body = dict(body) body["stats_version"] = ver body["refreshing"] = refreshing body["updated_at"] = updated_at body["error"] = err body["from_cache"] = True body["poll_interval_sec"] = _poll_interval_sec() body.setdefault("ok", True) return body def event_dict(self) -> dict[str, Any]: with self._lock: return { "stats_version": self.version, "updated_at": self.updated_at, "refreshing": self.refreshing, "ok": True, "error": self.last_error, "poll_interval_sec": _poll_interval_sec(), } def _loop(self) -> None: while not self._stop.is_set(): self._refresh_once() if self._stop.is_set(): break self._refresh.clear() self._refresh.wait(timeout=_poll_interval_sec()) def _current_snaps(self) -> list[dict[str, Any]]: try: from lib.overview_cache_lib import overview_store snap = overview_store.snapshot_dict() rows = snap.get("instances") if isinstance(snap, dict) else None if isinstance(rows, list) and rows: return rows except Exception: pass try: from lib.bridge_client_lib import pull_all_enabled return pull_all_enabled() except Exception: return [] def _refresh_once(self) -> None: with self._lock: self.refreshing = True self._broadcast() try: snaps = self._current_snaps() summaries: dict[str, dict[str, Any]] = {} curves: dict[str, dict[str, Any]] = {} # 多窗口并行刷;每窗内部 pull_all_stats 已并行拉机 with ThreadPoolExecutor(max_workers=len(STATS_WINDOWS)) as pool: futs = { pool.submit(build_stats_summary, w, snaps=snaps): ("summary", w) for w in STATS_WINDOWS } futs.update( { pool.submit(build_stats_curve, w): ("curve", w) for w in STATS_WINDOWS } ) for fut in as_completed(futs): kind, w = futs[fut] try: data = fut.result() except Exception as e: data = {"ok": False, "msg": str(e) or "failed", "window": w} if kind == "summary": summaries[w] = data if isinstance(data, dict) else {"ok": False} else: curves[w] = data if isinstance(data, dict) else {"ok": False, "total": [], "series": []} with self._lock: self.version += 1 self.summaries = summaries self.curves = curves self.updated_at = time.strftime("%Y-%m-%d %H:%M:%S") self.last_error = None self.refreshing = False self._ready = True except Exception as e: with self._lock: self.version += 1 self.last_error = str(e) or "stats_refresh_failed" self.refreshing = False self._ready = True self._broadcast() def _broadcast(self, *, close: bool = False) -> None: with self._lock: subs = list(self._subscribers) event = None if close else json.dumps(self.event_dict(), ensure_ascii=False) dead: list[queue.Queue[str | None]] = [] for q in subs: try: q.put_nowait(None if close else event) except queue.Full: try: q.get_nowait() except queue.Empty: pass try: q.put_nowait(event) except queue.Full: dead.append(q) except Exception: dead.append(q) if dead: with self._lock: for q in dead: if q in self._subscribers: self._subscribers.remove(q) def iter_sse(self) -> Iterator[str]: q: queue.Queue[str | None] = queue.Queue(maxsize=32) with self._lock: self._subscribers.append(q) try: yield _sse_frame(self.event_dict()) while True: try: raw = q.get(timeout=_sse_heartbeat_sec()) except queue.Empty: yield ": heartbeat\n\n" continue if raw is None: break try: data = json.loads(raw) except Exception: data = self.event_dict() yield _sse_frame(data) finally: with self._lock: if q in self._subscribers: self._subscribers.remove(q) def _sse_frame(data: dict[str, Any]) -> str: body = json.dumps(data, ensure_ascii=False) return f"event: stats\ndata: {body}\n\n" stats_store = StatsStore()