diff --git a/packages/tracker/meshnet_tracker/cli.py b/packages/tracker/meshnet_tracker/cli.py index b9b05b0..64ca93e 100644 --- a/packages/tracker/meshnet_tracker/cli.py +++ b/packages/tracker/meshnet_tracker/cli.py @@ -33,6 +33,12 @@ def main() -> None: default=None, help="This tracker's own URL as seen by peers (auto-derived from --host/--port if omitted)", ) + start_cmd.add_argument( + "--stats-db", + default=None, + metavar="PATH", + help="SQLite database path for persistent model usage statistics", + ) args = parser.parse_args() @@ -44,6 +50,7 @@ def main() -> None: heartbeat_timeout=args.heartbeat_timeout, cluster_peers=cluster_peers or None, cluster_self_url=args.self_url, + stats_db=getattr(args, "stats_db", None), ) port = server.start() print(f"meshnet-tracker listening on http://{args.host}:{port}", flush=True) diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 63f5d47..e96aa72 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -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//heartbeat -> ['', 'v1', 'nodes', '', '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 diff --git a/tests/test_tracker_routing.py b/tests/test_tracker_routing.py index ea71639..fa92ddd 100644 --- a/tests/test_tracker_routing.py +++ b/tests/test_tracker_routing.py @@ -1089,3 +1089,99 @@ def test_two_node_pipeline_via_tracker(): node_a.stop() node_b.stop() tracker.stop() + + +# ---------------------------------------------------------------- US-025 stats + +def test_stats_endpoint_returns_empty_models_initially(): + tracker = TrackerServer() + port = tracker.start() + try: + resp = _get_json(f"http://127.0.0.1:{port}/v1/stats") + assert "models" in resp + assert isinstance(resp["models"], dict) + finally: + tracker.stop() + + +def test_stats_rolling_counter_rpm_reflects_recorded_requests(): + from meshnet_tracker.server import _RollingCounter + + counter = _RollingCounter(num_buckets=60, bucket_seconds=60) + now = 1_000_000.0 + for _ in range(60): + counter.record(now) + + rpm = counter.rpm(now) + assert rpm == 1.0 # 60 requests over 60 minutes = 1 rpm + + +def test_stats_collector_records_and_returns_rpms(): + from meshnet_tracker.server import _StatsCollector + + collector = _StatsCollector() + now = 1_000_000.0 + for _ in range(30): + collector.record_request("my-model", now=now) + + combined = collector.get_combined_stats(now=now) + assert "my-model" in combined + assert combined["my-model"]["rpm_last_hour"] > 0 + assert combined["my-model"]["rpm_last_day"] > 0 + assert combined["my-model"]["rpm_last_month"] > 0 + + +def test_stats_collector_merges_peer_rpms_additively(): + from meshnet_tracker.server import _StatsCollector + + collector = _StatsCollector() + now = 1_000_000.0 + collector.record_request("modelX", now=now) # 1 local request in the current minute bucket + + peer_rpms = {"modelX": {"rpm_last_hour": 5.0, "rpm_last_day": 3.0, "rpm_last_month": 1.0}} + collector.merge_peer_rpms("http://peer-tracker:8080", peer_rpms) + + combined = collector.get_combined_stats(now=now) + assert combined["modelX"]["rpm_last_hour"] > 5.0 # local + peer + assert combined["modelX"]["rpm_last_day"] > 3.0 + assert combined["modelX"]["rpm_last_month"] >= 1.0 # local rounds to 0 at 4dp, peer is 1.0 + + +def test_stats_sqlite_persistence_survives_restart(tmp_path): + from meshnet_tracker.server import _StatsCollector + + db = str(tmp_path / "stats.db") + now = 1_000_000.0 + + collector = _StatsCollector(db_path=db) + for _ in range(10): + collector.record_request("persistent-model", now=now) + collector.save_to_db() + + collector2 = _StatsCollector(db_path=db) + combined = collector2.get_combined_stats(now=now) + assert "persistent-model" in combined + assert combined["persistent-model"]["rpm_last_hour"] > 0 + + +def test_stats_gossip_endpoint_merges_peer_slice(): + tracker = TrackerServer() + port = tracker.start() + try: + peer_payload = { + "tracker_url": "http://peer:9000", + "stats": { + "remote-model": { + "rpm_last_hour": 2.5, + "rpm_last_day": 1.0, + "rpm_last_month": 0.5, + } + }, + } + _post_json(f"http://127.0.0.1:{port}/v1/stats/gossip", peer_payload) + + combined = _get_json(f"http://127.0.0.1:{port}/v1/stats") + assert "remote-model" in combined["models"] + assert combined["models"]["remote-model"]["rpm_last_hour"] == 2.5 + finally: + tracker.stop()