feat(metrics): record requests and httpx durations as a histogram

.sampo/changesets/network-metrics.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Add the `network` option to the `metrics` client config. When set, the SDK records the duration of every HTTP request the application makes with `requests` or `httpx` as the `http.client.request.duration` histogram, with `method`, `host`, templated `path` and `status_class` attributes. `name` and `attributes` functions customise what is recorded. The SDK's own requests are skipped, and the wrappers are removed on `shutdown()`.

posthog/__init__.py

@@ -339,7 +339,15 @@ def get_tags() -> Dict[str, Any]:
     metrics: Config dict for the ``client.metrics`` API (``service_name``,
         ``service_version``, ``environment``, ``flush_interval``, ...). Applied
         when ``setup()`` builds the global client, or on a later ``setup()``
-        call if the metrics API hasn't been used yet.
+        call if the metrics API hasn't been used yet. Set ``network`` to
+        ``True`` to record the duration of every HTTP request the application
+        makes with ``requests`` or ``httpx`` as the
+        ``http.client.request.duration`` histogram, with ``method``, ``host``,
+        templated ``path`` and ``status_class`` attributes. Pass a dict with
+        ``name`` (a string, or a function of the request that returns the name
+        or ``None`` to skip it) and ``attributes`` (a function of the request
+        and response whose result is merged over the defaults) to customise it.
+        The SDK's own requests are not recorded.
     enable_exception_autocapture: Automatically capture uncaught exceptions.
     log_captured_exceptions: Also log exceptions captured by error tracking.
     project_root: Root path used to determine in-app exception stack frames.
@@ -1298,7 +1306,7 @@ def setup() -> Client:
     # module-attr assignment (e.g. a Django ready() hook running after something
     # already forced setup()) still applies until the metrics API is first used.
     if default_client._metrics is None:
-        default_client._metrics_config = metrics
+        default_client._configure_metrics(metrics)
 
     return default_client
 

posthog/_async_request.py

@@ -12,6 +12,7 @@
 
 from .capture_compression import CaptureCompression
 from .capture_v1 import _parse_retry_after, _send_v1_batch
+from .network_metrics import _mark_internal
 from .request import (
     APIError,
     DatetimeSerializer,
@@ -38,7 +39,9 @@ def _require_httpx():
 def _build_client(host: Optional[str] = None):
     httpx_module = _require_httpx()
     base_url = remove_trailing_slash(normalize_host(host))
-    return httpx_module.AsyncClient(base_url=base_url, follow_redirects=False)
+    return _mark_internal(
+        httpx_module.AsyncClient(base_url=base_url, follow_redirects=False)
+    )
 
 
 def _serialize_v0_body(

posthog/client.py

@@ -1077,6 +1077,7 @@ def __init__(
             )
 
         self._warn_if_duplicate_async_client()
+        self._configure_metrics(metrics)
 
     def _set_library_identity(self, library_id: str, library_version: str) -> None:
         """Override the SDK identity stamped on events and outbound requests."""
@@ -2466,6 +2467,12 @@ def metrics(self) -> PostHogMetrics:
                         self._metrics = PostHogMetrics(self, None)
         return self._metrics
 
+    def _configure_metrics(self, metrics: Optional[dict]) -> None:
+        self._metrics_config = metrics
+        if isinstance(metrics, dict) and metrics.get("network"):
+            # Building the metrics API installs the network request wrappers.
+            _ = 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.
@@ -2681,6 +2688,11 @@ def _shutdown_once(self, errors: list[Exception]) -> None:
             self._flush_or_discard_queues(errors)
 
         if self._metrics is not None:
+            self._run_lifecycle_cleanup(
+                "Failed to stop network metrics on shutdown",
+                self._metrics._stop_network_metrics,
+                errors,
+            )
             self._run_lifecycle_cleanup(
                 "Failed to flush metrics on shutdown", self._metrics.flush, errors
             )

posthog/metrics_capture.py

@@ -33,6 +33,7 @@
 
 import requests
 
+from posthog.network_metrics import _NetworkMetrics
 from posthog.request import _get_session
 from posthog.utils import remove_trailing_slash
 from posthog.version import VERSION
@@ -285,6 +286,11 @@ def __init__(self, client, config: Optional[dict] = None):
         self._type_by_name: dict = {}
         self._type_collision_warned: set = set()
 
+        network = config.get("network")
+        self._network: Optional[_NetworkMetrics] = (
+            _NetworkMetrics(self, network) if network else None
+        )
+
     def count(
         self,
         name: str,
@@ -329,6 +335,11 @@ def reset(self) -> None:
             self._type_by_name = {}
             self._type_collision_warned = set()
 
+    def _stop_network_metrics(self) -> None:
+        if self._network is not None:
+            self._network.stop()
+            self._network = None
+
     def _guarded_capture(
         self,
         metric_type: str,

posthog/network_metrics.py

@@ -0,0 +1,193 @@
+"""Automatic duration metrics for the HTTP requests an application makes.
+
+Enabled with the ``metrics={"network": ...}`` client option. Wraps
+``requests.Session.send`` and, when httpx is installed, ``httpx.Client.send``
+and ``httpx.AsyncClient.send``. The wrappers only observe: they call the
+original with the same arguments, return its result or re-raise its error,
+and never let a recording failure reach the caller. The SDK marks its own
+sessions with ``_mark_internal`` so PostHog's uploads are not recorded.
+"""
+
+import contextvars
+import functools
+import logging
+import re
+import time
+from typing import Any, Callable, List, Optional, Tuple
+from urllib.parse import urlsplit
+
+import requests
+
+try:
+    import httpx
+except ImportError:  # pragma: no cover
+    httpx = None
+
+log = logging.getLogger("posthog")
+
+DEFAULT_METRIC_NAME = "http.client.request.duration"
+
+_INTERNAL_MARKER = "_posthog_internal"
+
+# ``requests`` sends each redirect hop through ``Session.send`` again. Only the
+# outermost send in a thread or task is recorded, so a redirected request counts once.
+_in_flight: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "posthog_network_metrics_in_flight", default=False
+)
+
+_ALL_DIGITS = re.compile(r"^\d+$")
+_HEX_WITH_A_DIGIT = re.compile(r"^[0-9a-f-]*\d[0-9a-f-]*$", re.IGNORECASE)
+
+
+def _mark_internal(http_client):
+    """Marks a ``requests.Session`` or httpx client as the SDK's own.
+
+    Its requests are never recorded as network metrics.
+    """
+    setattr(http_client, _INTERNAL_MARKER, True)
+    return http_client
+
+
+def _is_internal(http_client) -> bool:
+    return getattr(http_client, _INTERNAL_MARKER, False) is True
+
+
+def _is_id_like(segment: str) -> bool:
+    return bool(_ALL_DIGITS.match(segment)) or (
+        len(segment) >= 8 and bool(_HEX_WITH_A_DIGIT.match(segment))
+    )
+
+
+def _template_path(path: str) -> str:
+    """Replaces each all-digit or uuid-like path segment with ``:id``."""
+    return "/".join(
+        ":id" if _is_id_like(segment) else segment for segment in path.split("/")
+    )
+
+
+def _status_class(status: Optional[int]) -> str:
+    return "{}xx".format(status // 100) if status else "missing"
+
+
+def _parse_config(config: Any) -> Tuple[Any, Optional[Callable]]:
+    if config is True:
+        config = {}
+    if not isinstance(config, dict):
+        log.warning(
+            "Ignoring metrics network config: expected True or a dict, got %s",
+            type(config).__name__,
+        )
+        config = {}
+    name = config.get("name", DEFAULT_METRIC_NAME)
+    if not (isinstance(name, str) or callable(name)):
+        log.warning("Ignoring metrics network name: expected a string or a callable")
+        name = DEFAULT_METRIC_NAME
+    attributes = config.get("attributes")
+    if attributes is not None and not callable(attributes):
+        log.warning("Ignoring metrics network attributes: expected a callable")
+        attributes = None
+    return name, attributes
+
+
+def _patch(target, attribute: str, make_wrapper: Callable) -> Callable[[], None]:
+    original = getattr(target, attribute)
+    wrapper = make_wrapper(original)
+    setattr(target, attribute, wrapper)
+
+    def restore() -> None:
+        # Another wrapper layered on top keeps ours in place as a pass-through.
+        if getattr(target, attribute) is wrapper:
+            setattr(target, attribute, original)
+
+    return restore
+
+
+class _NetworkMetrics:
+    """Installs the request wrappers for one metrics client; ``stop()`` removes them."""
+
+    def __init__(self, metrics, config: Any):
+        self._metrics = metrics
+        self._name, self._attributes = _parse_config(config)
+        self._active = True
+        self._record_error_warned = False
+        self._restores: List[Callable[[], None]] = [
+            _patch(requests.Session, "send", self._wrap_sync)
+        ]
+        if httpx is not None:
+            self._restores.append(_patch(httpx.Client, "send", self._wrap_sync))
+            self._restores.append(_patch(httpx.AsyncClient, "send", self._wrap_async))
+
+    def stop(self) -> None:
+        self._active = False
+        for restore in self._restores:
+            restore()
+
+    def _observes(self, http_client) -> bool:
+        return self._active and not _in_flight.get() and not _is_internal(http_client)
+
+    def _wrap_sync(self, original: Callable) -> Callable:
+        @functools.wraps(original)
+        def send(http_client, request, *args, **kwargs):
+            if not self._observes(http_client):
+                return original(http_client, request, *args, **kwargs)
+            token = _in_flight.set(True)
+            start = time.perf_counter()
+            try:
+                response = original(http_client, request, *args, **kwargs)
+            except Exception:
+                self._record(request, None, start)
+                raise
+            finally:
+                _in_flight.reset(token)
+            self._record(request, response, start)
+            return response
+
+        return send
+
+    def _wrap_async(self, original: Callable) -> Callable:
+        @functools.wraps(original)
+        async def send(http_client, request, *args, **kwargs):
+            if not self._observes(http_client):
+                return await original(http_client, request, *args, **kwargs)
+            token = _in_flight.set(True)
+            start = time.perf_counter()
+            try:
+                response = await original(http_client, request, *args, **kwargs)
+            except Exception:
+                self._record(request, None, start)
+                raise
+            finally:
+                _in_flight.reset(token)
+            self._record(request, response, start)
+            return response
+
+        return send
+
+    def _record(self, request, response, start: float) -> None:
+        try:
+            duration_ms = (time.perf_counter() - start) * 1000
+            url = str(request.url)
+            parts = urlsplit(url)
+            if parts.scheme not in ("http", "https"):
+                return
+            observed = {"url": url, "method": str(request.method).upper()}
+            name = self._name(observed) if callable(self._name) else self._name
+            if not name:
+                return
+            status = getattr(response, "status_code", None)
+            attributes = {
+                "method": observed["method"],
+                "host": parts.hostname or "",
+                "path": _template_path(parts.path),
+                "status_class": _status_class(status),
+            }
+            if self._attributes is not None:
+                extra = self._attributes(
+                    observed, {"status": status, "duration_ms": duration_ms}
+                )
+                attributes.update(extra or {})
+            self._metrics.histogram(name, duration_ms, unit="ms", attributes=attributes)
+        except Exception as e:
+            if not self._record_error_warned:
+                self._record_error_warned = True
+                log.warning("Failed to record network metric: %s", e)

posthog/request.py

@@ -16,6 +16,7 @@
 from urllib3.util.retry import Retry
 
 from posthog._logging import _configure_posthog_logging
+from posthog.network_metrics import _mark_internal
 from posthog.utils import remove_trailing_slash
 from posthog.version import VERSION
 
@@ -84,7 +85,7 @@ def _build_session(socket_options: Optional[SocketOptions] = None) -> requests.S
         ),
         socket_options=socket_options,
     )
-    session = requests.Session()
+    session = _mark_internal(requests.Session())
     session.mount("https://", adapter)
     return session
 
@@ -101,7 +102,7 @@ def _build_flags_session(
         max_retries=Retry(total=0, connect=0, read=0, status=0),
         socket_options=socket_options,
     )
-    session = requests.Session()
+    session = _mark_internal(requests.Session())
     session.mount("https://", adapter)
     return session