| """ |
| Agent Q3 [HQ] — Metrics |
| Prometheus-compatible /metrics endpoint + internal tracking. |
| """ |
| import time |
| from collections import defaultdict |
|
|
| _counters = defaultdict(int) |
| _latencies = defaultdict(list) |
| _start_time = time.time() |
|
|
| def track_request(target: str, latency_s: float): |
| _counters[f"requests_total_{target}"] += 1 |
| _latencies[target].append(latency_s) |
|
|
| def get_metrics() -> dict: |
| uptime = time.time() - _start_time |
| avg_latency = { |
| t: round(sum(v)/len(v), 3) if v else 0 |
| for t, v in _latencies.items() |
| } |
| return { |
| "uptime_seconds": round(uptime, 1), |
| "request_counts": dict(_counters), |
| "avg_latency_seconds": avg_latency, |
| } |
|
|
| def prometheus_format() -> str: |
| lines = [f"# Agent Q3 [HQ] Metrics\nagentq3_uptime_seconds {round(time.time()-_start_time,1)}"] |
| for k, v in _counters.items(): |
| lines.append(f"agentq3_{k} {v}") |
| return "\n".join(lines) |
|
|
