dekun
269 lines
9.0 KiB
Python
269 lines
9.0 KiB
Python
"""数据统计热缓存:后台按窗口并行刷新 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()
|