diff --git a/.sampo/changesets/metrics-review-followups.md b/.sampo/changesets/metrics-review-followups.md new file mode 100644 index 00000000..59446aac --- /dev/null +++ b/.sampo/changesets/metrics-review-followups.md @@ -0,0 +1,9 @@ +--- +pypi/posthog: patch +--- + +Harden the alpha `posthog.metrics` client based on review follow-ups. + +- Metric attributes are now deep-snapshotted at capture time, so mutating a nested list/dict value after `count()`/`gauge()`/`histogram()` can no longer rewrite an already-recorded series' attributes on the wire. +- Failed metric flushes now retry with exponential backoff (first retry at the base interval, then doubling per consecutive failure, capped at 64x the flush interval — the shared JS logs ramp) instead of the fixed cadence, and the buffered window is dropped loudly after 8 consecutive failed flushes — previously documented as 3 but effectively 4. +- Invalid `metrics` client config (non-dict config or `resource_attributes`, non-numeric `flush_interval`, non-integer `max_series_per_flush`, non-callable `before_send`) now degrades to defaults with a warning instead of raising from the first `client.metrics.count()` call, matching the client's no-throw contract. diff --git a/posthog/client.py b/posthog/client.py index 275ace1a..161660ab 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -1826,7 +1826,18 @@ def metrics(self) -> PostHogMetrics: if self._metrics is None: with self._metrics_lock: if self._metrics is None: - self._metrics = PostHogMetrics(self, self._metrics_config) + # Same no-throw semantics as the rest of the public client surface: + # a bad metrics config degrades to defaults instead of raising from + # the first chained metrics.count() call (raise only in debug mode). + try: + self._metrics = PostHogMetrics(self, self._metrics_config) + except Exception as e: + if self.debug: + raise e + self.log.exception( + f"Error initializing metrics, using default configuration: {e}" + ) + self._metrics = PostHogMetrics(self, None) return self._metrics def flush(self, timeout_seconds: Optional[float] = 10) -> None: diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py index a1bcf016..6e454842 100644 --- a/posthog/metrics_capture.py +++ b/posthog/metrics_capture.py @@ -14,10 +14,13 @@ 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. +double-count that window's deltas. Failed flushes retry with exponential +backoff capped at ``_MAX_RETRY_BACKOFF_MULTIPLIER`` times the flush interval +(the policy the shared JS logs implementation uses); the window is dropped +loudly once ``_MAX_CONSECUTIVE_SEND_FAILURES`` consecutive flushes have failed. """ +import copy import gzip import json import logging @@ -61,13 +64,28 @@ _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 +# memory and payload growth against a permanently unreachable endpoint. The series +# cap already bounds the buffered window, and backoff spaces the attempts out, so +# the budget covers a real outage (~21 min at the default 10s interval). +_MAX_CONSECUTIVE_SEND_FAILURES = 8 +# Retry delays grow 2x per consecutive failure, capped at this multiple of the +# flush interval — the same ceiling the shared JS logs implementation uses. +_MAX_RETRY_BACKOFF_MULTIPLIER = 64 _DEFAULT_FLUSH_INTERVAL_SECONDS = 10.0 _DEFAULT_MAX_SERIES_PER_FLUSH = 1000 _SCOPE_NAME = "posthog-python" +def _snapshot_attribute_value(value: Any) -> Any: + # Per-value fallback: one un-deepcopyable exotic value must not degrade the + # whole snapshot to shallow, leaving the other (mutable) values shared with + # the caller after the series key was computed. + try: + return copy.deepcopy(value) + except Exception: + return value + + 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): @@ -157,9 +175,22 @@ def __init__( 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 + # Deep snapshot: the series key was computed from these values, so a caller + # mutating the dict — or a nested list/dict value — after capture must not + # change the stored series. + if attributes: + try: + self.attributes: Optional[dict] = { + key: _snapshot_attribute_value(value) + for key, value in attributes.items() + } + except Exception: + # A hostile mapping whose iteration itself raises: a shallow + # snapshot still isolates the top-level dict, and the encoder + # stringifies whatever remains. + self.attributes = dict(attributes) + else: + self.attributes = None self.window_start_ms = int(time.time() * 1000) self.total: Optional[float] = None self.last: Optional[float] = None @@ -181,8 +212,27 @@ class PostHogMetrics: def __init__(self, client, config: Optional[dict] = None): self._client = client - config = config or {} - resource_attributes = config.get("resource_attributes") or {} + # client.metrics sits outside the client's no-throw guards, so invalid nested + # config must degrade to defaults (with a warning) instead of raising into + # the host application from the first metrics.count() call. The Any-typed + # local keeps the runtime defense visible to mypy despite the annotation. + raw_config: Any = config + if not isinstance(raw_config, dict): + if raw_config is not None: + log.warning( + "Ignoring metrics config: expected a dict, got %s", + type(raw_config).__name__, + ) + raw_config = {} + config = raw_config + resource_attributes = config.get("resource_attributes") + if not isinstance(resource_attributes, dict): + if resource_attributes is not None: + log.warning( + "Ignoring metrics resource_attributes: expected a dict, got %s", + type(resource_attributes).__name__, + ) + resource_attributes = {} self._service_name: Optional[str] = resource_attributes.get( "service.name" ) or config.get("service_name") @@ -193,13 +243,35 @@ def __init__(self, client, config: Optional[dict] = None): "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") + flush_interval = config.get("flush_interval", _DEFAULT_FLUSH_INTERVAL_SECONDS) + if ( + not isinstance(flush_interval, (int, float)) + or isinstance(flush_interval, bool) + or not flush_interval > 0 + ): + log.warning( + "Ignoring metrics flush_interval %r: expected a positive number of seconds", + flush_interval, + ) + flush_interval = _DEFAULT_FLUSH_INTERVAL_SECONDS + self._flush_interval: float = float(flush_interval) + max_series = config.get("max_series_per_flush", _DEFAULT_MAX_SERIES_PER_FLUSH) + if ( + not isinstance(max_series, int) + or isinstance(max_series, bool) + or max_series <= 0 + ): + log.warning( + "Ignoring metrics max_series_per_flush %r: expected a positive integer", + max_series, + ) + max_series = _DEFAULT_MAX_SERIES_PER_FLUSH + self._max_series_per_flush: int = max_series + before_send = config.get("before_send") + if before_send is not None and not callable(before_send): + log.warning("Ignoring metrics before_send: expected a callable") + before_send = None + self._before_send: Optional[Callable] = before_send self._lock = threading.Lock() self._pid = os.getpid() @@ -427,16 +499,31 @@ def _fold(self, state: _SeriesState, value: float) -> None: _bucket_index_for(value, DEFAULT_HISTOGRAM_BOUNDS) ] += 1 - def _arm_flush_timer(self) -> None: + def _arm_flush_timer( + self, delay: Optional[float] = None, replace: bool = False + ) -> None: if self._flush_timer is not None: - return - timer = threading.Timer(self._flush_interval, self._timer_flush) + if not replace: + return + # A failed explicit flush must reschedule the already-armed timer, + # or it fires at the base cadence and bypasses the retry backoff. + self._flush_timer.cancel() + self._flush_timer = None + timer = threading.Timer( + delay if delay is not None else self._flush_interval, + lambda: self._timer_flush(timer), + ) timer.daemon = True self._flush_timer = timer timer.start() - def _timer_flush(self) -> None: + def _timer_flush(self, fired: Optional[threading.Timer] = None) -> None: with self._lock: + # A timer whose thread already started can't be cancelled: if a newer + # timer replaced this one meanwhile (failed explicit flush arming the + # backoff timer), the stale body must not clear it or flush again. + if fired is not None and fired is not self._flush_timer: + return self._flush_timer = None try: self.flush() @@ -470,7 +557,7 @@ def _do_flush(self) -> None: if outcome == "retry-later": with self._lock: self._consecutive_send_failures += 1 - if self._consecutive_send_failures > _MAX_CONSECUTIVE_SEND_FAILURES: + 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( @@ -482,15 +569,24 @@ def _do_flush(self) -> None: 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. + # next flush instead of being lost — and re-arm the timer with capped + # exponential backoff, so a real outage isn't hammered at the base + # cadence. New captures see the armed timer and don't shorten it. + # First retry at the base interval, then doubling — the shared JS + # logs ramp (exponent is failures - 1), so the drop budget works + # out to the documented ~21 minutes at the default 10s interval. + delay = self._flush_interval * min( + 2 ** (self._consecutive_send_failures - 1), + _MAX_RETRY_BACKOFF_MULTIPLIER, + ) log.warning( - "Metrics flush failed (attempt %s of %s); will retry with the next window", + "Metrics flush failed (attempt %s of %s); retrying in %.0fs", self._consecutive_send_failures, - _MAX_CONSECUTIVE_SEND_FAILURES + 1, + _MAX_CONSECUTIVE_SEND_FAILURES, + delay, ) self._merge_window_back(window) - self._arm_flush_timer() + self._arm_flush_timer(delay, replace=True) elif outcome == "too-large": log.warning( "Metrics batch exceeded the server size limit and was dropped. " diff --git a/posthog/test/test_metrics.py b/posthog/test/test_metrics.py index b6783204..d77b4409 100644 --- a/posthog/test/test_metrics.py +++ b/posthog/test/test_metrics.py @@ -7,7 +7,12 @@ import posthog from posthog.client import Client -from posthog.metrics_capture import DEFAULT_HISTOGRAM_BOUNDS +from posthog.metrics_capture import ( + _DEFAULT_FLUSH_INTERVAL_SECONDS, + _DEFAULT_MAX_SERIES_PER_FLUSH, + _MAX_CONSECUTIVE_SEND_FAILURES, + DEFAULT_HISTOGRAM_BOUNDS, +) from posthog.version import VERSION FAKE_API_KEY = "phc_test_key" @@ -343,6 +348,86 @@ def test_non_string_attribute_keys_stringify_on_the_wire(self, client): assert keys == {"a", "2"} assert all(isinstance(attr["key"], str) for attr in dp["attributes"]) + def test_uncopyable_attribute_does_not_unshare_other_values(self, client): + # One un-deepcopyable value must not degrade the whole snapshot to a + # shallow copy: the other, perfectly copyable values would then stay + # shared with the caller and mutate after the series key was computed. + class Uncopyable: + def __deepcopy__(self, memo): + raise TypeError("nope") + + tags = ["a"] + client.metrics.count("m", 1, attributes={"bad": Uncopyable(), "tags": tags}) + tags.append("b") + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + (dp,) = metric["sum"]["dataPoints"] + attrs = {a["key"]: a["value"] for a in dp["attributes"]} + wire_tags = [v["stringValue"] for v in attrs["tags"]["arrayValue"]["values"]] + assert wire_tags == ["a"] + + def test_nested_attribute_values_snapshot_at_capture(self, client): + # The series key is computed at capture time; a caller mutating a nested + # value afterwards must not rewrite the stored series' attributes, or the + # wire payload diverges from the identity the series was keyed under. + tags = ["a"] + client.metrics.count("m", 1, attributes={"tags": tags}) + tags.append("b") + client.metrics.count("m", 1, attributes={"tags": tags}) + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + points = metric["sum"]["dataPoints"] + assert len(points) == 2 + wire_tags = sorted( + [ + v["stringValue"] + for v in dp["attributes"][0]["value"]["arrayValue"]["values"] + ] + for dp in points + ) + assert wire_tags == [["a"], ["a", "b"]] + + @pytest.mark.parametrize( + "bad_config", + [ + "not-a-dict", + {"resource_attributes": "bad"}, + {"flush_interval": "10"}, + {"max_series_per_flush": "many"}, + {"before_send": "not-callable"}, + ], + ids=[ + "config-not-dict", + "resource-attributes-not-dict", + "flush-interval-not-number", + "series-cap-not-int", + "before-send-not-callable", + ], + ) + def test_hostile_metrics_config_does_not_raise_and_still_records(self, bad_config): + # client.metrics is reached before the guarded capture path, so bad nested + # config must fall back to defaults instead of raising into the host app. + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics=bad_config, + ) + c.metrics.count("m", 1) # the complete public call must not raise + + assert c.metrics._flush_interval == _DEFAULT_FLUSH_INTERVAL_SECONDS + assert c.metrics._max_series_per_flush == _DEFAULT_MAX_SERIES_PER_FLUSH + + payload, _, _ = flush_and_capture(c) + c.metrics.reset() + + (metric,) = metrics_from(payload) + assert metric["name"] == "m" + def test_list_attribute_records_as_array_value(self, client): client.metrics.count("arr", 1, attributes={"tags": ["a", "b"]}) @@ -475,7 +560,7 @@ def test_window_dropped_after_consecutive_failures(self, client): with mock.patch( "posthog.metrics_capture._get_session", return_value=mock_session(503) ): - for _ in range(4): + for _ in range(_MAX_CONSECUTIVE_SEND_FAILURES): client.metrics.flush() payload, _, _ = flush_and_capture(client) @@ -483,3 +568,100 @@ def test_window_dropped_after_consecutive_failures(self, client): assert ( payload is None ) # budget exhausted → window dropped, nothing left to send + + def test_window_survives_failures_until_the_drop_limit(self, client): + # One failure short of the budget the window must still deliver in full — + # dropping earlier than documented silently loses data during outages. + client.metrics.count("m", 3) + with mock.patch( + "posthog.metrics_capture._get_session", return_value=mock_session(503) + ): + for _ in range(_MAX_CONSECUTIVE_SEND_FAILURES - 1): + client.metrics.flush() + + payload, _, _ = flush_and_capture(client) + + (metric,) = metrics_from(payload) + (dp,) = metric["sum"]["dataPoints"] + assert dp["asDouble"] == 3.0 + + def test_explicit_flush_failure_reschedules_armed_timer_with_backoff(self): + # A capture arms the base-interval timer; if an explicit flush() then + # fails, the retry must happen at the backoff delay — the already-armed + # timer has to be rescheduled, not left to fire at the base cadence. + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics={"flush_interval": 1.0}, + ) + c.metrics.count("m", 1) + assert c.metrics._flush_timer.interval == 1.0 + + with mock.patch( + "posthog.metrics_capture._get_session", return_value=mock_session(503) + ): + c.metrics.flush() + c.metrics.flush() + + assert c.metrics._flush_timer is not None + assert c.metrics._flush_timer.interval == 2.0 + c.metrics.reset() + + def test_stale_timer_callback_does_not_clobber_newer_timer(self): + # A timer whose thread already started can't be cancelled; when its body + # runs late it must not clear (and duplicate-flush over) a newer backoff + # timer armed in the meantime by a failed explicit flush. + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics={"flush_interval": 1.0}, + ) + c.metrics.count("m", 1) + stale = c.metrics._flush_timer + + with mock.patch( + "posthog.metrics_capture._get_session", return_value=mock_session(503) + ) as session: + c.metrics.flush() # cancels `stale`, arms the backoff timer + newer = c.metrics._flush_timer + assert newer is not stale + sends_before = session.post.call_count + + c.metrics._timer_flush(stale) # the already-started stale thread body + + assert c.metrics._flush_timer is newer + assert session.post.call_count == sends_before + c.metrics.reset() + + def test_failed_flushes_back_off_exponentially_capped(self): + # Retrying a down endpoint at the fixed cadence hammers it for the whole + # outage; retry delays must grow exponentially and cap at 64x the base + # interval (matching the shared JS logs policy), then reset on success. + c = Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics={"flush_interval": 1.0}, + ) + c.metrics.count("m", 1) + + intervals = [] + with mock.patch( + "posthog.metrics_capture._get_session", return_value=mock_session(503) + ): + for _ in range(_MAX_CONSECUTIVE_SEND_FAILURES - 1): + c.metrics._timer_flush() # what the flush timer thread invokes + intervals.append(c.metrics._flush_timer.interval) + # First retry at the base interval, then doubling to the 64x cap — the + # shared JS logs ramp (exponent is failures - 1), giving the documented + # ~21 minute outage budget at the default 10s interval. + assert intervals == [1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0] + + # A successful flush resets the backoff: the next capture arms the base interval. + flush_and_capture(c) + c.metrics.reset() + c.metrics.count("m", 1) + assert c.metrics._flush_timer.interval == 1.0 + c.metrics.reset()