From 66da54fbfcac0731c45c338be13c8c34e76d8ef6 Mon Sep 17 00:00:00 2001 From: Cagri Yonca Date: Fri, 7 Aug 2026 12:39:37 +0200 Subject: [PATCH] feat(runtime): Add pauseMs and runCount metrics to runtime Signed-off-by: Cagri Yonca --- src/instana/collector/helpers/runtime.py | 87 ++++++++++++++++--- .../helpers/test_collector_runtime.py | 80 ++++++++++++++++- 2 files changed, 154 insertions(+), 13 deletions(-) diff --git a/src/instana/collector/helpers/runtime.py b/src/instana/collector/helpers/runtime.py index 8156ffd9..2b7296ff 100644 --- a/src/instana/collector/helpers/runtime.py +++ b/src/instana/collector/helpers/runtime.py @@ -1,16 +1,18 @@ -# (c) Copyright IBM Corp. 2021 +# (c) Copyright IBM Corp. 2021, 2026 # (c) Copyright Instana Inc. 2020 """Collection helper for the Python runtime""" +import contextlib import gc import importlib.metadata import os import platform import sys import threading +import time from types import ModuleType -from typing import Any, Callable, Dict, List, Union +from typing import Any, Callable, Union from instana.collector.base import BaseCollector from instana.collector.helpers.base import BaseHelper @@ -49,7 +51,16 @@ def __init__( else: self.previous_gc_count = None - def collect_metrics(self, **kwargs: Dict[str, Any]) -> List[Dict[str, Any]]: + # GC pause accumulators — flushed on every _collect_gc_metrics() call. + # Populated by _gc_callback() via gc.callbacks (Python 3.3+). + self._gc_start_times = {} + self._gc_pause_total_ms = 0.0 + self._gc_run_count = 0 + + if gc.isenabled(): + gc.callbacks.append(self._gc_callback) + + def collect_metrics(self, with_snapshot: bool = False, **kwargs: object) -> list[dict[str, Any]]: # noqa: ARG002 plugin_data = dict() try: plugin_data["name"] = "com.instana.plugin.python" @@ -64,7 +75,6 @@ def collect_metrics(self, **kwargs: Dict[str, Any]) -> List[Dict[str, Any]]: else: plugin_data["data"]["pid"] = str(os.getpid()) - with_snapshot = kwargs.get("with_snapshot", False) self._collect_runtime_metrics(plugin_data, with_snapshot) if with_snapshot: @@ -75,13 +85,14 @@ def collect_metrics(self, **kwargs: Dict[str, Any]) -> List[Dict[str, Any]]: def _collect_runtime_metrics( self, - plugin_data: Dict[str, Any], + plugin_data: dict[str, Any], with_snapshot: bool, ) -> None: if os.environ.get("INSTANA_DISABLE_METRICS_COLLECTION", False): return """ Collect up and return the runtime metrics """ + rusage = self.previous_rusage try: rusage = get_resource_usage() if gc.isenabled(): @@ -230,7 +241,39 @@ def _collect_runtime_metrics( finally: self.previous_rusage = rusage - def _collect_gc_metrics(self, plugin_data, with_snapshot): + def _gc_callback(self, phase: str, info: dict[str, Any]) -> None: + """Accumulate GC pause time and run count via gc.callbacks. + + Called by CPython twice per GC cycle: once with phase='start' and once + with phase='stop'. Only primitive operations are performed here — no + new Python objects are allocated — so the callback cannot trigger + additional GC cycles. + """ + generation = info["generation"] + if phase == "start": + self._gc_start_times[generation] = time.perf_counter() + elif phase == "stop" and generation in self._gc_start_times: + elapsed_ms = ( + time.perf_counter() - self._gc_start_times.pop(generation) + ) * 1000 + self._gc_pause_total_ms += elapsed_ms + self._gc_run_count += 1 + + def close(self) -> None: + """Remove the GC callback registered in __init__. + + Must be called when the helper is torn down (agent reconnect, test + teardown) to prevent stale callbacks accumulating in gc.callbacks, + which is a process-level list. + """ + with contextlib.suppress(ValueError): + gc.callbacks.remove(self._gc_callback) + + def _collect_gc_metrics( + self, + plugin_data: dict[str, Any], + with_snapshot: bool, + ) -> None: try: gc_count = gc.get_count() gc_threshold = gc.get_threshold() @@ -278,12 +321,35 @@ def _collect_gc_metrics(self, plugin_data, with_snapshot): "threshold2", with_snapshot, ) + + # Flush accumulated pause metrics atomically: copy to locals first + # so any GC cycle firing between read and reset is attributed to + # the next window rather than being lost. + pause_ms = self._gc_pause_total_ms + run_count = self._gc_run_count + self._gc_pause_total_ms = 0.0 + self._gc_run_count = 0 + + self.apply_delta( + pause_ms, + self.previous["data"]["metrics"]["gc"], + plugin_data["data"]["metrics"]["gc"], + "pauseMs", + with_snapshot, + ) + self.apply_delta( + run_count, + self.previous["data"]["metrics"]["gc"], + plugin_data["data"]["metrics"]["gc"], + "runCount", + with_snapshot, + ) except Exception: logger.debug("_collect_gc_metrics", exc_info=True) def _collect_thread_metrics( self, - plugin_data: Dict[str, Any], + plugin_data: dict[str, Any], with_snapshot: bool, ) -> None: try: @@ -321,7 +387,7 @@ def _collect_thread_metrics( def _collect_runtime_snapshot( self, - plugin_data: Dict[str, Any], + plugin_data: dict[str, Any], ) -> None: """Gathers Python specific Snapshot information for this process""" snapshot_payload = {} @@ -359,7 +425,7 @@ def _collect_runtime_snapshot( plugin_data["data"]["snapshot"] = snapshot_payload - def gather_python_packages(self) -> Dict[str, Any]: + def gather_python_packages(self) -> dict[str, Any]: """Collect up the list of modules in use""" if os.environ.get("INSTANA_DISABLE_PYTHON_PACKAGE_COLLECTION"): return {"instana": VERSION} @@ -408,7 +474,7 @@ def gather_python_packages(self) -> Dict[str, Any]: def jsonable( self, - value: Union[Callable[[], Any], ModuleType, Any], + value: Union[Callable[[], str], ModuleType, object], ) -> str: try: if callable(value): @@ -423,3 +489,4 @@ def jsonable( return str(result) except Exception: logger.debug("jsonable: ", exc_info=True) + return "" diff --git a/tests/collector/helpers/test_collector_runtime.py b/tests/collector/helpers/test_collector_runtime.py index 959f7611..3e94f7dd 100644 --- a/tests/collector/helpers/test_collector_runtime.py +++ b/tests/collector/helpers/test_collector_runtime.py @@ -20,14 +20,18 @@ def _resource(self) -> Generator[None, None, None]: ), ) yield + self.helper.close() self.helper = None def test_default_while_gc_disabled(self) -> None: import gc gc.disable() - helper = RuntimeHelper(collector=HostCollector(HostAgent())) - assert helper.previous_gc_count is None + try: + helper = RuntimeHelper(collector=HostCollector(HostAgent())) + assert helper.previous_gc_count is None + finally: + gc.enable() def test_collect_metrics(self) -> None: response = self.helper.collect_metrics() @@ -66,7 +70,77 @@ def test_collect_gc_metrics(self) -> None: plugin_data = self.helper.collect_metrics() self.helper._collect_gc_metrics(plugin_data[0], True) - assert len(self.helper.previous["data"]["metrics"]["gc"]) == 6 + assert len(self.helper.previous["data"]["metrics"]["gc"]) == 8 + + def test_gc_callback_registered(self) -> None: + import gc + + gc.enable() + helper = RuntimeHelper(collector=HostCollector(HostAgent())) + try: + assert helper._gc_callback in gc.callbacks + finally: + helper.close() + + def test_gc_callback_removed_on_close(self) -> None: + import gc + + gc.enable() + helper = RuntimeHelper(collector=HostCollector(HostAgent())) + helper.close() + assert helper._gc_callback not in gc.callbacks + + def test_gc_callback_not_registered_when_gc_disabled(self) -> None: + import gc + + gc.disable() + try: + helper = RuntimeHelper(collector=HostCollector(HostAgent())) + assert helper._gc_callback not in gc.callbacks + finally: + gc.enable() + + def test_gc_callback_accumulates_pause(self) -> None: + import gc + + gc.enable() + helper = RuntimeHelper(collector=HostCollector(HostAgent())) + try: + assert helper._gc_pause_total_ms == 0.0 + assert helper._gc_run_count == 0 + + # Simulate one complete GC cycle (gen-0) + helper._gc_callback("start", {"generation": 0}) + helper._gc_callback("stop", {"generation": 0}) + + assert helper._gc_pause_total_ms > 0.0 + assert helper._gc_run_count == 1 + finally: + helper.close() + + def test_gc_callback_flushes_on_collect(self) -> None: + import gc + + gc.enable() + helper = RuntimeHelper(collector=HostCollector(HostAgent())) + try: + # Simulate a GC pause before collection + helper._gc_callback("start", {"generation": 0}) + helper._gc_callback("stop", {"generation": 0}) + assert helper._gc_run_count == 1 + + plugin_data = helper.collect_metrics() + helper._collect_gc_metrics(plugin_data[0], True) + + # Accumulators must be reset after flush + assert helper._gc_pause_total_ms == 0.0 + assert helper._gc_run_count == 0 + + gc_data = plugin_data[0]["data"]["metrics"]["gc"] + assert "pauseMs" in gc_data + assert "runCount" in gc_data + finally: + helper.close() def test_collect_runtime_metrics(self) -> None: """Test that _collect_runtime_metrics properly collects metrics"""