diff --git a/posthog/__init__.py b/posthog/__init__.py index 0431bc88..d4e26647 100644 --- a/posthog/__init__.py +++ b/posthog/__init__.py @@ -337,9 +337,12 @@ 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``, + ``autocapture_interval``, ...). Applied when ``setup()`` builds the + global client, or on a later ``setup()`` call if the metrics API hasn't + been used yet. + metrics_autocapture: When True, periodically sample process/host runtime + metrics and report them through the ``metrics`` client. Off by default. enable_exception_autocapture: Automatically capture uncaught exceptions. log_captured_exceptions: Also log exceptions captured by error tracking. project_root: Root path used to determine in-app exception stack frames. @@ -396,6 +399,7 @@ def get_tags() -> Dict[str, Any]: feature_flags_request_max_retries = 1 # type: int super_properties = None # type: Optional[Dict] metrics = None # type: Optional[Dict] +metrics_autocapture = False # type: bool enable_exception_autocapture = False # type: bool log_captured_exceptions = False # type: bool # Used to determine in app paths for exception autocapture. Defaults to the current working directory @@ -1259,6 +1263,7 @@ def setup() -> Client: feature_flags_request_max_retries=feature_flags_request_max_retries, super_properties=super_properties, metrics=metrics, + metrics_autocapture=metrics_autocapture, # TODO: Currently this monitoring begins only when the Client is initialised (which happens when you do something with the SDK) # This kind of initialisation is very annoying for exception capture. We need to figure out a way around this, # or deprecate this proxy option fully (it's already in the process of deprecation, no new clients should be using this method since like 5-6 months) diff --git a/posthog/client.py b/posthog/client.py index 437eca2d..d39155d6 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -19,6 +19,10 @@ from posthog._disabled_lane_queue import _DisabledLaneQueue from posthog.args import ID_TYPES, ExceptionArg, OptionalCaptureArgs, OptionalSetArgs from posthog.metrics_capture import PostHogMetrics +from posthog.runtime_metrics import ( + MetricsAutocapture, + _DEFAULT_SAMPLE_INTERVAL_SECONDS, +) from posthog.capture_compression import ( CaptureCompression, _resolve_capture_compression, @@ -708,6 +712,9 @@ def __init__( capture_trace_context=False, _use_ai_lane=False, _enable_multimodal_capture=False, + # Keyword-only and last so it never shifts an existing positional slot. + *, + metrics_autocapture: bool = False, ): """ Initialize a new PostHog client instance. @@ -773,6 +780,15 @@ def __init__( skips string truncation and passes media (base64/data URIs) through unredacted. ``privacy_mode`` always wins. Defaults to False. + metrics_autocapture: When True, periodically sample a small, + low-cardinality set of process/host runtime metrics (CPU time, + memory, threads, GC, load average; richer values when ``psutil`` + is installed) and report them through the ``metrics`` client to + PostHog's Metrics product. Off by default and no required + dependencies. Configure the cadence with + ``metrics={"autocapture_interval": 30}`` (seconds, default 60) + and identity via the ``metrics`` resource attributes (a per-process + ``service.instance.id`` of ``hostname-pid`` is set when absent). before_send: Optional callback that can modify or drop events before upload. Return ``None`` to drop an event. flag_fallback_cache_url: Optional feature flag fallback cache URL, @@ -901,6 +917,8 @@ def __init__( self._metrics_config = metrics self._metrics: Optional[PostHogMetrics] = None self._metrics_lock = threading.Lock() + self._metrics_autocapture_enabled = metrics_autocapture + self._metrics_autocapture: Optional[MetricsAutocapture] = None # `_use_ai_lane` / `_enable_multimodal_capture` are deprecated aliases. self.enable_full_ai_capture = ( enable_full_ai_capture is True @@ -1012,6 +1030,9 @@ def __init__( refill_interval_seconds=self.exception_autocapture_refill_interval_seconds, ) + if self._metrics_autocapture_enabled and not self.disabled: + self._start_metrics_autocapture() + if not sync_mode and send: # On program exit, allow the consumer threads to exit cleanly. # This prevents exceptions and a messy shutdown when the @@ -2217,6 +2238,10 @@ def _reinit_after_fork(self): self._metrics_lock = threading.Lock() if self._metrics is not None: self._metrics._reinit_after_fork() + # The sampler thread does not survive fork(); restart it (with a fresh + # per-process instance id) so each preforked worker reports on itself. + if self._metrics_autocapture is not None: + self._metrics_autocapture._reinit_after_fork() # If using Redis cache, we must reinitialize to get a fresh connection (fork-safe). # If using Memory cache, we keep it as-is to benefit from the inherited warm cache. @@ -2439,6 +2464,24 @@ def metrics(self) -> PostHogMetrics: self._metrics = PostHogMetrics(self, None) return self._metrics + def _start_metrics_autocapture(self) -> None: + # Enabling autocapture never raises into the host application: a bad + # config or a platform quirk degrades to no sampling (raise only in debug). + try: + config = ( + self._metrics_config if isinstance(self._metrics_config, dict) else {} + ) + interval = config.get( + "autocapture_interval", _DEFAULT_SAMPLE_INTERVAL_SECONDS + ) + autocapture = MetricsAutocapture(self.metrics, interval=interval) + autocapture.start() + self._metrics_autocapture = autocapture + except Exception as e: + if self.debug: + raise e + self.log.exception(f"Error starting metrics autocapture: {e}") + def flush(self, timeout_seconds: Optional[float] = 10) -> None: """ Force a flush from the internal queue to the server. Do not use directly, call `shutdown()` instead. @@ -2653,12 +2696,23 @@ def _shutdown_once(self, errors: list[Exception]) -> None: ) self._flush_or_discard_queues(errors) + if self._metrics_autocapture is not None: + # Stop before the metrics flush and take one last sample, so the + # final window carries fresh readings instead of a blind spot. + self._run_lifecycle_cleanup( + "Failed to stop metrics autocapture on shutdown", + self._metrics_autocapture.stop, + errors, + ) if self._metrics is not None: self._run_lifecycle_cleanup( "Failed to flush metrics on shutdown", self._metrics.flush, errors ) + # _close (not reset): mark the metrics client terminal so a late + # autocapture sample that outran the join above can't re-seed the + # window and schedule a network flush after shutdown. self._run_lifecycle_cleanup( - "Failed to reset metrics on shutdown", self._metrics.reset, errors + "Failed to close metrics on shutdown", self._metrics._close, errors ) self._join_once(errors, flush_queues=False, lanes_prepared=True) self._run_lifecycle_cleanup( diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py index 6e454842..d2b63e78 100644 --- a/posthog/metrics_capture.py +++ b/posthog/metrics_capture.py @@ -277,6 +277,11 @@ def __init__(self, client, config: Optional[dict] = None): self._pid = os.getpid() self._consecutive_send_failures = 0 self._capture_error_warned = False + # Terminal flag set by _close() at client shutdown. Once set, captures are + # dropped so a straggler (e.g. a runtime-metrics sample whose collector or + # before_send hook outran the autocapture join timeout) can't repopulate + # the cleared window or arm a post-shutdown flush timer. + self._closed = False # Serializes flushes so a manual flush() can't race a timer flush for the same window. self._flush_lock = threading.Lock() self._series: dict = {} @@ -329,6 +334,19 @@ def reset(self) -> None: self._type_by_name = {} self._type_collision_warned = set() + def _close(self) -> None: + """Terminal shutdown: cancel the timer, drop the window, and refuse + further captures. Unlike reset() the instance does not accept new samples + afterwards, so a sampler thread that outran the shutdown join can't + schedule a post-shutdown flush by folding a late sample back in.""" + with self._lock: + self._closed = True + self._clear_flush_timer() + self._series = {} + self._series_cap_warned = False + self._type_by_name = {} + self._type_collision_warned = set() + def _guarded_capture( self, metric_type: str, @@ -416,6 +434,12 @@ def _capture( with self._lock: self._reset_after_fork_locked() + if self._closed: + # Shutdown already flushed and cleared the window. A late sample + # must not re-seed the window or arm a new flush timer, or it + # would fire a network request after the client shut down. + return + state = self._series.get(key) if state is None: if len(self._series) >= self._max_series_per_flush: @@ -475,6 +499,9 @@ def _drop_inherited_window(self) -> None: self._type_by_name = {} self._type_collision_warned = set() self._consecutive_send_failures = 0 + # A forked child is a fresh process: even if the parent had shut its + # metrics client down, the child's captures must be live again. + self._closed = False def _fold(self, state: _SeriesState, value: float) -> None: if state.type == "count": diff --git a/posthog/runtime_metrics.py b/posthog/runtime_metrics.py new file mode 100644 index 00000000..c0c23189 --- /dev/null +++ b/posthog/runtime_metrics.py @@ -0,0 +1,307 @@ +"""Opt-in runtime metrics autocapture — samples host/process health on a timer +and feeds it into the existing ``client.metrics`` pipe (``/i/v1/metrics``). + +Off by default. When enabled (``metrics_autocapture=True``) a single daemon +thread wakes every ``interval`` seconds, reads a small, low-cardinality set of +process metrics from the standard library — and, when ``psutil`` is installed, +a few richer ones — and records them through ``client.metrics``. Nothing new is +on the wire path: the metrics client already pre-aggregates and exports OTLP/JSON +with correct temporality, so this module only decides *what* to sample and *how +often*, never *how to send*. + +Counters that the platform reports cumulatively (CPU seconds, GC collections) +are emitted as per-interval *deltas* through ``metrics.count`` — the metrics +client uses delta temporality, so each data point stands alone and a process +restart needs no cross-run state. Everything else is a ``gauge``. + +Fork safety mirrors the metrics client: the sampler thread does not survive +``fork()``, so preforking servers (gunicorn, celery) restart it per worker via +the client's ``os.register_at_fork`` hook, and a fresh ``service.instance.id`` +(``hostname-pid``) is stamped per process so workers don't collapse into one +series and clobber each other's gauges. +""" + +import gc +import logging +import os +import socket +import sys +import threading +from typing import TYPE_CHECKING, Any, Optional + +# Typed Any so the `resource is None` guard stays reachable under mypy on +# platforms (macOS/Linux) where the module always imports. +resource: Any +try: + import resource +except ImportError: # Windows has no `resource` module. + resource = None + +if TYPE_CHECKING: + from posthog.metrics_capture import PostHogMetrics + +log = logging.getLogger("posthog") + +_DEFAULT_SAMPLE_INTERVAL_SECONDS = 60.0 +# A floor so a misconfigured tiny interval can't turn a health probe into a +# busy loop that itself dominates the process's CPU/thread numbers. +_MIN_SAMPLE_INTERVAL_SECONDS = 1.0 +# Bounds how long shutdown waits for the sampler thread to notice the stop +# signal; it only ever sleeps on an Event, so this is reached in practice. +_THREAD_JOIN_TIMEOUT_SECONDS = 5.0 + + +def _import_psutil() -> Any: + try: + import psutil + + return psutil + except Exception: + return None + + +def _default_instance_id() -> str: + # `hostname-pid`: unique per process so preforked workers keep separate + # series. gethostname() can raise on exotic hosts — fall back to the pid. + try: + host = socket.gethostname() or "unknown-host" + except Exception: + host = "unknown-host" + return f"{host}-{os.getpid()}" + + +class MetricsAutocapture: + """Periodic runtime-metrics sampler feeding ``client.metrics``. + + Created by the client only when ``metrics_autocapture=True``. Public surface + is just ``start``/``stop`` plus the fork hook the client calls. + """ + + def __init__( + self, + metrics: "PostHogMetrics", + interval: float = _DEFAULT_SAMPLE_INTERVAL_SECONDS, + ): + self._metrics = metrics + if ( + not isinstance(interval, (int, float)) + or isinstance(interval, bool) + or not interval > 0 + ): + log.warning( + "Ignoring metrics autocapture interval %r: expected a positive number of seconds", + interval, + ) + interval = _DEFAULT_SAMPLE_INTERVAL_SECONDS + self._interval = max(float(interval), _MIN_SAMPLE_INTERVAL_SECONDS) + + self._lock = threading.Lock() + self._stop_event = threading.Event() + self._thread: Optional[threading.Thread] = None + # True between start() and stop(): the intent to be running. Consulted on + # fork so a child of an already-stopped client doesn't spawn a thread. + self._active = False + self._pid = os.getpid() + # Last cumulative reading per counter series, so we can emit deltas. + self._prev: dict = {} + self._psutil = _import_psutil() + self._psutil_process: Any = None + # Manage a default instance id only if the user didn't set one, so their + # override survives; remember it so the fork hook can refresh the pid. + self._manages_instance_id = ( + "service.instance.id" not in self._metrics._resource_attributes + ) + self._apply_instance_id() + + def _apply_instance_id(self) -> None: + if not self._manages_instance_id: + return + try: + self._metrics._resource_attributes["service.instance.id"] = ( + _default_instance_id() + ) + except Exception: + pass + + def start(self) -> None: + with self._lock: + if self._thread is not None and self._thread.is_alive(): + return + self._active = True + self._stop_event = threading.Event() + # Prime cumulative baselines so the first interval emits a real delta + # rather than the whole since-process-start total. + self._prev = {} + self._prime_baselines() + thread = threading.Thread( + target=self._run, + name="posthog-metrics-autocapture", + daemon=True, + ) + self._thread = thread + thread.start() + + def stop(self, final_sample: bool = True) -> None: + """Stop the sampler. By default takes one last sample so the final + window (flushed by the client on shutdown) isn't a blind spot.""" + with self._lock: + self._active = False + self._stop_event.set() + thread = self._thread + self._thread = None + if thread is not None and thread is not threading.current_thread(): + thread.join(timeout=_THREAD_JOIN_TIMEOUT_SECONDS) + if final_sample: + self._sample_guarded() + + def _run(self) -> None: + # wait() returns True only when the stop Event is set, so this exits + # promptly on stop() and otherwise ticks once per interval. + while not self._stop_event.wait(self._interval): + self._sample_guarded() + + def _reinit_after_fork(self) -> None: + # Runs in a forked child (via the client's fork hook) before user code. + # The inherited thread does not exist here and the lock may be held by a + # vanished parent thread — replace state without acquiring anything. + was_active = self._active + self._lock = threading.Lock() + self._stop_event = threading.Event() + self._thread = None + self._pid = os.getpid() + self._prev = {} + # psutil.Process caches the pid it was built for; drop it so the child + # rebuilds against its own process. + self._psutil_process = None + self._apply_instance_id() + if was_active: + self.start() + + def _sample_guarded(self) -> None: + # A telemetry timer must never raise into the host application, and one + # failing collector must not skip the rest — each collector guards itself. + for collector in ( + self._collect_rusage, + self._collect_threads, + self._collect_gc, + self._collect_loadavg, + self._collect_psutil, + ): + try: + collector() + except Exception as e: + log.debug("metrics autocapture collector failed: %s", e) + + def _prime_baselines(self) -> None: + # Record current cumulative values without emitting, so the first real + # sample's deltas are measured from start(), not from process boot. + for primer in (self._prime_rusage, self._prime_gc): + try: + primer() + except Exception: + pass + + def _emit_counter_delta( + self, key: str, name: str, current: float, unit=None, attributes=None + ) -> None: + prev = self._prev.get(key) + self._prev[key] = current + if prev is None: + return + delta = current - prev + # A negative delta means the cumulative source reset (should not happen + # for these) — skip it rather than feed the monotonic counter a drop. + if delta <= 0: + return + self._metrics.count(name, delta, unit=unit, attributes=attributes) + + def _maxrss_bytes(self, ru_maxrss: int) -> int: + # getrusage reports ru_maxrss in bytes on macOS but kilobytes on Linux + # and the BSDs; normalize to bytes. (Ternary, not an if/return, so mypy's + # host-platform narrowing doesn't flag one branch as unreachable.) + multiplier = 1 if sys.platform == "darwin" else 1024 + return int(ru_maxrss) * multiplier + + def _prime_rusage(self) -> None: + if resource is None: + return + usage = resource.getrusage(resource.RUSAGE_SELF) + self._prev["cpu.user"] = usage.ru_utime + self._prev["cpu.system"] = usage.ru_stime + + def _collect_rusage(self) -> None: + if resource is None: + return + usage = resource.getrusage(resource.RUSAGE_SELF) + self._emit_counter_delta( + "cpu.user", + "process.cpu.time", + usage.ru_utime, + unit="s", + attributes={"state": "user"}, + ) + self._emit_counter_delta( + "cpu.system", + "process.cpu.time", + usage.ru_stime, + unit="s", + attributes={"state": "system"}, + ) + self._metrics.gauge( + "process.memory.peak_rss", + self._maxrss_bytes(usage.ru_maxrss), + unit="By", + ) + + def _collect_threads(self) -> None: + self._metrics.gauge("process.thread.count", threading.active_count()) + + def _prime_gc(self) -> None: + for generation, stats in enumerate(gc.get_stats()): + self._prev[f"gc.collections.{generation}"] = stats.get("collections", 0) + + def _collect_gc(self) -> None: + for generation, stats in enumerate(gc.get_stats()): + self._emit_counter_delta( + f"gc.collections.{generation}", + "process.runtime.gc_collections", + stats.get("collections", 0), + attributes={"generation": generation}, + ) + + def _collect_loadavg(self) -> None: + getloadavg = getattr(os, "getloadavg", None) + if getloadavg is None: # Not available on Windows. + return + one, five, fifteen = getloadavg() + self._metrics.gauge("system.cpu.load_average.1m", one) + self._metrics.gauge("system.cpu.load_average.5m", five) + self._metrics.gauge("system.cpu.load_average.15m", fifteen) + + def _collect_psutil(self) -> None: + if self._psutil is None: + return + if self._psutil_process is None: + self._psutil_process = self._psutil.Process() + proc = self._psutil_process + + try: + # interval=None returns utilization since the previous call; the + # priming call at Process() creation makes the first real value meaningful. + self._metrics.gauge( + "process.cpu.utilization", proc.cpu_percent(None), unit="%" + ) + except Exception: + pass + try: + mem = proc.memory_info() + self._metrics.gauge("process.memory.usage", mem.rss, unit="By") + self._metrics.gauge("process.memory.virtual", mem.vms, unit="By") + except Exception: + pass + num_fds = getattr(proc, "num_fds", None) + if num_fds is not None: # POSIX only. + try: + self._metrics.gauge("process.open_file_descriptors", num_fds()) + except Exception: + pass diff --git a/posthog/test/test_client.py b/posthog/test/test_client.py index 087a79ca..79a7827f 100644 --- a/posthog/test/test_client.py +++ b/posthog/test/test_client.py @@ -2902,7 +2902,7 @@ def test_shutdown_after_join_runs_shutdown_only_cleanup(self): client.shutdown() metrics.flush.assert_called_once() - metrics.reset.assert_called_once() + metrics._close.assert_called_once() exception_capture.close.assert_called_once() self.assertTrue(client._shutdown_complete) @@ -3261,7 +3261,7 @@ def test_shutdown_failure_is_terminal_and_not_retried(self): def test_shutdown_failure_is_raised_after_later_cleanup_in_debug_mode(self): client = Client(FAKE_TEST_API_KEY, flush_interval=0.01, debug=True) metrics = mock.Mock() - metrics.reset.side_effect = Exception("reset failed") + metrics._close.side_effect = Exception("close failed") dedupe_cache = mock.Mock() dedupe_cache.clear.side_effect = Exception("clear failed") exception_capture = mock.Mock() @@ -3269,13 +3269,13 @@ def test_shutdown_failure_is_raised_after_later_cleanup_in_debug_mode(self): client.distinct_ids_feature_flags_reported = dedupe_cache client.exception_capture = exception_capture - with self.assertRaisesRegex(Exception, "reset failed"): + with self.assertRaisesRegex(Exception, "close failed"): client.shutdown() with self.assertRaisesRegex(RuntimeError, "client lifecycle cleanup failed"): client.shutdown() metrics.flush.assert_called_once_with() - metrics.reset.assert_called_once_with() + metrics._close.assert_called_once_with() dedupe_cache.clear.assert_called_once_with() exception_capture.close.assert_called_once_with() self.assertTrue(client._workers_joined) diff --git a/posthog/test/test_client_fork.py b/posthog/test/test_client_fork.py index 442322eb..6cc1d1dc 100644 --- a/posthog/test/test_client_fork.py +++ b/posthog/test/test_client_fork.py @@ -419,6 +419,54 @@ def child_probe(): ) self.assertEqual(result, "ok") + def test_register_at_fork_restarts_metrics_autocapture_in_child_process(self): + # The sampler thread does not survive fork(); each preforked worker must + # get its own live sampler and a fresh per-process service.instance.id. + client = Client( + FAKE_TEST_API_KEY, + host="https://fork-autocapture.example.com", + sync_mode=True, + send=False, + metrics_autocapture=True, + ) + try: + self.assertIsNotNone(client._metrics_autocapture) + parent_instance_id = client.metrics._resource_attributes[ + "service.instance.id" + ] + parent_thread = client._metrics_autocapture._thread + + def child_probe(): + signal.alarm(5) + try: + autocapture = client._metrics_autocapture + thread = autocapture._thread + child_instance_id = client.metrics._resource_attributes[ + "service.instance.id" + ] + if thread is None or not thread.is_alive(): + return "child sampler thread is not running" + if thread is parent_thread: + return "child inherited the parent's dead thread" + if child_instance_id == parent_instance_id: + return "child kept the parent's service.instance.id" + if not child_instance_id.endswith(f"-{os.getpid()}"): + return f"instance id not repinned to child pid: {child_instance_id}" + autocapture.stop(final_sample=False) + finally: + signal.alarm(0) + return "ok" + + status, result = self._run_fork_probe(child_probe) + finally: + client._metrics_autocapture.stop(final_sample=False) + client.metrics.reset() + + self.assertTrue( + os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, msg=result + ) + self.assertEqual(result, "ok") + def test_register_at_fork_reinitializes_poller_and_sessions_in_child_process(self): client = Client( FAKE_TEST_API_KEY, diff --git a/posthog/test/test_metrics.py b/posthog/test/test_metrics.py index 0575c2aa..dbc22c89 100644 --- a/posthog/test/test_metrics.py +++ b/posthog/test/test_metrics.py @@ -1,6 +1,7 @@ import gzip import json import math +import threading from unittest import mock import pytest @@ -603,6 +604,82 @@ def test_fork_drops_inherited_window_and_rearms(self, client): dp["asDouble"] == 5.0 ) # inherited window dropped, only the child's sample remains + def test_close_drops_late_captures_and_arms_no_timer(self, client): + # After _close(), a straggler sample (e.g. an autocapture collector that + # outran the shutdown join) must not re-seed the window or arm a flush + # timer that would fire a network request post-shutdown. + client.metrics._close() + + client.metrics.count("late.count", 1) + client.metrics.gauge("late.gauge", 5) + + assert client.metrics._series == {} + assert client.metrics._flush_timer is None + + def test_close_prevents_post_shutdown_flush_from_straggler(self): + # End-to-end: a sample that lands after the client shut its metrics down + # must not post to the network. + c = Client(FAKE_API_KEY, host="https://us.example.com", sync_mode=True) + c.metrics.count("m", 1) + + session = mock_session() + with mock.patch("posthog.metrics_capture._get_session", return_value=session): + c.shutdown() + post_calls_after_shutdown = session.post.call_count + + # Straggler sample after shutdown completed. + c.metrics.count("straggler", 1) + c.metrics.flush() + + assert session.post.call_count == post_calls_after_shutdown + assert c.metrics._flush_timer is None + + def test_blocked_sample_after_shutdown_does_not_emit(self): + # The shutdown race the reviewer flagged: a sample already in flight and + # blocked in a before_send hook (a stand-in for a runtime-metrics + # collector that outran the sampler's join timeout) must not emit or arm + # a flush timer once shutdown has flushed and closed the window. + entered = threading.Event() + release = threading.Event() + + def blocking_hook(sample): + entered.set() + release.wait(5) + return sample + + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics={"before_send": blocking_hook}, + ) + # before_send runs before the capture lock, so this thread blocks there + # while shutdown flushes and closes the metrics client. + sampler = threading.Thread(target=lambda: c.metrics.gauge("late", 1)) + sampler.start() + assert entered.wait(5) + + session = mock_session() + with mock.patch("posthog.metrics_capture._get_session", return_value=session): + c.shutdown() + posts_after_shutdown = session.post.call_count + release.set() + sampler.join(5) + + assert session.post.call_count == posts_after_shutdown + assert c.metrics._flush_timer is None + + def test_fork_reopens_closed_metrics(self, client): + # A forked child is a fresh process: even if the parent's metrics were + # closed at shutdown, the child must record again. + client.metrics._close() + client.metrics._pid -= 1 # simulate being in a fork child + + client.metrics.count("m", 2) + + assert client.metrics._series != {} + client.metrics.reset() + def test_merge_back_respects_series_cap(self): c = Client( FAKE_API_KEY, diff --git a/posthog/test/test_runtime_metrics.py b/posthog/test/test_runtime_metrics.py new file mode 100644 index 00000000..bed6286a --- /dev/null +++ b/posthog/test/test_runtime_metrics.py @@ -0,0 +1,292 @@ +import gzip +import json +from unittest import mock + +import pytest + +from posthog.client import Client +from posthog.runtime_metrics import ( + _DEFAULT_SAMPLE_INTERVAL_SECONDS, + _MIN_SAMPLE_INTERVAL_SECONDS, + MetricsAutocapture, + _default_instance_id, +) + +FAKE_API_KEY = "phc_test_key" + + +class RecordingMetrics: + """Stand-in for PostHogMetrics that records what the sampler pushes.""" + + def __init__(self, resource_attributes=None): + self.counts = [] + self.gauges = [] + self._resource_attributes = resource_attributes or {} + + def count(self, name, value=1, unit=None, attributes=None): + self.counts.append((name, value, unit, attributes)) + + def gauge(self, name, value, unit=None, attributes=None): + self.gauges.append((name, value, unit, attributes)) + + +def mock_session(status_code=200): + session = mock.Mock() + session.post.return_value = mock.Mock(status_code=status_code) + return session + + +class TestCounterDeltas: + def test_first_reading_is_baseline_only(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m) + ac._emit_counter_delta("k", "process.cpu.time", 5.0) + assert m.counts == [] + + def test_second_reading_emits_delta_not_cumulative(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m) + ac._emit_counter_delta("k", "process.cpu.time", 5.0) + ac._emit_counter_delta("k", "process.cpu.time", 7.5, unit="s") + assert m.counts == [("process.cpu.time", 2.5, "s", None)] + + def test_zero_delta_is_not_emitted(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m) + ac._prev["k"] = 5.0 + ac._emit_counter_delta("k", "process.cpu.time", 5.0) + assert m.counts == [] + + def test_counter_reset_is_skipped(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m) + ac._prev["k"] = 10.0 + ac._emit_counter_delta("k", "process.cpu.time", 3.0) + assert m.counts == [] + # The reset value still becomes the new baseline. + assert ac._prev["k"] == 3.0 + + +class TestInstanceId: + def test_default_instance_id_is_hostname_pid(self): + m = RecordingMetrics() + MetricsAutocapture(m) + instance_id = m._resource_attributes["service.instance.id"] + assert instance_id + assert instance_id.endswith(f"-{__import__('os').getpid()}") + + def test_user_supplied_instance_id_is_preserved(self): + m = RecordingMetrics({"service.instance.id": "mine"}) + MetricsAutocapture(m) + assert m._resource_attributes["service.instance.id"] == "mine" + + def test_default_instance_id_helper_survives_hostname_failure(self): + with mock.patch( + "posthog.runtime_metrics.socket.gethostname", + side_effect=OSError("boom"), + ): + assert _default_instance_id().startswith("unknown-host-") + + +class TestInterval: + def test_invalid_interval_falls_back_to_default(self): + m = RecordingMetrics() + assert ( + MetricsAutocapture(m, interval=0)._interval + == _DEFAULT_SAMPLE_INTERVAL_SECONDS + ) + assert ( + MetricsAutocapture(m, interval=-5)._interval + == _DEFAULT_SAMPLE_INTERVAL_SECONDS + ) + assert ( + MetricsAutocapture(m, interval=True)._interval + == _DEFAULT_SAMPLE_INTERVAL_SECONDS + ) + + def test_tiny_interval_is_clamped_to_floor(self): + m = RecordingMetrics() + assert ( + MetricsAutocapture(m, interval=0.001)._interval + == _MIN_SAMPLE_INTERVAL_SECONDS + ) + + +class TestCollectors: + def test_sample_emits_core_stdlib_metrics(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m) + ac._prime_baselines() + ac._sample_guarded() + gauge_names = {g[0] for g in m.gauges} + assert "process.thread.count" in gauge_names + assert "process.memory.peak_rss" in gauge_names + + def test_maxrss_normalized_to_bytes_per_platform(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m) + with mock.patch("posthog.runtime_metrics.sys.platform", "darwin"): + assert ac._maxrss_bytes(1000) == 1000 + with mock.patch("posthog.runtime_metrics.sys.platform", "linux"): + assert ac._maxrss_bytes(1000) == 1024000 + + def test_one_failing_collector_does_not_skip_the_rest(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m) + with mock.patch.object(ac, "_collect_rusage", side_effect=RuntimeError("boom")): + ac._sample_guarded() + assert any(g[0] == "process.thread.count" for g in m.gauges) + + def test_psutil_metrics_collected_when_available(self): + fake_proc = mock.Mock() + fake_proc.cpu_percent.return_value = 12.0 + fake_proc.memory_info.return_value = mock.Mock(rss=1000, vms=2000) + fake_proc.num_fds.return_value = 7 + fake_psutil = mock.Mock() + fake_psutil.Process.return_value = fake_proc + + m = RecordingMetrics() + ac = MetricsAutocapture(m) + ac._psutil = fake_psutil + ac._collect_psutil() + + by_name = {g[0]: g[1] for g in m.gauges} + assert by_name["process.cpu.utilization"] == 12.0 + assert by_name["process.memory.usage"] == 1000 + assert by_name["process.memory.virtual"] == 2000 + assert by_name["process.open_file_descriptors"] == 7 + + def test_psutil_absent_is_a_no_op(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m) + ac._psutil = None + ac._collect_psutil() + assert m.gauges == [] + + +class TestLifecycle: + def test_start_spawns_thread_stop_removes_it(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m, interval=60) + ac.start() + try: + assert ac._thread is not None and ac._thread.is_alive() + finally: + ac.stop(final_sample=False) + assert ac._thread is None + + def test_stop_takes_a_final_sample(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m, interval=60) + ac.start() + m.gauges.clear() + m.counts.clear() + ac.stop() + assert len(m.gauges) > 0 + + def test_reinit_after_fork_restarts_when_active(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m, interval=60) + ac.start() + try: + ac._prev["stale"] = 1 + ac._reinit_after_fork() + assert "stale" not in ac._prev + assert ac._thread is not None and ac._thread.is_alive() + finally: + ac.stop(final_sample=False) + + def test_reinit_after_fork_does_not_restart_when_stopped(self): + m = RecordingMetrics() + ac = MetricsAutocapture(m, interval=60) + ac._reinit_after_fork() + assert ac._thread is None + + +class TestClientWiring: + def test_off_by_default(self): + c = Client(FAKE_API_KEY, host="https://us.example.com", sync_mode=True) + assert c._metrics_autocapture is None + + def test_disabled_client_does_not_start_autocapture(self): + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + disabled=True, + metrics_autocapture=True, + ) + assert c._metrics_autocapture is None + + def test_enabled_creates_a_running_sampler(self): + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics_autocapture=True, + ) + try: + assert c._metrics_autocapture is not None + assert c._metrics_autocapture._thread.is_alive() + finally: + c._metrics_autocapture.stop(final_sample=False) + c.metrics.reset() + + def test_interval_read_from_metrics_config(self): + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics_autocapture=True, + metrics={"autocapture_interval": 30}, + ) + try: + assert c._metrics_autocapture._interval == 30 + finally: + c._metrics_autocapture.stop(final_sample=False) + c.metrics.reset() + + def test_samples_flow_through_to_the_metrics_wire(self): + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics_autocapture=True, + ) + try: + c._metrics_autocapture._sample_guarded() + session = mock_session() + with mock.patch( + "posthog.metrics_capture._get_session", return_value=session + ): + c.metrics.flush() + assert session.post.called + _, kwargs = session.post.call_args + payload = json.loads(gzip.decompress(kwargs["data"]).decode("utf-8")) + resource = payload["resourceMetrics"][0]["resource"]["attributes"] + keys = {kv["key"] for kv in resource} + assert "service.instance.id" in keys + metric_names = { + mtr["name"] + for mtr in payload["resourceMetrics"][0]["scopeMetrics"][0]["metrics"] + } + assert "process.thread.count" in metric_names + finally: + c._metrics_autocapture.stop(final_sample=False) + c.metrics.reset() + + +@pytest.mark.skipif( + not hasattr(__import__("os"), "getloadavg"), + reason="os.getloadavg is not available on this platform", +) +def test_loadavg_emits_three_windows(): + m = RecordingMetrics() + ac = MetricsAutocapture(m) + ac._collect_loadavg() + names = {g[0] for g in m.gauges} + assert { + "system.cpu.load_average.1m", + "system.cpu.load_average.5m", + "system.cpu.load_average.15m", + } <= names diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index 01e2ff66..77f742ab 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -256,6 +256,7 @@ alias posthog.client.FlagsResponse -> posthog.types.FlagsResponse alias posthog.client.ID_TYPES -> posthog.args.ID_TYPES alias posthog.client.InconclusiveMatchError -> posthog.feature_flags.InconclusiveMatchError alias posthog.client.MAX_MSG_SIZE -> posthog.consumer.MAX_MSG_SIZE +alias posthog.client.MetricsAutocapture -> posthog.runtime_metrics.MetricsAutocapture alias posthog.client.OptionalCaptureArgs -> posthog.args.OptionalCaptureArgs alias posthog.client.OptionalSetArgs -> posthog.args.OptionalSetArgs alias posthog.client.Poller -> posthog.poller.Poller @@ -364,6 +365,7 @@ alias posthog.metrics_capture.VERSION -> posthog.version.VERSION alias posthog.metrics_capture.remove_trailing_slash -> posthog.utils.remove_trailing_slash alias posthog.request.VERSION -> posthog.version.VERSION alias posthog.request.remove_trailing_slash -> posthog.utils.remove_trailing_slash +alias posthog.runtime_metrics.PostHogMetrics -> posthog.metrics_capture.PostHogMetrics alias posthog.set_socket_options -> posthog.request.set_socket_options attribute posthog.__version__ = VERSION attribute posthog.ai.anthropic.anthropic.Anthropic.messages = WrappedMessages(self) @@ -797,6 +799,7 @@ attribute posthog.mcp.types.UserIdentity.groups: Optional[Dict[str, str]] = None attribute posthog.mcp.types.UserIdentity.properties: Optional[JsonRecord] = None attribute posthog.mcp.version.__version__ = '0.3.0' attribute posthog.metrics = None +attribute posthog.metrics_autocapture = False attribute posthog.metrics_capture.DEFAULT_HISTOGRAM_BOUNDS = [0, 5, 10, 25, 50, 75, 100, 250, 500, 750, 1000, 2500, 5000, 7500, 10000] attribute posthog.metrics_capture.MetricAttributeValue = Union[str, int, float, bool] attribute posthog.metrics_capture.log = logging.getLogger('posthog') @@ -832,6 +835,7 @@ attribute posthog.request.RequestsTimeout = requests.exceptions.Timeout attribute posthog.request.SocketOptions = List[Tuple[int, int, Union[int, bytes]]] attribute posthog.request.USER_AGENT = 'posthog-python/' + VERSION attribute posthog.request.US_INGESTION_ENDPOINT = 'https://us.i.posthog.com' +attribute posthog.runtime_metrics.log = logging.getLogger('posthog') attribute posthog.secret_key = None attribute posthog.send = True attribute posthog.super_properties = None @@ -958,7 +962,7 @@ class posthog.bucketed_rate_limiter.BucketedRateLimiter(bucket_size: Number, ref class posthog.capture_compression.CaptureCompression class posthog.capture_mode.CaptureMode class posthog.capture_v1.CaptureV1Error(status: int | str, message: str, *, retry_after: Optional[float] = None, request_id: Optional[str] = None, attempts: Optional[int] = None, retry_exhausted: Optional[list[str]] = None, drops: Optional[list[tuple[str, Optional[str]]]] = None) -class posthog.client.Client(project_api_key: str, host=None, debug=False, max_queue_size=10000, send=True, on_error=None, flush_at=100, flush_interval=5.0, gzip=False, max_retries=3, sync_mode=False, timeout=15, thread=1, poll_interval=30, personal_api_key=None, disabled=False, disable_geoip=True, is_server=True, historical_migration=False, feature_flags_request_timeout_seconds=3, feature_flags_request_max_retries=1, super_properties=None, enable_exception_autocapture=False, log_captured_exceptions=False, project_root=None, privacy_mode=False, before_send=None, flag_fallback_cache_url=None, enable_local_evaluation=True, flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None, capture_exception_code_variables=False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: list[str] | None = None, enable_exception_autocapture_rate_limiting=False, exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, capture_mode: Optional[Union[CaptureMode, str]] = None, capture_compression: Optional[Union[CaptureCompression, str]] = None, secret_key=None, metrics: Optional[dict] = None, enable_full_ai_capture=False, capture_trace_context=False, _use_ai_lane=False, _enable_multimodal_capture=False) +class posthog.client.Client(project_api_key: str, host=None, debug=False, max_queue_size=10000, send=True, on_error=None, flush_at=100, flush_interval=5.0, gzip=False, max_retries=3, sync_mode=False, timeout=15, thread=1, poll_interval=30, personal_api_key=None, disabled=False, disable_geoip=True, is_server=True, historical_migration=False, feature_flags_request_timeout_seconds=3, feature_flags_request_max_retries=1, super_properties=None, enable_exception_autocapture=False, log_captured_exceptions=False, project_root=None, privacy_mode=False, before_send=None, flag_fallback_cache_url=None, enable_local_evaluation=True, flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None, capture_exception_code_variables=False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: list[str] | None = None, enable_exception_autocapture_rate_limiting=False, exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, capture_mode: Optional[Union[CaptureMode, str]] = None, capture_compression: Optional[Union[CaptureCompression, str]] = None, secret_key=None, metrics: Optional[dict] = None, enable_full_ai_capture=False, capture_trace_context=False, _use_ai_lane=False, _enable_multimodal_capture=False, *, metrics_autocapture: bool = False) class posthog.consumer.Consumer(queue, api_key, flush_at=100, host=None, on_error=None, flush_interval=5.0, gzip=False, retries=10, timeout=15, historical_migration=False, endpoint=EVENTS_ENDPOINT, max_msg_size=MAX_MSG_SIZE, capture_mode=CaptureMode.V0, capture_compression=CaptureCompression.NONE) class posthog.contexts.ContextScope(parent=None, fresh: bool = False, capture_exceptions: bool = True, client: Optional[Client] = None) class posthog.exception_capture.ExceptionCapture(client: Client, rate_limiting_enabled=False, bucket_size=DEFAULT_BUCKET_SIZE, refill_rate=DEFAULT_REFILL_RATE, refill_interval_seconds=DEFAULT_REFILL_INTERVAL_SECONDS) @@ -990,6 +994,7 @@ class posthog.request.DatetimeSerializer class posthog.request.GetResponse(data: Any, etag: Optional[str] = None, not_modified: bool = False) class posthog.request.HTTPAdapterWithSocketOptions(*args, socket_options: Optional[SocketOptions] = None, **kwargs) class posthog.request.QuotaLimitError +class posthog.runtime_metrics.MetricsAutocapture(metrics: PostHogMetrics, interval: float = _DEFAULT_SAMPLE_INTERVAL_SECONDS) class posthog.types.FeatureFlag(key: str, enabled: bool, variant: Optional[str], reason: Optional[FlagReason], metadata: Union[FlagMetadata, LegacyFlagMetadata]) class posthog.types.FeatureFlagError class posthog.types.FeatureFlagResult(key: str, enabled: bool, variant: Optional[str], payload: Optional[Any], reason: Optional[str]) @@ -1402,6 +1407,8 @@ method posthog.poller.Poller.run() method posthog.poller.Poller.stop() method posthog.request.DatetimeSerializer.default(obj: Any) method posthog.request.HTTPAdapterWithSocketOptions.init_poolmanager(*args, **kwargs) +method posthog.runtime_metrics.MetricsAutocapture.start() -> None +method posthog.runtime_metrics.MetricsAutocapture.stop(final_sample: bool = True) -> None method posthog.types.FeatureFlag.from_json(resp: Any) -> FeatureFlag method posthog.types.FeatureFlag.from_value_and_payload(key: str, value: FlagValue, payload: Any) -> FeatureFlag method posthog.types.FeatureFlag.get_value() -> FlagValue @@ -1490,6 +1497,7 @@ module posthog.mcp.version module posthog.metrics_capture module posthog.poller module posthog.request +module posthog.runtime_metrics module posthog.types module posthog.utils module posthog.version