rename Metric Classes and change Folder Structure
This commit is contained in:
@@ -0,0 +1,21 @@
|
||||
from server.metrics.backends.Template import StoreTemplate
|
||||
|
||||
class MemoryMetricsStore(StoreTemplate):
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(backend="memory", **kwargs)
|
||||
self._snapshots:dict[str, dict] = {}
|
||||
|
||||
async def publish(self, worker_id:str, snapshot:dict) -> None:
|
||||
self._snapshots[worker_id] = dict(snapshot)
|
||||
|
||||
async def load_all(self) -> list[dict]:
|
||||
return [dict(value) for value in self._snapshots.values()]
|
||||
|
||||
async def clear_all(self) -> None:
|
||||
self._snapshots.clear()
|
||||
|
||||
async def _acquire_startup_cleanup_lock(self, lock_key:str, ttl_seconds:int=300) -> bool:
|
||||
return True
|
||||
|
||||
async def close(self) -> None:
|
||||
return None
|
||||
@@ -0,0 +1,73 @@
|
||||
from server.metrics.backends.Template import StoreTemplate
|
||||
|
||||
import inspect, json
|
||||
|
||||
class RedisMetricsStore(StoreTemplate):
|
||||
def __init__(self, redis_url:str="redis://localhost:6379/0", key_prefix:str="snake:metrics:worker", ttl_seconds:int|None=None, **kwargs):
|
||||
super().__init__(backend="redis", key_prefix=key_prefix, **kwargs)
|
||||
self.redis_url = redis_url
|
||||
self.key_prefix = key_prefix
|
||||
self.ttl_seconds = ttl_seconds
|
||||
self._redis = None
|
||||
|
||||
async def _get_redis(self):
|
||||
if self._redis is not None:
|
||||
return self._redis
|
||||
|
||||
try:
|
||||
import redis.asyncio as aioredis # type: ignore[import-not-found]
|
||||
except ImportError as error: # pragma: no cover
|
||||
raise RuntimeError("Metrics backend set to redis but 'redis' package is not installed") from error
|
||||
|
||||
self._redis = aioredis.from_url(self.redis_url)
|
||||
return self._redis
|
||||
|
||||
def _key(self, worker_id:str) -> str:
|
||||
return f"{self.key_prefix}:{worker_id}"
|
||||
|
||||
async def publish(self, worker_id:str, snapshot:dict) -> None:
|
||||
redis = await self._get_redis()
|
||||
await redis.set(self._key(worker_id), json.dumps(snapshot), ex=self.ttl_seconds)
|
||||
|
||||
async def load_all(self) -> list[dict]:
|
||||
redis = await self._get_redis()
|
||||
keys = await redis.keys(f"{self.key_prefix}:*")
|
||||
snapshots = []
|
||||
for key in keys:
|
||||
payload = await redis.get(key)
|
||||
if not payload:
|
||||
continue
|
||||
try:
|
||||
snapshots.append(json.loads(payload))
|
||||
except Exception:
|
||||
continue
|
||||
return snapshots
|
||||
|
||||
async def clear_all(self) -> None:
|
||||
redis = await self._get_redis()
|
||||
keys = await redis.keys(f"{self.key_prefix}:*")
|
||||
if keys:
|
||||
await redis.delete(*keys)
|
||||
|
||||
async def _acquire_startup_cleanup_lock(self, lock_key:str, ttl_seconds:int=300) -> bool:
|
||||
redis = await self._get_redis()
|
||||
locked = await redis.set(lock_key, '1', ex=max(1, int(ttl_seconds)), nx=True)
|
||||
return bool(locked)
|
||||
|
||||
async def close(self) -> None:
|
||||
if self._redis is None:
|
||||
return
|
||||
|
||||
aclose_method = getattr(self._redis, "aclose", None)
|
||||
if callable(aclose_method):
|
||||
maybe_result = aclose_method()
|
||||
if inspect.isawaitable(maybe_result):
|
||||
await maybe_result
|
||||
else:
|
||||
close_method = getattr(self._redis, "close", None)
|
||||
if callable(close_method):
|
||||
close_result = close_method()
|
||||
if inspect.isawaitable(close_result):
|
||||
await close_result
|
||||
|
||||
self._redis = None
|
||||
@@ -0,0 +1,131 @@
|
||||
from typing import Any, Awaitable, cast
|
||||
import inspect, os
|
||||
|
||||
class StoreTemplate:
|
||||
def __init__(self, backend:str="memory", key_prefix:str="snake:metrics:worker", worker_id:str|None=None, **kwargs):
|
||||
self.backend = (backend or "memory").strip().lower()
|
||||
self.key_prefix = key_prefix
|
||||
self.worker_id = worker_id or f"{os.getpid()}"
|
||||
self.store = self
|
||||
|
||||
async def publish_only(self, snapshot:dict) -> None:
|
||||
await self.store.publish(self.worker_id, snapshot)
|
||||
|
||||
async def snapshot(self, local_snapshot:dict) -> dict:
|
||||
await self.store.publish(self.worker_id, local_snapshot)
|
||||
|
||||
if self.backend != "redis":
|
||||
return local_snapshot
|
||||
|
||||
snapshots = await self.store.load_all()
|
||||
if not snapshots:
|
||||
return local_snapshot
|
||||
return self._merge_snapshots(snapshots)
|
||||
|
||||
async def close(self) -> None:
|
||||
if self.store is self:
|
||||
return
|
||||
close_store = getattr(self.store, "close", None)
|
||||
if not callable(close_store):
|
||||
return
|
||||
maybe_result = close_store()
|
||||
if inspect.isawaitable(maybe_result):
|
||||
await cast(Awaitable[Any], maybe_result)
|
||||
|
||||
async def clear_all_workers(self) -> None:
|
||||
clear_all = getattr(self.store, "clear_all", None)
|
||||
if callable(clear_all):
|
||||
maybe_result = clear_all()
|
||||
if inspect.isawaitable(maybe_result):
|
||||
await cast(Awaitable[Any], maybe_result)
|
||||
|
||||
async def acquire_startup_cleanup_lock(self, ttl_seconds:int=300) -> bool:
|
||||
if self.backend != "redis":
|
||||
return True
|
||||
|
||||
acquire_lock = getattr(self.store, "_acquire_startup_cleanup_lock", None)
|
||||
if not callable(acquire_lock):
|
||||
acquire_lock = getattr(self.store, "acquire_startup_cleanup_lock", None)
|
||||
if not callable(acquire_lock):
|
||||
return True
|
||||
|
||||
lock_key = f"{self.key_prefix}:startup_cleanup_lock"
|
||||
maybe_result = acquire_lock(lock_key, ttl_seconds)
|
||||
if inspect.isawaitable(maybe_result):
|
||||
return bool(await cast(Awaitable[Any], maybe_result))
|
||||
return bool(maybe_result)
|
||||
|
||||
def _merge_snapshots(self, snapshots:list[dict]) -> dict:
|
||||
merged = {
|
||||
"games_started": 0,
|
||||
"games_ended": 0,
|
||||
"wins": 0,
|
||||
"losses": 0,
|
||||
"total_moves": 0,
|
||||
"total_turns": 0,
|
||||
"max_turn": 0,
|
||||
"active_games_peak": 0,
|
||||
"games_autocreated": 0,
|
||||
"http_requests_total": 0,
|
||||
"move_response_time_ms_total": 0.0,
|
||||
"move_response_time_ms_max": 0.0,
|
||||
"last_game_start_unix": 0,
|
||||
"last_game_end_unix": 0,
|
||||
"last_move_unix": 0,
|
||||
"games_stuck_removed": 0,
|
||||
"game_state_local_cache_enabled": False,
|
||||
"metrics_backend": "redis",
|
||||
"active_games": 0,
|
||||
"tracked_games": 0,
|
||||
"oldest_active_game_age_sec": 0,
|
||||
"stale_game_timeout_sec": 0,
|
||||
"active_games_stale": 0,
|
||||
"http_requests_by_endpoint": {"info": 0, "start": 0, "move": 0, "end": 0},
|
||||
"move_direction_counts": {
|
||||
"up": 0,
|
||||
"down": 0,
|
||||
"left": 0,
|
||||
"right": 0,
|
||||
"unknown": 0,
|
||||
},
|
||||
}
|
||||
|
||||
for worker in snapshots:
|
||||
for metric_name in (
|
||||
"games_started",
|
||||
"games_ended",
|
||||
"wins",
|
||||
"losses",
|
||||
"total_moves",
|
||||
"total_turns",
|
||||
"games_autocreated",
|
||||
"http_requests_total",
|
||||
"games_stuck_removed",
|
||||
"active_games",
|
||||
"tracked_games",
|
||||
"active_games_stale",
|
||||
):
|
||||
merged[metric_name] += int(worker.get(metric_name, 0))
|
||||
|
||||
merged["move_response_time_ms_total"] += float(worker.get("move_response_time_ms_total", 0.0))
|
||||
merged["max_turn"] = max(merged["max_turn"], int(worker.get("max_turn", 0)))
|
||||
merged["active_games_peak"] = max(merged["active_games_peak"], int(worker.get("active_games_peak", 0)))
|
||||
merged["move_response_time_ms_max"] = max(merged["move_response_time_ms_max"], float(worker.get("move_response_time_ms_max", 0.0)))
|
||||
merged["last_game_start_unix"] = max(merged["last_game_start_unix"], int(worker.get("last_game_start_unix", 0)))
|
||||
merged["last_game_end_unix"] = max(merged["last_game_end_unix"], int(worker.get("last_game_end_unix", 0)))
|
||||
merged["last_move_unix"] = max(merged["last_move_unix"], int(worker.get("last_move_unix", 0)))
|
||||
merged["oldest_active_game_age_sec"] = max(merged["oldest_active_game_age_sec"], int(worker.get("oldest_active_game_age_sec", 0)))
|
||||
merged["stale_game_timeout_sec"] = max(merged["stale_game_timeout_sec"], int(worker.get("stale_game_timeout_sec", 0)))
|
||||
merged["game_state_local_cache_enabled"] = merged["game_state_local_cache_enabled"] or bool(worker.get("game_state_local_cache_enabled", False))
|
||||
|
||||
for endpoint in merged["http_requests_by_endpoint"]:
|
||||
merged["http_requests_by_endpoint"][endpoint] += int(worker.get("http_requests_by_endpoint", {}).get(endpoint, 0))
|
||||
for direction in merged["move_direction_counts"]:
|
||||
merged["move_direction_counts"][direction] += int(worker.get("move_direction_counts", {}).get(direction, 0))
|
||||
|
||||
games_ended = merged["games_ended"]
|
||||
total_moves = merged["total_moves"]
|
||||
merged["avg_turns_per_game"] = round((merged["total_turns"] / games_ended) if games_ended else 0.0, 2)
|
||||
merged["win_rate"] = round((merged["wins"] / games_ended) if games_ended else 0.0, 4)
|
||||
merged["avg_move_response_ms"] = round((merged["move_response_time_ms_total"] / total_moves) if total_moves else 0.0, 2)
|
||||
return merged
|
||||
Reference in New Issue
Block a user