Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 77 additions & 10 deletions src/instana/collector/helpers/runtime.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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"
Expand All @@ -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:
Expand All @@ -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():
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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 = {}
Expand Down Expand Up @@ -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}
Expand Down Expand Up @@ -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):
Expand All @@ -423,3 +489,4 @@ def jsonable(
return str(result)
except Exception:
logger.debug("jsonable: ", exc_info=True)
return ""
80 changes: 77 additions & 3 deletions tests/collector/helpers/test_collector_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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"""
Expand Down
Loading