popov/neuron-tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(us-025): model usage statistics with rolling RPM windows and SQLite persistence

Browse Source 
Adds _RollingCounter and _StatsCollector to the tracker: three rolling windows
(hour=60×1min, day=24×1hr, month=30×1day) track request RPMs per model.
GET /v1/stats returns combined local + peer stats. POST /v1/stats/gossip lets
trackers push their local slice for additive merge in hive mode.
SQLite persistence via --stats-db flag; stats survive tracker restarts.
Records a stat for each proxied /v1/chat/completions request.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dobromir Popov
2026-06-30 13:37:33 +03:00
parent fff11c4d8b
commit d9110b623b
3 changed files with 333 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

230
packages/tracker/meshnet_tracker/server.py
View File

@@ -25,6 +25,7 @@ import http.server
import hashlib
import json
import socketserver
import sqlite3
import threading
import time
import urllib.parse
@@ -53,6 +54,171 @@ DEFAULT_QUANTIZATIONS = ["bfloat16"]
DEFAULT_BENCHMARK_TOKENS_PER_SEC = 1.0
class _RollingCounter:
"""Circular-bucket request counter.
Tracks events in `num_buckets` slots, each covering `bucket_seconds` of wall time.
Old buckets are silently discarded when their epoch expires.
"""
def __init__(self, num_buckets: int, bucket_seconds: int) -> None:
self._num = num_buckets
self._bsec = bucket_seconds
self._counts: list[int] = [0] * num_buckets
self._epochs: list[int] = [-1] * num_buckets # which epoch each slot holds
def _epoch(self, now: float) -> int:
return int(now) // self._bsec
def record(self, now: float | None = None) -> None:
t = now if now is not None else time.time()
ep = self._epoch(t)
idx = ep % self._num
if self._epochs[idx] != ep:
self._counts[idx] = 0
self._epochs[idx] = ep
self._counts[idx] += 1
def rpm(self, now: float | None = None) -> float:
t = now if now is not None else time.time()
cutoff = self._epoch(t) - self._num # epochs <= this are stale
total = sum(self._counts[i] for i in range(self._num) if self._epochs[i] > cutoff)
window_minutes = (self._num * self._bsec) / 60.0
return total / window_minutes if window_minutes > 0 else 0.0
def buckets(self) -> list[tuple[int, int]]:
return [(self._epochs[i], self._counts[i]) for i in range(self._num)]
def restore_buckets(self, data: list[tuple[int, int]]) -> None:
for i, (ep, cnt) in enumerate(data):
if i < self._num:
self._epochs[i] = ep
self._counts[i] = cnt
class _ModelStats:
"""Three rolling windows for one model: last hour, last day, last month."""
def __init__(self) -> None:
self.per_minute = _RollingCounter(60, 60) # 60 × 1-min buckets = 1 hour
self.per_hour = _RollingCounter(24, 3600) # 24 × 1-hr buckets = 1 day
self.per_day = _RollingCounter(30, 86400) # 30 × 1-day buckets = ~1 month
def record(self, now: float | None = None) -> None:
t = now if now is not None else time.time()
self.per_minute.record(t)
self.per_hour.record(t)
self.per_day.record(t)
def rpms(self, now: float | None = None) -> dict:
t = now if now is not None else time.time()
return {
"rpm_last_hour": round(self.per_minute.rpm(t), 4),
"rpm_last_day": round(self.per_hour.rpm(t), 4),
"rpm_last_month": round(self.per_day.rpm(t), 4),
}
class _StatsCollector:
"""Thread-safe model request stats with SQLite persistence and peer slice merging."""
SAVE_INTERVAL = 60.0 # seconds between DB saves
def __init__(self, db_path: str | None = None) -> None:
self._lock = threading.Lock()
self._local: dict[str, _ModelStats] = {}
self._peer_rpms: dict[str, dict[str, dict]] = {} # tracker_url -> model -> rpms
self._db_path = db_path
if db_path:
self._init_db()
self._load_from_db()
# ---------- public API ----------
def record_request(self, model: str, now: float | None = None) -> None:
t = now if now is not None else time.time()
with self._lock:
if model not in self._local:
self._local[model] = _ModelStats()
self._local[model].record(t)
def get_local_rpms(self, now: float | None = None) -> dict[str, dict]:
t = now if now is not None else time.time()
with self._lock:
return {m: s.rpms(t) for m, s in self._local.items()}
def merge_peer_rpms(self, tracker_url: str, rpms: dict[str, dict]) -> None:
with self._lock:
self._peer_rpms[tracker_url] = dict(rpms)
def get_combined_stats(self, now: float | None = None) -> dict:
t = now if now is not None else time.time()
with self._lock:
combined: dict[str, dict] = {}
for model, ms in self._local.items():
combined[model] = ms.rpms(t)
for _url, peer in self._peer_rpms.items():
for model, rpms in peer.items():
if model not in combined:
combined[model] = {"rpm_last_hour": 0.0, "rpm_last_day": 0.0, "rpm_last_month": 0.0}
for key in ("rpm_last_hour", "rpm_last_day", "rpm_last_month"):
combined[model][key] = round(combined[model].get(key, 0.0) + rpms.get(key, 0.0), 4)
return combined
# ---------- persistence ----------
def _init_db(self) -> None:
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
con.execute(
"CREATE TABLE IF NOT EXISTS model_rpm_buckets "
"(model TEXT, window TEXT, bucket_idx INTEGER, bucket_epoch INTEGER, count INTEGER, "
"PRIMARY KEY (model, window, bucket_idx))"
)
con.commit()
con.close()
def save_to_db(self) -> None:
if not self._db_path:
return
with self._lock:
rows = []
for model, ms in self._local.items():
for window_name, counter in (
("hour", ms.per_minute),
("day", ms.per_hour),
("month", ms.per_day),
):
for idx, (ep, cnt) in enumerate(counter.buckets()):
if ep >= 0:
rows.append((model, window_name, idx, ep, cnt))
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
con.executemany(
"INSERT OR REPLACE INTO model_rpm_buckets VALUES (?,?,?,?,?)", rows
)
con.commit()
con.close()
def _load_from_db(self) -> None:
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
rows = con.execute("SELECT model, window, bucket_idx, bucket_epoch, count FROM model_rpm_buckets").fetchall()
con.close()
grouped: dict[str, dict[str, list[tuple[int, int]]]] = {}
for model, window, idx, ep, cnt in rows:
grouped.setdefault(model, {}).setdefault(window, []).append((idx, ep, cnt))
for model, windows in grouped.items():
ms = _ModelStats()
for window_name, entries in windows.items():
counter = {"hour": ms.per_minute, "day": ms.per_hour, "month": ms.per_day}.get(window_name)
if counter is None:
continue
data = [(0, -1)] * counter._num
for idx, ep, cnt in entries:
if 0 <= idx < counter._num:
data[idx] = (ep, cnt)
counter.restore_buckets(data)
self._local[model] = ms
class _NodeEntry:
__slots__ = (
"node_id", "endpoint", "shard_start", "shard_end",
@@ -523,6 +689,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
minimum_stake: int,
raft: "RaftNode | None" = None,
gossip: "NodeGossip | None" = None,
stats: "_StatsCollector | None" = None,
) -> None:
super().__init__(addr, handler)
self.registry = registry
@@ -533,6 +700,7 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
self.minimum_stake = minimum_stake
self.raft = raft
self.gossip = gossip
self.stats: _StatsCollector | None = stats
class _TrackerHandler(http.server.BaseHTTPRequestHandler):
@@ -582,6 +750,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
if self.path == "/v1/gossip":
self._handle_gossip()
return
if self.path == "/v1/stats/gossip":
self._handle_stats_gossip()
return
parts = self.path.split("/")
# /v1/nodes/<node_id>/heartbeat -> ['', 'v1', 'nodes', '<id>', 'heartbeat']
if len(parts) == 5 and parts[1] == "v1" and parts[2] == "nodes" and parts[4] == "heartbeat":
@@ -610,6 +781,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
self._handle_tracker_nodes(model)
elif parsed.path == "/v1/raft/status":
self._handle_raft_status()
elif parsed.path == "/v1/stats":
self._handle_stats()
elif parsed.path == "/v1/health":
self._send_json(200, {"status": "ok"})
else:
@@ -772,6 +945,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
model: str = body.get("model", "")
is_stream: bool = bool(body.get("stream", False))
if model and server.stats is not None:
server.stats.record_request(model)
# Find a live tracker-mode node for this model
with server.lock:
self._purge_expired_nodes()
@@ -1209,6 +1385,24 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
server.gossip.merge({k: float(v) for k, v in body.items()})
self._send_json(200, {})
def _handle_stats(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if server.stats is None:
self._send_json(200, {"models": {}})
return
self._send_json(200, {"models": server.stats.get_combined_stats()})
def _handle_stats_gossip(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
body = self._read_json_body()
if body is None:
return
tracker_url = body.get("tracker_url", "")
rpms = body.get("stats", {})
if server.stats is not None and tracker_url and isinstance(rpms, dict):
server.stats.merge_peer_rpms(tracker_url, rpms)
self._send_json(200, {})
def _handle_assign(self, parsed: urllib.parse.ParseResult):
"""Return an optimal shard assignment for a node given its hardware profile.
@@ -1556,6 +1750,7 @@ class TrackerServer:
minimum_stake: int = 0,
cluster_peers: list[str] | None = None,
cluster_self_url: str | None = None,
stats_db: str | None = None,
) -> None:
self._host = host
self._requested_port = port
@@ -1576,6 +1771,9 @@ class TrackerServer:
self._rebalance_thread: threading.Thread | None = None
self._raft: RaftNode | None = None
self._gossip: NodeGossip | None = None
self._stats: _StatsCollector | None = _StatsCollector(db_path=stats_db) if stats_db else _StatsCollector()
self._stats_stop = threading.Event()
self._stats_thread: threading.Thread | None = None
self.port: int | None = None
def start(self) -> int:
@@ -1592,6 +1790,7 @@ class TrackerServer:
self._model_presets,
self._contracts,
self._minimum_stake,
stats=self._stats,
)
self.port = self._server.server_address[1]
@@ -1610,10 +1809,13 @@ class TrackerServer:
self._gossip.start()
self._rebalance_stop.clear()
self._stats_stop.clear()
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
self._thread.start()
self._rebalance_thread = threading.Thread(target=self._rebalance_loop, daemon=True)
self._rebalance_thread.start()
self._stats_thread = threading.Thread(target=self._stats_loop, daemon=True)
self._stats_thread.start()
return self.port
def _raft_apply(self, command: str, payload: dict) -> None:
@@ -1657,6 +1859,28 @@ class TrackerServer:
_purge_expired_nodes_locked(server)
_rebalance_all_locked(server)
def _stats_loop(self) -> None:
"""Periodically save stats to DB and push local slice to cluster peers."""
while not self._stats_stop.wait(_StatsCollector.SAVE_INTERVAL):
if self._stats is not None:
self._stats.save_to_db()
if self._stats is not None and self._cluster_peers:
self_url = self._cluster_self_url or f"http://{self._host}:{self.port}"
local_rpms = self._stats.get_local_rpms()
body = json.dumps({"tracker_url": self_url, "stats": local_rpms}).encode()
for peer in self._cluster_peers:
try:
req = urllib.request.Request(
f"{peer}/v1/stats/gossip",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=2.0) as r:
r.read()
except Exception:
pass
def stop(self) -> None:
if self._raft is not None:
self._raft.stop()
@@ -1665,15 +1889,21 @@ class TrackerServer:
if self._server is None:
return
self._rebalance_stop.set()
self._stats_stop.set()
if self._stats is not None:
self._stats.save_to_db()
self._server.shutdown()
self._server.server_close()
if self._thread is not None:
self._thread.join(timeout=1)
if self._rebalance_thread is not None:
self._rebalance_thread.join(timeout=1)
if self._stats_thread is not None:
self._stats_thread.join(timeout=1)
self._server = None
self._thread = None
self._rebalance_thread = None
self._stats_thread = None
self._raft = None
self._gossip = None
self.port = None