| """
|
| metrics_service.py — LibBee v3.1
|
|
|
| Fixes applied:
|
| 1. All incr / incr_bucket calls are now pure in-memory (no disk I/O per request).
|
| Counters accumulate in _counters / _buckets dicts.
|
| 2. flush() merges in-memory state to disk atomically using a temp-file rename.
|
| Called by a background task in app.py every 30 seconds and on shutdown.
|
| 3. threading.Lock replaced with asyncio.Lock — consistent with FastAPI's async model.
|
| 4. Corrupt metrics are reported without copying file contents into logs. |
| 5. _write() uses os.replace (atomic rename) to prevent corrupt state on crash. |
| 6. Failed writes retain pending counts for the next flush. |
| """
|
| import asyncio
|
| import json
|
| import logging
|
| import os
|
| import time |
| from copy import deepcopy |
| from pathlib import Path
|
| from typing import Any, Dict
|
|
|
| logger = logging.getLogger(__name__)
|
|
|
| _DEFAULT_STATE: Dict[str, Any] = {
|
| "agent_requests": 0,
|
| "search_requests": 0,
|
| "feedback_total": 0,
|
| "intents": {},
|
| "errors": {},
|
| "summary_fallbacks": 0,
|
| "follow_up_hits": 0,
|
| "search_handoffs": 0,
|
| "last_updated": 0.0,
|
| }
|
|
|
| _MAX_QUERY_LOG = 200
|
|
|
|
|
| class MetricsService:
|
| def __init__(self, path: Path):
|
| self.path = Path(path)
|
| self.path.parent.mkdir(parents=True, exist_ok=True)
|
| self._lock = asyncio.Lock()
|
|
|
|
|
| self._counters: Dict[str, int] = {}
|
| self._buckets: Dict[str, Dict[str, int]] = {}
|
| self._dirty = False
|
|
|
|
|
| self._query_log: list = []
|
|
|
|
|
| if not self.path.exists():
|
| self._write_to_disk(deepcopy(_DEFAULT_STATE)) |
|
|
|
|
|
|
| def log_query(
|
| self,
|
| question: str,
|
| intent: str = "",
|
| tool: str = "",
|
| model: str = "",
|
| response_time: float = 0.0,
|
| result_count: int = 0,
|
| error: str = "",
|
| ) -> None:
|
| """Append a query record to the in-memory log (capped at _MAX_QUERY_LOG)."""
|
| import datetime
|
| record = {
|
| "timestamp": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), |
| "question": (question or "")[:120],
|
| "intent": intent or "",
|
| "tool": tool or "",
|
| "model": model or "",
|
| "response_time": round(response_time, 2),
|
| "result_count": result_count,
|
| "error": (error or "")[:80],
|
| }
|
| self._query_log.append(record)
|
| if len(self._query_log) > _MAX_QUERY_LOG:
|
| self._query_log = self._query_log[-_MAX_QUERY_LOG:]
|
|
|
| def recent_queries(self, limit: int = 50) -> list: |
| """Return the most recent `limit` query records, newest first.""" |
| if limit <= 0: |
| return [] |
| return list(reversed(self._query_log[-limit:])) |
|
|
|
|
|
|
| def incr(self, key: str, amount: int = 1) -> None:
|
| """Increment a top-level counter in memory."""
|
| self._counters[key] = self._counters.get(key, 0) + amount
|
| self._dirty = True
|
|
|
| def incr_bucket(self, bucket: str, key: str, amount: int = 1) -> None:
|
| """Increment a nested bucket counter in memory."""
|
| if bucket not in self._buckets:
|
| self._buckets[bucket] = {}
|
| self._buckets[bucket][key] = self._buckets[bucket].get(key, 0) + amount
|
| self._dirty = True
|
|
|
| def snapshot(self) -> Dict[str, Any]:
|
| """Return current on-disk state merged with in-memory pending counters."""
|
| data = self._read_from_disk()
|
|
|
| for k, v in self._counters.items():
|
| data[k] = int(data.get(k, 0)) + v
|
| for bucket, keys in self._buckets.items():
|
| b = data.setdefault(bucket, {})
|
| for k, v in keys.items():
|
| b[k] = int(b.get(k, 0)) + v
|
| return data
|
|
|
|
|
|
|
| async def flush(self) -> None:
|
| """Merge in-memory counters to disk. No-op if nothing changed."""
|
| if not self._dirty:
|
| return
|
| async with self._lock:
|
| if not self._dirty:
|
| return
|
| data = self._read_from_disk()
|
| for k, v in self._counters.items():
|
| data[k] = int(data.get(k, 0)) + v
|
| for bucket, keys in self._buckets.items():
|
| b = data.setdefault(bucket, {})
|
| for k, v in keys.items():
|
| b[k] = int(b.get(k, 0)) + v
|
| if not self._write_to_disk(data): |
| |
| return |
| self._counters.clear()
|
| self._buckets.clear()
|
| self._dirty = False
|
|
|
|
|
|
|
| def _read_from_disk(self) -> Dict[str, Any]:
|
| if not self.path.exists():
|
| return deepcopy(_DEFAULT_STATE) |
| try: |
| loaded = json.loads(self.path.read_text(encoding="utf-8")) |
| if not isinstance(loaded, dict): |
| raise ValueError("metrics state must be a JSON object") |
| data = deepcopy(_DEFAULT_STATE) |
| for key, value in loaded.items(): |
| if isinstance(value, dict): |
| if key in _DEFAULT_STATE and not isinstance(_DEFAULT_STATE[key], dict): |
| continue |
| data[key] = { |
| str(k): v for k, v in value.items() |
| if isinstance(v, int) and not isinstance(v, bool) and v >= 0 |
| } |
| elif isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0: |
| if key not in {"intents", "errors"}: |
| data[key] = value |
| return data |
| except Exception as exc:
|
| logger.warning( |
| "MetricsService: failed to parse metrics file %s: %s", |
| self.path, exc, |
| ) |
| return deepcopy(_DEFAULT_STATE) |
|
|
| def _write_to_disk(self, data: Dict[str, Any]) -> bool: |
| data["last_updated"] = time.time()
|
| tmp = self.path.with_suffix(".tmp")
|
| try:
|
| tmp.write_text(
|
| json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
|
| )
|
| os.replace(tmp, self.path) |
| return True |
| except Exception as exc:
|
| logger.error("MetricsService: failed to write metrics: %s", exc)
|
| try:
|
| tmp.unlink(missing_ok=True)
|
| except Exception: |
| pass |
| return False |
|
|
