File size: 6,653 Bytes
37ae25d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | """
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. _read() logs a warning (with raw content) on JSON parse failure instead of
silently returning zeros, so corrupt files are visible in logs.
5. _write() uses os.replace (atomic rename) to prevent corrupt state on crash.
"""
import asyncio
import json
import logging
import os
import time
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 # keep last 200 queries in memory
class MetricsService:
def __init__(self, path: Path):
self.path = Path(path)
self.path.parent.mkdir(parents=True, exist_ok=True)
self._lock = asyncio.Lock()
# In-memory accumulators — flushed to disk periodically
self._counters: Dict[str, int] = {}
self._buckets: Dict[str, Dict[str, int]] = {}
self._dirty = False
# In-memory query log (last N queries, never flushed to disk)
self._query_log: list = []
# Seed disk file if it doesn't exist
if not self.path.exists():
self._write_to_disk(dict(_DEFAULT_STATE))
# ── Public API (sync, no I/O) ──────────────────────────────────────────────
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.utcnow().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."""
return list(reversed(self._query_log[-limit:]))
# ── Public API (sync, no I/O) ──────────────────────────────────────────────
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()
# Merge pending in-memory counters without flushing
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 flush (called by background task + shutdown) ─────────────────────
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 # double-check after acquiring lock
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
self._write_to_disk(data)
self._counters.clear()
self._buckets.clear()
self._dirty = False
# ── Internal disk helpers ──────────────────────────────────────────────────
def _read_from_disk(self) -> Dict[str, Any]:
if not self.path.exists():
return dict(_DEFAULT_STATE)
try:
return json.loads(self.path.read_text(encoding="utf-8"))
except Exception as exc:
raw = ""
try:
raw = self.path.read_text(encoding="utf-8", errors="replace")[:200]
except Exception:
pass
logger.warning(
"MetricsService: failed to parse metrics file %s: %s | raw: %r",
self.path, exc, raw,
)
return dict(_DEFAULT_STATE)
def _write_to_disk(self, data: Dict[str, Any]) -> None:
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) # atomic on POSIX / Windows
except Exception as exc:
logger.error("MetricsService: failed to write metrics: %s", exc)
try:
tmp.unlink(missing_ok=True)
except Exception:
pass
|