Spaces:
Running
Running
| from __future__ import annotations | |
| from collections import Counter | |
| from threading import Lock | |
| def metric_name(name: str) -> str: return f'matrix_builder_{name}' | |
| class MetricsRegistry: | |
| def __init__(self) -> None: self._lock=Lock(); self._counters: Counter[str]=Counter() | |
| def inc(self, name: str, amount: int=1) -> None: | |
| with self._lock: self._counters[metric_name(name)] += amount | |
| def render_prometheus(self) -> str: | |
| with self._lock: | |
| lines=['# HELP matrix_builder_info Matrix Builder service information','# TYPE matrix_builder_info gauge','matrix_builder_info{service="matrix-builder-api"} 1'] | |
| for name,value in sorted(self._counters.items()): lines += [f'# HELP {name} Matrix Builder counter', f'# TYPE {name} counter', f'{name} {value}'] | |
| return '\n'.join(lines)+'\n' | |
| metrics_registry=MetricsRegistry() | |