1// utils.py
2
3import gc
4import time
5
6from opentelemetry import metrics
7
8
9def collect_gc_metrics():
10    meter = metrics.get_meter("monitor", "1.0.0")
11
12    gc_duration = meter.create_counter(
13        name="process.runtime.cpython.gc.duration",
14        description="Runtime cpython GC duration per generation",
15        unit="s",
16    )
17    gc_counter = meter.create_counter(
18        name="process.runtime.cpython.gc.collections",
19        description="Number of GC collections per generation",
20    )
21
22    gc_start_times = {}
23
24    def gc_callback(phase, info):
25        gen = info.get("generation")
26        if phase == "start":
27            gc_start_times[gen] = time.perf_counter()
28        elif phase == "stop":
29            start = gc_start_times.pop(gen, None)
30            gc_counter.add(1, attributes={"count": gen})
31            if start is not None:
32                duration = time.perf_counter() - start
33                gc_duration.add(duration, attributes={"count": gen})
34
35    gc.callbacks.append(gc_callback)