diff --git a/.sampo/changesets/macos-runtime-metrics.md b/.sampo/changesets/macos-runtime-metrics.md new file mode 100644 index 000000000..ef1f97f7e --- /dev/null +++ b/.sampo/changesets/macos-runtime-metrics.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: minor +--- + +Add opt-in, dependency-free macOS process, garbage collection, and host metrics with automatic process hostname attribution. diff --git a/_example_runtime_metrics.py b/_example_runtime_metrics.py new file mode 100644 index 000000000..80de3f532 --- /dev/null +++ b/_example_runtime_metrics.py @@ -0,0 +1,82 @@ +"""Bounded CPU and memory workload for example.py.""" + +import gc +import hashlib +import os +import time +from threading import Event, Thread + + +_MIB = 1024 * 1024 +_BLOCK_SIZE = 8 * _MIB + + +def _burn_cpu(stop: Event) -> None: + payload = b"x" * _MIB + # Hashing more than 2047 bytes releases the GIL, allowing parallel CPU work. + while not stop.is_set(): + hashlib.sha256(payload).digest() + + +def _resize_memory(blocks: list[bytearray], target_mib: int) -> None: + target_blocks = target_mib * _MIB // _BLOCK_SIZE + del blocks[target_blocks:] + while len(blocks) < target_blocks: + block = bytearray(_BLOCK_SIZE) + block[::4096] = b"\x01" * (len(block) // 4096) + blocks.append(block) + + +def run_workload(duration: float = 30 * 60, phase_duration: float = 30) -> None: + cpu_count = getattr(os, "process_cpu_count", os.cpu_count)() or 1 + max_workers = max(1, min(8, cpu_count // 2)) + try: + physical_memory = os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE") + memory_mib = max(0, min(1024, physical_memory // 32 // _MIB)) + except (AttributeError, OSError, ValueError): + memory_mib = 256 + phases = [ + ("idle", 0, memory_mib // 8), + ("ramp-up", 1, memory_mib // 4), + ("CPU burst", max_workers, memory_mib // 4), + ("CPU and memory peak", max_workers, memory_mib), + ("recovery", max(1, max_workers // 2), memory_mib // 2), + ("idle", 0, memory_mib // 8), + ] + blocks: list[bytearray] = [] + start = time.monotonic() + deadline = start + duration + phase_index = 0 + try: + while time.monotonic() < deadline: + name, worker_count, target_mib = phases[phase_index % len(phases)] + phase_end = min(deadline, start + (phase_index + 1) * phase_duration) + _resize_memory(blocks, target_mib) + for _ in range(10000): + cycle: list[object] = [] + cycle.append(cycle) + del cycle + gc.collect() + print( + f"Phase: {name}; CPU workers: {worker_count}; " + f"target memory: {target_mib} MiB; " + f"elapsed: {time.monotonic() - start:.0f}s / {duration:.0f}s", + flush=True, + ) + stop = Event() + workers = [] + try: + for _ in range(worker_count): + worker = Thread(target=_burn_cpu, args=(stop,), daemon=True) + worker.start() + workers.append(worker) + while time.monotonic() < phase_end: + time.sleep(min(1, max(0, phase_end - time.monotonic()))) + finally: + stop.set() + for worker in workers: + worker.join() + phase_index += 1 + finally: + blocks.clear() + gc.collect() diff --git a/example.py b/example.py index 8561fdc94..a426cc8d0 100644 --- a/example.py +++ b/example.py @@ -11,6 +11,9 @@ # 2. Run this script and choose from the interactive menu import os +import sys +from datetime import datetime, timezone +from uuid import uuid4 import posthog @@ -71,8 +74,9 @@ def load_env_file(): print(f"4. Flag dependencies examples{local_eval_note}") print("5. Context management and tagging examples") print("6. Run all examples") -print("7. Exit") -choice = input("\nEnter your choice (1-7): ").strip() +print("7. Process, host, and GC metrics (30-minute workload, macOS)") +print("8. Exit") +choice = input("\nEnter your choice (1-8): ").strip() if choice == "1": print("\n" + "=" * 60) @@ -487,13 +491,67 @@ def process_payment(payment_id): posthog.capture("demo_completed") print("✅ Demo completed with context tags") +elif choice == "7" and sys.platform == "darwin": + from _example_runtime_metrics import run_workload + + run_id = uuid4().hex + + def print_runtime_sample(sample): + labels = ", ".join( + f"{key}={value}" + for key, value in sample["attributes"].items() + if key + in ( + "cpu.mode", + "cpu.logical_number", + "cpython.gc.generation", + "system.paging.direction", + "system.paging.state", + "system.memory.state", + "process.context_switch.type", + ) + ) + name = sample["name"] + (f"{{{labels}}}" if labels else "") + print( + f"{datetime.now(timezone.utc).isoformat()} " + f"{name} = {sample['value']:.2f} {sample['unit']}", + flush=True, + ) + return sample + + posthog.metrics = { + "capture_runtime_metrics": True, + "capture_host_metrics": True, + "service_name": "python-runtime-metrics-demo", + "flush_interval": 5, + "resource_attributes": {"demo.run_id": run_id}, + "before_send": print_runtime_sample, + } + posthog.setup() + print(f"Service: python-runtime-metrics-demo; demo.run_id: {run_id}", flush=True) + print( + "Running for 30 minutes with 30-second phases. Press Ctrl+C to stop.", + flush=True, + ) + try: + run_workload() + except KeyboardInterrupt: + print("\nMetrics workload stopped.", flush=True) + finally: + posthog.shutdown() + elif choice == "7": + print("Automatic runtime metrics are currently supported only on macOS.") + posthog.shutdown() + exit(1) + +elif choice == "8": print("👋 Goodbye!") posthog.shutdown() exit() else: - print("❌ Invalid choice. Please run again and select 1-7.") + print("❌ Invalid choice. Please run again and select 1-8.") posthog.shutdown() exit() diff --git a/posthog/__init__.py b/posthog/__init__.py index d718906f0..02a651d1e 100644 --- a/posthog/__init__.py +++ b/posthog/__init__.py @@ -339,9 +339,13 @@ def get_tags() -> Dict[str, Any]: Set to 0 to disable retries. super_properties: Properties merged into every captured event. metrics: Config dict for the ``client.metrics`` API (``service_name``, - ``service_version``, ``environment``, ``flush_interval``, ...). Applied - when ``setup()`` builds the global client, or on a later ``setup()`` - call if the metrics API hasn't been used yet. + ``service_version``, ``environment``, ``flush_interval``, ...). + ``capture_runtime_metrics=True`` automatically collects process CPU, + RSS memory and its fraction of total physical RAM, native thread count, + CPython GC statistics, and total context switches on macOS; disabled by + default. Applied when + ``setup()`` builds the global client, or on a later ``setup()`` call + if the metrics API hasn't been used yet. traces: Config dict for distributed tracing: ``service_name``, ``service_version``, ``environment``, ``resource_attributes``, ``flush_interval`` (5 s), ``max_queue_size`` (2048), @@ -1444,6 +1448,7 @@ def setup() -> Client: # already forced setup()) still applies until the metrics API is first used. if default_client._metrics is None: default_client._metrics_config = metrics + default_client._start_runtime_metrics() # traces=None means off, so the module option applies only where the client # has none; False is the latch of an init that failed and stays off. if traces is not None and default_client._traces_config is None: diff --git a/posthog/_runtime_metrics.py b/posthog/_runtime_metrics.py new file mode 100644 index 000000000..513fd2958 --- /dev/null +++ b/posthog/_runtime_metrics.py @@ -0,0 +1,584 @@ +import ctypes +import logging +import os +import sys +from collections import deque +from dataclasses import dataclass +from gc import callbacks as gc_callbacks +from gc import get_stats +from resource import RUSAGE_SELF, getrusage +from socket import gethostname +from threading import Lock +from time import monotonic, monotonic_ns, time +from uuid import uuid4 +from weakref import finalize + +log = logging.getLogger("posthog") + + +class _ProcTaskInfo(ctypes.Structure): + # ABI layout of proc_taskinfo in macOS . + _fields_ = [ + ("pti_virtual_size", ctypes.c_uint64), + ("pti_resident_size", ctypes.c_uint64), + ("pti_total_user", ctypes.c_uint64), + ("pti_total_system", ctypes.c_uint64), + ("pti_threads_user", ctypes.c_uint64), + ("pti_threads_system", ctypes.c_uint64), + ("pti_policy", ctypes.c_int32), + ("pti_faults", ctypes.c_int32), + ("pti_pageins", ctypes.c_int32), + ("pti_cow_faults", ctypes.c_int32), + ("pti_messages_sent", ctypes.c_int32), + ("pti_messages_received", ctypes.c_int32), + ("pti_syscalls_mach", ctypes.c_int32), + ("pti_syscalls_unix", ctypes.c_int32), + ("pti_csw", ctypes.c_int32), + ("pti_threadnum", ctypes.c_int32), + ("pti_numrunning", ctypes.c_int32), + ("pti_priority", ctypes.c_int32), + ] + + +@dataclass(frozen=True) +class _RuntimeMetric: + name: str + value: float + unit: str + metric_type: str = "observable_counter" + attributes: dict | None = None + start_time_ms: int | None = None + + +_MAX_GC_DURATION_SAMPLES = 4096 + + +class _GCDurationMetrics: + def __init__(self) -> None: + self._pid = os.getpid() + self._lock = Lock() + self._starts: list[int | None] = [None, None, None] + self._pending: deque[tuple[int, float]] = deque() + self._window_start_ms = int(time() * 1000) + self._skipped = False + self._warned = False + self._active = True + self._callback = self._on_gc + gc_callbacks.append(self._callback) + + def close(self) -> None: + self._active = False + try: + gc_callbacks.remove(self._callback) + except ValueError: + pass + + def _on_gc(self, phase: str, info: dict) -> None: + if not self._active or os.getpid() != self._pid: + return + generation = info.get("generation", -1) + if generation not in (0, 1, 2) or phase not in ("start", "stop"): + return + # GC may interrupt a thread draining this buffer. Never wait on its lock + # or call hooks, logging, or transport from inside the callback. + if not self._lock.acquire(False): + self._starts[generation] = None + self._skipped = True + return + try: + if phase == "start": + self._starts[generation] = monotonic_ns() + else: + start = self._starts[generation] + self._starts[generation] = None + if start is not None: + elapsed = monotonic_ns() - start + if elapsed >= 0: + if len(self._pending) < _MAX_GC_DURATION_SAMPLES: + self._pending.append((generation, elapsed / 1_000_000)) + else: + self._skipped = True + except Exception: + self._starts[generation] = None + self._skipped = True + finally: + self._lock.release() + + def reset(self) -> None: + pending: deque[tuple[int, float]] = deque() + now = int(time() * 1000) + with self._lock: + self._pending = pending + self._starts[0] = self._starts[1] = self._starts[2] = None + self._window_start_ms = now + self._skipped = False + + def collect(self) -> list[_RuntimeMetric]: + pending: deque[tuple[int, float]] = deque() + now = int(time() * 1000) + with self._lock: + pending, self._pending = self._pending, pending + start, self._window_start_ms = self._window_start_ms, now + skipped, self._skipped = self._skipped, False + if skipped and not self._warned: + self._warned = True + log.warning( + "GC duration observations skipped: busy collector or full buffer" + ) + return [ + _RuntimeMetric( + "app.cpython.gc.duration", + duration, + "ms", + "histogram", + {"cpython.gc.generation": generation}, + start, + ) + for generation, duration in pending + ] + + +class _MacOSProcessMetrics: + def __init__(self) -> None: + libproc = ctypes.CDLL("/usr/lib/libproc.dylib", use_errno=True) + self._proc_pidinfo = libproc.proc_pidinfo + self._proc_pidinfo.argtypes = [ + ctypes.c_int, + ctypes.c_int, + ctypes.c_uint64, + ctypes.c_void_p, + ctypes.c_int, + ] + self._proc_pidinfo.restype = ctypes.c_int + self._native_error_warned = False + self._reset() + + def close(self) -> None: + cleanup = getattr(self, "_gc_cleanup", None) + if cleanup is not None: + cleanup() + + def reset_gc_duration(self) -> None: + if self._gc_duration is not None: + self._gc_duration.reset() + + def _reset(self) -> None: + self.close() + self._pid = os.getpid() + self.attributes = {"process.pid": self._pid} + self.resource_attributes = { + "service.instance.id": uuid4().hex, + "host.name": gethostname(), + } + self.start_time_ms = int(time() * 1000) + usage = getrusage(RUSAGE_SELF) + self._cpu_start_user = self._cpu_time_user = usage.ru_utime + self._cpu_start_sys = self._cpu_time_sys = usage.ru_stime + self._ctx_start = (usage.ru_nvcsw, usage.ru_nivcsw) + self._gc_start = get_stats() if sys.implementation.name == "cpython" else [] + self._gc_duration = ( + _GCDurationMetrics() if sys.implementation.name == "cpython" else None + ) + self._gc_cleanup = ( + finalize(self, self._gc_duration.close) + if self._gc_duration is not None + else None + ) + self._time = monotonic() + + def _read_task_info(self) -> _ProcTaskInfo: + info = _ProcTaskInfo() + size = ctypes.sizeof(info) + read = self._proc_pidinfo(self._pid, 4, 0, ctypes.byref(info), size) + if read != size: + raise OSError( + "proc_pidinfo(PROC_PIDTASKINFO) returned incomplete process information" + ) + return info + + def collect(self) -> list[_RuntimeMetric]: + if self._pid != os.getpid(): + self._reset() + usage = getrusage(RUSAGE_SELF) + now = monotonic() + elapsed = now - self._time + user_delta = usage.ru_utime - self._cpu_time_user + sys_delta = usage.ru_stime - self._cpu_time_sys + self._cpu_time_user = usage.ru_utime + self._cpu_time_sys = usage.ru_stime + self._time = now + + samples = [ + _RuntimeMetric( + "process.cpu.time", + usage.ru_utime - self._cpu_start_user, + "s", + attributes={"cpu.mode": "user"}, + ), + _RuntimeMetric( + "process.cpu.time", + usage.ru_stime - self._cpu_start_sys, + "s", + attributes={"cpu.mode": "system"}, + ), + ] + if elapsed > 0 and user_delta >= 0 and sys_delta >= 0: + cpus = getattr(os, "process_cpu_count", os.cpu_count)() + if cpus is not None and cpus > 0: + for mode, delta in (("user", user_delta), ("system", sys_delta)): + samples.append( + _RuntimeMetric( + "process.cpu.utilization", + delta / elapsed / cpus, + "1", + "gauge", + {"cpu.mode": mode}, + ) + ) + for kind, current, baseline in zip( + ("voluntary", "involuntary"), + (usage.ru_nvcsw, usage.ru_nivcsw), + self._ctx_start, + ): + samples.append( + _RuntimeMetric( + "process.context_switches", + current - baseline, + "{context_switch}", + attributes={"process.context_switch.type": kind}, + ) + ) + if sys.implementation.name == "cpython": + for generation, stats in enumerate(get_stats()): + for name, field, unit in ( + ("collections", "collections", "{collection}"), + ("collected_objects", "collected", "{object}"), + ("uncollectable_objects", "uncollectable", "{object}"), + ): + samples.append( + _RuntimeMetric( + f"cpython.gc.{name}", + stats[field] - self._gc_start[generation][field], + unit, + attributes={"cpython.gc.generation": generation}, + ) + ) + try: + info = self._read_task_info() + samples.append( + _RuntimeMetric( + "process.memory.usage", + info.pti_resident_size, + "By", + "observable_updowncounter", + ) + ) + try: + physical_pages = os.sysconf("SC_PHYS_PAGES") + page_size = os.sysconf("SC_PAGE_SIZE") + except (OSError, ValueError): + pass + else: + if physical_pages > 0 and page_size > 0: + samples.append( + _RuntimeMetric( + "process.memory.utilization", + info.pti_resident_size / (physical_pages * page_size), + "1", + "gauge", + ) + ) + samples.append( + _RuntimeMetric( + "process.thread.count", + info.pti_threadnum, + "{thread}", + "observable_updowncounter", + ) + ) + except OSError as e: + if not self._native_error_warned: + self._native_error_warned = True + log.warning("Unable to collect native process metrics: %s", e) + if self._gc_duration is not None: + samples.extend(self._gc_duration.collect()) + return samples + + +class _HostCPULoadInfo(ctypes.Structure): + _fields_ = [("cpu_ticks", ctypes.c_uint32 * 4)] + + +class _VMStatistics64(ctypes.Structure): + # Revision-1 prefix, not the growing latest vm_statistics64 ABI. + _fields_ = ( + [ + (name, ctypes.c_uint32) + for name in ("free_count", "active_count", "inactive_count", "wire_count") + ] + + [ + (name, ctypes.c_uint64) + for name in ( + "zero_fill_count", + "reactivations", + "pageins", + "pageouts", + "faults", + "cow_faults", + "lookups", + "hits", + "purges", + ) + ] + + [(name, ctypes.c_uint32) for name in ("purgeable_count", "speculative_count")] + + [ + (name, ctypes.c_uint64) + for name in ("decompressions", "compressions", "swapins", "swapouts") + ] + + [ + (name, ctypes.c_uint32) + for name in ( + "compressor_page_count", + "throttled_count", + "external_page_count", + "internal_page_count", + ) + ] + + [("total_uncompressed_pages_in_compressor", ctypes.c_uint64)] + ) + + +class _SwapUsage(ctypes.Structure): + _fields_ = [ + ("total", ctypes.c_uint64), + ("available", ctypes.c_uint64), + ("used", ctypes.c_uint64), + ("page_size", ctypes.c_uint32), + ("encrypted", ctypes.c_int32), + ] + + +class _MacOSHostMetrics: + def __init__(self) -> None: + self._lib = ctypes.CDLL("/usr/lib/libSystem.B.dylib", use_errno=True) + for name, arguments in ( + ("mach_host_self", []), + ("mach_task_self", []), + ("mach_port_deallocate", [ctypes.c_uint32, ctypes.c_uint32]), + ("mach_vm_deallocate", [ctypes.c_uint32, ctypes.c_uint64, ctypes.c_uint64]), + ( + "host_processor_info", + [ + ctypes.c_uint32, + ctypes.c_int32, + ctypes.POINTER(ctypes.c_uint32), + ctypes.POINTER(ctypes.POINTER(ctypes.c_int32)), + ctypes.POINTER(ctypes.c_uint32), + ], + ), + ( + "host_statistics64", + [ + ctypes.c_uint32, + ctypes.c_int32, + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_uint32), + ], + ), + ("host_page_size", [ctypes.c_uint32, ctypes.POINTER(ctypes.c_size_t)]), + ( + "sysctlbyname", + [ + ctypes.c_char_p, + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_size_t), + ctypes.c_void_p, + ctypes.c_size_t, + ], + ), + ): + function = getattr(self._lib, name) + function.argtypes = arguments + function.restype = ( + ctypes.c_uint32 + if name in ("mach_host_self", "mach_task_self") + else ctypes.c_int32 + ) + self._task_port = self._lib.mach_task_self() + self.attributes = {"host.name": gethostname()} + self.start_time_ms = int(time() * 1000) + self._core_ticks: list[tuple[int, ...]] | None = None + self._swap_start: tuple[int, int] | None = None + self._warned: set[str] = set() + self.collect() + + def _read_statistics(self, info: ctypes.Structure) -> int: + host = self._lib.mach_host_self() + count = ctypes.c_uint32(ctypes.sizeof(info) // 4) + page_size = ctypes.c_size_t() + try: + result = self._lib.host_statistics64( + host, 4, ctypes.byref(info), ctypes.byref(count) + ) + if result != 0 or count.value != ctypes.sizeof(info) // 4: + raise OSError( + f"host statistics flavor 4: result={result}, count={count.value}" + ) + result = self._lib.host_page_size(host, ctypes.byref(page_size)) + if result != 0 or page_size.value == 0: + raise OSError(f"host_page_size failed: result={result}") + finally: + self._lib.mach_port_deallocate(self._task_port, host) + return page_size.value + + def _sysctl(self, name: bytes, value) -> None: + size = ctypes.c_size_t(ctypes.sizeof(value)) + result = self._lib.sysctlbyname( + name, ctypes.byref(value), ctypes.byref(size), None, 0 + ) + if result != 0: + error = ctypes.get_errno() + raise OSError(error, f"{name.decode()}: {os.strerror(error)}") + if size.value != ctypes.sizeof(value): + raise OSError(f"{name.decode()} returned incomplete information") + + def _read_core_ticks(self) -> list[tuple[int, ...]]: + host = self._lib.mach_host_self() + cpus = ctypes.c_uint32() + count = ctypes.c_uint32() + info = ctypes.POINTER(ctypes.c_int32)() + try: + result = self._lib.host_processor_info( + host, 2, ctypes.byref(cpus), ctypes.byref(info), ctypes.byref(count) + ) + if ( + result != 0 + or not info + or cpus.value == 0 + or count.value != cpus.value * 4 + ): + raise OSError( + f"host_processor_info(PROCESSOR_CPU_LOAD_INFO): result={result}, " + f"cpus={cpus.value}, count={count.value}" + ) + loads = ctypes.cast(info, ctypes.POINTER(_HostCPULoadInfo)) + return [tuple(loads[cpu].cpu_ticks) for cpu in range(cpus.value)] + finally: + try: + # Mach returns an out-of-line allocation, not a borrowed buffer. + if info: + result = self._lib.mach_vm_deallocate( + self._task_port, + ctypes.cast(info, ctypes.c_void_p).value, + count.value * ctypes.sizeof(ctypes.c_int32), + ) + if result != 0: + raise OSError(f"mach_vm_deallocate failed: result={result}") + finally: + self._lib.mach_port_deallocate(self._task_port, host) + + def _per_core_cpu(self) -> list[_RuntimeMetric]: + current = self._read_core_ticks() + previous, self._core_ticks = self._core_ticks, current + if previous is None or len(previous) != len(current): + return [] + samples = [] + for cpu, (now, before) in enumerate(zip(current, previous)): + deltas = [(tick - old) % (1 << 32) for tick, old in zip(now, before)] + total = sum(deltas) + if total == 0: + continue + for mode, delta in zip(("user", "system", "idle", "nice"), deltas): + samples.append( + _RuntimeMetric( + "system.cpu.utilization", + delta / total, + "1", + "gauge", + {"cpu.logical_number": cpu, "cpu.mode": mode}, + ) + ) + return samples + + def _memory(self) -> list[_RuntimeMetric]: + info = _VMStatistics64() + page_size = self._read_statistics(info) + if self._swap_start is None: + self._swap_start = (info.swapins, info.swapouts) + return [ + _RuntimeMetric( + "system.memory.usage", + (info.free_count - info.speculative_count) * page_size, + "By", + "observable_updowncounter", + {"system.memory.state": "free"}, + ), + _RuntimeMetric( + "system.memory.usage", + (info.active_count + info.wire_count) * page_size, + "By", + "observable_updowncounter", + {"system.memory.state": "used"}, + ), + *[ + _RuntimeMetric( + "app.system.swap.io", + (current - baseline) * page_size, + "By", + attributes={"system.paging.direction": direction}, + ) + for direction, current, baseline in zip( + ("in", "out"), (info.swapins, info.swapouts), self._swap_start + ) + ], + ] + + def _memory_limit(self) -> list[_RuntimeMetric]: + total = ctypes.c_uint64() + self._sysctl(b"hw.memsize", total) + return [ + _RuntimeMetric( + "system.memory.limit", total.value, "By", "observable_updowncounter" + ) + ] + + def _swap_usage(self) -> list[_RuntimeMetric]: + info = _SwapUsage() + self._sysctl(b"vm.swapusage", info) + return [ + _RuntimeMetric( + "system.paging.usage", + info.used, + "By", + "observable_updowncounter", + {"system.paging.state": "used"}, + ) + ] + + def _pressure(self) -> list[_RuntimeMetric]: + level = ctypes.c_uint32() + self._sysctl(b"kern.memorystatus_vm_pressure_level", level) + if level.value not in (1, 2, 4): + raise OSError(f"Unknown memory pressure level: {level.value}") + return [ + _RuntimeMetric( + "app.system.memory.pressure", level.value, "{level}", "gauge" + ) + ] + + def collect(self) -> list[_RuntimeMetric]: + samples = [] + for name, read in ( + ("per-CPU", self._per_core_cpu), + ("VM", self._memory), + ("memory limit", self._memory_limit), + ("swap usage", self._swap_usage), + ("memory pressure", self._pressure), + ): + try: + samples.extend(read()) + except OSError as error: + if name not in self._warned: + self._warned.add(name) + log.warning("Unable to collect host %s metrics: %s", name, error) + return samples diff --git a/posthog/client.py b/posthog/client.py index d74ea56e3..4cafe56bb 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -803,6 +803,19 @@ def __init__( ``$trace_id``/``$span_id`` values passed in ``properties`` win. Exception events (``capture_exception``) always attach these IDs regardless of this setting. Defaults to False. + metrics: Config dict for ``client.metrics``. Set + ``capture_runtime_metrics=True`` to collect process CPU usage, + user/system CPU time, current RSS memory and its fraction of total + physical RAM, native thread count, CPython GC statistics, and + total context switches on macOS without additional dependencies. + Separately, ``capture_host_metrics=True`` collects host CPU, + estimated available memory, native memory pressure, swap usage, + and swap transfers. Enable host collection in only one client + per host; inherited clients do not restart it after fork. + Sampling uses ``flush_interval`` (10 s by default), including + with ``sync_mode=True``. Disabled by default; unsupported + platforms warn and do not collect. Call ``shutdown()`` to + collect a final sample, flush, and stop the timer. traces: Config dict for distributed tracing: ``service_name``, ``service_version``, ``environment``, ``resource_attributes``, ``flush_interval`` (5 s), ``max_queue_size`` (2048), @@ -1116,6 +1129,18 @@ def __init__( ) self._warn_if_duplicate_async_client() + self._start_runtime_metrics() + + def _start_runtime_metrics(self) -> None: + if ( + isinstance(self._metrics_config, dict) + and ( + self._metrics_config.get("capture_runtime_metrics") is True + or self._metrics_config.get("capture_host_metrics") is True + ) + and not self._shutdown_requested + ): + _ = self.metrics def _set_library_identity(self, library_id: str, library_version: str) -> None: """Override the SDK identity stamped on events and outbound requests.""" @@ -2579,6 +2604,7 @@ def metrics(self) -> PostHogMetrics: # the first chained metrics.count() call (raise only in debug mode). try: self._metrics = PostHogMetrics(self, self._metrics_config) + self._metrics._start_runtime_metrics() except Exception as e: if self.debug: raise e @@ -2970,6 +2996,11 @@ def _shutdown_once(self, errors: list[Exception]) -> None: self._flush_or_discard_queues(errors) if self._metrics is not None: + self._run_lifecycle_cleanup( + "Failed to stop runtime metrics on shutdown", + self._metrics._stop_runtime_metrics, + errors, + ) self._run_lifecycle_cleanup( "Failed to flush metrics on shutdown", self._metrics.flush, errors ) diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py index 781aa89b3..ec578f89b 100644 --- a/posthog/metrics_capture.py +++ b/posthog/metrics_capture.py @@ -3,10 +3,11 @@ Samples fold into per-series aggregates in memory (counts sum, gauges keep the last value, histograms accumulate buckets) and flush as one OTLP/JSON data point per series per window to ``/i/v1/metrics`` — a burst of 10k ``count()`` -calls costs one data point on the wire. Sums and histograms use delta -temporality, so each data point stands alone and process restarts need no -cross-window state. Mirrors the ``posthog-js`` core implementation so every -SDK speaks the same wire shape. +calls costs one data point on the wire. Manually recorded sums and histograms +use delta temporality, so each data point stands alone and process restarts +need no cross-window state. Automatic runtime counters and up/down counters +instead export the latest cumulative observations. Manual metrics mirror the +``posthog-js`` core implementation so every SDK speaks the same wire shape. Deliberately unlike event capture, no per-user context (distinct ID, session) is attached: every attribute value creates a new series, and per-user series @@ -31,9 +32,10 @@ import logging import math import os +import sys import threading import time -from typing import Any, Callable, Optional, Union +from typing import TYPE_CHECKING, Any, Callable, Optional, Union from urllib.parse import quote import requests @@ -44,6 +46,9 @@ from .tracing._traceparent import TRACE_FLAGS_SAMPLED, parse_traceparent +if TYPE_CHECKING: + from ._runtime_metrics import _MacOSHostMetrics, _MacOSProcessMetrics + log = logging.getLogger("posthog") MetricAttributeValue = Union[str, int, float, bool] @@ -69,7 +74,9 @@ ] _OTLP_TEMPORALITY_DELTA = 1 +_OTLP_TEMPORALITY_CUMULATIVE = 2 _VALID_METRIC_TYPES = ("count", "gauge", "histogram") +_RUNTIME_SUM_TYPES = ("observable_counter", "observable_updowncounter") # Consecutive failed flushes before the buffered window is dropped (loudly) — bounds # memory and payload growth against a permanently unreachable endpoint. The series # cap already bounds the buffered window, and backoff spaces the attempts out, so @@ -232,9 +239,11 @@ class _SeriesState: "type", "unit", "attributes", + "resource_attributes", "window_start_ms", "total", "last", + "observation_time_ms", "hist", "exemplar", ) @@ -245,7 +254,9 @@ def __init__( metric_type: str, unit: Optional[str], attributes: Optional[dict], + resource_attributes: Optional[dict] = None, ): + self.resource_attributes = dict(resource_attributes or {}) self.name = name self.type = metric_type self.unit = unit @@ -268,6 +279,7 @@ def __init__( self.window_start_ms = int(time.time() * 1000) self.total: Optional[float] = None self.last: Optional[float] = None + self.observation_time_ms = self.window_start_ms self.hist: Optional[dict] = None self.exemplar: Optional[dict] = None @@ -283,6 +295,77 @@ class PostHogMetrics: client.metrics.count("invoices.processed", 1, attributes={"plan": "pro"}) client.metrics.gauge("queue.depth", 42) client.metrics.histogram("job.duration", 187, unit="ms") + + Set ``capture_runtime_metrics=True`` in the config to automatically collect + these metrics on macOS without extra dependencies: + + * ``process.cpu.time``: cumulative CPU seconds, with ``cpu.mode`` set to + ``user`` or ``system``. + * ``process.cpu.utilization``: utilization fractions per ``cpu.mode``, + normalized by the CPUs available to the process (falling back to the + system CPU count on older Python versions). Omitted if that count is unknown. + * ``process.memory.usage``: current resident memory in bytes, exported as + a cumulative non-monotonic sum (ObservableUpDownCounter). + * ``process.memory.utilization``: RSS divided by total physical RAM, + exported as a gauge fraction. This is not a container-limit measurement. + * ``process.thread.count``: current native thread count, including threads + created by C extensions, also an ObservableUpDownCounter. + * ``cpython.gc.collections``, ``cpython.gc.collected_objects``, and + ``cpython.gc.uncollectable_objects``: cumulative ``gc.get_stats()`` + counters, distinguished by ``cpython.gc.generation`` (0, 1, 2). + These are CPython-only, and measure neither heap size nor GC pause duration. + * ``app.cpython.gc.duration``: CPython collection-duration histograms in + milliseconds, distinguished by ``cpython.gc.generation``. Measures wall + time between GC callbacks, not exact application stop-the-world time. + Callbacks only buffer observations; hooks and export run at flush time. + At most 4096 observations are buffered per collector; contention or a full + buffer can omit observations, with a bounded warning on collection. + * ``process.context_switches``: cumulative native ``getrusage()`` counts, + with ``process.context_switch.type=voluntary`` or ``involuntary``. + Per-core-equivalent CPU percentage can be derived from + ``rate(process.cpu.time) * 100``, summing the CPU modes. + + Only this process is measured, not children or host totals. Samples use the + configured flush interval (10 seconds by default) and pass through + ``before_send``. Automatic cumulative sums use sample types + ``observable_counter`` and ``observable_updowncounter``; manual ``count`` + calls remain delta sums. CPU, GC, and context-switch counters exclude work + before collection starts. Automatic process samples carry ``process.pid`` + and resource attributes ``service.instance.id`` (generated) and ``host.name`` + (the same native hostname used by host collection, unless explicitly configured + in ``resource_attributes``). Forked workers get a new instance identity and + counter baselines. Automatic process resources do not change manual or host + metrics' configured resource attributes. + ``shutdown()`` collects a final sample and stops collection. + + Independently opt in to macOS host collection with ``capture_host_metrics=True``: + + * ``system.cpu.utilization``: per-logical-CPU utilization fractions, with + integer ``cpu.logical_number`` (0-based) and ``cpu.mode`` set to ``user``, + ``system``, ``idle``, or ``nice``. To chart busy utilization, exclude idle + and sum the other modes per CPU. Initial, unchanged, and topology-change + readings are omitted until a valid counter delta is available. + * ``system.memory.usage``: native memory bytes by ``system.memory.state``. + Following psutil's macOS accounting, ``free`` excludes speculative pages + and ``used`` is active plus wired pages. Unsupported ``cached`` and + ``buffers`` states are omitted. These values are not available memory + and need not sum to total RAM (for example, compressed storage is separate). + * ``system.memory.limit``: installed physical RAM bytes from ``hw.memsize``. + Neither memory metric represents a container or process memory limit. + * ``app.system.memory.pressure``: native level, 1=normal, 2=warning, 4=critical. + This is categorical, not a percentage. The optional, undocumented polling + sysctl may be unavailable or denied; failed and unknown readings are omitted. + * ``system.paging.usage``: current used swap bytes, a non-monotonic sum with + ``system.paging.state=used``. Not total disk capacity. + * ``app.system.swap.io``: cumulative page-accounted compressed-segment swap + transfer bytes since collection starts, with ``system.paging.direction`` + set to ``in`` or ``out``. Not I/O operation counts or physical disk traffic. + + The ``app.system.*`` names are custom, not portable OpenTelemetry instruments. + Host samples carry ``host.name``, without automatic process identity. Configure + only one host collector per host to avoid duplicate reporting. An inherited + client does not restart host collection after fork; a newly created client + can explicitly opt in. Sources fail independently and recover on later reads. """ def __init__(self, client, config: Optional[dict] = None): @@ -356,9 +439,129 @@ def __init__(self, client, config: Optional[dict] = None): self._flush_lock = threading.Lock() self._series: dict = {} self._flush_timer: Optional[threading.Timer] = None + self._timer_flush_in_progress = False self._series_cap_warned = False self._type_by_name: dict = {} self._type_collision_warned: set = set() + self._capture_runtime_metrics = config.get("capture_runtime_metrics", False) + self._runtime_metrics: Optional["_MacOSProcessMetrics"] = None + self._capture_host_metrics = config.get("capture_host_metrics", False) + self._host_metrics: Optional["_MacOSHostMetrics"] = None + self._host_metrics_owner_pid = self._pid + self._runtime_lock = threading.RLock() + self._runtime_collecting = False + self._runtime_error_warned = False + + def _start_runtime_metrics(self) -> None: + for name, value in ( + ("capture_runtime_metrics", self._capture_runtime_metrics), + ("capture_host_metrics", self._capture_host_metrics), + ): + if value is not True and value is not False: + log.warning("Ignoring %s: expected a boolean", name) + capture_runtime = self._capture_runtime_metrics is True + capture_host = ( + self._capture_host_metrics is True + and os.getpid() == self._host_metrics_owner_pid + ) + if not capture_runtime and not capture_host: + return + with self._runtime_lock: + if ( + self._client.disabled + or not self._client.send + or self._client._shutdown_requested + or self._client._join_requested + ): + return + if sys.platform == "darwin": + runtime = None + host = None + if capture_runtime: + try: + from ._runtime_metrics import _MacOSProcessMetrics + + runtime = _MacOSProcessMetrics() + except Exception as e: + log.warning("Unable to start runtime metrics: %s", e) + if capture_host: + try: + from ._runtime_metrics import _MacOSHostMetrics + + host = _MacOSHostMetrics() + except Exception as e: + log.warning("Unable to start host metrics: %s", e) + if ( + self._client.disabled + or not self._client.send + or self._client._shutdown_requested + or self._client._join_requested + ): + if runtime is not None: + runtime.close() + return + if self._runtime_metrics is not None: + self._runtime_metrics.close() + self._runtime_metrics = runtime + self._host_metrics = host + if runtime is not None or host is not None: + with self._lock: + self._arm_flush_timer() + else: + for enabled, kind in ( + (capture_runtime, "runtime"), + (capture_host, "host"), + ): + if enabled: + log.warning( + "Automatic %s metrics are currently supported only on macOS", + kind, + ) + + def _collect_runtime_metrics(self) -> None: + with self._runtime_lock: + if self._runtime_collecting or self._client.disabled: + return + self._runtime_collecting = True + callback_context = self._client._lifecycle_callback_context + token = callback_context.set(True) + try: + for kind, collector in ( + ("runtime", self._runtime_metrics), + ("host", self._host_metrics), + ): + if collector is None or ( + kind == "host" and os.getpid() != self._host_metrics_owner_pid + ): + continue + try: + for sample in collector.collect(): + self._guarded_capture( + sample.metric_type, + sample.name, + sample.value, + sample.unit, + {**collector.attributes, **(sample.attributes or {})}, + collector, + sample.start_time_ms, + ) + except Exception as e: + if not self._runtime_error_warned: + self._runtime_error_warned = True + log.warning("Unable to collect %s metrics: %s", kind, e) + finally: + callback_context.reset(token) + self._runtime_collecting = False + + def _stop_runtime_metrics(self) -> None: + with self._runtime_lock: + if self._runtime_metrics is not None: + self._runtime_metrics.close() + self._collect_runtime_metrics() + with self._lock: + self._runtime_metrics = None + self._host_metrics = None + self._clear_flush_timer() def count( self, @@ -391,18 +594,28 @@ def histogram( self._guarded_capture("histogram", name, value, unit, attributes) def flush(self) -> None: - """Sends everything aggregated so far without waiting for the flush interval.""" - with self._flush_lock: - self._do_flush() + """Sends buffered metrics, collecting enabled runtime metrics first.""" + self._collect_runtime_metrics() + try: + with self._flush_lock: + self._do_flush() + finally: + with self._lock: + if self._runtime_metrics is not None or self._host_metrics is not None: + self._arm_flush_timer() def reset(self) -> None: - """Clears the flush timer and drops the current window.""" - with self._lock: + """Drops the current window; enabled runtime collection continues.""" + with self._runtime_lock, self._lock: + if self._runtime_metrics is not None: + self._runtime_metrics.reset_gc_duration() self._clear_flush_timer() self._series = {} self._series_cap_warned = False self._type_by_name = {} self._type_collision_warned = set() + if self._runtime_metrics is not None or self._host_metrics is not None: + self._arm_flush_timer() def _guarded_capture( self, @@ -411,10 +624,22 @@ def _guarded_capture( value: float, unit: Optional[str], attributes: Optional[dict], + runtime_collector: Optional[ + Union["_MacOSProcessMetrics", "_MacOSHostMetrics"] + ] = None, + runtime_start_time_ms: Optional[int] = None, ) -> None: # A telemetry call must never raise into the host application, whatever the input. try: - self._capture(metric_type, name, value, unit, attributes) + self._capture( + metric_type, + name, + value, + unit, + attributes, + runtime_collector, + runtime_start_time_ms, + ) except Exception as e: if not self._capture_error_warned: self._capture_error_warned = True @@ -427,6 +652,10 @@ def _capture( value: float, unit: Optional[str], attributes: Optional[dict], + runtime_collector: Optional[ + Union["_MacOSProcessMetrics", "_MacOSHostMetrics"] + ] = None, + runtime_start_time_ms: Optional[int] = None, ) -> None: if getattr(self._client, "disabled", False): return @@ -460,7 +689,9 @@ def _capture( unit = sample_dict.get("unit") attributes = sample_dict.get("attributes") - if metric_type not in _VALID_METRIC_TYPES: + if metric_type not in _VALID_METRIC_TYPES and not ( + runtime_collector is not None and metric_type in _RUNTIME_SUM_TYPES + ): log.warning( "Dropping metric '%s': unknown metric type '%s'", name, metric_type ) @@ -476,7 +707,7 @@ def _capture( ): log.warning("Dropping metric '%s': value must be a finite number", name) return - if metric_type == "count" and value < 0: + if metric_type in ("count", "observable_counter") and value < 0: log.warning( "Dropping count '%s': counters are monotonic, value must be >= 0", name ) @@ -486,11 +717,34 @@ def _capture( # None-valued attributes are stripped from the wire, so strip them from the # series identity too — otherwise two indistinguishable data points emit. attributes = {k: v for k, v in attributes.items() if v is not None} + resource_attributes = ( + getattr(runtime_collector, "resource_attributes", None) + if runtime_collector is not None + else None + ) + if resource_attributes and "host.name" in self._resource_attributes: + resource_attributes = { + **resource_attributes, + "host.name": self._resource_attributes["host.name"], + } key = _series_key(metric_type, name, unit, attributes) - trace_context = _active_trace_context(self._client) + if resource_attributes: + key += "\x00" + _series_key("", "", None, resource_attributes) + trace_context = ( + None + if runtime_collector is not None and metric_type == "histogram" + else _active_trace_context(self._client) + ) with self._lock: self._reset_after_fork_locked() + # A before_send hook can stop collection or fork while a sample is in flight. + if ( + runtime_collector is not None + and runtime_collector is not self._runtime_metrics + and runtime_collector is not self._host_metrics + ): + return state = self._series.get(key) if state is None: @@ -503,7 +757,15 @@ def _capture( self._max_series_per_flush, ) return - state = _SeriesState(name, metric_type, unit, attributes) + state = _SeriesState( + name, metric_type, unit, attributes, resource_attributes + ) + if runtime_collector is not None and metric_type in _RUNTIME_SUM_TYPES: + state.window_start_ms = runtime_collector.start_time_ms + elif ( + runtime_collector is not None and runtime_start_time_ms is not None + ): + state.window_start_ms = runtime_start_time_ms self._series[key] = state # Bookkeeping only for admitted samples, so name-cardinality misuse (IDs in @@ -537,8 +799,15 @@ def _reinit_after_fork(self) -> None: # exist in the child, so replace them without ever acquiring them. self._lock = threading.Lock() self._flush_lock = threading.Lock() + self._runtime_lock = threading.RLock() + self._runtime_collecting = False + if self._runtime_metrics is not None: + self._runtime_metrics.close() + self._runtime_metrics = None + self._host_metrics = None self._pid = os.getpid() self._drop_inherited_window() + self._start_runtime_metrics() def _reset_after_fork_locked(self) -> None: # PID-guard fallback for platforms without os.register_at_fork: a forked child @@ -552,7 +821,9 @@ def _reset_after_fork_locked(self) -> None: self._drop_inherited_window() def _drop_inherited_window(self) -> None: + self._host_metrics = None self._flush_timer = None + self._timer_flush_in_progress = False self._series = {} self._series_cap_warned = False self._type_by_name = {} @@ -564,6 +835,9 @@ def _fold(self, state: _SeriesState, value: float) -> None: state.total = (state.total or 0.0) + value elif state.type == "gauge": state.last = value + elif state.type in _RUNTIME_SUM_TYPES: + state.last = value + state.observation_time_ms = int(time.time() * 1000) else: hist = state.hist if hist is None: @@ -598,20 +872,33 @@ def _arm_flush_timer( ) timer.daemon = True self._flush_timer = timer - timer.start() + self._start_flush_timer() + + def _start_flush_timer(self) -> None: + if self._flush_timer is not None and not self._timer_flush_in_progress: + self._flush_timer.start() def _timer_flush(self, fired: Optional[threading.Timer] = None) -> None: with self._lock: # A timer whose thread already started can't be cancelled: if a newer # timer replaced this one meanwhile (failed explicit flush arming the # backoff timer), the stale body must not clear it or flush again. - if fired is not None and fired is not self._flush_timer: + if self._timer_flush_in_progress or ( + fired is not None and fired is not self._flush_timer + ): return + self._timer_flush_in_progress = True self._flush_timer = None + pid = os.getpid() try: self.flush() except Exception as e: log.error("Metrics flush failed: %s", e) + finally: + if os.getpid() == pid: + with self._lock: + self._timer_flush_in_progress = False + self._start_flush_timer() def _clear_flush_timer(self) -> None: if self._flush_timer is not None: @@ -739,7 +1026,7 @@ def _merge_window_back(self, window: dict) -> None: current.hist["max"] = max(current.hist["max"], old.hist["max"]) for i, count in enumerate(old.hist["bucket_counts"]): current.hist["bucket_counts"][i] += count - # Gauge: the live window's value is newer — keep it. + # Gauges and cumulative observations keep the live window's newer value. if dropped: log.warning( "Dropped %s unsent metric series while merging a failed flush back (series cap %s)", @@ -759,19 +1046,29 @@ def _build_payload(self, window: dict) -> dict: resource_attributes["telemetry.sdk.name"] = _SCOPE_NAME resource_attributes["telemetry.sdk.version"] = VERSION + groups: dict = {} + for key, state in window.items(): + resource_key = _series_key("", "", None, state.resource_attributes) + if resource_key not in groups: + groups[resource_key] = (state.resource_attributes, {}) + groups[resource_key][1][key] = state + return { "resourceMetrics": [ { "resource": { - "attributes": _to_otlp_key_value_list(resource_attributes) + "attributes": _to_otlp_key_value_list( + {**resource_attributes, **extra_attributes} + ) }, "scopeMetrics": [ { "scope": {"name": _SCOPE_NAME, "version": VERSION}, - "metrics": self._build_metrics(window), + "metrics": self._build_metrics(series), } ], } + for extra_attributes, series in (groups.values() or [({}, {})]) ] } @@ -787,10 +1084,14 @@ def _build_metrics(self, window: dict) -> list: metric = {"name": state.name} if state.unit: metric["unit"] = state.unit - if state.type == "count": + if state.type == "count" or state.type in _RUNTIME_SUM_TYPES: metric["sum"] = { - "aggregationTemporality": _OTLP_TEMPORALITY_DELTA, - "isMonotonic": True, + "aggregationTemporality": ( + _OTLP_TEMPORALITY_DELTA + if state.type == "count" + else _OTLP_TEMPORALITY_CUMULATIVE + ), + "isMonotonic": state.type != "observable_updowncounter", "dataPoints": [], } elif state.type == "gauge": @@ -806,13 +1107,20 @@ def _build_metrics(self, window: dict) -> list: start_nano = _ms_to_unix_nano(state.window_start_ms) exemplars = _build_exemplars(state.exemplar) - if state.type == "count": + if state.type == "count" or state.type in _RUNTIME_SUM_TYPES: metric["sum"]["dataPoints"].append( { "attributes": attributes, "startTimeUnixNano": start_nano, - "timeUnixNano": now_nano, - "asDouble": state.total or 0.0, + "timeUnixNano": ( + now_nano + if state.type == "count" + else _ms_to_unix_nano(state.observation_time_ms) + ), + "asDouble": ( + state.total if state.type == "count" else state.last + ) + or 0.0, **exemplars, } ) diff --git a/posthog/test/test_example_runtime_metrics.py b/posthog/test/test_example_runtime_metrics.py new file mode 100644 index 000000000..6dbf4f1df --- /dev/null +++ b/posthog/test/test_example_runtime_metrics.py @@ -0,0 +1,105 @@ +from unittest import mock + +import pytest + +import _example_runtime_metrics as workload + + +def test_default_duration_is_thirty_minutes(): + assert workload.run_workload.__defaults__ == (1800, 30) + + +@pytest.mark.parametrize( + "cpus,ram_mib,expected_workers,expected_memory", + [(1, 128, 1, 4), (8, 8192, 4, 256), (32, 65536, 8, 1024)], +) +def test_workload_repeats_phases_with_bounded_resources( + monkeypatch, cpus, ram_mib, expected_workers, expected_memory +): + clock = mock.Mock(elapsed=0.0) + monkeypatch.setattr(workload.time, "monotonic", lambda: clock.elapsed) + monkeypatch.setattr( + workload.time, + "sleep", + lambda seconds: setattr(clock, "elapsed", clock.elapsed + seconds), + ) + monkeypatch.setattr(workload.os, "cpu_count", lambda: cpus) + monkeypatch.setattr(workload.os, "process_cpu_count", lambda: cpus, raising=False) + monkeypatch.setattr( + workload.os, + "sysconf", + lambda key: ram_mib * 256 if key == "SC_PHYS_PAGES" else 4096, + ) + resize = mock.Mock() + threads = mock.Mock() + monkeypatch.setattr(workload, "_resize_memory", resize) + monkeypatch.setattr(workload, "Thread", threads) + monkeypatch.setattr(workload.gc, "collect", mock.Mock()) + workload.run_workload(duration=7, phase_duration=1) + assert clock.elapsed == 7 + targets = [call.args[1] for call in resize.call_args_list] + assert targets == [ + expected_memory // 8, + expected_memory // 4, + expected_memory // 4, + expected_memory, + expected_memory // 2, + expected_memory // 8, + expected_memory // 8, + ] + assert threads.call_count == 1 + 2 * expected_workers + max( + 1, expected_workers // 2 + ) + assert threads.return_value.start.call_count == threads.call_count + assert threads.return_value.join.call_count == threads.call_count + assert all(call.kwargs["args"][0].is_set() for call in threads.call_args_list) + + +def test_memory_resize_touches_pages_and_releases_excess_blocks(): + blocks = [] + workload._resize_memory(blocks, 16) + assert sum(map(len, blocks)) == 16 * 1024 * 1024 + assert all(set(block[::4096]) == {1} for block in blocks) + first = blocks[0] + workload._resize_memory(blocks, 8) + assert blocks == [first] + workload._resize_memory(blocks, 0) + assert blocks == [] + + +def test_cpu_worker_hashes_large_buffers_until_stopped(monkeypatch): + stop = mock.Mock() + stop.is_set.side_effect = [False, False, True] + hash_buffer = mock.Mock() + monkeypatch.setattr(workload.hashlib, "sha256", hash_buffer) + workload._burn_cpu(stop) + assert hash_buffer.call_count == 2 + assert len(hash_buffer.call_args.args[0]) > 2047 + + +def test_interrupt_stops_workers_and_releases_memory(monkeypatch): + blocks = [] + + def resize(allocations, target): + allocations.append(bytearray(8)) + blocks.append(allocations) + + monkeypatch.setattr(workload, "_resize_memory", resize) + monkeypatch.setattr(workload.gc, "collect", mock.Mock()) + threads = mock.Mock() + monkeypatch.setattr(workload, "Thread", threads) + clock = mock.Mock(elapsed=0.0) + monkeypatch.setattr(workload.time, "monotonic", lambda: clock.elapsed) + + def sleep(seconds): + if clock.elapsed: + raise KeyboardInterrupt + clock.elapsed += seconds + + monkeypatch.setattr(workload.time, "sleep", sleep) + with pytest.raises(KeyboardInterrupt): + workload.run_workload(duration=30, phase_duration=1) + assert threads.call_count == 1 + assert all(not allocation for allocation in blocks) + assert all(call.kwargs["args"][0].is_set() for call in threads.call_args_list) + assert threads.return_value.join.call_count == threads.return_value.start.call_count diff --git a/posthog/test/test_gc_duration_metrics.py b/posthog/test/test_gc_duration_metrics.py new file mode 100644 index 000000000..75fa5eb9b --- /dev/null +++ b/posthog/test/test_gc_duration_metrics.py @@ -0,0 +1,348 @@ +import gc +import os +import sys +import weakref +from types import SimpleNamespace +from unittest import mock + +import pytest +from google.protobuf.json_format import ParseDict +from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( + ExportMetricsServiceRequest, +) + +from .. import metrics_capture +from . import test_runtime_metrics as runtime_tests +from .test_metrics import sent_payload +from .test_runtime_metrics import attributes, exported_metrics + +native = runtime_tests.native +runtime_client = runtime_tests.runtime_client + +NAME = "app.cpython.gc.duration" + + +def collection(module, monkeypatch, generation=0, duration_ms=2.5): + clock = mock.Mock(return_value=100_000_000) + monkeypatch.setattr(module, "monotonic_ns", clock, raising=False) + callbacks = list(module.gc_callbacks) + for callback in callbacks: + callback("start", {"generation": generation}) + clock.return_value += int(duration_ms * 1_000_000) + for callback in callbacks: + callback("stop", {"generation": generation}) + + +def test_gc_duration_is_deferred_histogram_with_process_identity( + runtime_client, native, monkeypatch +): + module = native[0] + monkeypatch.setattr(module, "time", lambda: 1234.5) + client, session = runtime_client() + assert len(module.gc_callbacks) == 1 + start_ms = client.metrics._runtime_metrics.start_time_ms + collection(module, monkeypatch, generation=2, duration_ms=12.5) + assert not session.post.called + client.metrics.flush() + metric = exported_metrics(session)[NAME] + assert metric["unit"] == "ms" + assert metric["histogram"]["aggregationTemporality"] == 1 + point = metric["histogram"]["dataPoints"][0] + assert point["startTimeUnixNano"] == f"{start_ms}000000" + assert point["count"] == 1 + assert point["sum"] == point["min"] == point["max"] == 12.5 + assert sum(point["bucketCounts"]) == 1 + assert attributes(point) == { + **client.metrics._runtime_metrics.attributes, + "cpython.gc.generation": 2, + } + ParseDict(sent_payload(session), ExportMetricsServiceRequest()) + client.metrics.flush() + assert NAME not in exported_metrics(session) + + +def test_shutdown_removes_only_own_callback_and_flushes_duration( + runtime_client, native, monkeypatch +): + module = native[0] + application_callback = mock.Mock() + module.gc_callbacks.append(application_callback) + client, session = runtime_client() + collection(module, monkeypatch) + client.shutdown() + assert module.gc_callbacks == [application_callback] + assert exported_metrics(session)[NAME]["histogram"]["dataPoints"][0]["count"] == 1 + client.shutdown() + assert module.gc_callbacks == [application_callback] + + +def test_gc_callbacks_not_installed_without_runtime_opt_in(runtime_client, native): + runtime_client(metrics={}) + assert native[0].gc_callbacks == [] + + +def test_gc_duration_callback_does_not_wait_for_collection_lock( + runtime_client, native, monkeypatch +): + client, session = runtime_client() + monitor = client.metrics._runtime_metrics._gc_duration + with monitor._lock: + collection(native[0], monkeypatch) + client.metrics.flush() + assert NAME not in exported_metrics(session) + + +def test_gc_durations_survive_send_retry(runtime_client, native, monkeypatch): + client, session = runtime_client() + collection(native[0], monkeypatch, duration_ms=2) + session.post.return_value.status_code = 500 + client.metrics.flush() + collection(native[0], monkeypatch, duration_ms=7) + session.post.return_value.status_code = 200 + client.metrics.flush() + point = exported_metrics(session)[NAME]["histogram"]["dataPoints"][0] + assert (point["count"], point["sum"], point["min"], point["max"]) == (2, 9, 2, 7) + assert sum(point["bucketCounts"]) == 2 + + +def test_generations_are_distinct_and_hooks_are_deferred( + runtime_client, native, monkeypatch +): + hook = mock.Mock(side_effect=lambda sample: sample) + client, session = runtime_client( + metrics={"capture_runtime_metrics": True, "before_send": hook} + ) + for generation in range(3): + collection(native[0], monkeypatch, generation, generation + 1) + assert not hook.called + client.metrics.flush() + points = exported_metrics(session)[NAME]["histogram"]["dataPoints"] + assert {attributes(p)["cpython.gc.generation"]: p["sum"] for p in points} == { + 0: 1, + 1: 2, + 2: 3, + } + assert all(p["count"] == 1 for p in points) + assert sum(call.args[0]["name"] == NAME for call in hook.call_args_list) == 3 + + +def test_reset_drops_pending_and_inflight_collections( + runtime_client, native, monkeypatch +): + client, session = runtime_client() + module = native[0] + collection(module, monkeypatch) + callback = module.gc_callbacks[0] + callback("start", {"generation": 1}) + client.metrics.reset() + callback("stop", {"generation": 1}) + client.metrics.flush() + assert NAME not in exported_metrics(session) + collection(module, monkeypatch, duration_ms=4) + client.metrics.flush() + assert exported_metrics(session)[NAME]["histogram"]["dataPoints"][0]["sum"] == 4 + + +def test_buffer_is_bounded_and_recovers_with_one_warning( + runtime_client, native, monkeypatch, caplog +): + module = native[0] + monkeypatch.setattr(module, "_MAX_GC_DURATION_SAMPLES", 2) + client, session = runtime_client() + for _ in range(4): + collection(module, monkeypatch) + assert len(client.metrics._runtime_metrics._gc_duration._pending) == 2 + client.metrics.flush() + assert exported_metrics(session)[NAME]["histogram"]["dataPoints"][0]["count"] == 2 + for _ in range(3): + collection(module, monkeypatch) + client.metrics.flush() + assert caplog.text.count("GC duration observations skipped") == 1 + collection(module, monkeypatch) + client.metrics.flush() + assert exported_metrics(session)[NAME]["histogram"]["dataPoints"][0]["count"] == 1 + + +def test_clients_have_independent_callbacks(runtime_client, native, monkeypatch): + first, _ = runtime_client() + second, session = runtime_client() + assert len(native[0].gc_callbacks) == 2 + first.shutdown() + assert len(native[0].gc_callbacks) == 1 + collection(native[0], monkeypatch) + second.metrics.flush() + assert exported_metrics(session)[NAME]["histogram"]["dataPoints"][0]["count"] == 1 + second.shutdown() + assert native[0].gc_callbacks == [] + + +def test_collector_garbage_collection_removes_callback(native): + collector = native[0]._MacOSProcessMetrics() + reference = weakref.ref(collector) + assert len(native[0].gc_callbacks) == 1 + del collector + gc.collect() + assert reference() is None + assert native[0].gc_callbacks == [] + + +@pytest.mark.parametrize("phase,generation", [("other", 0), ("start", -1), ("stop", 3)]) +def test_invalid_gc_notifications_are_ignored( + runtime_client, native, phase, generation +): + client, session = runtime_client() + callback = native[0].gc_callbacks[0] + callback(phase, {"generation": generation}) + callback("stop", {"generation": 0}) + client.metrics.flush() + assert NAME not in exported_metrics(session) + + +def test_clock_failure_and_backwards_clock_do_not_escape( + runtime_client, native, monkeypatch +): + client, session = runtime_client() + clock = mock.Mock(side_effect=[RuntimeError("clock unavailable"), 100, 90]) + monkeypatch.setattr(native[0], "monotonic_ns", clock) + callback = native[0].gc_callbacks[0] + callback("start", {"generation": 0}) + callback("start", {"generation": 0}) + callback("stop", {"generation": 0}) + client.metrics.flush() + assert NAME not in exported_metrics(session) + collection(native[0], monkeypatch) + client.metrics.flush() + assert NAME in exported_metrics(session) + + +def test_gc_durations_do_not_inherit_flush_span_exemplars( + runtime_client, native, monkeypatch +): + client, session = runtime_client() + monkeypatch.setattr( + metrics_capture, "_active_trace_context", lambda _: ("a" * 32, "b" * 16) + ) + collection(native[0], monkeypatch) + client.metrics.flush() + assert ( + "exemplars" not in exported_metrics(session)[NAME]["histogram"]["dataPoints"][0] + ) + + +def test_non_cpython_does_not_install_gc_callback(native, monkeypatch): + module = native[0] + monkeypatch.setattr( + module, "sys", SimpleNamespace(implementation=SimpleNamespace(name="pypy")) + ) + collector = module._MacOSProcessMetrics() + assert module.gc_callbacks == [] + assert all(sample.name != NAME for sample in collector.collect()) + collector.close() + + +def test_histogram_windows_advance_after_each_drain( + runtime_client, native, monkeypatch +): + wall_time = mock.Mock(return_value=1000) + monkeypatch.setattr(native[0], "time", wall_time) + client, session = runtime_client() + collection(native[0], monkeypatch) + wall_time.return_value = 2000 + client.metrics.flush() + assert ( + exported_metrics(session)[NAME]["histogram"]["dataPoints"][0][ + "startTimeUnixNano" + ] + == "1000000000000" + ) + collection(native[0], monkeypatch) + wall_time.return_value = 3000 + client.metrics.flush() + assert ( + exported_metrics(session)[NAME]["histogram"]["dataPoints"][0][ + "startTimeUnixNano" + ] + == "2000000000000" + ) + + +@pytest.mark.skipif( + sys.platform != "darwin" or sys.implementation.name != "cpython", + reason="requires macOS CPython", +) +def test_native_gc_callback_records_actual_collection(): + from .._runtime_metrics import _MacOSProcessMetrics + + original_callbacks = list(gc.callbacks) + was_enabled = gc.isenabled() + gc.disable() + collector = _MacOSProcessMetrics() + try: + gc.collect(2) + durations = [sample for sample in collector.collect() if sample.name == NAME] + assert len(durations) == 1 + assert durations[0].value > 0 + assert durations[0].unit == "ms" + assert durations[0].attributes == {"cpython.gc.generation": 2} + finally: + collector.close() + if was_enabled: + gc.enable() + assert gc.callbacks == original_callbacks + + +def test_runtime_replacement_removes_previous_callback(runtime_client, native): + client, _ = runtime_client() + original = native[0].gc_callbacks[0] + client.metrics._start_runtime_metrics() + assert len(native[0].gc_callbacks) == 1 + assert native[0].gc_callbacks[0] is not original + original("start", {"generation": 0}) + original("stop", {"generation": 0}) + assert not client.metrics._runtime_metrics._gc_duration._pending + + +def test_shutdown_during_initialization_cleans_callback( + runtime_client, native, monkeypatch +): + client, _ = runtime_client(metrics={}) + constructor = native[0]._MacOSProcessMetrics + + def initialize(): + collector = constructor() + client.shutdown() + return collector + + monkeypatch.setattr(native[0], "_MacOSProcessMetrics", initialize) + client.metrics._capture_runtime_metrics = True + client.metrics._start_runtime_metrics() + assert client.metrics._runtime_metrics is None + assert native[0].gc_callbacks == [] + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires fork") +def test_fork_replaces_callback_and_drops_parent_durations( + runtime_client, native, monkeypatch +): + client, session = runtime_client() + collection(native[0], monkeypatch, duration_ms=50) + with client.metrics._runtime_metrics._gc_duration._lock: + pid = os.fork() + if pid == 0: + try: + assert len(native[0].gc_callbacks) == 1 + collection(native[0], monkeypatch, duration_ms=3) + client.metrics.flush() + point = exported_metrics(session)[NAME]["histogram"]["dataPoints"][0] + assert (point["count"], point["sum"]) == (1, 3) + assert attributes(point)["process.pid"] == os.getpid() + client.shutdown() + assert native[0].gc_callbacks == [] + os._exit(0) + except BaseException: + os._exit(1) + _, status = os.waitpid(pid, 0) + assert os.waitstatus_to_exitcode(status) == 0 + assert len(native[0].gc_callbacks) == 1 + client.metrics.flush() + assert exported_metrics(session)[NAME]["histogram"]["dataPoints"][0]["sum"] == 50 diff --git a/posthog/test/test_host_metrics.py b/posthog/test/test_host_metrics.py new file mode 100644 index 000000000..15c174d10 --- /dev/null +++ b/posthog/test/test_host_metrics.py @@ -0,0 +1,589 @@ +import ctypes +import errno +import os +import signal +from types import SimpleNamespace +from unittest import mock + +import pytest +from google.protobuf.json_format import ParseDict +from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( + ExportMetricsServiceRequest, +) + +from .. import metrics_capture +from . import test_runtime_metrics as runtime_tests +from .test_metrics import sent_payload +from .test_runtime_metrics import attributes, exported_metrics + +native = runtime_tests.native +runtime_client = runtime_tests.runtime_client + + +def test_host_opt_in_collects_without_process_metrics( + runtime_client, native, monkeypatch +): + module = native[0] + collector = SimpleNamespace( + attributes={"host.name": "test-host"}, + start_time_ms=1234, + collect=mock.Mock( + return_value=[ + module._RuntimeMetric( + "system.cpu.utilization", + 0.25, + "1", + "gauge", + {"cpu.logical_number": 0, "cpu.mode": "user"}, + ) + ] + ), + ) + constructor = mock.Mock(return_value=collector) + monkeypatch.setattr(module, "_MacOSHostMetrics", constructor, raising=False) + client, session = runtime_client(metrics={"capture_host_metrics": True}) + assert client._metrics is not None + assert client.metrics._runtime_metrics is None + client.metrics._flush_timer.fire() + constructor.assert_called_once() + metric = exported_metrics(session)["system.cpu.utilization"] + point = metric["gauge"]["dataPoints"][0] + assert point["asDouble"] == 0.25 + assert attributes(point) == { + "host.name": "test-host", + "cpu.logical_number": 0, + "cpu.mode": "user", + } + assert client.metrics._flush_timer.active + client.shutdown() + assert client.metrics._host_metrics is None + assert client.metrics._flush_timer is None + + +@pytest.fixture +def host_native(native, monkeypatch): + module = native[0] + state = SimpleNamespace( + free=100, + active=200, + wired=30, + total_memory=8 * 1024**3, + total_error=0, + total_size=8, + inactive=50, + speculative=20, + page_size=16384, + swapins=123, + swapouts=456, + swap_used=98765, + pressure=1, + vm_result=0, + page_result=0, + vm_count=38, + pressure_error=0, + swap_error=0, + pressure_size=4, + swap_size=32, + core_ticks=[[100, 200, 300, 400], [1000, 2000, 3000, 4000]], + processor_result=0, + processor_count=None, + processor_info_count=None, + processor_null=False, + deallocate_result=0, + allocations={}, + ) + + def statistics(host, flavor, output, count): + assert host == 42 + size = ctypes.cast(count, ctypes.POINTER(ctypes.c_uint32)).contents + assert flavor == 4 and size.value == 38 + info = ctypes.cast(output, ctypes.POINTER(module._VMStatistics64)).contents + info.active_count = state.active + info.wire_count = state.wired + info.free_count = state.free + info.inactive_count = state.inactive + info.speculative_count = state.speculative + info.pageins = 1000000 + info.pageouts = 2000000 + info.swapins = state.swapins + info.swapouts = state.swapouts + size.value = state.vm_count + return state.vm_result + + def page_size(host, output): + assert host == 42 + ctypes.cast( + output, ctypes.POINTER(ctypes.c_size_t) + ).contents.value = state.page_size + return state.page_result + + def sysctl(name, output, size_pointer, new_value, new_size): + assert new_value is None and new_size == 0 + size = ctypes.cast(size_pointer, ctypes.POINTER(ctypes.c_size_t)).contents + if name == b"hw.memsize": + assert size.value == 8 + ctypes.cast( + output, ctypes.POINTER(ctypes.c_uint64) + ).contents.value = state.total_memory + size.value = state.total_size + error = state.total_error + elif name == b"vm.swapusage": + assert size.value == 32 + info = ctypes.cast(output, ctypes.POINTER(module._SwapUsage)).contents + info.used = state.swap_used + size.value = state.swap_size + error = state.swap_error + else: + assert name == b"kern.memorystatus_vm_pressure_level" + assert size.value == 4 + ctypes.cast( + output, ctypes.POINTER(ctypes.c_uint32) + ).contents.value = state.pressure + size.value = state.pressure_size + error = state.pressure_error + ctypes.set_errno(error) + return -1 if error else 0 + + def processors(host, flavor, number, output, count): + assert host == 42 and flavor == 2 + ctypes.cast(number, ctypes.POINTER(ctypes.c_uint32)).contents.value = ( + len(state.core_ticks) + if state.processor_count is None + else state.processor_count + ) + ctypes.cast(count, ctypes.POINTER(ctypes.c_uint32)).contents.value = ( + len(state.core_ticks) * 4 + if state.processor_info_count is None + else state.processor_info_count + ) + pointer = ctypes.cast(output, ctypes.POINTER(ctypes.POINTER(ctypes.c_int32))) + if state.processor_null: + pointer[0] = ctypes.POINTER(ctypes.c_int32)() + else: + values = [tick for core in state.core_ticks for tick in core] + array = (ctypes.c_int32 * len(values))(*values) + state.allocations[ctypes.addressof(array)] = array + pointer[0] = ctypes.cast(array, ctypes.POINTER(ctypes.c_int32)) + return state.processor_result + + def deallocate(task, address, size): + assert task == 43 + assert address in state.allocations + expected_count = ( + len(state.core_ticks) * 4 + if state.processor_info_count is None + else state.processor_info_count + ) + assert size == expected_count * ctypes.sizeof(ctypes.c_int32) + del state.allocations[address] + return state.deallocate_result + + library = mock.Mock( + mach_host_self=mock.Mock(return_value=42), + mach_task_self=mock.Mock(return_value=43), + mach_port_deallocate=mock.Mock(return_value=0), + host_statistics64=mock.Mock(side_effect=statistics), + host_page_size=mock.Mock(side_effect=page_size), + sysctlbyname=mock.Mock(side_effect=sysctl), + host_processor_info=mock.Mock(side_effect=processors), + mach_vm_deallocate=mock.Mock(side_effect=deallocate), + ) + original = module.ctypes.CDLL + + def load(path, **kwargs): + if path == "/usr/lib/libSystem.B.dylib": + return library + return original(path, **kwargs) + + monkeypatch.setattr(module.ctypes, "CDLL", load) + monkeypatch.setattr(module, "gethostname", lambda: "test-host") + return module, state, library + + +def host_values(collector): + return {sample.name: sample.value for sample in collector.collect()} + + +def test_native_host_abi(host_native): + module = host_native[0] + assert ctypes.sizeof(module._HostCPULoadInfo) == 16 + assert ctypes.sizeof(module._VMStatistics64) == 152 + assert module._VMStatistics64.swapins.offset == 112 + assert module._VMStatistics64.total_uncompressed_pages_in_compressor.offset == 144 + assert ctypes.sizeof(module._SwapUsage) == 32 + assert module._SwapUsage.encrypted.offset == 28 + + +@pytest.mark.parametrize("page_size", [4096, 16384]) +def test_native_host_values_and_cumulative_baselines(host_native, page_size): + module, state, library = host_native + state.page_size = page_size + collector = module._MacOSHostMetrics() + state.core_ticks[0][0] += 100 + state.swapins += 3 + state.swapouts += 5 + samples = collector.collect() + values = {sample.name: sample.value for sample in samples} + assert values["system.cpu.utilization"] == 0 + assert values["system.memory.usage"] == (state.active + state.wired) * page_size + assert values["system.memory.limit"] == state.total_memory + assert values["app.system.memory.pressure"] == 1 + assert values["system.paging.usage"] == 98765 + swap = [sample for sample in samples if sample.name == "app.system.swap.io"] + assert [(sample.attributes, sample.value) for sample in swap] == [ + ({"system.paging.direction": "in"}, 3 * page_size), + ({"system.paging.direction": "out"}, 5 * page_size), + ] + assert all( + sample.unit == "By" and sample.metric_type == "observable_counter" + for sample in swap + ) + assert collector.attributes == {"host.name": "test-host"} + assert library.mach_port_deallocate.call_args_list == [mock.call(43, 42)] * 4 + + +def test_host_uses_per_core_cpu_instead_of_custom_aggregate(host_native): + module, state, _ = host_native + collector = module._MacOSHostMetrics() + assert "app.system.cpu.utilization" not in host_values(collector) + state.core_ticks[0][0] += 100 + assert "system.cpu.utilization" in host_values(collector) + assert "app.system.cpu.utilization" not in host_values(collector) + + +@pytest.mark.parametrize("level", [1, 2, 4]) +def test_host_pressure_uses_native_categorical_levels(host_native, level): + module, state, _ = host_native + state.pressure = level + collector = module._MacOSHostMetrics() + sample = next( + sample + for sample in collector.collect() + if sample.name == "app.system.memory.pressure" + ) + assert sample.value == level + assert sample.metric_type == "gauge" + assert sample.unit == "{level}" + + +@pytest.mark.parametrize("level", [0, 3, 8]) +def test_unknown_pressure_is_omitted_and_recovers(host_native, caplog, level): + module, state, _ = host_native + state.pressure = level + collector = module._MacOSHostMetrics() + assert "app.system.memory.pressure" not in host_values(collector) + assert "app.system.memory.pressure" not in host_values(collector) + assert caplog.text.count("Unable to collect host memory pressure") == 1 + state.pressure = 2 + assert host_values(collector)["app.system.memory.pressure"] == 2 + + +@pytest.mark.parametrize( + "field,bad,good,omitted", + [ + ("total_error", errno.EPERM, 0, {"system.memory.limit"}), + ("total_error", errno.ENOENT, 0, {"system.memory.limit"}), + ("total_size", 4, 8, {"system.memory.limit"}), + ( + "vm_result", + 5, + 0, + {"system.memory.usage", "app.system.swap.io"}, + ), + ( + "vm_count", + 24, + 38, + {"system.memory.usage", "app.system.swap.io"}, + ), + ( + "page_result", + 5, + 0, + {"system.memory.usage", "app.system.swap.io"}, + ), + ( + "page_size", + 0, + 16384, + {"system.memory.usage", "app.system.swap.io"}, + ), + ("pressure_error", errno.EPERM, 0, {"app.system.memory.pressure"}), + ("pressure_error", errno.ENOENT, 0, {"app.system.memory.pressure"}), + ("pressure_size", 3, 4, {"app.system.memory.pressure"}), + ("swap_error", errno.EPERM, 0, {"system.paging.usage"}), + ("swap_size", 24, 32, {"system.paging.usage"}), + ], +) +def test_host_source_failures_are_isolated_and_recover( + host_native, caplog, field, bad, good, omitted +): + module, state, library = host_native + collector = module._MacOSHostMetrics() + setattr(state, field, bad) + expected = { + "system.cpu.utilization", + "system.memory.limit", + "system.memory.usage", + "app.system.memory.pressure", + "app.system.swap.io", + "system.paging.usage", + } + for _ in range(2): + state.core_ticks[0][0] += 100 + assert set(host_values(collector)) == expected - omitted + assert caplog.text.count("Unable to collect host") == 1 + assert library.mach_host_self.call_count == library.mach_port_deallocate.call_count + setattr(state, field, good) + state.core_ticks[0][0] += 100 + assert set(host_values(collector)) == expected + + +def test_failed_initial_vm_read_seeds_swap_baseline_on_recovery(host_native): + module, state, _ = host_native + state.vm_result = 5 + collector = module._MacOSHostMetrics() + state.swapins += 100 + state.vm_result = 0 + samples = collector.collect() + assert [ + sample.value for sample in samples if sample.name == "app.system.swap.io" + ] == [0, 0] + state.swapins += 2 + samples = collector.collect() + assert [ + sample.value for sample in samples if sample.name == "app.system.swap.io" + ] == [32768, 0] + + +def test_host_and_process_exports_keep_separate_identity_and_retry_values( + runtime_client, host_native, native +): + client, session = runtime_client( + metrics={"capture_runtime_metrics": True, "capture_host_metrics": True} + ) + native[1].time += 2 + native[1].user += 1 + host_native[1].core_ticks[0][0] += 100 + host_native[1].swapins += 2 + session.post.return_value.status_code = 503 + client.metrics._consecutive_send_failures = 1 + client.metrics._flush_timer.fire() + assert client.metrics._flush_timer.interval == 20 + first = exported_metrics(session)["app.system.swap.io"]["sum"]["dataPoints"] + host_native[1].swapins += 3 + host_native[1].swap_used = 0 + session.post.return_value.status_code = 200 + client.metrics.flush() + metrics = exported_metrics(session) + ParseDict(sent_payload(session), ExportMetricsServiceRequest()) + assert len(metrics) == 15 + for name, metric in metrics.items(): + for point in metric.get("sum", metric.get("gauge"))["dataPoints"]: + tags = attributes(point) + if name.startswith(("app.system.", "system.")): + assert tags["host.name"] == "test-host" + assert "process.pid" not in tags and "process.runtime_id" not in tags + else: + assert tags["process.pid"] == os.getpid() + assert "host.name" not in tags + swap = metrics["app.system.swap.io"]["sum"] + assert swap["aggregationTemporality"] == 2 and swap["isMonotonic"] + inbound = next( + point + for point in swap["dataPoints"] + if attributes(point)["system.paging.direction"] == "in" + ) + assert inbound["asDouble"] == 5 * 16384 + assert inbound["startTimeUnixNano"] == first[0]["startTimeUnixNano"] + usage = metrics["system.paging.usage"]["sum"] + assert usage["aggregationTemporality"] == 2 and not usage["isMonotonic"] + assert usage["dataPoints"][0]["asDouble"] == 0 + + +@pytest.mark.parametrize("flag", [False, None, 1, "true", {}, []]) +def test_host_opt_in_requires_exact_true(runtime_client, native, monkeypatch, flag): + constructor = mock.Mock(side_effect=AssertionError("unexpected host collection")) + monkeypatch.setattr(native[0], "_MacOSHostMetrics", constructor) + client, _ = runtime_client(metrics={"capture_host_metrics": flag}) + assert client.metrics._host_metrics is None + constructor.assert_not_called() + + +@pytest.mark.parametrize("options", [{"disabled": True}, {"send": False}]) +def test_host_collection_skips_non_sending_clients( + runtime_client, host_native, options +): + client, session = runtime_client(metrics={"capture_host_metrics": True}, **options) + assert client.metrics._host_metrics is None + assert client.metrics._flush_timer is None + host_native[2].mach_host_self.assert_not_called() + host_native[2].sysctlbyname.assert_not_called() + session.post.assert_not_called() + + +def test_host_collection_warns_on_unsupported_platform( + runtime_client, monkeypatch, caplog +): + monkeypatch.setattr(metrics_capture.sys, "platform", "linux") + client, _ = runtime_client(metrics={"capture_host_metrics": True}) + assert client.metrics._host_metrics is None + assert "Automatic host metrics are currently supported only on macOS" in caplog.text + + +@pytest.mark.parametrize("option", ["capture_host_metrics", "capture_runtime_metrics"]) +def test_optional_native_import_failure_preserves_manual_config(runtime_client, option): + original_import = __import__ + + def load(name, *args, **kwargs): + if name == "_runtime_metrics": + raise ImportError("native module unavailable") + return original_import(name, *args, **kwargs) + + hook = mock.Mock(side_effect=lambda sample: sample) + with mock.patch("builtins.__import__", side_effect=load): + client, session = runtime_client( + metrics={option: True, "service_name": "configured", "before_send": hook} + ) + client.metrics.count("manual") + client.metrics.flush() + assert ( + attributes(sent_payload(session)["resourceMetrics"][0]["resource"])[ + "service.name" + ] + == "configured" + ) + hook.assert_called_once() + assert client.metrics._host_metrics is None + assert client.metrics._runtime_metrics is None + + +@pytest.mark.parametrize("failed", ["_MacOSHostMetrics", "_MacOSProcessMetrics"]) +def test_collector_initialization_failures_do_not_disable_other_collector( + runtime_client, host_native, monkeypatch, failed +): + monkeypatch.setattr( + host_native[0], failed, mock.Mock(side_effect=OSError("unavailable")) + ) + client, session = runtime_client( + metrics={"capture_host_metrics": True, "capture_runtime_metrics": True} + ) + client.metrics.flush() + names = exported_metrics(session) + assert ("process.cpu.time" in names) == (failed != "_MacOSProcessMetrics") + assert ("system.memory.usage" in names) == (failed != "_MacOSHostMetrics") + + +def test_shutdown_during_host_initialization_does_not_install_collector( + runtime_client, host_native, monkeypatch +): + client, session = runtime_client(metrics=None) + client._metrics_config = {"capture_host_metrics": True} + constructor = host_native[0]._MacOSHostMetrics + + def initialize(): + client.shutdown() + return constructor() + + monkeypatch.setattr(host_native[0], "_MacOSHostMetrics", initialize) + client._start_runtime_metrics() + assert client._shutdown_complete + assert client.metrics._host_metrics is None + assert client.metrics._flush_timer is None + session.post.assert_not_called() + + +def test_host_before_send_can_shutdown_and_reset_keeps_collection( + runtime_client, host_native +): + closed = False + + def hook(sample): + if closed: + client.shutdown() + return sample + + client, session = runtime_client( + metrics={"capture_host_metrics": True, "before_send": hook} + ) + first = client.metrics._flush_timer + client.metrics.reset() + assert not first.active and client.metrics._flush_timer.active + closed = True + client.metrics.flush() + client.shutdown() + assert client._shutdown_complete + assert client.metrics._host_metrics is None + assert client.metrics._flush_timer is None + assert not client.metrics._series + assert "app.system.memory.pressure" in exported_metrics(session) + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires fork") +def test_fork_in_process_hook_skips_inherited_host_sampler( + runtime_client, host_native, monkeypatch +): + parent = os.getpid() + children = [] + + def hook(sample): + if not children: + children.append(os.fork()) + if os.getpid() != parent: + signal.alarm(5) + return sample + + client, _ = runtime_client( + metrics={ + "capture_runtime_metrics": True, + "capture_host_metrics": True, + "before_send": hook, + } + ) + old_host = client.metrics._host_metrics + read = mock.Mock(wraps=old_host.collect) + monkeypatch.setattr(old_host, "collect", read) + try: + client.metrics.flush() + if os.getpid() != parent: + read.assert_not_called() + assert client.metrics._host_metrics is None + except BaseException: + if os.getpid() != parent: + os._exit(1) + raise + if os.getpid() != parent: + os._exit(0) + _, status = os.waitpid(children[0], 0) + assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0 + read.assert_called_once() + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires fork") +@pytest.mark.parametrize("runtime", [False, True]) +def test_fork_does_not_duplicate_host_collection(runtime_client, host_native, runtime): + client, session = runtime_client( + metrics={"capture_host_metrics": True, "capture_runtime_metrics": runtime} + ) + parent_host = client.metrics._host_metrics + pid = os.fork() + if pid == 0: + signal.alarm(5) + try: + assert client.metrics._host_metrics is None + assert (client.metrics._runtime_metrics is not None) == runtime + assert (client.metrics._flush_timer is not None) == runtime + client.metrics.flush() + if runtime: + assert "app.system.swap.io" not in exported_metrics(session) + else: + session.post.assert_not_called() + explicit, _ = runtime_client(metrics={"capture_host_metrics": True}) + assert explicit.metrics._host_metrics is not None + explicit.shutdown() + except BaseException: + os._exit(1) + os._exit(0) + _, status = os.waitpid(pid, 0) + assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0 + assert client.metrics._host_metrics is parent_host diff --git a/posthog/test/test_per_core_cpu_metrics.py b/posthog/test/test_per_core_cpu_metrics.py new file mode 100644 index 000000000..cc2f83272 --- /dev/null +++ b/posthog/test/test_per_core_cpu_metrics.py @@ -0,0 +1,238 @@ +import ctypes +import os +import sys +import time + +import pytest +from google.protobuf.json_format import ParseDict +from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( + ExportMetricsServiceRequest, +) + +from . import test_host_metrics as host_tests +from .test_metrics import sent_payload +from .test_runtime_metrics import attributes, exported_metrics + +native = host_tests.native +host_native = host_tests.host_native +runtime_client = host_tests.runtime_client + +NAME = "system.cpu.utilization" + + +def core_samples(collector): + return [sample for sample in collector.collect() if sample.name == NAME] + + +def advance(state): + state.core_ticks[0] = [130, 220, 300, 450] + state.core_ticks[1] = [1000, 2000, 3100, 4000] + + +def test_per_core_native_function_signatures(host_native): + module, _, library = host_native + module._MacOSHostMetrics() + assert library.host_processor_info.argtypes == [ + ctypes.c_uint32, + ctypes.c_int32, + ctypes.POINTER(ctypes.c_uint32), + ctypes.POINTER(ctypes.POINTER(ctypes.c_int32)), + ctypes.POINTER(ctypes.c_uint32), + ] + assert library.mach_vm_deallocate.argtypes == [ + ctypes.c_uint32, + ctypes.c_uint64, + ctypes.c_uint64, + ] + assert ( + library.host_processor_info.restype + == library.mach_vm_deallocate.restype + == ctypes.c_int32 + ) + + +def test_per_core_modes_without_custom_aggregate(host_native): + module, state, library = host_native + collector = module._MacOSHostMetrics() + advance(state) + samples = collector.collect() + cores = [sample for sample in samples if sample.name == NAME] + assert len(cores) == 8 + values = { + ( + sample.attributes["cpu.logical_number"], + sample.attributes["cpu.mode"], + ): sample.value + for sample in cores + } + assert values == { + (0, "user"): 0.3, + (0, "system"): 0.2, + (0, "idle"): 0, + (0, "nice"): 0.5, + (1, "user"): 0, + (1, "system"): 0, + (1, "idle"): 1, + (1, "nice"): 0, + } + assert all(sample.unit == "1" and sample.metric_type == "gauge" for sample in cores) + assert not any(sample.name == "app.system.cpu.utilization" for sample in samples) + assert not state.allocations + assert ( + library.host_processor_info.call_count + == library.mach_vm_deallocate.call_count + == 2 + ) + assert library.mach_host_self.call_count == library.mach_port_deallocate.call_count + + +def test_per_core_omits_initial_and_unchanged_counters(host_native): + module, state, library = host_native + collector = module._MacOSHostMetrics() + assert library.host_processor_info.call_count == 1 + assert core_samples(collector) == [] + state.core_ticks[1][0] += 100 + samples = core_samples(collector) + assert len(samples) == 4 + assert all(sample.attributes["cpu.logical_number"] == 1 for sample in samples) + assert sum(sample.value for sample in samples) == 1 + assert core_samples(collector) == [] + assert not state.allocations + + +@pytest.mark.parametrize("before", [2**31 - 10, 2**32 - 10]) +def test_per_core_ticks_are_unsigned_and_wrap_without_spikes(host_native, before): + module, state, _ = host_native + state.core_ticks = [[before] * 4] + collector = module._MacOSHostMetrics() + state.core_ticks = [[(before + delta) % 2**32 for delta in (20, 20, 60, 0)]] + samples = core_samples(collector) + assert {sample.attributes["cpu.mode"]: sample.value for sample in samples} == { + "user": 0.2, + "system": 0.2, + "idle": 0.6, + "nice": 0, + } + + +def test_topology_change_rebaselines_all_logical_cpus(host_native): + module, state, _ = host_native + collector = module._MacOSHostMetrics() + advance(state) + state.core_ticks.append([10, 20, 30, 40]) + assert core_samples(collector) == [] + for ticks in state.core_ticks: + ticks[0] += 10 + assert len(core_samples(collector)) == 12 + state.core_ticks.pop() + assert core_samples(collector) == [] + for ticks in state.core_ticks: + ticks[2] += 10 + samples = core_samples(collector) + assert len(samples) == 8 + assert all( + sample.value == (1 if sample.attributes["cpu.mode"] == "idle" else 0) + for sample in samples + ) + + +@pytest.mark.parametrize( + "field,bad,good", + [ + ("processor_result", 5, 0), + ("processor_count", 0, None), + ("processor_info_count", 7, None), + ("processor_info_count", 9, None), + ("processor_null", True, False), + ("deallocate_result", 5, 0), + ], +) +def test_per_core_failure_is_isolated_frees_resources_and_recovers( + host_native, caplog, field, bad, good +): + module, state, library = host_native + collector = module._MacOSHostMetrics() + advance(state) + setattr(state, field, bad) + for _ in range(2): + samples = collector.collect() + assert not any(sample.name == NAME for sample in samples) + assert any(sample.name == "app.system.memory.pressure" for sample in samples) + assert not state.allocations + assert caplog.text.count("Unable to collect host per-CPU metrics") == 1 + setattr(state, field, good) + assert len(core_samples(collector)) == 8 + assert library.mach_host_self.call_count == library.mach_port_deallocate.call_count + assert ( + library.mach_vm_deallocate.call_count + == library.host_processor_info.call_count + - (2 if field == "processor_null" else 0) + ) + + +def test_initial_per_core_failure_seeds_on_first_success(host_native): + module, state, _ = host_native + state.processor_result = 5 + collector = module._MacOSHostMetrics() + state.processor_result = 0 + advance(state) + assert core_samples(collector) == [] + state.core_ticks[0][0] += 10 + assert len(core_samples(collector)) == 4 + + +def test_otlp_per_core_gauges_keep_host_identity_and_latest_reading( + runtime_client, host_native +): + client, session = runtime_client(metrics={"capture_host_metrics": True}) + state = host_native[1] + advance(state) + session.post.return_value.status_code = 503 + client.metrics.flush() + first = exported_metrics(session)[NAME] + assert len(first["gauge"]["dataPoints"]) == 8 + state.core_ticks[0][2] += 100 + state.core_ticks[1][0] += 100 + session.post.return_value.status_code = 200 + client.metrics.flush() + points = exported_metrics(session)[NAME]["gauge"]["dataPoints"] + assert len(points) == 8 + for point in points: + tags = attributes(point) + assert set(tags) == {"host.name", "cpu.logical_number", "cpu.mode"} + assert tags["host.name"] == "test-host" + assert type(tags["cpu.logical_number"]) is int + expected = ( + 1 + if (tags["cpu.logical_number"], tags["cpu.mode"]) + in ((0, "idle"), (1, "user")) + else 0 + ) + assert point["asDouble"] == expected + ParseDict(sent_payload(session), ExportMetricsServiceRequest()) + client.shutdown() + assert not state.allocations + assert client.metrics._flush_timer is None + + +@pytest.mark.skipif(sys.platform != "darwin", reason="requires macOS") +def test_native_per_core_cpu_readings(): + from .._runtime_metrics import _MacOSHostMetrics + + collector = _MacOSHostMetrics() + start = time.monotonic() + while time.monotonic() - start < 0.1: + sum(i * i for i in range(1000)) + samples = core_samples(collector) + assert samples + cores = {} + for sample in samples: + cpu = sample.attributes["cpu.logical_number"] + assert 0 <= cpu < os.cpu_count() + assert sample.attributes["cpu.mode"] in ("user", "system", "idle", "nice") + assert 0 <= sample.value <= 1 + cores.setdefault(cpu, []).append(sample.value) + assert all( + len(values) == 4 and sum(values) == pytest.approx(1) + for values in cores.values() + ) diff --git a/posthog/test/test_runtime_metrics.py b/posthog/test/test_runtime_metrics.py new file mode 100644 index 000000000..ff77115d8 --- /dev/null +++ b/posthog/test/test_runtime_metrics.py @@ -0,0 +1,670 @@ +import ctypes +import importlib +import os +import signal +import sys +import threading +import time +from types import SimpleNamespace +from unittest import mock + +import pytest + +import posthog +from posthog import metrics_capture +from posthog.client import Client +from posthog.test.test_metrics import mock_session, sent_payload + + +class ManualTimer: + def __init__(self, interval, callback): + self.interval = interval + self.callback = callback + self.daemon = False + self.active = False + self.started = False + + def start(self): + assert not self.started + self.started = True + self.active = True + + def cancel(self): + self.active = False + + def fire(self): + self.active = False + self.callback() + + +@pytest.fixture +def native(monkeypatch): + pytest.importorskip("resource") + module = importlib.import_module("posthog._runtime_metrics") + state = SimpleNamespace( + user=1.5, + system=0.5, + time=100.0, + rss=123456, + size=96, + threads=7, + switches=100, + voluntary=50, + involuntary=100, + cpus=8, + physical_pages=4096, + page_size=4096, + gc_stats=[ + {"collections": 10, "collected": 25, "uncollectable": 1}, + {"collections": 3, "collected": 18, "uncollectable": 0}, + {"collections": 1, "collected": 5, "uncollectable": 2}, + ], + ) + + def read_process(pid, flavor, arg, pointer, size): + assert pid == os.getpid() + assert flavor == 4 + assert arg == 0 + assert size == 96 + info = ctypes.cast(pointer, ctypes.POINTER(module._ProcTaskInfo)).contents + info.pti_resident_size = state.rss + info.pti_threadnum = state.threads + info.pti_csw = state.switches + return state.size + + proc_pidinfo = mock.Mock(side_effect=read_process) + library = mock.Mock(proc_pidinfo=proc_pidinfo) + monkeypatch.setattr(module.ctypes, "CDLL", mock.Mock(return_value=library)) + monkeypatch.setattr( + module, + "getrusage", + lambda who: SimpleNamespace( + ru_utime=state.user, + ru_stime=state.system, + ru_nvcsw=state.voluntary, + ru_nivcsw=state.involuntary, + ), + ) + monkeypatch.setattr(module, "gethostname", lambda: "test-host") + monkeypatch.setattr(module, "monotonic", lambda: state.time) + monkeypatch.setattr(module, "gc_callbacks", [], raising=False) + monkeypatch.setattr( + module, "get_stats", lambda: [stats.copy() for stats in state.gc_stats] + ) + monkeypatch.setattr( + module.os, + "sysconf", + lambda name: { + "SC_PHYS_PAGES": state.physical_pages, + "SC_PAGE_SIZE": state.page_size, + }[name], + ) + monkeypatch.setattr( + module.os, "process_cpu_count", lambda: state.cpus, raising=False + ) + return module, state, proc_pidinfo + + +@pytest.fixture +def runtime_client(native, monkeypatch): + monkeypatch.setattr(metrics_capture.sys, "platform", "darwin") + monkeypatch.setattr(metrics_capture.threading, "Timer", ManualTimer) + session = mock_session() + monkeypatch.setattr(metrics_capture, "_get_session", lambda: session) + clients = [] + + def create(**kwargs): + kwargs.setdefault("sync_mode", True) + kwargs.setdefault( + "metrics", {"capture_runtime_metrics": True, "service_name": "worker"} + ) + client = Client("phc_test_key", host="https://us.example.com", **kwargs) + clients.append(client) + return client, session + + yield create + for client in clients: + client.shutdown() + + +def exported_metrics(session): + return { + metric["name"]: metric + for group in sent_payload(session)["resourceMetrics"] + for scope in group["scopeMetrics"] + for metric in scope["metrics"] + } + + +def attributes(point): + return { + item["key"]: next(iter(item["value"].values())) for item in point["attributes"] + } + + +@pytest.mark.parametrize("before", [2**31 - 2, 2**32 - 2]) +def test_context_switch_native_counters_preserve_large_values(native, before): + module, state, _ = native + state.involuntary = before + collector = module._MacOSProcessMetrics() + state.involuntary += 5 + state.time += 10 + samples = {sample.name: sample.value for sample in collector.collect()} + assert samples["process.context_switches"] == 5 + + +def test_native_read_failure_omits_native_metrics_and_warns_once(native, caplog): + module, state, _ = native + collector = module._MacOSProcessMetrics() + state.size = 8 + for _ in range(2): + state.time += 10 + samples = {sample.name: sample.value for sample in collector.collect()} + assert "process.cpu.time" in samples + assert "cpython.gc.collections" in samples + assert "process.memory.usage" not in samples + assert "process.memory.utilization" not in samples + assert "process.thread.count" not in samples + assert "process.context_switches" in samples + state.time += 10 + state.size = 96 + state.involuntary += 10000 + samples = {sample.name: sample.value for sample in collector.collect()} + assert samples["process.thread.count"] == 7 + assert samples["process.context_switches"] == 10000 + assert caplog.text.count("Unable to collect native process metrics") == 1 + + +def test_native_baseline_failure_does_not_disable_cpu_and_gc(native): + module, state, _ = native + state.size = 8 + collector = module._MacOSProcessMetrics() + state.time += 10 + samples = {sample.name: sample.value for sample in collector.collect()} + assert "process.cpu.time" in samples + assert "cpython.gc.collections" in samples + state.time += 10 + state.size = 96 + samples = {sample.name: sample.value for sample in collector.collect()} + assert samples["process.memory.usage"] == state.rss + assert samples["process.context_switches"] == 0 + state.involuntary += 3 + samples = {sample.name: sample.value for sample in collector.collect()} + assert samples["process.context_switches"] == 3 + + +def test_opt_in_starts_collection_without_manual_metrics_calls(runtime_client): + client, _ = runtime_client() + assert client._metrics is not None + assert client._metrics._runtime_metrics is not None + assert client._metrics._flush_timer.active + assert client._metrics._flush_timer.interval == 10 + + +@pytest.mark.parametrize("metrics", [None, {}, {"capture_runtime_metrics": False}]) +def test_disabled_by_default(metrics): + client = Client("phc_test_key", sync_mode=True, metrics=metrics) + try: + assert client._metrics is None + assert client.metrics._runtime_metrics is None + assert client.metrics._flush_timer is None + finally: + client.shutdown() + + +@pytest.mark.parametrize("value", ["true", 1, {}, None]) +def test_option_requires_true_boolean(runtime_client, value, caplog): + client, _ = runtime_client(metrics={"capture_runtime_metrics": value}) + assert client.metrics._runtime_metrics is None + assert "expected a boolean" in caplog.text + + +@pytest.mark.parametrize("kwargs", [{"disabled": True}, {"send": False}]) +def test_disabled_or_non_sending_client_does_not_collect( + runtime_client, native, kwargs +): + client, _ = runtime_client(**kwargs) + assert client.metrics._runtime_metrics is None + assert client.metrics._flush_timer is None + native[2].assert_not_called() + + +def test_unsupported_platform_is_noop(monkeypatch, caplog): + monkeypatch.setattr(metrics_capture.sys, "platform", "linux") + client = Client( + "phc_test_key", sync_mode=True, metrics={"capture_runtime_metrics": True} + ) + try: + assert client.metrics._runtime_metrics is None + assert client.metrics._flush_timer is None + assert "only on macOS" in caplog.text + finally: + client.shutdown() + + +def test_timer_collects_process_metrics_and_continues(runtime_client, native): + client, session = runtime_client( + metrics={"capture_runtime_metrics": True, "flush_interval": 3} + ) + state = native[1] + state.time += 2 + state.user += 1 + state.system += 2 + timer = client.metrics._flush_timer + timer.fire() + metrics = exported_metrics(session) + cpu = metrics["process.cpu.time"] + memory = metrics["process.memory.usage"] + assert cpu["unit"] == "s" + assert cpu["sum"]["dataPoints"][0]["asDouble"] == 1 + assert memory["unit"] == "By" + assert memory["sum"]["dataPoints"][0]["asDouble"] == state.rss + tags = attributes(memory["sum"]["dataPoints"][0]) + assert tags["process.pid"] == os.getpid() + assert attributes(sent_payload(session)["resourceMetrics"][0]["resource"])[ + "service.instance.id" + ] + assert "process.runtime_id" not in tags + assert attributes(cpu["sum"]["dataPoints"][0]) == { + **tags, + "cpu.mode": "user", + } + assert client.metrics._flush_timer is not timer + assert client.metrics._flush_timer.active + assert client.metrics._flush_timer.interval == 3 + + +@pytest.mark.parametrize("runtime_enabled", [False, True]) +@pytest.mark.parametrize("outcome", ["ok", "retry-later"]) +@pytest.mark.parametrize("reset_during_export", [False, True]) +def test_timer_waits_for_export_before_starting_next_timer( + runtime_client, monkeypatch, runtime_enabled, outcome, reset_during_export +): + client, _ = runtime_client(metrics={"capture_runtime_metrics": runtime_enabled}) + metrics = client.metrics + metrics.count("manual.before") + metrics._consecutive_send_failures = 1 + entered, release = threading.Event(), threading.Event() + + def send(payload): + entered.set() + assert release.wait(5) + return outcome + + sender = mock.Mock(side_effect=send) + monkeypatch.setattr(metrics, "_send", sender) + timer = metrics._flush_timer + worker = threading.Thread(target=timer.fire) + worker.start() + try: + assert entered.wait(5) + if reset_during_export: + metrics.reset() + metrics.count("manual.during") + pending = metrics._flush_timer + assert pending is None or not pending.active + finally: + release.set() + worker.join(5) + assert not worker.is_alive() + assert sender.call_count == 1 + assert metrics._flush_timer.active + assert metrics._flush_timer.interval == metrics._flush_interval * ( + 2 if outcome == "retry-later" else 1 + ) + + +def test_runtime_samples_use_before_send(runtime_client, native): + hook = mock.Mock(side_effect=lambda sample: None) + client, session = runtime_client( + metrics={"capture_runtime_metrics": True, "before_send": hook} + ) + native[1].time += 10 + client.metrics._flush_timer.fire() + assert hook.call_count == 18 + session.post.assert_not_called() + assert client.metrics._flush_timer.active + + +def test_shutdown_collects_final_samples_and_stops_timer(runtime_client, native): + client, session = runtime_client() + timer = client.metrics._flush_timer + native[1].time += 10 + native[1].user += 2.5 + client.shutdown() + assert ( + exported_metrics(session)["process.cpu.time"]["sum"]["dataPoints"][0][ + "asDouble" + ] + == 2.5 + ) + assert not timer.active + assert client.metrics._runtime_metrics is None + assert client.metrics._flush_timer is None + session.post.reset_mock() + timer.fire() + client.metrics.flush() + session.post.assert_not_called() + + +def test_shutdown_from_before_send_stops_collection(runtime_client, native): + client, _ = runtime_client() + client.metrics._before_send = lambda sample: (client.shutdown(), sample)[1] + native[1].time += 10 + client.metrics._flush_timer.fire() + client.shutdown() + assert client._shutdown_complete + assert client.metrics._flush_timer is None + assert not client.metrics._series + + +def test_shutdown_waits_for_inflight_collection(runtime_client, native): + client, session = runtime_client() + entered, release = threading.Event(), threading.Event() + + def hook(sample): + entered.set() + assert release.wait(5) + return sample + + client.metrics._before_send = hook + native[1].time += 10 + sampler = threading.Thread(target=client.metrics._flush_timer.fire) + sampler.start() + assert entered.wait(5) + shutdown = threading.Thread(target=client.shutdown) + shutdown.start() + release.set() + sampler.join(5) + shutdown.join(5) + assert not sampler.is_alive() + assert not shutdown.is_alive() + assert client._shutdown_complete + assert client.metrics._runtime_metrics is None + assert client.metrics._flush_timer is None + assert not client.metrics._series + assert session.post.called + + +def test_reset_clears_window_but_collection_continues(runtime_client, native): + client, session = runtime_client() + client.metrics.count("manual.metric") + old_timer = client.metrics._flush_timer + client.metrics.reset() + assert not old_timer.active + assert client.metrics._flush_timer.active + native[1].time += 10 + client.metrics._flush_timer.fire() + metrics = exported_metrics(session) + assert "manual.metric" not in metrics + assert "process.cpu.utilization" in metrics + assert "process.memory.usage" in metrics + + +def test_runtime_collection_recovers_from_read_error( + runtime_client, native, monkeypatch, caplog +): + client, session = runtime_client() + module, state, _ = native + with monkeypatch.context() as patch: + patch.setattr( + module, "getrusage", mock.Mock(side_effect=OSError("unavailable")) + ) + client.metrics._flush_timer.fire() + client.metrics._flush_timer.fire() + session.post.assert_not_called() + assert caplog.text.count("Unable to collect runtime metrics") == 1 + assert client.metrics._flush_timer.active + state.time += 10 + client.metrics._flush_timer.fire() + assert "process.memory.usage" in exported_metrics(session) + + +def test_initialization_failure_does_not_break_client( + runtime_client, native, monkeypatch, caplog +): + monkeypatch.setattr( + native[0].ctypes, "CDLL", mock.Mock(side_effect=OSError("missing libproc")) + ) + client, _ = runtime_client() + assert client.metrics._runtime_metrics is None + assert client.metrics._flush_timer is None + assert "Unable to start runtime metrics" in caplog.text + client.metrics.count("manual.metric") + client.metrics.flush() + assert client.metrics._flush_timer is not None + + +def test_late_module_configuration_starts_collection(runtime_client, monkeypatch): + for key, value in { + "default_client": None, + "api_key": "phc_test_key", + "sync_mode": True, + "metrics": None, + }.items(): + monkeypatch.setattr(posthog, key, value) + client = posthog.setup() + try: + assert client._metrics is None + posthog.metrics = {"capture_runtime_metrics": True} + assert posthog.setup() is client + assert client.metrics._runtime_metrics is not None + assert client.metrics._flush_timer.active + finally: + client.shutdown() + + +def test_shutdown_waits_for_runtime_metrics_initialization( + runtime_client, native, monkeypatch +): + client, _ = runtime_client(metrics=None) + client._metrics_config = {"capture_runtime_metrics": True} + entered, release = threading.Event(), threading.Event() + stopping, stopped = threading.Event(), threading.Event() + constructor = native[0]._MacOSProcessMetrics + original_stop = metrics_capture.PostHogMetrics._stop_runtime_metrics + + def initialize(): + entered.set() + assert release.wait(5) + return constructor() + + def stop(metrics): + stopping.set() + original_stop(metrics) + stopped.set() + + monkeypatch.setattr(native[0], "_MacOSProcessMetrics", initialize) + monkeypatch.setattr(metrics_capture.PostHogMetrics, "_stop_runtime_metrics", stop) + startup = threading.Thread(target=client._start_runtime_metrics) + shutdown = threading.Thread(target=client.shutdown) + startup.start() + try: + assert entered.wait(5) + shutdown.start() + assert stopping.wait(5) + assert not stopped.wait(0.1) + finally: + release.set() + startup.join(5) + if shutdown.ident is not None: + shutdown.join(5) + assert not startup.is_alive() and not shutdown.is_alive() + assert client._shutdown_complete + assert client._metrics._runtime_metrics is None + assert client._metrics._flush_timer is None + + +def test_shutdown_during_initialization_prevents_collector_install( + runtime_client, native, monkeypatch +): + client, session = runtime_client(metrics=None) + client._metrics_config = {"capture_runtime_metrics": True} + constructor = native[0]._MacOSProcessMetrics + + def initialize(): + client.shutdown() + return constructor() + + monkeypatch.setattr(native[0], "_MacOSProcessMetrics", initialize) + client._start_runtime_metrics() + assert client._shutdown_complete + assert client._metrics._runtime_metrics is None + assert client._metrics._flush_timer is None + session.post.assert_not_called() + + +def test_zero_elapsed_time_skips_cpu_utilization(native): + collector = native[0]._MacOSProcessMetrics() + samples = {sample.name: sample.value for sample in collector.collect()} + assert "app.process.cpu.percent" not in samples + assert "process.cpu.utilization" not in samples + assert "process.cpu.time" in samples + assert samples["process.memory.usage"] == native[1].rss + assert samples["process.thread.count"] == 7 + assert samples["process.context_switches"] == 0 + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires fork") +def test_fork_replaces_locks_identity_and_cpu_baseline(runtime_client, native): + client, _ = runtime_client() + parent_id = client.metrics._runtime_metrics.resource_attributes[ + "service.instance.id" + ] + native[1].time += 100 + native[1].user += 50 + native[1].involuntary += 100000 + locks = [ + client.metrics._lock, + client.metrics._flush_lock, + client.metrics._runtime_lock, + ] + for lock in locks: + lock.acquire() + pid = os.fork() + if pid == 0: + signal.alarm(5) + try: + collector = client.metrics._runtime_metrics + assert collector.attributes["process.pid"] == os.getpid() + assert collector.resource_attributes["service.instance.id"] != parent_id + assert collector.resource_attributes["host.name"] == "test-host" + assert client.metrics._flush_timer.active + collected = collector.collect() + samples = {sample.name: sample.value for sample in collected} + assert "app.process.cpu.percent" not in samples + assert "process.cpu.utilization" not in samples + assert all( + sample.value == 0 + for sample in collected + if sample.metric_type == "observable_counter" + ) + assert samples["process.memory.usage"] == native[1].rss + native[1].time += 10 + native[1].user += 1 + client.metrics.flush() + client.shutdown() + except BaseException: + os._exit(1) + os._exit(0) + for lock in locks: + lock.release() + _, status = os.waitpid(pid, 0) + assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0 + assert ( + client.metrics._runtime_metrics.resource_attributes["service.instance.id"] + == parent_id + ) + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires fork") +def test_fork_during_timer_export_preserves_child_timer(runtime_client, monkeypatch): + client, _ = runtime_client() + metrics = client.metrics + parent_pid = os.getpid() + children = [] + + def send(payload): + pid = os.fork() + if pid == 0: + signal.alarm(5) + children.append(pid) + return "ok" + + monkeypatch.setattr(metrics, "_send", send) + try: + metrics._flush_timer.fire() + if os.getpid() != parent_pid: + assert not metrics._timer_flush_in_progress + assert metrics._flush_timer.active + except BaseException: + if os.getpid() != parent_pid: + os._exit(1) + raise + if os.getpid() != parent_pid: + os._exit(0) + _, status = os.waitpid(children[0], 0) + assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0 + monkeypatch.setattr(metrics, "_send", lambda payload: "ok") + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires fork") +def test_shutdown_from_hook_during_concurrent_shutdown_does_not_deadlock( + runtime_client, native +): + client, _ = runtime_client() + pid = os.fork() + if pid == 0: + signal.alarm(5) + try: + entered = threading.Event() + + def hook(sample): + entered.set() + while client._lifecycle_owner is None: + time.sleep(0.001) + client.shutdown() + return sample + + client.metrics._before_send = hook + native[1].time += 10 + sampler = threading.Thread(target=client.metrics._flush_timer.fire) + sampler.start() + assert entered.wait(2) + shutdown = threading.Thread(target=client.shutdown) + shutdown.start() + sampler.join(2) + shutdown.join(2) + assert not sampler.is_alive() and not shutdown.is_alive() + assert client._shutdown_complete + except BaseException: + os._exit(1) + os._exit(0) + _, status = os.waitpid(pid, 0) + assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0 + + +@pytest.mark.skipif(sys.platform != "darwin", reason="requires macOS libproc") +def test_native_macos_current_process_reading(): + module = importlib.import_module("posthog._runtime_metrics") + collector = module._MacOSProcessMetrics() + samples = { + sample.name: (sample.value, sample.unit) for sample in collector.collect() + } + rss, unit = samples["process.memory.usage"] + assert rss > 0 + assert unit == "By" + utilization, unit = samples["process.memory.utilization"] + assert unit == "1" + assert utilization == rss / ( + os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE") + ) + assert 0 < utilization <= 1 + assert samples["process.thread.count"][0] > 0 + assert samples["process.cpu.time"][0] >= 0 + assert samples["process.context_switches"][0] >= 0 + assert samples["cpython.gc.collections"][0] >= 0 + assert samples["cpython.gc.collected_objects"][0] >= 0 + assert samples["cpython.gc.uncollectable_objects"][0] >= 0 + assert ctypes.sizeof(module._ProcTaskInfo) == 96 + assert module._ProcTaskInfo.pti_resident_size.offset == 8 diff --git a/posthog/test/test_runtime_metrics_alignment.py b/posthog/test/test_runtime_metrics_alignment.py new file mode 100644 index 000000000..ea77232b3 --- /dev/null +++ b/posthog/test/test_runtime_metrics_alignment.py @@ -0,0 +1,211 @@ +from types import SimpleNamespace + +import pytest +from google.protobuf.json_format import ParseDict +from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( + ExportMetricsServiceRequest, +) + +from . import test_host_metrics as host_tests +from . import test_runtime_metrics as runtime_tests +from .test_metrics import sent_payload +from .test_runtime_metrics import attributes + +native = runtime_tests.native +runtime_client = runtime_tests.runtime_client +host_native = host_tests.host_native + + +def metric_groups(session): + payload = sent_payload(session) + ParseDict(payload, ExportMetricsServiceRequest()) + return [ + ( + attributes(group["resource"]), + { + metric["name"]: metric + for scope in group["scopeMetrics"] + for metric in scope["metrics"] + }, + ) + for group in payload["resourceMetrics"] + ] + + +def test_context_switches_use_native_split_not_combined_libproc( + runtime_client, native, monkeypatch +): + module, state, _ = native + counts = SimpleNamespace(voluntary=20, involuntary=300) + monkeypatch.setattr( + module, + "getrusage", + lambda _: SimpleNamespace( + ru_utime=state.user, + ru_stime=state.system, + ru_nvcsw=counts.voluntary, + ru_nivcsw=counts.involuntary, + ), + ) + client, session = runtime_client() + counts.voluntary += 4 + counts.involuntary += 9 + state.switches += 99999 + state.size = 8 + client.metrics.flush() + metrics = metric_groups(session)[0][1] + metric = metrics["process.context_switches"] + assert metric["unit"] == "{context_switch}" + assert metric["sum"]["isMonotonic"] is True + assert metric["sum"]["aggregationTemporality"] == 2 + assert { + attributes(point)["process.context_switch.type"]: point["asDouble"] + for point in metric["sum"]["dataPoints"] + } == {"voluntary": 4, "involuntary": 9} + assert "app.process.cpu.ctx_switch.total" not in metrics + assert "app.process.cpu.percent" not in metrics + + +@pytest.mark.parametrize("configured_host", [None, "configured-host"]) +def test_runtime_instance_is_resource_scoped_without_changing_manual_or_host( + runtime_client, host_native, configured_host +): + client, session = runtime_client( + metrics={ + "capture_runtime_metrics": True, + "capture_host_metrics": True, + "resource_attributes": { + "deployment.region": "test", + "service.instance.id": "configured", + **({"host.name": configured_host} if configured_host else {}), + }, + } + ) + collector = client.metrics._runtime_metrics + client.metrics.gauge( + "process.cpu.utilization", + 42, + unit="1", + attributes={ + **collector.attributes, + "cpu.mode": "user", + }, + ) + host_native[1].core_ticks[0][0] += 100 + client.metrics.flush() + groups = metric_groups(session) + assert len(groups) == 2 + manual_resource, manual = next( + group for group in groups if group[0]["service.instance.id"] == "configured" + ) + runtime_resource, runtime = next( + group for group in groups if "process.cpu.time" in group[1] + ) + assert ( + runtime_resource["service.instance.id"] + == collector.resource_attributes["service.instance.id"] + ) + assert runtime_resource["service.instance.id"] != "configured" + assert ( + runtime_resource["deployment.region"] + == manual_resource["deployment.region"] + == "test" + ) + assert collector.resource_attributes["host.name"] == "test-host" + assert runtime_resource["host.name"] == (configured_host or "test-host") + assert manual_resource.get("host.name") == configured_host + assert manual["process.cpu.utilization"]["gauge"]["dataPoints"][0]["asDouble"] == 42 + assert "system.cpu.utilization" in manual + assert { + attributes(point)["host.name"] + for point in manual["system.cpu.utilization"]["gauge"]["dataPoints"] + } == {"test-host"} + for metric in runtime.values(): + for point in metric.get("sum", metric.get("gauge", metric.get("histogram")))[ + "dataPoints" + ]: + assert ( + attributes(point)["process.pid"] == collector.attributes["process.pid"] + ) + assert "process.runtime_id" not in attributes(point) + assert "service.instance.id" not in attributes(point) + assert "host.name" not in attributes(point) + + +def test_runtime_host_resource_without_host_collection(runtime_client, native): + client, session = runtime_client() + client.metrics.gauge("manual.sample", 1) + client.metrics.flush() + assert client.metrics._host_metrics is None + runtime_resource, _ = next( + group for group in metric_groups(session) if "process.cpu.time" in group[1] + ) + manual_resource, _ = next( + group for group in metric_groups(session) if "manual.sample" in group[1] + ) + assert runtime_resource["host.name"] == native[0].gethostname() + assert "host.name" not in manual_resource + + +def test_failed_export_preserves_resource_identity_when_collector_is_replaced( + runtime_client, native, monkeypatch +): + monkeypatch.setattr(native[0], "gethostname", lambda: "first-host") + client, session = runtime_client() + first = client.metrics._runtime_metrics + native[1].user += 1 + session.post.return_value.status_code = 503 + client.metrics.flush() + monkeypatch.setattr(native[0], "gethostname", lambda: "second-host") + client.metrics._start_runtime_metrics() + second = client.metrics._runtime_metrics + native[1].user += 2 + session.post.return_value.status_code = 200 + client.metrics.flush() + groups = metric_groups(session) + assert {resource["service.instance.id"] for resource, _ in groups} == { + first.resource_attributes["service.instance.id"], + second.resource_attributes["service.instance.id"], + } + assert { + resource["service.instance.id"]: resource["host.name"] for resource, _ in groups + } == { + first.resource_attributes["service.instance.id"]: "first-host", + second.resource_attributes["service.instance.id"]: "second-host", + } + assert sorted( + next( + point["asDouble"] + for point in metrics["process.cpu.time"]["sum"]["dataPoints"] + if attributes(point)["cpu.mode"] == "user" + ) + for _, metrics in groups + ) == [1, 2] + + +def test_host_memory_states_and_total_without_inventing_unsupported_states(host_native): + module, state, _ = host_native + state.active = 200 + state.wired = 30 + state.speculative = 10 + collector = module._MacOSHostMetrics() + samples = collector.collect() + memory = [sample for sample in samples if sample.name == "system.memory.usage"] + assert { + sample.attributes["system.memory.state"]: sample.value for sample in memory + } == { + "free": (state.free - state.speculative) * state.page_size, + "used": (state.active + state.wired) * state.page_size, + } + assert all( + sample.unit == "By" and sample.metric_type == "observable_updowncounter" + for sample in memory + ) + limit = next(sample for sample in samples if sample.name == "system.memory.limit") + assert limit.value == state.total_memory + assert limit.unit == "By" and limit.metric_type == "observable_updowncounter" + names = {sample.name for sample in samples} + assert "app.system.memory.available.estimated" not in names + assert "app.system.cpu.utilization" not in names + assert "system.paging.operations" not in names + assert {"app.system.swap.io", "app.system.memory.pressure"} <= names diff --git a/posthog/test/test_runtime_metrics_semconv.py b/posthog/test/test_runtime_metrics_semconv.py new file mode 100644 index 000000000..dc63f77c4 --- /dev/null +++ b/posthog/test/test_runtime_metrics_semconv.py @@ -0,0 +1,376 @@ +from unittest import mock + +import pytest +from google.protobuf.json_format import ParseDict +from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( + ExportMetricsServiceRequest, +) + +from .. import metrics_capture +from . import test_runtime_metrics as runtime_tests +from .test_metrics import sent_payload +from .test_runtime_metrics import attributes, exported_metrics + +native = runtime_tests.native +runtime_client = runtime_tests.runtime_client + + +def points(metric): + return metric.get("sum", metric.get("gauge"))["dataPoints"] + + +def by_attribute(metric, key): + return {attributes(point)[key]: point for point in points(metric)} + + +def test_otel_types_units_cpu_modes_gc_generations_and_identity(runtime_client, native): + client, session = runtime_client() + state = native[1] + state.time += 2 + state.user += 2 + state.system += 1 + state.involuntary += 9 + state.voluntary += 4 + for generation, stats in enumerate(state.gc_stats): + for factor, field in enumerate( + ("collections", "collected", "uncollectable"), 1 + ): + stats[field] += (generation + 1) * factor + client.metrics.flush() + metrics = exported_metrics(session) + expected = { + "process.cpu.time": ("s", True), + "process.cpu.utilization": ("1", None), + "process.memory.usage": ("By", False), + "process.memory.utilization": ("1", None), + "process.thread.count": ("{thread}", False), + "cpython.gc.collections": ("{collection}", True), + "cpython.gc.collected_objects": ("{object}", True), + "cpython.gc.uncollectable_objects": ("{object}", True), + "process.context_switches": ("{context_switch}", True), + } + assert set(metrics) == set(expected) + identity = client.metrics._runtime_metrics.attributes + for name, (unit, monotonic) in expected.items(): + metric = metrics[name] + assert metric["unit"] == unit + if monotonic is None: + assert "gauge" in metric + else: + assert metric["sum"]["isMonotonic"] is monotonic + assert metric["sum"]["aggregationTemporality"] == 2 + for point in points(metric): + assert attributes(point).items() >= identity.items() + if monotonic is not None: + assert int(point["startTimeUnixNano"]) <= int(point["timeUnixNano"]) + assert sum(len(points(metric)) for metric in metrics.values()) == 18 + cpu_time = by_attribute(metrics["process.cpu.time"], "cpu.mode") + assert set(cpu_time) == {"user", "system"} + assert cpu_time["user"]["asDouble"] == 2 + assert cpu_time["system"]["asDouble"] == 1 + utilization = by_attribute(metrics["process.cpu.utilization"], "cpu.mode") + assert set(utilization) == {"user", "system"} + assert utilization["user"]["asDouble"] == 0.125 + assert utilization["system"]["asDouble"] == 0.0625 + assert points(metrics["process.memory.usage"])[0]["asDouble"] == state.rss + assert points(metrics["process.thread.count"])[0]["asDouble"] == state.threads + switches = by_attribute( + metrics["process.context_switches"], "process.context_switch.type" + ) + assert switches["voluntary"]["asDouble"] == 4 + assert switches["involuntary"]["asDouble"] == 9 + for name, factor in ( + ("cpython.gc.collections", 1), + ("cpython.gc.collected_objects", 2), + ("cpython.gc.uncollectable_objects", 3), + ): + generations = by_attribute(metrics[name], "cpython.gc.generation") + assert set(generations) == {0, 1, 2} + for generation in range(3): + assert generations[generation]["asDouble"] == (generation + 1) * factor + decoded = ParseDict(sent_payload(session), ExportMetricsServiceRequest()) + assert len(decoded.resource_metrics[0].scope_metrics[0].metrics) == 9 + + +@pytest.mark.parametrize( + "name", + [ + "process.cpu.time", + "cpython.gc.collections", + "cpython.gc.collected_objects", + "cpython.gc.uncollectable_objects", + "process.context_switches", + ], +) +def test_cumulative_counters_exclude_work_before_collection_starts( + runtime_client, name +): + client, session = runtime_client() + client.metrics.flush() + metric = exported_metrics(session)[name] + assert all(point["asDouble"] == 0 for point in points(metric)) + assert all( + int(point["startTimeUnixNano"]) + == client.metrics._runtime_metrics.start_time_ms * 1_000_000 + for point in points(metric) + ) + + +def test_memory_utilization_tracks_rss_and_physical_memory(runtime_client, native): + client, session = runtime_client() + state = native[1] + for rss, physical_pages in ((128, 1024), (256, 1024), (256, 2048), (0, 2048)): + state.rss = rss + state.physical_pages = physical_pages + state.time += 10 + client.metrics.flush() + metrics = exported_metrics(session) + utilization = metrics["process.memory.utilization"] + assert utilization["unit"] == "1" + assert points(utilization)[0]["asDouble"] == rss / ( + physical_pages * state.page_size + ) + assert "gauge" in utilization + assert points(metrics["process.memory.usage"])[0]["asDouble"] == rss + + +@pytest.mark.parametrize("key", ["SC_PHYS_PAGES", "SC_PAGE_SIZE"]) +@pytest.mark.parametrize( + "unavailable", [-1, 0, OSError("unavailable"), ValueError("unknown")] +) +def test_unavailable_physical_memory_omits_utilization_and_recovers( + runtime_client, native, monkeypatch, key, unavailable +): + module, state, _ = native + original_sysconf = module.os.sysconf + + def sysconf(name): + if name == key: + if isinstance(unavailable, Exception): + raise unavailable + return unavailable + return original_sysconf(name) + + monkeypatch.setattr(module.os, "sysconf", sysconf) + client, session = runtime_client() + state.time += 10 + client.metrics.flush() + metrics = exported_metrics(session) + assert "process.memory.utilization" not in metrics + assert "process.memory.usage" in metrics + assert "process.thread.count" in metrics + assert "process.cpu.time" in metrics + assert "process.context_switches" in metrics + monkeypatch.setattr(module.os, "sysconf", original_sysconf) + state.time += 10 + client.metrics.flush() + utilization = exported_metrics(session)["process.memory.utilization"] + assert points(utilization)[0]["asDouble"] == state.rss / ( + state.physical_pages * state.page_size + ) + + +def test_cpu_utilization_tracks_available_cpus(runtime_client, native): + client, session = runtime_client() + state = native[1] + for cpus in (8, 4): + state.cpus = cpus + state.time += 10 + state.user += 10 + client.metrics.flush() + metrics = exported_metrics(session) + utilization = by_attribute(metrics["process.cpu.utilization"], "cpu.mode") + assert utilization["user"]["asDouble"] == 1 / cpus + assert utilization["system"]["asDouble"] == 0 + assert "app.process.cpu.percent" not in metrics + + +@pytest.mark.parametrize("cpus", [None, 0]) +def test_unknown_cpu_availability_omits_normalized_utilization( + runtime_client, native, cpus +): + client, session = runtime_client() + native[1].cpus = cpus + native[1].time += 10 + native[1].user += 10 + client.metrics.flush() + metrics = exported_metrics(session) + assert "process.cpu.utilization" not in metrics + assert "process.cpu.time" in metrics + assert "app.process.cpu.percent" not in metrics + + +def test_cpu_availability_falls_back_on_older_python( + runtime_client, native, monkeypatch +): + monkeypatch.delattr(native[0].os, "process_cpu_count") + monkeypatch.setattr(native[0].os, "cpu_count", lambda: 4) + client, session = runtime_client() + native[1].time += 10 + native[1].user += 10 + client.metrics.flush() + utilization = by_attribute( + exported_metrics(session)["process.cpu.utilization"], "cpu.mode" + ) + assert utilization["user"]["asDouble"] == 0.25 + + +def test_cumulative_observations_keep_latest_value_and_stable_start( + runtime_client, native, monkeypatch +): + client, session = runtime_client() + state = native[1] + wall_start = metrics_capture.time.time() + monkeypatch.setattr(metrics_capture.time, "time", lambda: wall_start + state.time) + state.time += 10 + state.user += 2 + client.metrics.flush() + first = exported_metrics(session) + state.time += 10 + state.user += 0.5 + state.system += 0.25 + state.involuntary += 2 + state.rss = 100 + state.threads = 3 + state.gc_stats[0]["collections"] += 1 + client.metrics._collect_runtime_metrics() + state.time += 10 + client.metrics.flush() + second = exported_metrics(session) + cpu = by_attribute(second["process.cpu.time"], "cpu.mode") + assert cpu["user"]["asDouble"] == 2.5 + assert cpu["system"]["asDouble"] == 0.25 + assert points(second["process.memory.usage"])[0]["asDouble"] == 100 + assert points(second["process.thread.count"])[0]["asDouble"] == 3 + assert ( + by_attribute(second["process.context_switches"], "process.context_switch.type")[ + "involuntary" + ]["asDouble"] + == 2 + ) + assert ( + by_attribute(second["cpython.gc.collections"], "cpython.gc.generation")[0][ + "asDouble" + ] + == 1 + ) + for name, metric in first.items(): + if "sum" in metric: + assert ( + points(metric)[0]["startTimeUnixNano"] + == points(second[name])[0]["startTimeUnixNano"] + ) + + +def test_retry_merge_preserves_newer_cumulative_observations(runtime_client, native): + client, session = runtime_client() + state = native[1] + state.time += 10 + state.user += 2 + + def capture_during_failed_send(*args, **kwargs): + state.time += 10 + state.user += 0.5 + state.rss = 100 + state.threads = 3 + state.gc_stats[0]["collected"] += 20 + state.involuntary += 7 + client.metrics._collect_runtime_metrics() + return mock.Mock(status_code=500) + + session.post.side_effect = capture_during_failed_send + client.metrics.flush() + first = exported_metrics(session) + session.post.side_effect = None + session.post.return_value.status_code = 200 + state.time += 10 + client.metrics.flush() + metrics = exported_metrics(session) + assert ( + by_attribute(metrics["process.cpu.time"], "cpu.mode")["user"]["asDouble"] == 2.5 + ) + assert points(metrics["process.memory.usage"])[0]["asDouble"] == 100 + assert points(metrics["process.thread.count"])[0]["asDouble"] == 3 + assert ( + by_attribute(metrics["cpython.gc.collected_objects"], "cpython.gc.generation")[ + 0 + ]["asDouble"] + == 20 + ) + assert ( + by_attribute( + metrics["process.context_switches"], "process.context_switch.type" + )["involuntary"]["asDouble"] + == 7 + ) + assert ( + points(first["process.cpu.time"])[0]["startTimeUnixNano"] + == points(metrics["process.cpu.time"])[0]["startTimeUnixNano"] + ) + + +def test_retry_without_new_reading_preserves_observation_timestamp( + runtime_client, native, monkeypatch +): + client, session = runtime_client() + state = native[1] + wall_start = metrics_capture.time.time() + monkeypatch.setattr(metrics_capture.time, "time", lambda: wall_start + state.time) + state.time += 10 + session.post.return_value.status_code = 500 + client.metrics.flush() + first = exported_metrics(session) + monkeypatch.setattr( + native[0], "getrusage", mock.Mock(side_effect=OSError("unavailable")) + ) + state.time += 10 + session.post.return_value.status_code = 200 + client.metrics.flush() + second = exported_metrics(session) + for name, metric in first.items(): + if "sum" in metric: + assert points(metric) == points(second[name]) + + +def test_non_cpython_omits_cpython_metrics_only(runtime_client, native, monkeypatch): + client, session = runtime_client() + with monkeypatch.context() as patch: + patch.setattr(native[0].sys.implementation, "name", "pypy") + native[1].time += 10 + client.metrics.flush() + metrics = exported_metrics(session) + assert not any(name.startswith("cpython.") for name in metrics) + assert "process.cpu.time" in metrics + assert "process.memory.usage" in metrics + + +def test_native_failure_preserves_context_switch_counter_across_gap( + runtime_client, native +): + client, session = runtime_client() + state = native[1] + state.time += 10 + client.metrics.flush() + state.size = 8 + state.time += 10 + state.involuntary += 100 + client.metrics.flush() + metrics = exported_metrics(session) + assert "process.memory.usage" not in metrics + assert "process.memory.utilization" not in metrics + switches = by_attribute( + metrics["process.context_switches"], "process.context_switch.type" + ) + assert switches["involuntary"]["asDouble"] == 100 + assert "cpython.gc.collections" in metrics + state.size = 96 + state.time += 10 + state.involuntary += 2 + client.metrics.flush() + assert ( + by_attribute( + exported_metrics(session)["process.context_switches"], + "process.context_switch.type", + )["involuntary"]["asDouble"] + == 102 + )