chore(tracing): emit five CPython GC collection and pause runtime metrics (PROF-15855)

ddtrace/internal/runtime/constants.py

@@ -2,6 +2,15 @@
 GC_COUNT_GEN1 = "runtime.python.gc.count.gen1"
 GC_COUNT_GEN2 = "runtime.python.gc.count.gen2"
 
+# Deltas of gc.get_stats() collections over the flush interval.
+GC_COLLECTIONS_GEN0: str = "runtime.python.gc.collections.gen0"
+GC_COLLECTIONS_GEN1: str = "runtime.python.gc.collections.gen1"
+GC_COLLECTIONS_GEN2: str = "runtime.python.gc.collections.gen2"
+
+# Stop-the-world pause over the flush interval, nanoseconds.
+GC_PAUSE_TIME: str = "runtime.python.gc.pause.time"
+GC_PAUSE_MAX: str = "runtime.python.gc.pause.max"
+
 THREAD_COUNT = "runtime.python.thread_count"
 MEM_RSS = "runtime.python.mem.rss"
 # `runtime.python.cpu.time.sys` metric is used to auto-enable runtime metrics dashboards in DD backend
@@ -11,7 +20,18 @@
 CTX_SWITCH_VOLUNTARY = "runtime.python.cpu.ctx_switch.voluntary"
 CTX_SWITCH_INVOLUNTARY = "runtime.python.cpu.ctx_switch.involuntary"
 
-GC_RUNTIME_METRICS = set([GC_COUNT_GEN0, GC_COUNT_GEN1, GC_COUNT_GEN2])
+GC_RUNTIME_METRICS: set[str] = set(
+    [
+        GC_COUNT_GEN0,
+        GC_COUNT_GEN1,
+        GC_COUNT_GEN2,
+        GC_COLLECTIONS_GEN0,
+        GC_COLLECTIONS_GEN1,
+        GC_COLLECTIONS_GEN2,
+        GC_PAUSE_TIME,
+        GC_PAUSE_MAX,
+    ]
+)
 
 NATIVE_PROCESS_RUNTIME_METRICS = set(
     [THREAD_COUNT, MEM_RSS, CTX_SWITCH_VOLUNTARY, CTX_SWITCH_INVOLUNTARY, CPU_TIME_SYS, CPU_TIME_USER, CPU_PERCENT]

ddtrace/internal/runtime/gc_monitor.py

@@ -0,0 +1,212 @@
+"""Process-wide CPython GC pause observer.
+
+One gc.callbacks subscriber. Install is refcounted via acquire/release.
+Runtime metrics drain a snapshot on each flush.
+"""
+
+from __future__ import annotations
+
+import gc
+import logging
+import sys
+import threading
+import time
+from types import FrameType
+from typing import Callable
+from typing import NamedTuple
+from typing import Optional
+
+from ddtrace.internal import forksafe
+from ddtrace.internal.logger import get_logger
+
+
+log: logging.Logger = get_logger(__name__)
+
+GEN_COUNT: int = 3
+
+# (generation, pause_ns, start_ns, triggering frame or None)
+PauseListener = Callable[[int, int, int, Optional[FrameType]], None]
+
+
+class _GenWindow:
+    __slots__ = ("count", "total_ns", "max_ns")
+    count: int
+    total_ns: int
+    max_ns: int
+
+    def __init__(self) -> None:
+        self.count = 0
+        self.total_ns = 0
+        self.max_ns = 0
+
+
+class GCPauseSnapshot(NamedTuple):
+    n_pauses: int
+    total_ns: int
+    max_ns: int
+    # (n_pauses, total_ns, max_ns) per generation
+    per_gen: tuple[tuple[int, int, int], ...]
+
+    @classmethod
+    def zeros(cls) -> GCPauseSnapshot:
+        empty: tuple[int, int, int] = (0, 0, 0)
+        return cls(0, 0, 0, (empty, empty, empty))
+
+
+class GCPauseMonitor:
+    """Single gc.callbacks subscriber with refcounted install."""
+
+    _lock: threading.RLock
+    _refcount: int
+    _fork_registered: bool
+    _start_ns: list[int]
+    _count: int
+    _total_ns: int
+    _max_ns: int
+    _per_gen: list[_GenWindow]
+    _listeners: list[PauseListener]
+
+    def __init__(self) -> None:
+        # RLock: snapshot_and_reset allocates, which can reenter GC in this thread.
+        self._lock = threading.RLock()
+        self._refcount = 0
+        self._fork_registered = False
+        self._start_ns = [0] * GEN_COUNT
+        self._count = 0
+        self._total_ns = 0
+        self._max_ns = 0
+        self._per_gen = [_GenWindow() for _ in range(GEN_COUNT)]
+        self._listeners = []
+
+    def acquire(self) -> None:
+        with self._lock:
+            self._refcount += 1
+            if self._refcount == 1:
+                if self._on_gc not in gc.callbacks:
+                    gc.callbacks.append(self._on_gc)
+                if not self._fork_registered:
+                    forksafe.register(self.reset)
+                    self._fork_registered = True
+
+    def release(self) -> None:
+        with self._lock:
+            if self._refcount <= 0:
+                return
+            self._refcount -= 1
+            if self._refcount == 0:
+                try:
+                    gc.callbacks.remove(self._on_gc)
+                except ValueError:
+                    pass
+                # Drop in-flight starts so a later re-acquire cannot pair a
+                # new stop with a stale timestamp from before uninstall.
+                self._start_ns = [0] * GEN_COUNT
+                self._clear_window()
+
+    def add_listener(self, listener: PauseListener) -> None:
+        with self._lock:
+            self._listeners.append(listener)
+
+    def remove_listener(self, listener: PauseListener) -> None:
+        with self._lock:
+            try:
+                self._listeners.remove(listener)
+            except ValueError:
+                pass
+
+    def reset(self) -> None:
+        """Drop in-flight starts and the current window. Used after fork."""
+        with self._lock:
+            self._start_ns = [0] * GEN_COUNT
+            self._clear_window()
+
+    def snapshot_and_reset(self) -> GCPauseSnapshot:
+        # Copy primitives, then clear, then allocate. A reentrant GC callback
+        # during NamedTuple/tuple construction must land in the next window.
+        with self._lock:
+            n_pauses: int = self._count
+            total_ns: int = self._total_ns
+            max_ns: int = self._max_ns
+            g0: _GenWindow = self._per_gen[0]
+            g1: _GenWindow = self._per_gen[1]
+            g2: _GenWindow = self._per_gen[2]
+            c0: int = g0.count
+            t0: int = g0.total_ns
+            m0: int = g0.max_ns
+            c1: int = g1.count
+            t1: int = g1.total_ns
+            m1: int = g1.max_ns
+            c2: int = g2.count
+            t2: int = g2.total_ns
+            m2: int = g2.max_ns
+            self._clear_window()
+        per_gen: tuple[tuple[int, int, int], ...] = ((c0, t0, m0), (c1, t1, m1), (c2, t2, m2))
+        return GCPauseSnapshot(n_pauses, total_ns, max_ns, per_gen)
+
+    def _clear_window(self) -> None:
+        self._count = 0
+        self._total_ns = 0
+        self._max_ns = 0
+        for w in self._per_gen:
+            w.count = 0
+            w.total_ns = 0
+            w.max_ns = 0
+
+    def _on_gc(self, phase: str, info: dict[str, int]) -> None:
+        # Do not allocate on the metrics-only path: object creation in a GC
+        # callback can recurse.
+        try:
+            gen: int = info.get("generation", 0)
+            if not 0 <= gen < GEN_COUNT:
+                return
+            if phase == "start":
+                with self._lock:
+                    self._start_ns[gen] = time.monotonic_ns()
+                return
+            if phase != "stop":
+                return
+            with self._lock:
+                start: int = self._start_ns[gen]
+                if start == 0:
+                    return
+                self._start_ns[gen] = 0
+                pause_ns: int = time.monotonic_ns() - start
+                if pause_ns < 0:
+                    return
+                window: _GenWindow = self._per_gen[gen]
+                window.count += 1
+                window.total_ns += pause_ns
+                if pause_ns > window.max_ns:
+                    window.max_ns = pause_ns
+                self._count += 1
+                self._total_ns += pause_ns
+                if pause_ns > self._max_ns:
+                    self._max_ns = pause_ns
+                listeners: Optional[tuple[PauseListener, ...]] = tuple(self._listeners) if self._listeners else None
+            if not listeners:
+                return
+            frame: Optional[FrameType]
+            try:
+                frame = sys._getframe(1)
+            except ValueError:
+                frame = None
+            for listener in listeners:
+                try:
+                    listener(gen, pause_ns, start, frame)
+                except Exception:
+                    log.debug("GC pause listener failed", exc_info=True)
+        except Exception:
+            log.debug("GC pause monitor callback failed", exc_info=True)
+
+
+_MONITOR: Optional[GCPauseMonitor] = None
+_MONITOR_LOCK: threading.Lock = threading.Lock()
+
+
+def gc_pause_monitor() -> GCPauseMonitor:
+    """Process-wide monitor."""
+    global _MONITOR
+    with _MONITOR_LOCK:
+        if _MONITOR is None:
+            _MONITOR = GCPauseMonitor()
+        return _MONITOR

ddtrace/internal/runtime/metric_collectors.py

@@ -1,5 +1,7 @@
 import time
+from types import ModuleType
 from typing import NamedTuple
+from typing import Optional
 
 from .. import forksafe
 from .collector import ValueCollector
@@ -8,34 +10,107 @@
 from .constants import CPU_TIME_USER
 from .constants import CTX_SWITCH_INVOLUNTARY
 from .constants import CTX_SWITCH_VOLUNTARY
+from .constants import GC_COLLECTIONS_GEN0
+from .constants import GC_COLLECTIONS_GEN1
+from .constants import GC_COLLECTIONS_GEN2
 from .constants import GC_COUNT_GEN0
 from .constants import GC_COUNT_GEN1
 from .constants import GC_COUNT_GEN2
+from .constants import GC_PAUSE_MAX
+from .constants import GC_PAUSE_TIME
 from .constants import MEM_RSS
 from .constants import THREAD_COUNT
+from .gc_monitor import GEN_COUNT
+from .gc_monitor import GCPauseMonitor
+from .gc_monitor import GCPauseSnapshot
+from .gc_monitor import gc_pause_monitor
 
 
 class RuntimeMetricCollector(ValueCollector):
     value = []  # type: list[tuple[str, str]]
     periodic = True
 
 
+def _read_gc_collections(gc_mod: ModuleType) -> list[int]:
+    """Return per-generation collections counts from gc.get_stats()."""
+    stats: list[dict[str, int]] = gc_mod.get_stats()
+    collections: list[int] = []
+    for i in range(GEN_COUNT):
+        row: dict[str, int] = stats[i] if i < len(stats) else {}
+        collections.append(int(row.get("collections", 0)))
+    return collections
+
+
+def _delta(current: list[int], previous: list[int]) -> list[int]:
+    return [max(0, c - p) for c, p in zip(current, previous)]
+
+
 class GCRuntimeMetricCollector(RuntimeMetricCollector):
-    """Collector for garbage collection generational counts
+    """Collector for CPython GC counts, collection stats, and STW pause time.
 
-    More information at https://docs.python.org/3/library/gc.html
+    gc.count.genN remains the gc.get_count() allocation counters.
+    Collection/pause metrics are interval deltas, matching CPU time.
     """
 
     required_modules = ["gc"]
+    _monitor: Optional[GCPauseMonitor] = None
+    _prev_collections: list[int]
 
-    def collect_fn(self, keys):
-        gc = self.modules.get("gc")
+    def _on_modules_load(self) -> None:
+        monitor: Optional[GCPauseMonitor] = None
+        try:
+            gc_mod: ModuleType = self.modules["gc"]
+            # Seed collections before gc.callbacks is installed so an enable-time
+            # collection is not a pause without a matching collections delta.
+            self._prev_collections: list[int] = _read_gc_collections(gc_mod)
+            monitor = gc_pause_monitor()
+            monitor.acquire()
+            forksafe.register(self._reset_state)
+        except Exception:
+            if monitor is not None:
+                monitor.release()
+            self.enabled = False
+            return
+        self._monitor = monitor
 
-        counts = gc.get_count()
-        metrics = [
+    def _reset_state(self) -> None:
+        gc_mod: Optional[ModuleType] = self.modules.get("gc")
+        if gc_mod is None:
+            return
+        self._prev_collections: list[int] = _read_gc_collections(gc_mod)
+
+    def stop(self) -> None:
+        monitor: Optional[GCPauseMonitor] = self._monitor
+        if monitor is not None:
+            self._monitor = None
+            monitor.release()
+            forksafe.unregister(self._reset_state)
+
+    def collect_fn(self, keys: Optional[set[str]]) -> list[tuple[str, int]]:
+        # Snapshot first so flush allocations are not attributed to this window,
+        # and so stop() cannot None-out _monitor between the check and the call.
+        monitor: Optional[GCPauseMonitor] = self._monitor
+        pause: Optional[GCPauseSnapshot]
+        if monitor is not None:
+            pause = monitor.snapshot_and_reset()
+        else:
+            pause = None
+
+        gc_mod: ModuleType = self.modules["gc"]
+        counts: tuple[int, int, int] = gc_mod.get_count()
+        collections: list[int] = _read_gc_collections(gc_mod)
+        d_collections: list[int] = _delta(collections, self._prev_collections)
+        self._prev_collections: list[int] = collections
+
+        metrics: list[tuple[str, int]] = [
             (GC_COUNT_GEN0, counts[0]),
             (GC_COUNT_GEN1, counts[1]),
             (GC_COUNT_GEN2, counts[2]),
+            (GC_COLLECTIONS_GEN0, d_collections[0]),
+            (GC_COLLECTIONS_GEN1, d_collections[1]),
+            (GC_COLLECTIONS_GEN2, d_collections[2]),
+            (GC_PAUSE_TIME, 0 if pause is None else pause.total_ns),
+            (GC_PAUSE_MAX, 0 if pause is None else pause.max_ns),
         ]
 
         return metrics

ddtrace/internal/runtime/runtime_metrics.py

@@ -1,4 +1,5 @@
 import itertools
+from typing import Callable
 from typing import ClassVar  # noqa:F401
 from typing import Optional  # noqa:F401
 
@@ -36,6 +37,12 @@ def __iter__(self):
         collected = (collector.collect(self._enabled) for collector in self._collectors)
         return itertools.chain.from_iterable(collected)
 
+    def stop(self) -> None:
+        for collector in self._collectors:
+            stop: Optional[Callable[[], None]] = getattr(collector, "stop", None)
+            if callable(stop):
+                stop()
+
     def __repr__(self):
         return "{}(enabled={})".format(
             self.__class__.__name__,
@@ -124,6 +131,7 @@ def disable(cls) -> None:
             #    _eintr_retry_call (/usr/lib/python2.7/subprocess.py:125)
             # which is the eventual cause of the deadlock.
             cls._instance.join(1)
+            cls._instance._runtime_metrics.stop()
             cls._instance = None
             cls.enabled = False