fix(metrics): harden attribute snapshots, retry backoff, and config validation

.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.

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:

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. "