Instrumenting The Python Garbage Collector with OpenTelemetry

gc_instrument.py

import gc  
import time  
from collections import defaultdict  
  
import attrs  
from opentelemetry import metrics  
from opentelemetry.metrics import CallbackOptions, Observation  
  
  
@attrs.define  
class GCInstrumenter:  
    # collected synchronously in the callback  
    duration_hist_ns: metrics.Histogram = attrs.field()  
    collected_objs_hist: metrics.Histogram = attrs.field()  
    uncollected_objs_hist: metrics.Histogram = attrs.field()  
  
    # collected asynchronously w/ observations  
    total_collection_time_by_gen: dict[int, int] = attrs.field(  
        factory=lambda: defaultdict(int)  
    )  
  
    last_collection_start: int | None = None  
  
    def callback(self, phase: str, info: dict[str, int]) -> None:  
        if phase == "start":  
            self.last_collection_start = time.monotonic_ns()  
        elif phase == "stop" and self.last_collection_start is not None:  
            now = time.monotonic_ns()  
            duration_ns = now - self.last_collection_start  
            gen = info["generation"]  
            self.total_collection_time_by_gen[gen] += duration_ns  
            self.last_collection_start = None  
  
            self.collected_objs_hist.record(  
                info["collected"], {"generation": gen}  
            )  
            self.uncollected_objs_hist.record(  
                info["uncollectable"], {"generation": gen}  
            )  
            self.duration_hist_ns.record(duration_ns, {"generation": gen})  
  
    def observe_total_gc_time(  
        self, _options: CallbackOptions  
    ) -> list[Observation]:  
        return [  
            Observation(ns / 1e9, attributes={"generation": gen})  
            for gen, ns in self.total_collection_time_by_gen.items()  
        ]  
  
  
def setup_gc_metrics() -> None:  
    meter = metrics.get_meter(__name__)  
    gc_duration_hist = meter.create_histogram(  
        name="cio_gc_duration",  
        description="measures the duration of garbage collection",  
        unit="ns",  
    )  
    gc_collectable_hist = meter.create_histogram(  
        name="cio_gc_collectable_objs",  
        description="The number of collectable objects in each gc run",  
        unit="objects",  
    )  
    gc_uncollectable_hist = meter.create_histogram(  
        name="cio_gc_uncollectable_objs",  
        description="The number of uncollectable objects in each gc run",  
        unit="objects",  
    )  
  
    gc_instrumenter = GCInstrumenter(  
        duration_hist_ns=gc_duration_hist,  
        collected_objs_hist=gc_collectable_hist,  
        uncollected_objs_hist=gc_uncollectable_hist,  
    )  
  
    gc.callbacks.append(gc_instrumenter.callback)  
  
    # Note: observation is implemented as an asynchronous counter so that  
    # we can track the GC time internally in nanoseconds (as ints), but
    # report observation metrics in seconds (as floats) without losing
    # precision.
    meter.create_observable_counter(  
        name="cio_gc_time",  
        description="Time spent doing garbage collection (s)",  
        callbacks=[gc_instrumenter.observe_total_gc_time],  
    )