feat(metrics): add posthog.metrics count/gauge/histogram API (alpha)

.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 SDK ships, with no OpenTelemetry setup:
+
+```python
+client = Posthog("<ph_project_api_key>", 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()`; 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.

posthog/__init__.py

@@ -1132,8 +1132,7 @@ def shutdown() -> None:
     Category:
         Client management
     """
-    _proxy("flush")
-    _proxy("join")
+    _proxy("shutdown")
 
 
 def setup() -> Client:

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
@@ -1599,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):
@@ -1715,6 +1726,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("<ph_project_api_key>", 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 +1824,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()