From 697a3f6ec3bbdd3b9581da8f97ccf4d99c259e17 Mon Sep 17 00:00:00 2001 From: Daniel Visca Date: Tue, 14 Jul 2026 10:04:42 -0700 Subject: [PATCH 1/5] feat(metrics): add posthog.metrics count/gauge/histogram API (alpha) Port of the posthog-js core metrics client: statsd-style pre-aggregation (counts sum, gauges keep last, histograms bucket with the OTel default bounds), one OTLP/JSON data point per series per flush window, delta temporality, posted to /i/v1/metrics with the project token. Thread-safe; a daemon timer flushes every 10s and shutdown() drains the window. Includes the cardinality guardrail (max_series_per_flush, default 1000), the type-collision warning, transient-failure window merge-back, and the same resource-attribute layering (SDK keys over user keys) as posthog-js. Generated-By: PostHog Code Task-Id: 0d65d94c-c18a-4d84-93f6-18eb54876551 --- .sampo/changesets/metrics-python-alpha.md | 16 + posthog/client.py | 35 ++ posthog/metrics_capture.py | 502 ++++++++++++++++++++++ posthog/test/test_metrics.py | 239 ++++++++++ 4 files changed, 792 insertions(+) create mode 100644 .sampo/changesets/metrics-python-alpha.md create mode 100644 posthog/metrics_capture.py create mode 100644 posthog/test/test_metrics.py diff --git a/.sampo/changesets/metrics-python-alpha.md b/.sampo/changesets/metrics-python-alpha.md new file mode 100644 index 00000000..7f60f223 --- /dev/null +++ b/.sampo/changesets/metrics-python-alpha.md @@ -0,0 +1,16 @@ +--- +pypi/posthog: minor +--- + +Add the `posthog.metrics` API (`count`, `gauge`, `histogram`) — alpha. + +Backend services can now record metrics through the same statsd-style pre-aggregating client the browser and Node SDKs ship, with no OpenTelemetry setup: + +```python +client = Posthog("", metrics={"service_name": "billing-worker"}) +client.metrics.count("invoices.processed", 1, attributes={"plan": "pro"}) +client.metrics.gauge("queue.depth", 42) +client.metrics.histogram("job.duration", 187, unit="ms") +``` + +Samples aggregate in memory and flush as OTLP/JSON to `/i/v1/metrics` (one data point per series per window, delta temporality). Pending metrics are flushed on `shutdown()`. The `metrics` client option accepts `service_name`, `service_version`, `environment`, `resource_attributes`, `flush_interval`, `max_series_per_flush` (cardinality guardrail, default 1000), and a `before_send` hook. diff --git a/posthog/client.py b/posthog/client.py index 7bcfb591..d06f979f 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -16,6 +16,7 @@ from posthog._async_utils import _BackgroundEventLoopRunner from posthog.args import ID_TYPES, ExceptionArg, OptionalCaptureArgs, OptionalSetArgs +from posthog.metrics_capture import PostHogMetrics from posthog.capture_compression import ( CaptureCompression, _resolve_capture_compression, @@ -270,6 +271,7 @@ def __init__( capture_compression: Optional[Union[CaptureCompression, str]] = None, secret_key=None, _dedicated_ai_endpoint=False, + metrics: Optional[dict] = None, ): """ Initialize a new PostHog client instance. @@ -423,6 +425,9 @@ def __init__( self._flag_definition_cache_provider_async_runner_lock = threading.Lock() self.disabled = disabled or not self.api_key self.disable_geoip = disable_geoip + self._metrics_config = metrics + self._metrics: Optional[PostHogMetrics] = None + self._metrics_lock = threading.Lock() self.is_server = is_server self.historical_migration = historical_migration # Selects the capture wire protocol (V0 legacy `/batch/` vs V1 @@ -1715,6 +1720,30 @@ def _enqueue(self, msg, disable_geoip): self.log.warning("analytics-python queue is full") return None + @property + def metrics(self) -> PostHogMetrics: + """ + The `posthog.metrics` API: a statsd-style pre-aggregating metrics client — alpha. + + Samples fold into per-series aggregates in memory and flush as one OTLP + data point per series per window, so recording from hot paths is cheap. + Configure via the ``metrics`` client option; pending metrics flush on + ``shutdown()``. + + Examples: + ```python + client = Posthog("", metrics={"service_name": "billing-worker"}) + client.metrics.count("invoices.processed", 1, attributes={"plan": "pro"}) + client.metrics.gauge("queue.depth", 42) + client.metrics.histogram("job.duration", 187, unit="ms") + ``` + """ + if self._metrics is None: + with self._metrics_lock: + if self._metrics is None: + self._metrics = PostHogMetrics(self, self._metrics_config) + return self._metrics + 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. @@ -1789,6 +1818,12 @@ def shutdown(self) -> None: ``` """ self.flush(timeout_seconds=None) + if self._metrics is not None: + try: + self._metrics.flush() + except Exception: + self.log.exception("Failed to flush metrics on shutdown") + self._metrics.reset() self.join() self.distinct_ids_feature_flags_reported.clear() diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py new file mode 100644 index 00000000..e16b8c5e --- /dev/null +++ b/posthog/metrics_capture.py @@ -0,0 +1,502 @@ +"""Statsd-style pre-aggregating metrics client (`client.metrics`) — alpha. + +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. + +Deliberately unlike event capture, no per-user context (distinct ID, session) +is attached: every attribute value creates a new series, and per-user series +are the canonical metrics-cardinality explosion. +""" + +import json +import logging +import math +import threading +import time +from typing import Any, Callable, Optional, Union + +import requests + +from posthog.utils import remove_trailing_slash +from posthog.version import VERSION + +log = logging.getLogger("posthog") + +MetricAttributeValue = Union[str, int, float, bool] + +# OpenTelemetry SDK default bucket boundaries — usable resolution for common +# latency/size ranges without per-metric configuration. Must match posthog-js. +DEFAULT_HISTOGRAM_BOUNDS = [ + 0, + 5, + 10, + 25, + 50, + 75, + 100, + 250, + 500, + 750, + 1000, + 2500, + 5000, + 7500, + 10000, +] + +_OTLP_TEMPORALITY_DELTA = 1 +_DEFAULT_FLUSH_INTERVAL_SECONDS = 10.0 +_DEFAULT_MAX_SERIES_PER_FLUSH = 1000 +_SCOPE_NAME = "posthog-python" + + +def _to_otlp_any_value(value: Any) -> dict: + # bool before int: Python bool is an int subclass and must not encode as intValue. + if isinstance(value, bool): + return {"boolValue": value} + if isinstance(value, int): + return {"intValue": value} + if isinstance(value, float): + # proto3 JSON has no representation for non-finite floats; keep the + # human-readable signal as a string regardless of downstream parser. + if not math.isfinite(value): + return {"stringValue": str(value)} + return {"doubleValue": value} + if isinstance(value, str): + return {"stringValue": value} + if isinstance(value, (list, tuple)): + return {"arrayValue": {"values": [_to_otlp_any_value(v) for v in value]}} + try: + return {"stringValue": json.dumps(value)} + except (TypeError, ValueError): + return {"stringValue": str(value)} + + +def _to_otlp_key_value_list(attributes: dict) -> list: + return [ + {"key": key, "value": _to_otlp_any_value(value)} + for key, value in attributes.items() + if value is not None + ] + + +def _ms_to_unix_nano(ms: int) -> str: + # OTLP requires nanoseconds as a decimal string (uint64). + return f"{ms}000000" + + +def _bucket_index_for(value: float, bounds: list) -> int: + for i, bound in enumerate(bounds): + if value <= bound: + return i + return len(bounds) + + +class _SeriesState: + __slots__ = ( + "name", + "type", + "unit", + "attributes", + "window_start_ms", + "total", + "last", + "hist", + ) + + def __init__( + self, + name: str, + metric_type: str, + unit: Optional[str], + attributes: Optional[dict], + ): + self.name = name + self.type = metric_type + self.unit = unit + # Snapshot: the series key was computed from these values, so a caller + # mutating the dict after capture must not change the stored series. + self.attributes = dict(attributes) if attributes else None + self.window_start_ms = int(time.time() * 1000) + self.total: Optional[float] = None + self.last: Optional[float] = None + self.hist: Optional[dict] = None + + +class PostHogMetrics: + """The ``client.metrics`` API: ``count``, ``gauge``, ``histogram``, ``flush``. + + Thread-safe; safe to call from hot paths. Configure via the ``metrics`` + client option:: + + client = Client("phc_...", metrics={"service_name": "billing-worker"}) + client.metrics.count("invoices.processed", 1, attributes={"plan": "pro"}) + client.metrics.gauge("queue.depth", 42) + client.metrics.histogram("job.duration", 187, unit="ms") + """ + + def __init__(self, client, config: Optional[dict] = None): + self._client = client + config = config or {} + resource_attributes = config.get("resource_attributes") or {} + self._service_name: Optional[str] = resource_attributes.get( + "service.name" + ) or config.get("service_name") + self._service_version: Optional[str] = resource_attributes.get( + "service.version" + ) or config.get("service_version") + self._environment: Optional[str] = resource_attributes.get( + "deployment.environment" + ) or config.get("environment") + self._resource_attributes: dict = resource_attributes + self._flush_interval: float = config.get( + "flush_interval", _DEFAULT_FLUSH_INTERVAL_SECONDS + ) + self._max_series_per_flush: int = config.get( + "max_series_per_flush", _DEFAULT_MAX_SERIES_PER_FLUSH + ) + self._before_send: Optional[Callable] = config.get("before_send") + + self._lock = threading.RLock() + # Serializes flushes so a manual flush() can't race a timer flush for the same window. + self._flush_lock = threading.Lock() + self._series: dict = {} + self._flush_timer: Optional[threading.Timer] = None + self._series_cap_warned = False + self._type_by_name: dict = {} + self._type_collision_warned: set = set() + + def count( + self, + name: str, + value: float = 1, + unit: Optional[str] = None, + attributes: Optional[dict] = None, + ) -> None: + """Record an increment for a monotonic counter (things that only go up).""" + self._capture("count", name, value, unit, attributes) + + def gauge( + self, + name: str, + value: float, + unit: Optional[str] = None, + attributes: Optional[dict] = None, + ) -> None: + """Record the current value of something that goes up and down.""" + self._capture("gauge", name, value, unit, attributes) + + def histogram( + self, + name: str, + value: float, + unit: Optional[str] = None, + attributes: Optional[dict] = None, + ) -> None: + """Record one observation of a distribution (durations, sizes).""" + self._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() + + def reset(self) -> None: + """Clears the flush timer and drops the current window.""" + with self._lock: + self._clear_flush_timer() + self._series = {} + self._series_cap_warned = False + self._type_by_name = {} + self._type_collision_warned = set() + + def _capture( + self, + metric_type: str, + name: str, + value: float, + unit: Optional[str], + attributes: Optional[dict], + ) -> None: + if getattr(self._client, "disabled", False): + return + + sample = { + "name": name, + "type": metric_type, + "value": value, + "unit": unit, + "attributes": attributes, + } + if self._before_send is not None: + try: + filtered = self._before_send(sample) + except Exception as e: + log.error("Error in metrics before_send: %s", e) + return + if not filtered: + return + sample = filtered + name = sample.get("name") + metric_type = sample.get("type", metric_type) + value = sample.get("value") + unit = sample.get("unit") + attributes = sample.get("attributes") + + if not name or not isinstance(name, str): + log.warning("Dropping metric with empty name") + return + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not math.isfinite(value) + ): + log.warning("Dropping metric '%s': value must be a finite number", name) + return + if metric_type == "count" and value < 0: + log.warning( + "Dropping count '%s': counters are monotonic, value must be >= 0", name + ) + return + + attrs_key = tuple(sorted(attributes.items())) if attributes else () + key = (metric_type, name, unit or "", attrs_key) + + with self._lock: + seen_type = self._type_by_name.get(name) + if seen_type is None: + self._type_by_name[name] = metric_type + elif seen_type != metric_type and name not in self._type_collision_warned: + self._type_collision_warned.add(name) + log.warning( + "Metric name '%s' is already used as a %s; recording it as a %s too will blend " + "both series in charts. Use a distinct name.", + name, + seen_type, + metric_type, + ) + + state = self._series.get(key) + if state is None: + if len(self._series) >= self._max_series_per_flush: + if not self._series_cap_warned: + self._series_cap_warned = True + log.warning( + "Metric series cap reached (%s per flush window); dropping new series " + "until the next flush. Reduce attribute cardinality.", + self._max_series_per_flush, + ) + return + state = _SeriesState(name, metric_type, unit, attributes) + self._series[key] = state + + self._fold(state, float(value)) + self._arm_flush_timer() + + def _fold(self, state: _SeriesState, value: float) -> None: + if state.type == "count": + state.total = (state.total or 0.0) + value + elif state.type == "gauge": + state.last = value + else: + hist = state.hist + if hist is None: + hist = state.hist = { + "count": 0, + "sum": 0.0, + "min": value, + "max": value, + "bucket_counts": [0] * (len(DEFAULT_HISTOGRAM_BOUNDS) + 1), + } + hist["count"] += 1 + hist["sum"] += value + hist["min"] = min(hist["min"], value) + hist["max"] = max(hist["max"], value) + hist["bucket_counts"][ + _bucket_index_for(value, DEFAULT_HISTOGRAM_BOUNDS) + ] += 1 + + def _arm_flush_timer(self) -> None: + if self._flush_timer is not None: + return + timer = threading.Timer(self._flush_interval, self._timer_flush) + timer.daemon = True + self._flush_timer = timer + timer.start() + + def _timer_flush(self) -> None: + with self._lock: + self._flush_timer = None + try: + self.flush() + except Exception as e: + log.error("Metrics flush failed: %s", e) + + def _clear_flush_timer(self) -> None: + if self._flush_timer is not None: + self._flush_timer.cancel() + self._flush_timer = None + + def _do_flush(self) -> None: + # Snapshot and reset the window under the lock; send outside it so + # captures during the request fold into a fresh window. + with self._lock: + if not self._series: + return + window = self._series + self._series = {} + self._series_cap_warned = False + self._type_by_name = {} + self._type_collision_warned = set() + + payload = self._build_payload(window) + outcome = self._send(payload) + if outcome == "retry-later": + # Transient failure: merge the unsent window back so the data rides + # the next flush instead of being lost — and re-arm the timer, since + # with no new captures nothing else would schedule that flush. + with self._lock: + self._merge_window_back(window) + self._arm_flush_timer() + elif outcome == "too-large": + log.warning("Metrics batch exceeded the server size limit and was dropped") + + def _send(self, payload: dict) -> str: + url = "{}/i/v1/metrics?token={}".format( + remove_trailing_slash(self._client.host), self._client.api_key + ) + try: + response = requests.post( + url, json=payload, timeout=getattr(self._client, "timeout", 15) + ) + except requests.RequestException: + return "retry-later" + if response.status_code < 300: + return "ok" + if response.status_code == 413: + return "too-large" + if response.status_code >= 500 or response.status_code == 429: + return "retry-later" + log.error("Failed to send metrics batch: HTTP %s", response.status_code) + return "fatal" + + def _merge_window_back(self, window: dict) -> None: + for key, old in window.items(): + current = self._series.get(key) + if current is None: + self._series[key] = old + continue + current.window_start_ms = min(current.window_start_ms, old.window_start_ms) + if current.type == "count": + current.total = (current.total or 0.0) + (old.total or 0.0) + elif current.type == "histogram" and old.hist: + if current.hist is None: + current.hist = old.hist + else: + current.hist["count"] += old.hist["count"] + current.hist["sum"] += old.hist["sum"] + current.hist["min"] = min(current.hist["min"], old.hist["min"]) + 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. + + def _build_payload(self, window: dict) -> dict: + # User resource attributes first, SDK-controlled keys layered on top so + # a stray user key can't clobber attribution. + resource_attributes = dict(self._resource_attributes) + resource_attributes["service.name"] = self._service_name or "unknown_service" + if self._environment: + resource_attributes["deployment.environment"] = self._environment + if self._service_version: + resource_attributes["service.version"] = self._service_version + resource_attributes["telemetry.sdk.name"] = _SCOPE_NAME + resource_attributes["telemetry.sdk.version"] = VERSION + + return { + "resourceMetrics": [ + { + "resource": { + "attributes": _to_otlp_key_value_list(resource_attributes) + }, + "scopeMetrics": [ + { + "scope": {"name": _SCOPE_NAME, "version": VERSION}, + "metrics": self._build_metrics(window), + } + ], + } + ] + } + + def _build_metrics(self, window: dict) -> list: + # One OTLP metric entry per (type, name, unit), one data point per attribute set. + now_nano = _ms_to_unix_nano(int(time.time() * 1000)) + by_metric: dict = {} + + for state in window.values(): + metric_key = (state.type, state.name, state.unit or "") + metric = by_metric.get(metric_key) + if metric is None: + metric = {"name": state.name} + if state.unit: + metric["unit"] = state.unit + if state.type == "count": + metric["sum"] = { + "aggregationTemporality": _OTLP_TEMPORALITY_DELTA, + "isMonotonic": True, + "dataPoints": [], + } + elif state.type == "gauge": + metric["gauge"] = {"dataPoints": []} + else: + metric["histogram"] = { + "aggregationTemporality": _OTLP_TEMPORALITY_DELTA, + "dataPoints": [], + } + by_metric[metric_key] = metric + + attributes = _to_otlp_key_value_list(state.attributes or {}) + start_nano = _ms_to_unix_nano(state.window_start_ms) + + if state.type == "count": + metric["sum"]["dataPoints"].append( + { + "attributes": attributes, + "startTimeUnixNano": start_nano, + "timeUnixNano": now_nano, + "asDouble": state.total or 0.0, + } + ) + elif state.type == "gauge": + metric["gauge"]["dataPoints"].append( + { + "attributes": attributes, + "timeUnixNano": now_nano, + "asDouble": state.last or 0.0, + } + ) + elif state.hist is not None: + # Encoding pinned by the ingest's JSON deserializer: nano timestamps are decimal + # strings, but count/bucketCounts are plain JSON numbers — string-encoded u64s in + # those fields are silently dropped upstream (opentelemetry-rust#3328). + metric["histogram"]["dataPoints"].append( + { + "attributes": attributes, + "startTimeUnixNano": start_nano, + "timeUnixNano": now_nano, + "count": state.hist["count"], + "sum": state.hist["sum"], + "min": state.hist["min"], + "max": state.hist["max"], + "bucketCounts": state.hist["bucket_counts"], + "explicitBounds": DEFAULT_HISTOGRAM_BOUNDS, + } + ) + + return list(by_metric.values()) diff --git a/posthog/test/test_metrics.py b/posthog/test/test_metrics.py new file mode 100644 index 00000000..cf61f52d --- /dev/null +++ b/posthog/test/test_metrics.py @@ -0,0 +1,239 @@ +import math +from unittest import mock + +import pytest + +from posthog.client import Client +from posthog.metrics_capture import DEFAULT_HISTOGRAM_BOUNDS +from posthog.version import VERSION + +FAKE_API_KEY = "phc_test_key" + + +@pytest.fixture +def client(): + c = Client(FAKE_API_KEY, host="https://us.example.com", sync_mode=True) + yield c + c.metrics.reset() + + +def flush_and_capture(client): + """Flush metrics with the HTTP boundary mocked; returns (payload, url, kwargs) of the send.""" + with mock.patch("posthog.metrics_capture.requests.post") as post: + post.return_value = mock.Mock(status_code=200) + client.metrics.flush() + if not post.called: + return None, None, None + args, kwargs = post.call_args + return kwargs.get("json"), args[0] if args else kwargs.get("url"), kwargs + + +def metrics_from(payload): + return payload["resourceMetrics"][0]["scopeMetrics"][0]["metrics"] + + +class TestMetricsAggregation: + def test_count_burst_folds_to_one_delta_data_point(self, client): + # The whole point of pre-aggregation: 1000 count() calls must produce ONE data point + # whose value is the sum, marked delta + monotonic so the ingest diffs nothing. + for _ in range(1000): + client.metrics.count("jobs.processed") + client.metrics.count("jobs.processed", 5) + + payload, url, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + assert metric["name"] == "jobs.processed" + assert metric["sum"]["aggregationTemporality"] == 1 + assert metric["sum"]["isMonotonic"] is True + (dp,) = metric["sum"]["dataPoints"] + assert dp["asDouble"] == 1005.0 + assert url == "https://us.example.com/i/v1/metrics?token=phc_test_key" + + def test_gauge_keeps_last_value(self, client): + client.metrics.gauge("queue.depth", 10) + client.metrics.gauge("queue.depth", 3) + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + (dp,) = metric["gauge"]["dataPoints"] + assert dp["asDouble"] == 3.0 + # Gauges are instantaneous: no window start on the data point (matches posthog-js). + assert "startTimeUnixNano" not in dp + + def test_histogram_wire_shape(self, client): + # Pins the exact OTLP/JSON encoding the ingest's deserializer requires: nano timestamps + # as strings, but count/bucketCounts as plain JSON numbers (string-encoded u64s are + # silently dropped upstream — opentelemetry-rust#3328). + client.metrics.histogram("job.duration", 3, unit="ms") + client.metrics.histogram("job.duration", 40, unit="ms") + client.metrics.histogram("job.duration", 99999, unit="ms") + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + assert metric["unit"] == "ms" + assert metric["histogram"]["aggregationTemporality"] == 1 + (dp,) = metric["histogram"]["dataPoints"] + assert dp["count"] == 3 + assert dp["sum"] == 100042.0 + assert dp["min"] == 3.0 + assert dp["max"] == 99999.0 + assert dp["explicitBounds"] == DEFAULT_HISTOGRAM_BOUNDS + assert isinstance(dp["count"], int) + assert all(isinstance(c, int) for c in dp["bucketCounts"]) + assert len(dp["bucketCounts"]) == len(DEFAULT_HISTOGRAM_BOUNDS) + 1 + assert sum(dp["bucketCounts"]) == 3 + # 3 → first bucket with bound >= 3 (index 1: bound 5); 99999 → overflow bucket. + assert dp["bucketCounts"][1] == 1 + assert dp["bucketCounts"][-1] == 1 + assert isinstance(dp["timeUnixNano"], str) + assert isinstance(dp["startTimeUnixNano"], str) + + def test_attribute_sets_split_series_and_order_does_not(self, client): + client.metrics.count( + "http.requests", 1, attributes={"route": "/a", "status": "200"} + ) + client.metrics.count( + "http.requests", 1, attributes={"status": "200", "route": "/a"} + ) + client.metrics.count( + "http.requests", 1, attributes={"route": "/b", "status": "200"} + ) + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + points = metric["sum"]["dataPoints"] + assert len(points) == 2 + values = sorted(dp["asDouble"] for dp in points) + assert values == [1.0, 2.0] + + def test_attribute_value_otlp_encoding(self, client): + # bool must encode as boolValue, not intValue — Python bool is an int subclass, so a + # naive isinstance(int) check first silently miscodes True as 1. + client.metrics.count( + "encoded", 1, attributes={"s": "x", "b": True, "i": 7, "f": 1.5} + ) + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + (dp,) = metric["sum"]["dataPoints"] + by_key = {attr["key"]: attr["value"] for attr in dp["attributes"]} + assert by_key["s"] == {"stringValue": "x"} + assert by_key["b"] == {"boolValue": True} + assert by_key["i"] == {"intValue": 7} + assert by_key["f"] == {"doubleValue": 1.5} + + +class TestMetricsGuardrails: + def test_series_cap_drops_new_series_not_existing(self, client): + capped = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics={"max_series_per_flush": 2}, + ) + capped.metrics.count("m", 1, attributes={"k": "a"}) + capped.metrics.count("m", 1, attributes={"k": "b"}) + capped.metrics.count( + "m", 1, attributes={"k": "c"} + ) # new series past cap: dropped + capped.metrics.count( + "m", 1, attributes={"k": "a"} + ) # existing series: still folds + + payload, _, _ = flush_and_capture(capped) + capped.metrics.reset() + + (metric,) = metrics_from(payload) + points = metric["sum"]["dataPoints"] + assert len(points) == 2 + assert sorted(dp["asDouble"] for dp in points) == [1.0, 2.0] + + @pytest.mark.parametrize( + "record", + [ + lambda m: m.count("bad", -1), # counters are monotonic + lambda m: m.count("bad", math.nan), + lambda m: m.gauge("bad", math.inf), + lambda m: m.count("", 1), # empty name + ], + ) + def test_invalid_samples_dropped(self, client, record): + record(client.metrics) + + payload, _, _ = flush_and_capture(client) + + assert payload is None # nothing aggregated, nothing sent + + def test_disabled_client_records_nothing(self): + disabled = Client(FAKE_API_KEY, host="https://us.example.com", disabled=True) + disabled.metrics.count("m", 1) + + payload, _, _ = flush_and_capture(disabled) + + assert payload is None + + +class TestMetricsDelivery: + def test_resource_attributes_layer_sdk_keys_over_user_keys(self): + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics={ + "service_name": "billing-worker", + "environment": "production", + # A stray user key must not clobber SDK attribution. + "resource_attributes": { + "telemetry.sdk.name": "spoofed", + "team": "billing", + }, + }, + ) + c.metrics.count("m", 1) + + payload, _, _ = flush_and_capture(c) + c.metrics.reset() + + attrs = { + attr["key"]: attr["value"] + for attr in payload["resourceMetrics"][0]["resource"]["attributes"] + } + assert attrs["service.name"] == {"stringValue": "billing-worker"} + assert attrs["deployment.environment"] == {"stringValue": "production"} + assert attrs["team"] == {"stringValue": "billing"} + assert attrs["telemetry.sdk.name"] == {"stringValue": "posthog-python"} + assert attrs["telemetry.sdk.version"] == {"stringValue": VERSION} + scope = payload["resourceMetrics"][0]["scopeMetrics"][0]["scope"] + assert scope == {"name": "posthog-python", "version": VERSION} + + def test_transient_failure_merges_window_back(self, client): + # A 5xx must not lose the window: the counts ride the next flush, summed with new samples. + client.metrics.count("m", 3) + with mock.patch("posthog.metrics_capture.requests.post") as post: + post.return_value = mock.Mock(status_code=503) + client.metrics.flush() + + client.metrics.count("m", 4) + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + (dp,) = metric["sum"]["dataPoints"] + assert dp["asDouble"] == 7.0 + + def test_shutdown_flushes_pending_metrics(self): + c = Client(FAKE_API_KEY, host="https://us.example.com", sync_mode=True) + c.metrics.count("m", 1) + + with mock.patch("posthog.metrics_capture.requests.post") as post: + post.return_value = mock.Mock(status_code=200) + c.shutdown() + + assert post.called + payload = post.call_args.kwargs.get("json") + (metric,) = metrics_from(payload) + assert metric["name"] == "m" From 2f11c7ca00761983986223ef34f88496c26beb1d Mon Sep 17 00:00:00 2001 From: Daniel Visca Date: Tue, 14 Jul 2026 10:24:58 -0700 Subject: [PATCH 2/5] fix(metrics): harden the capture path and failure handling per QA review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Crash safety (a telemetry SDK must never raise into the host app): the series key is now JSON-based like the JS core's seriesKey, so list/dict attribute values and mixed-type keys aggregate instead of raising TypeError; before_send returns are validated (non-dict dropped, unknown metric types dropped instead of folding as histograms); and a catch-all guard drops-with-warning anything that still escapes. Fork safety: a forked child inherits the parent's window and a timer handle whose thread doesn't exist in the child — captures now detect the PID change, drop the inherited window (avoiding duplication), and re-arm. Failure handling: retry-later flushes log a warning and are bounded by a 3-consecutive-failure budget (then dropped loudly), merge-back respects the series cap so an outage with attribute churn can't grow memory without bound, and the send goes through the shared pooled session with a gzipped body and a URL-encoded token. Encoding parity with posthog-js: non-finite floats emit proto3 literals (Infinity/-Infinity/NaN), integral floats emit intValue, None-valued attributes are stripped before keying, and bool/int values stay distinct series via the JSON key. Generated-By: PostHog Code Task-Id: 0d65d94c-c18a-4d84-93f6-18eb54876551 --- .sampo/changesets/metrics-python-alpha.md | 4 +- posthog/metrics_capture.py | 195 +++++++++++++++++---- posthog/test/test_metrics.py | 196 ++++++++++++++++++++-- 3 files changed, 353 insertions(+), 42 deletions(-) diff --git a/.sampo/changesets/metrics-python-alpha.md b/.sampo/changesets/metrics-python-alpha.md index 7f60f223..3134b4f3 100644 --- a/.sampo/changesets/metrics-python-alpha.md +++ b/.sampo/changesets/metrics-python-alpha.md @@ -4,7 +4,7 @@ pypi/posthog: minor Add the `posthog.metrics` API (`count`, `gauge`, `histogram`) — alpha. -Backend services can now record metrics through the same statsd-style pre-aggregating client the browser and Node SDKs ship, with no OpenTelemetry setup: +Backend services can now record metrics through the same statsd-style pre-aggregating client the browser SDK ships, with no OpenTelemetry setup: ```python client = Posthog("", metrics={"service_name": "billing-worker"}) @@ -13,4 +13,4 @@ client.metrics.gauge("queue.depth", 42) client.metrics.histogram("job.duration", 187, unit="ms") ``` -Samples aggregate in memory and flush as OTLP/JSON to `/i/v1/metrics` (one data point per series per window, delta temporality). Pending metrics are flushed on `shutdown()`. The `metrics` client option accepts `service_name`, `service_version`, `environment`, `resource_attributes`, `flush_interval`, `max_series_per_flush` (cardinality guardrail, default 1000), and a `before_send` hook. +Samples aggregate in memory and flush as OTLP/JSON to `/i/v1/metrics` (one data point per series per window, delta temporality). Pending metrics are flushed on `shutdown()`; buffered windows are retried on transient failures and dropped loudly after 3 consecutive failed flushes. The `metrics` client option accepts `service_name`, `service_version`, `environment`, `resource_attributes`, `flush_interval` (seconds), `max_series_per_flush` (cardinality guardrail, default 1000), and a `before_send` hook. diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py index e16b8c5e..e2e9822a 100644 --- a/posthog/metrics_capture.py +++ b/posthog/metrics_capture.py @@ -11,17 +11,26 @@ Deliberately unlike event capture, no per-user context (distinct ID, session) is attached: every attribute value creates a new series, and per-user series are the canonical metrics-cardinality explosion. + +Delivery is at-least-once: a request that succeeds server-side but fails +client-side (e.g. a read timeout) is retried with the next window, which can +double-count that window's deltas. Windows are dropped loudly after +``_MAX_CONSECUTIVE_SEND_FAILURES`` failed flushes. """ +import gzip import json import logging import math +import os import threading import time from typing import Any, Callable, Optional, Union +from urllib.parse import quote import requests +from posthog.request import _get_session from posthog.utils import remove_trailing_slash from posthog.version import VERSION @@ -50,6 +59,10 @@ ] _OTLP_TEMPORALITY_DELTA = 1 +_VALID_METRIC_TYPES = ("count", "gauge", "histogram") +# Consecutive failed flushes before the buffered window is dropped (loudly) — bounds +# memory and payload growth against a permanently unreachable endpoint. +_MAX_CONSECUTIVE_SEND_FAILURES = 3 _DEFAULT_FLUSH_INTERVAL_SECONDS = 10.0 _DEFAULT_MAX_SERIES_PER_FLUSH = 1000 _SCOPE_NAME = "posthog-python" @@ -62,10 +75,15 @@ def _to_otlp_any_value(value: Any) -> dict: if isinstance(value, int): return {"intValue": value} if isinstance(value, float): - # proto3 JSON has no representation for non-finite floats; keep the - # human-readable signal as a string regardless of downstream parser. + # proto3 JSON has no representation for non-finite floats; encode the proto3 + # literal strings (not Python's "inf"/"nan") so both SDKs emit identical bytes. if not math.isfinite(value): - return {"stringValue": str(value)} + if math.isnan(value): + return {"stringValue": "NaN"} + return {"stringValue": "Infinity" if value > 0 else "-Infinity"} + # Integral floats encode as intValue, matching the JS Number.isInteger branch. + if value.is_integer(): + return {"intValue": int(value)} return {"doubleValue": value} if isinstance(value, str): return {"stringValue": value} @@ -97,6 +115,24 @@ def _bucket_index_for(value: float, bounds: list) -> int: return len(bounds) +def _series_key( + metric_type: str, name: str, unit: Optional[str], attributes: Optional[dict] +) -> str: + """Canonical, total series identity: JSON-encoded like the JS core's seriesKey, so any + attribute value the encoder accepts (lists, dicts, mixed keys) produces a hashable key, + and bool/int values stay distinct (json encodes true vs 1).""" + attrs_part = "" + if attributes: + items = sorted( + ((str(k), v) for k, v in attributes.items()), key=lambda kv: kv[0] + ) + attrs_part = ",".join( + f"{json.dumps(k)}:{json.dumps(v, sort_keys=True, default=str)}" + for k, v in items + ) + return "\x00".join((metric_type, name, unit or "", attrs_part)) + + class _SeriesState: __slots__ = ( "name", @@ -132,7 +168,8 @@ class PostHogMetrics: """The ``client.metrics`` API: ``count``, ``gauge``, ``histogram``, ``flush``. Thread-safe; safe to call from hot paths. Configure via the ``metrics`` - client option:: + client option (``flush_interval`` is in seconds, matching the client's own + ``flush_interval`` — unlike posthog-js, whose ``flushIntervalMs`` is milliseconds):: client = Client("phc_...", metrics={"service_name": "billing-worker"}) client.metrics.count("invoices.processed", 1, attributes={"plan": "pro"}) @@ -162,7 +199,10 @@ def __init__(self, client, config: Optional[dict] = None): ) self._before_send: Optional[Callable] = config.get("before_send") - self._lock = threading.RLock() + self._lock = threading.Lock() + self._pid = os.getpid() + self._consecutive_send_failures = 0 + self._capture_error_warned = 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 = {} @@ -179,7 +219,7 @@ def count( attributes: Optional[dict] = None, ) -> None: """Record an increment for a monotonic counter (things that only go up).""" - self._capture("count", name, value, unit, attributes) + self._guarded_capture("count", name, value, unit, attributes) def gauge( self, @@ -189,7 +229,7 @@ def gauge( attributes: Optional[dict] = None, ) -> None: """Record the current value of something that goes up and down.""" - self._capture("gauge", name, value, unit, attributes) + self._guarded_capture("gauge", name, value, unit, attributes) def histogram( self, @@ -199,7 +239,7 @@ def histogram( attributes: Optional[dict] = None, ) -> None: """Record one observation of a distribution (durations, sizes).""" - self._capture("histogram", name, value, unit, attributes) + self._guarded_capture("histogram", name, value, unit, attributes) def flush(self) -> None: """Sends everything aggregated so far without waiting for the flush interval.""" @@ -215,6 +255,22 @@ def reset(self) -> None: self._type_by_name = {} self._type_collision_warned = set() + def _guarded_capture( + self, + metric_type: str, + name: str, + value: float, + unit: Optional[str], + attributes: Optional[dict], + ) -> None: + # A telemetry call must never raise into the host application, whatever the input. + try: + self._capture(metric_type, name, value, unit, attributes) + except Exception as e: + if not self._capture_error_warned: + self._capture_error_warned = True + log.warning("Dropping metric '%s': %s", name, e) + def _capture( self, metric_type: str, @@ -241,6 +297,11 @@ def _capture( return if not filtered: return + if not isinstance(filtered, dict): + log.warning( + "Dropping metric: before_send must return the sample dict or a falsy value" + ) + return sample = filtered name = sample.get("name") metric_type = sample.get("type", metric_type) @@ -248,6 +309,12 @@ def _capture( unit = sample.get("unit") attributes = sample.get("attributes") + if metric_type not in _VALID_METRIC_TYPES: + log.warning( + "Dropping metric '%s': unknown metric type '%s'", name, metric_type + ) + return + if not name or not isinstance(name, str): log.warning("Dropping metric with empty name") return @@ -264,22 +331,14 @@ def _capture( ) return - attrs_key = tuple(sorted(attributes.items())) if attributes else () - key = (metric_type, name, unit or "", attrs_key) + if attributes: + # 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} + key = _series_key(metric_type, name, unit, attributes) with self._lock: - seen_type = self._type_by_name.get(name) - if seen_type is None: - self._type_by_name[name] = metric_type - elif seen_type != metric_type and name not in self._type_collision_warned: - self._type_collision_warned.add(name) - log.warning( - "Metric name '%s' is already used as a %s; recording it as a %s too will blend " - "both series in charts. Use a distinct name.", - name, - seen_type, - metric_type, - ) + self._reset_after_fork_locked() state = self._series.get(key) if state is None: @@ -295,9 +354,39 @@ def _capture( state = _SeriesState(name, metric_type, unit, attributes) self._series[key] = state + # Bookkeeping only for admitted samples, so name-cardinality misuse (IDs in + # metric names) can't grow this map past the series cap. + seen_type = self._type_by_name.get(name) + if seen_type is None: + self._type_by_name[name] = metric_type + elif seen_type != metric_type and name not in self._type_collision_warned: + self._type_collision_warned.add(name) + log.warning( + "Metric name '%s' is already used as a %s; recording it as a %s too will blend " + "both series in charts. Use a distinct name.", + name, + seen_type, + metric_type, + ) + self._fold(state, float(value)) self._arm_flush_timer() + def _reset_after_fork_locked(self) -> None: + # A forked child inherits the parent's window and a timer handle whose thread does + # not exist in the child — without this, the child never flushes (silent total + # loss) and would duplicate the parent's samples if it ever did. Drop both. + pid = os.getpid() + if pid == self._pid: + return + self._pid = pid + self._flush_timer = None + self._series = {} + self._series_cap_warned = False + self._type_by_name = {} + self._type_collision_warned = set() + self._consecutive_send_failures = 0 + def _fold(self, state: _SeriesState, value: float) -> None: if state.type == "count": state.total = (state.total or 0.0) + value @@ -357,22 +446,58 @@ def _do_flush(self) -> None: payload = self._build_payload(window) outcome = self._send(payload) if outcome == "retry-later": - # Transient failure: merge the unsent window back so the data rides - # the next flush instead of being lost — and re-arm the timer, since - # with no new captures nothing else would schedule that flush. with self._lock: + self._consecutive_send_failures += 1 + if self._consecutive_send_failures > _MAX_CONSECUTIVE_SEND_FAILURES: + # A persistently unreachable endpoint must not buffer forever: drop the + # window loudly instead of growing until a too-large drop loses more. + log.error( + "Dropping %s metric series after %s consecutive failed flushes — " + "check the endpoint and network configuration", + len(window), + self._consecutive_send_failures, + ) + self._consecutive_send_failures = 0 + return + # Transient failure: merge the unsent window back so the data rides the + # next flush instead of being lost — and re-arm the timer, since with no + # new captures nothing else would schedule that flush. + log.warning( + "Metrics flush failed (attempt %s of %s); will retry with the next window", + self._consecutive_send_failures, + _MAX_CONSECUTIVE_SEND_FAILURES + 1, + ) self._merge_window_back(window) self._arm_flush_timer() elif outcome == "too-large": - log.warning("Metrics batch exceeded the server size limit and was dropped") + log.warning( + "Metrics batch exceeded the server size limit and was dropped. " + "Reduce series count or attribute cardinality." + ) + with self._lock: + self._consecutive_send_failures = 0 + else: + with self._lock: + self._consecutive_send_failures = 0 def _send(self, payload: dict) -> str: url = "{}/i/v1/metrics?token={}".format( - remove_trailing_slash(self._client.host), self._client.api_key + remove_trailing_slash(self._client.host), + quote(self._client.api_key, safe=""), ) + body = gzip.compress(json.dumps(payload).encode("utf-8")) + timeout = getattr(self._client, "timeout", 15) or 15 try: - response = requests.post( - url, json=payload, timeout=getattr(self._client, "timeout", 15) + # The shared pooled session: keepalive between the 10s flushes, fork-safe + # reset, and the same adapter/proxy configuration as event capture. + response = _get_session().post( + url, + data=body, + headers={ + "Content-Type": "application/json", + "Content-Encoding": "gzip", + }, + timeout=timeout, ) except requests.RequestException: return "retry-later" @@ -386,9 +511,15 @@ def _send(self, payload: dict) -> str: return "fatal" def _merge_window_back(self, window: dict) -> None: + dropped = 0 for key, old in window.items(): current = self._series.get(key) if current is None: + # The cap applies through merge-back too, or a long outage with attribute + # churn grows the live window (and the retried payload) without bound. + if len(self._series) >= self._max_series_per_flush: + dropped += 1 + continue self._series[key] = old continue current.window_start_ms = min(current.window_start_ms, old.window_start_ms) @@ -405,6 +536,12 @@ def _merge_window_back(self, window: dict) -> None: 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. + if dropped: + log.warning( + "Dropped %s unsent metric series while merging a failed flush back (series cap %s)", + dropped, + self._max_series_per_flush, + ) def _build_payload(self, window: dict) -> dict: # User resource attributes first, SDK-controlled keys layered on top so diff --git a/posthog/test/test_metrics.py b/posthog/test/test_metrics.py index cf61f52d..fae5cc33 100644 --- a/posthog/test/test_metrics.py +++ b/posthog/test/test_metrics.py @@ -1,3 +1,5 @@ +import gzip +import json import math from unittest import mock @@ -17,15 +19,28 @@ def client(): c.metrics.reset() +def mock_session(status_code=200): + session = mock.Mock() + session.post.return_value = mock.Mock(status_code=status_code) + return session + + +def sent_payload(session): + """Decode the gzipped OTLP/JSON body of the last send through the mocked session.""" + args, kwargs = session.post.call_args + body = kwargs.get("data") + return json.loads(gzip.decompress(body).decode("utf-8")) + + def flush_and_capture(client): """Flush metrics with the HTTP boundary mocked; returns (payload, url, kwargs) of the send.""" - with mock.patch("posthog.metrics_capture.requests.post") as post: - post.return_value = mock.Mock(status_code=200) + session = mock_session() + with mock.patch("posthog.metrics_capture._get_session", return_value=session): client.metrics.flush() - if not post.called: + if not session.post.called: return None, None, None - args, kwargs = post.call_args - return kwargs.get("json"), args[0] if args else kwargs.get("url"), kwargs + args, kwargs = session.post.call_args + return sent_payload(session), args[0] if args else kwargs.get("url"), kwargs def metrics_from(payload): @@ -214,8 +229,9 @@ def test_resource_attributes_layer_sdk_keys_over_user_keys(self): def test_transient_failure_merges_window_back(self, client): # A 5xx must not lose the window: the counts ride the next flush, summed with new samples. client.metrics.count("m", 3) - with mock.patch("posthog.metrics_capture.requests.post") as post: - post.return_value = mock.Mock(status_code=503) + with mock.patch( + "posthog.metrics_capture._get_session", return_value=mock_session(503) + ): client.metrics.flush() client.metrics.count("m", 4) @@ -229,11 +245,169 @@ def test_shutdown_flushes_pending_metrics(self): c = Client(FAKE_API_KEY, host="https://us.example.com", sync_mode=True) c.metrics.count("m", 1) - with mock.patch("posthog.metrics_capture.requests.post") as post: - post.return_value = mock.Mock(status_code=200) + session = mock_session() + with mock.patch("posthog.metrics_capture._get_session", return_value=session): c.shutdown() - assert post.called - payload = post.call_args.kwargs.get("json") + assert session.post.called + payload = sent_payload(session) (metric,) = metrics_from(payload) assert metric["name"] == "m" + + +class TestMetricsCrashSafety: + # A telemetry SDK must never raise into the host application — these inputs all + # crashed the capture hot path before the series key became JSON-based and + # _capture gained validation (QA findings, reproduced). + + @pytest.mark.parametrize( + "attributes", + [ + {"tags": ["a", "b"]}, # unhashable value + {"nested": {"k": "v"}}, # unhashable value + {"a": 1, 2: "x"}, # unsortable mixed-type keys + ], + ) + def test_hostile_attributes_do_not_raise(self, client, attributes): + client.metrics.count("hostile", 1, attributes=attributes) # must not raise + + def test_list_attribute_records_as_array_value(self, client): + client.metrics.count("arr", 1, attributes={"tags": ["a", "b"]}) + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + (dp,) = metric["sum"]["dataPoints"] + (attr,) = dp["attributes"] + assert attr["value"] == { + "arrayValue": {"values": [{"stringValue": "a"}, {"stringValue": "b"}]} + } + + @pytest.mark.parametrize( + "hook", + [ + lambda s: True, # truthy non-dict return + lambda s: {**s, "type": "guage"}, # typo'd type must not fold as histogram + ], + ) + def test_misbehaving_before_send_drops_sample(self, hook): + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics={"before_send": hook}, + ) + c.metrics.count("m", 1) + + payload, _, _ = flush_and_capture(c) + c.metrics.reset() + + assert payload is None + + def test_bool_and_int_attribute_values_are_distinct_series(self, client): + # Python True == 1 collapsed these into one series with a tuple-based key. + client.metrics.count("m", 1, attributes={"k": True}) + client.metrics.count("m", 1, attributes={"k": 1}) + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + assert len(metric["sum"]["dataPoints"]) == 2 + + def test_none_attribute_values_dropped_before_keying(self, client): + # A None-valued attribute split the series key but was stripped from the wire, + # emitting two indistinguishable data points. + client.metrics.count("m", 1, attributes={"k": None}) + client.metrics.count("m", 1) + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + (dp,) = metric["sum"]["dataPoints"] + assert dp["asDouble"] == 2.0 + + +class TestMetricsEncodingParity: + # The two SDKs must emit byte-identical AnyValues for the same logical input. + + @pytest.mark.parametrize( + "value,expected", + [ + ( + math.inf, + {"stringValue": "Infinity"}, + ), # proto3 JSON literal, not Python's "inf" + (-math.inf, {"stringValue": "-Infinity"}), + (math.nan, {"stringValue": "NaN"}), + ( + 2.0, + {"intValue": 2}, + ), # integral floats encode as intValue, matching JS Number.isInteger + (2.5, {"doubleValue": 2.5}), + ], + ) + def test_any_value_encoding_matches_js(self, client, value, expected): + client.metrics.count("enc", 1, attributes={"v": value}) + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + (dp,) = metric["sum"]["dataPoints"] + (attr,) = dp["attributes"] + assert attr["value"] == expected + + +class TestMetricsFailureHandling: + def test_fork_drops_inherited_window_and_rearms(self, client): + # A forked child inherits a dead timer thread and the parent's window; without the + # PID guard it never flushes again and duplicates the parent's samples. + client.metrics.count("m", 1) + assert client.metrics._flush_timer is not None + client.metrics._pid -= 1 # simulate being in a fork child + + client.metrics.count("m", 5) + + payload, _, _ = flush_and_capture(client) + (metric,) = metrics_from(payload) + (dp,) = metric["sum"]["dataPoints"] + assert ( + dp["asDouble"] == 5.0 + ) # inherited window dropped, only the child's sample remains + + def test_merge_back_respects_series_cap(self): + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics={"max_series_per_flush": 2}, + ) + c.metrics.count("m", 1, attributes={"k": "a"}) + c.metrics.count("m", 1, attributes={"k": "b"}) + with mock.patch( + "posthog.metrics_capture._get_session", return_value=mock_session(503) + ): + c.metrics.flush() + # Attribute churn: two NEW series in the fresh window, then the failed window merges back. + c.metrics.count("m", 1, attributes={"k": "c"}) + c.metrics.count("m", 1, attributes={"k": "d"}) + + assert ( + len(c.metrics._series) <= 2 + ) # cap holds through merge-back; no unbounded backlog + c.metrics.reset() + + def test_window_dropped_after_consecutive_failures(self, client): + # A permanently-down endpoint must not buffer forever: after the retry budget the + # window is dropped loudly instead of growing until a 413 destroys everything. + client.metrics.count("m", 3) + with mock.patch( + "posthog.metrics_capture._get_session", return_value=mock_session(503) + ): + for _ in range(4): + client.metrics.flush() + + payload, _, _ = flush_and_capture(client) + + assert ( + payload is None + ) # budget exhausted → window dropped, nothing left to send From 11c557c5b7738fe9e2198a2681e19c2aba9acdaf Mon Sep 17 00:00:00 2001 From: Daniel Visca Date: Tue, 14 Jul 2026 10:46:40 -0700 Subject: [PATCH 3/5] fix(metrics): satisfy mypy and refresh the public API snapshot Typed extraction after the before_send isinstance narrow (dict[str, Any] + closed defaults the validation below drops), RequestException added to the vendored requests stub as the base the real hierarchy has, and the public API snapshot regenerated for the new Client.metrics surface. Generated-By: PostHog Code Task-Id: 0d65d94c-c18a-4d84-93f6-18eb54876551 --- posthog/metrics_capture.py | 16 +++++++++------- references/public_api_snapshot.txt | 16 +++++++++++++++- typings/requests/exceptions.pyi | 5 +++-- 3 files changed, 27 insertions(+), 10 deletions(-) diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py index e2e9822a..2371f32b 100644 --- a/posthog/metrics_capture.py +++ b/posthog/metrics_capture.py @@ -302,12 +302,14 @@ def _capture( "Dropping metric: before_send must return the sample dict or a falsy value" ) return - sample = filtered - name = sample.get("name") - metric_type = sample.get("type", metric_type) - value = sample.get("value") - unit = sample.get("unit") - attributes = sample.get("attributes") + sample_dict: dict[str, Any] = filtered + # Defaults keep the static types closed; a hook that removed the field + # produces a value the validation below drops. + name = sample_dict.get("name", "") + metric_type = sample_dict.get("type", metric_type) + value = sample_dict.get("value", math.nan) + unit = sample_dict.get("unit") + attributes = sample_dict.get("attributes") if metric_type not in _VALID_METRIC_TYPES: log.warning( @@ -499,7 +501,7 @@ def _send(self, payload: dict) -> str: }, timeout=timeout, ) - except requests.RequestException: + except requests.exceptions.RequestException: return "retry-later" if response.status_code < 300: return "ok" diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index 60072517..05256028 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -244,6 +244,7 @@ alias posthog.client.InconclusiveMatchError -> posthog.feature_flags.Inconclusiv alias posthog.client.OptionalCaptureArgs -> posthog.args.OptionalCaptureArgs alias posthog.client.OptionalSetArgs -> posthog.args.OptionalSetArgs alias posthog.client.Poller -> posthog.poller.Poller +alias posthog.client.PostHogMetrics -> posthog.metrics_capture.PostHogMetrics alias posthog.client.QuotaLimitError -> posthog.request.QuotaLimitError alias posthog.client.RedisFlagCache -> posthog.utils.RedisFlagCache alias posthog.client.RequestsConnectionError -> posthog.request.RequestsConnectionError @@ -332,6 +333,8 @@ alias posthog.mcp.__version__ -> posthog.mcp.version.__version__ alias posthog.mcp.derive_session_id_from_mcp_session -> posthog.mcp.session.derive_session_id_from_mcp_session alias posthog.mcp.get_more_tools_result -> posthog.mcp.tools.get_more_tools_result alias posthog.mcp.set_logger -> posthog.mcp.logger.set_logger +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.set_socket_options -> posthog.request.set_socket_options @@ -526,6 +529,7 @@ attribute posthog.client.Client.is_server = is_server attribute posthog.client.Client.log = logging.getLogger('posthog') attribute posthog.client.Client.log_captured_exceptions = log_captured_exceptions attribute posthog.client.Client.max_retries = max_retries +attribute posthog.client.Client.metrics: PostHogMetrics attribute posthog.client.Client.on_error = on_error attribute posthog.client.Client.personal_api_key = self.secret_key attribute posthog.client.Client.poll_interval = poll_interval @@ -701,6 +705,9 @@ attribute posthog.mcp.types.UserIdentity.distinct_id: str 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.1.0' +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') attribute posthog.on_error = None attribute posthog.personal_api_key = None attribute posthog.poll_interval = 30 @@ -855,7 +862,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, _dedicated_ai_endpoint=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, _dedicated_ai_endpoint=False, metrics: Optional[dict] = None) 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, dedicated_ai_endpoint=False, 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) @@ -878,6 +885,7 @@ class posthog.mcp.types.MCPAnalyticsContextOptions(description: Optional[str] = class posthog.mcp.types.MCPAnalyticsOptions(logger: Optional[LoggerFn] = None, report_missing: bool = False, missing_capability_tool_name: Optional[str] = None, enable_conversation_id: bool = False, enable_exception_autocapture: bool = True, context: Union[bool, MCPAnalyticsContextOptions] = True, identify: Optional[Union[IdentifyFn, UserIdentity]] = None, intent_fallback: Optional[IntentFallbackFn] = None, before_send: Optional[BeforeSendFn] = None, event_properties: Optional[EventPropertiesFn] = None) class posthog.mcp.types.PreparedToolCall(args: Optional[JsonRecord] = None, intent: Optional[str] = None, intent_source: Optional[str] = None, is_missing_capability: bool = False) class posthog.mcp.types.UserIdentity(distinct_id: str, properties: Optional[JsonRecord] = None, groups: Optional[Dict[str, str]] = None) +class posthog.metrics_capture.PostHogMetrics(client, config: Optional[dict] = None) class posthog.poller.Poller(interval, execute, *args, **kwargs) class posthog.request.APIError(status: Union[int, str], message: str, retry_after: Optional[float] = None) class posthog.request.DatetimeSerializer @@ -1264,6 +1272,11 @@ method posthog.mcp.posthog_mcp.PostHogMCP.flush(timeout_seconds: Optional[float] method posthog.mcp.posthog_mcp.PostHogMCP.prepare_tool_call(name: str, args: Optional[JsonRecord] = None) -> PreparedToolCall method posthog.mcp.posthog_mcp.PostHogMCP.prepare_tool_list(tools: List[Any], context: Union[bool, MCPAnalyticsContextOptions] = True, report_missing: bool = False) -> List[Any] method posthog.mcp.posthog_mcp.PostHogMCP.shutdown() -> None +method posthog.metrics_capture.PostHogMetrics.count(name: str, value: float = 1, unit: Optional[str] = None, attributes: Optional[dict] = None) -> None +method posthog.metrics_capture.PostHogMetrics.flush() -> None +method posthog.metrics_capture.PostHogMetrics.gauge(name: str, value: float, unit: Optional[str] = None, attributes: Optional[dict] = None) -> None +method posthog.metrics_capture.PostHogMetrics.histogram(name: str, value: float, unit: Optional[str] = None, attributes: Optional[dict] = None) -> None +method posthog.metrics_capture.PostHogMetrics.reset() -> None method posthog.poller.Poller.run() method posthog.poller.Poller.stop() method posthog.request.DatetimeSerializer.default(obj: Any) @@ -1347,6 +1360,7 @@ module posthog.mcp.session module posthog.mcp.tools module posthog.mcp.types module posthog.mcp.version +module posthog.metrics_capture module posthog.poller module posthog.request module posthog.types diff --git a/typings/requests/exceptions.pyi b/typings/requests/exceptions.pyi index 73ffa5cc..f7311f21 100644 --- a/typings/requests/exceptions.pyi +++ b/typings/requests/exceptions.pyi @@ -1,2 +1,3 @@ -class Timeout(Exception): ... -class ConnectionError(Exception): ... +class RequestException(Exception): ... +class Timeout(RequestException): ... +class ConnectionError(RequestException): ... From be5e7675ca44dbb097aa513d97a8142ee243c1e3 Mon Sep 17 00:00:00 2001 From: Daniel Visca Date: Tue, 14 Jul 2026 10:59:02 -0700 Subject: [PATCH 4/5] fix(metrics): stringify attribute keys at the OTLP wire boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Series identity already str()s keys, but the payload emitted the raw object — a numeric key violates OTLP's string KeyValue.key and strict decoders reject or drop the attribute. Generated-By: PostHog Code Task-Id: 0d65d94c-c18a-4d84-93f6-18eb54876551 --- posthog/metrics_capture.py | 4 +++- posthog/test/test_metrics.py | 13 +++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py index 2371f32b..1a79f43c 100644 --- a/posthog/metrics_capture.py +++ b/posthog/metrics_capture.py @@ -96,8 +96,10 @@ def _to_otlp_any_value(value: Any) -> dict: def _to_otlp_key_value_list(attributes: dict) -> list: + # str(key): OTLP KeyValue.key is a string field — strict decoders reject numeric + # keys — and the series identity already stringifies keys the same way. return [ - {"key": key, "value": _to_otlp_any_value(value)} + {"key": str(key), "value": _to_otlp_any_value(value)} for key, value in attributes.items() if value is not None ] diff --git a/posthog/test/test_metrics.py b/posthog/test/test_metrics.py index fae5cc33..72ad559b 100644 --- a/posthog/test/test_metrics.py +++ b/posthog/test/test_metrics.py @@ -271,6 +271,19 @@ class TestMetricsCrashSafety: def test_hostile_attributes_do_not_raise(self, client, attributes): client.metrics.count("hostile", 1, attributes=attributes) # must not raise + def test_non_string_attribute_keys_stringify_on_the_wire(self, client): + # Series identity str()s keys, but the wire must too — OTLP KeyValue.key is a + # string field and strict decoders reject numeric keys. + client.metrics.count("m", 1, attributes={"a": 1, 2: "x"}) + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + (dp,) = metric["sum"]["dataPoints"] + keys = {attr["key"] for attr in dp["attributes"]} + assert keys == {"a", "2"} + assert all(isinstance(attr["key"], str) for attr in dp["attributes"]) + def test_list_attribute_records_as_array_value(self, client): client.metrics.count("arr", 1, attributes={"tags": ["a", "b"]}) From 04d9d8b9d66d3fbcf2777eeb96fc0340bfb8e5ca Mon Sep 17 00:00:00 2001 From: Daniel Visca Date: Tue, 14 Jul 2026 14:47:16 -0700 Subject: [PATCH 5/5] fix(metrics): honor send=False, fork-safe locks, module shutdown flush Address review of the metrics capture surface: - _do_flush discards the window without transmitting when client.send is False, mirroring event capture (record locally, send nothing). - _reinit_after_fork replaces the metrics locks (_metrics_lock, _lock, _flush_lock) and drops the inherited window/timer in the forked child, so a child can't deadlock on a lock held by a vanished parent thread. - Module-level shutdown() delegates to Client.shutdown() so the final metrics window is flushed for apps using the global SDK lifecycle. Generated-By: PostHog Code Task-Id: 9606289a-ed0e-4b1b-828e-3b7f405d31e4 --- posthog/__init__.py | 3 +- posthog/client.py | 6 ++++ posthog/metrics_capture.py | 24 +++++++++++-- posthog/test/test_client_fork.py | 41 ++++++++++++++++++++++ posthog/test/test_metrics.py | 59 ++++++++++++++++++++++++++++++++ 5 files changed, 128 insertions(+), 5 deletions(-) diff --git a/posthog/__init__.py b/posthog/__init__.py index 9b664424..1842d467 100644 --- a/posthog/__init__.py +++ b/posthog/__init__.py @@ -1132,8 +1132,7 @@ def shutdown() -> None: Category: Client management """ - _proxy("flush") - _proxy("join") + _proxy("shutdown") def setup() -> Client: diff --git a/posthog/client.py b/posthog/client.py index d06f979f..2458af21 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -1604,6 +1604,12 @@ def _reinit_after_fork(self): self._flag_definition_cache_provider_async_runner = None self._flag_definition_cache_provider_async_runner_lock = threading.Lock() + # Metrics locks may have been held by a parent thread at fork time; replace + # them (never acquire them) so the child can't deadlock on a vanished holder. + self._metrics_lock = threading.Lock() + if self._metrics is not None: + self._metrics._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. if isinstance(self.flag_cache, RedisFlagCache): diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py index 1a79f43c..a1bcf016 100644 --- a/posthog/metrics_capture.py +++ b/posthog/metrics_capture.py @@ -376,14 +376,27 @@ def _capture( self._fold(state, float(value)) self._arm_flush_timer() + def _reinit_after_fork(self) -> None: + # Runs in a forked child (via the client's os.register_at_fork hook) before + # user code. The inherited locks may be held by parent threads that do not + # exist in the child, so replace them without ever acquiring them. + self._lock = threading.Lock() + self._flush_lock = threading.Lock() + self._pid = os.getpid() + self._drop_inherited_window() + def _reset_after_fork_locked(self) -> None: - # A forked child inherits the parent's window and a timer handle whose thread does - # not exist in the child — without this, the child never flushes (silent total - # loss) and would duplicate the parent's samples if it ever did. Drop both. + # PID-guard fallback for platforms without os.register_at_fork: a forked child + # inherits the parent's window and a timer handle whose thread does not exist + # in the child — without this, the child never flushes (silent total loss) and + # would duplicate the parent's samples if it ever did. Drop both. pid = os.getpid() if pid == self._pid: return self._pid = pid + self._drop_inherited_window() + + def _drop_inherited_window(self) -> None: self._flush_timer = None self._series = {} self._series_cap_warned = False @@ -447,6 +460,11 @@ def _do_flush(self) -> None: self._type_by_name = {} self._type_collision_warned = set() + # send=False mirrors event capture: recording succeeds locally, but + # nothing is transmitted — the flushed window is discarded. + if not getattr(self._client, "send", True): + return + payload = self._build_payload(window) outcome = self._send(payload) if outcome == "retry-later": diff --git a/posthog/test/test_client_fork.py b/posthog/test/test_client_fork.py index f5ed4279..eabae53b 100644 --- a/posthog/test/test_client_fork.py +++ b/posthog/test/test_client_fork.py @@ -1,5 +1,6 @@ import os import gc +import signal import unittest import warnings import weakref @@ -237,6 +238,46 @@ def child_probe(): ) self.assertEqual(result, "ok") + def test_register_at_fork_replaces_metrics_locks_in_child_process(self): + # Locks held at fork time are inherited locked, and their holders don't + # exist in the child — the metrics path must not deadlock on them. + client = Client(FAKE_TEST_API_KEY, send=False) + client.metrics.count("parent.metric") + + locks = [ + client._metrics_lock, + client.metrics._lock, + client.metrics._flush_lock, + ] + + def child_probe(): + # A deadlocked child gets killed by SIGALRM (a signaled exit fails + # the assertion below) instead of hanging the test forever. + signal.alarm(5) + try: + if not client._metrics_lock.acquire(blocking=False): + return "inherited _metrics_lock still held" + client._metrics_lock.release() + client.metrics.count("child.metric") + client.metrics.flush() + finally: + signal.alarm(0) + return "ok" + + for lock in locks: + lock.acquire() + try: + status, result = self._run_fork_probe(child_probe) + finally: + for lock in locks: + lock.release() + 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 72ad559b..b6783204 100644 --- a/posthog/test/test_metrics.py +++ b/posthog/test/test_metrics.py @@ -5,6 +5,7 @@ import pytest +import posthog from posthog.client import Client from posthog.metrics_capture import DEFAULT_HISTOGRAM_BOUNDS from posthog.version import VERSION @@ -241,6 +242,27 @@ def test_transient_failure_merges_window_back(self, client): (dp,) = metric["sum"]["dataPoints"] assert dp["asDouble"] == 7.0 + @pytest.mark.parametrize( + "flush_via", + [ + lambda m: m.flush(), # manual flush + lambda m: m._timer_flush(), # what the flush timer invokes + ], + ids=["manual", "timer"], + ) + def test_send_false_aggregates_but_never_posts(self, flush_via): + # send=False documents "queueing succeeds but events are not sent"; metrics + # must mirror that: fold locally, never hit the transport. + c = Client(FAKE_API_KEY, host="https://us.example.com", send=False, thread=0) + c.metrics.count("jobs.processed") + + session = mock_session() + with mock.patch("posthog.metrics_capture._get_session", return_value=session): + flush_via(c.metrics) + c.metrics.reset() + + assert not session.post.called + def test_shutdown_flushes_pending_metrics(self): c = Client(FAKE_API_KEY, host="https://us.example.com", sync_mode=True) c.metrics.count("m", 1) @@ -254,6 +276,43 @@ def test_shutdown_flushes_pending_metrics(self): (metric,) = metrics_from(payload) assert metric["name"] == "m" + def test_module_shutdown_flushes_default_client_metrics(self): + # Module-level shutdown() must delegate to Client.shutdown(), or the default + # client from posthog.setup() loses its final metrics window. + saved = ( + posthog.default_client, + posthog.api_key, + posthog.host, + posthog.sync_mode, + ) + posthog.default_client = None + posthog.api_key = FAKE_API_KEY + posthog.host = "https://us.example.com" + posthog.sync_mode = True + try: + posthog.setup().metrics.count("m", 1) + + session = mock_session() + with mock.patch( + "posthog.metrics_capture._get_session", return_value=session + ): + posthog.shutdown() + + assert session.post.called + (metric,) = metrics_from(sent_payload(session)) + assert metric["name"] == "m" + + # The window was flushed and cleared, not left pending. + payload, _, _ = flush_and_capture(posthog.default_client) + assert payload is None + finally: + ( + posthog.default_client, + posthog.api_key, + posthog.host, + posthog.sync_mode, + ) = saved + class TestMetricsCrashSafety: # A telemetry SDK must never raise into the host application — these inputs all