File size: 3,916 Bytes
0703e2b | 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 | """Minimal Prometheus metrics (text exposition format), no dependency."""
from __future__ import annotations
import threading
from typing import Dict, Iterable, List, Sequence, Tuple
Labels = Tuple[Tuple[str, str], ...]
def _fmt_labels(labels: Labels) -> str:
if not labels:
return ""
inner = ",".join(f'{k}="{str(v).replace(chr(92), chr(92) * 2).replace(chr(34), chr(92) + chr(34))}"' for k, v in labels)
return "{" + inner + "}"
class Counter:
kind = "counter"
def __init__(self, name: str, help: str) -> None:
self.name, self.help = name, help
self.values: Dict[Labels, float] = {}
self._lock = threading.Lock()
def inc(self, amount: float = 1.0, **labels: str) -> None:
key = tuple(sorted(labels.items()))
with self._lock:
self.values[key] = self.values.get(key, 0.0) + amount
def samples(self) -> Iterable[Tuple[str, Labels, float]]:
with self._lock:
return [(self.name, k, v) for k, v in self.values.items()]
class Gauge(Counter):
kind = "gauge"
def set(self, value: float, **labels: str) -> None:
key = tuple(sorted(labels.items()))
with self._lock:
self.values[key] = float(value)
class Histogram:
kind = "histogram"
def __init__(self, name: str, help: str, buckets: Sequence[float]) -> None:
self.name, self.help = name, help
self.buckets = sorted(buckets)
self.data: Dict[Labels, List[float]] = {} # per label set: bucket counts + [sum, count]
self._lock = threading.Lock()
def observe(self, x: float, **labels: str) -> None:
key = tuple(sorted(labels.items()))
with self._lock:
d = self.data.setdefault(key, [0.0] * (len(self.buckets) + 2))
for i, b in enumerate(self.buckets):
if x <= b:
d[i] += 1
d[-2] += x
d[-1] += 1
def samples(self):
out = []
with self._lock:
for key, d in self.data.items():
for i, b in enumerate(self.buckets):
out.append((self.name + "_bucket", key + (("le", repr(float(b))),), d[i]))
out.append((self.name + "_bucket", key + (("le", "+Inf"),), d[-1]))
out.append((self.name + "_sum", key, d[-2]))
out.append((self.name + "_count", key, d[-1]))
return out
class Registry:
def __init__(self) -> None:
self.metrics: List = []
self.collectors: List = [] # callables returning [(metric_name, kind, help, labels, value)]
def counter(self, name: str, help: str) -> Counter:
m = Counter(name, help)
self.metrics.append(m)
return m
def gauge(self, name: str, help: str) -> Gauge:
m = Gauge(name, help)
self.metrics.append(m)
return m
def histogram(self, name: str, help: str, buckets: Sequence[float]) -> Histogram:
m = Histogram(name, help, buckets)
self.metrics.append(m)
return m
def render(self) -> str:
lines: List[str] = []
for m in self.metrics:
lines.append(f"# HELP {m.name} {m.help}")
lines.append(f"# TYPE {m.name} {m.kind}")
for name, labels, v in m.samples():
lines.append(f"{name}{_fmt_labels(labels)} {v:g}")
families: Dict[str, Tuple[str, str, List[str]]] = {}
for collect in self.collectors:
for name, kind, help, labels, v in collect():
fam = families.setdefault(name, (kind, help, []))
fam[2].append(f"{name}{_fmt_labels(tuple(sorted(labels.items())))} {v:g}")
for name, (kind, help, samples) in families.items():
lines.append(f"# HELP {name} {help}")
lines.append(f"# TYPE {name} {kind}")
lines.extend(samples)
return "\n".join(lines) + "\n"
|