diff --git a/.sampo/changesets/network-metrics.md b/.sampo/changesets/network-metrics.md new file mode 100644 index 000000000..b1fca56fd --- /dev/null +++ b/.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()`. diff --git a/posthog/__init__.py b/posthog/__init__.py index 0431bc883..08b2054cc 100644 --- a/posthog/__init__.py +++ b/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 diff --git a/posthog/_async_request.py b/posthog/_async_request.py index ae750de9a..502efebe5 100644 --- a/posthog/_async_request.py +++ b/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( diff --git a/posthog/client.py b/posthog/client.py index 59ccb9dca..3903b99bc 100644 --- a/posthog/client.py +++ b/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 ) diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py index 6e454842c..19e1d871d 100644 --- a/posthog/metrics_capture.py +++ b/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, diff --git a/posthog/network_metrics.py b/posthog/network_metrics.py new file mode 100644 index 000000000..5772422f6 --- /dev/null +++ b/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) diff --git a/posthog/request.py b/posthog/request.py index 76df1fdc9..56c6fa905 100644 --- a/posthog/request.py +++ b/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 diff --git a/posthog/test/test_network_metrics.py b/posthog/test/test_network_metrics.py new file mode 100644 index 000000000..0f40034e9 --- /dev/null +++ b/posthog/test/test_network_metrics.py @@ -0,0 +1,373 @@ +import logging +from unittest import mock + +import httpx +import pytest +import requests +from requests.adapters import BaseAdapter + +import posthog +from posthog.client import Client +from posthog.network_metrics import _mark_internal, _template_path +from posthog.request import _get_flags_session, _get_session + +FAKE_API_KEY = "phc_test_key" + +ORIGINAL_REQUESTS_SEND = requests.Session.send +ORIGINAL_HTTPX_SEND = httpx.Client.send +ORIGINAL_HTTPX_ASYNC_SEND = httpx.AsyncClient.send + + +def make_client(network=True): + return Client( + FAKE_API_KEY, + host="https://us.example.com", + sync_mode=True, + metrics={"network": network}, + ) + + +@pytest.fixture +def client(): + c = make_client() + yield c + c.shutdown() + + +@pytest.fixture +def recorded(client): + with mock.patch.object(client.metrics, "histogram") as histogram: + yield histogram + + +def recorded_attributes(histogram): + return histogram.call_args.kwargs["attributes"] + + +class FakeAdapter(BaseAdapter): + """Answers each send with the next canned status, or raises the given error.""" + + def __init__(self, *statuses, error=None): + super().__init__() + self.statuses = list(statuses) + self.error = error + + def send(self, request, **kwargs): + if self.error is not None: + raise self.error + response = requests.Response() + response.status_code = self.statuses.pop(0) + response.request = request + response.url = request.url + response._content = b"" + response._content_consumed = True + if response.status_code in (301, 302): + response.headers["location"] = request.url + "next/" + return response + + def close(self): + pass + + +def session_with(adapter, scheme="https://"): + session = requests.Session() + session.mount(scheme, adapter) + return session + + +class TestRequestsRecording: + def test_records_a_duration_histogram_with_default_attributes(self, recorded): + session_with(FakeAdapter(200)).get("https://api.example.com/users/42/orders") + + recorded.assert_called_once() + name, duration_ms = recorded.call_args.args + assert name == "http.client.request.duration" + assert duration_ms >= 0 + assert recorded.call_args.kwargs["unit"] == "ms" + assert recorded_attributes(recorded) == { + "method": "GET", + "host": "api.example.com", + "path": "/users/:id/orders", + "status_class": "2xx", + } + + @pytest.mark.parametrize( + "status,status_class", + [(200, "2xx"), (304, "3xx"), (404, "4xx"), (503, "5xx")], + ) + def test_status_class_groups_the_status_code(self, recorded, status, status_class): + session_with(FakeAdapter(status)).get("https://api.example.com/") + + assert recorded_attributes(recorded)["status_class"] == status_class + + def test_a_failed_request_records_missing_and_still_raises(self, recorded): + session = session_with(FakeAdapter(error=requests.ConnectionError("boom"))) + + with pytest.raises(requests.ConnectionError): + session.post("https://api.example.com/") + + assert recorded_attributes(recorded) == { + "method": "POST", + "host": "api.example.com", + "path": "/", + "status_class": "missing", + } + + def test_a_redirected_request_is_recorded_once_with_its_final_status( + self, recorded + ): + session_with(FakeAdapter(302, 200)).get("https://api.example.com/a/") + + recorded.assert_called_once() + assert recorded_attributes(recorded)["status_class"] == "2xx" + + @pytest.mark.parametrize("scheme", ["file://", "ftp://"]) + def test_non_http_requests_are_not_recorded(self, recorded, scheme): + session_with(FakeAdapter(200), scheme).get(scheme + "example.com/thing") + + recorded.assert_not_called() + + @pytest.mark.parametrize("get_sdk_session", [_get_session, _get_flags_session]) + def test_the_sdks_own_requests_are_not_recorded(self, recorded, get_sdk_session): + session = get_sdk_session() + sdk_adapter = session.get_adapter("https://us.example.com/") + session.mount("https://", FakeAdapter(200)) + try: + session.get("https://us.example.com/batch/") + finally: + session.mount("https://", sdk_adapter) + + recorded.assert_not_called() + + def test_a_marked_session_is_not_recorded(self, recorded): + _mark_internal(session_with(FakeAdapter(200))).get("https://api.example.com/") + + recorded.assert_not_called() + + +class TestHttpxRecording: + def test_records_sync_httpx_requests(self, recorded): + transport = httpx.MockTransport(lambda request: httpx.Response(201)) + with httpx.Client(transport=transport) as http: + http.post("https://api.example.com/items") + + assert recorded_attributes(recorded) == { + "method": "POST", + "host": "api.example.com", + "path": "/items", + "status_class": "2xx", + } + + async def test_records_async_httpx_requests(self, recorded): + transport = httpx.MockTransport(lambda request: httpx.Response(404)) + async with httpx.AsyncClient(transport=transport) as http: + await http.get("https://api.example.com/items/9f8e7d6c5b4a") + + assert recorded_attributes(recorded) == { + "method": "GET", + "host": "api.example.com", + "path": "/items/:id", + "status_class": "4xx", + } + + async def test_a_failed_async_request_records_missing_and_still_raises( + self, recorded + ): + def fail(request): + raise httpx.ConnectError("boom", request=request) + + async with httpx.AsyncClient(transport=httpx.MockTransport(fail)) as http: + with pytest.raises(httpx.ConnectError): + await http.get("https://api.example.com/") + + assert recorded_attributes(recorded)["status_class"] == "missing" + + async def test_a_marked_async_client_is_not_recorded(self, recorded): + transport = httpx.MockTransport(lambda request: httpx.Response(200)) + async with _mark_internal(httpx.AsyncClient(transport=transport)) as http: + await http.get("https://api.example.com/") + + recorded.assert_not_called() + + +class TestConfig: + def test_a_string_name_is_used_for_every_request(self): + client = make_client({"name": "outbound.duration"}) + try: + with mock.patch.object(client.metrics, "histogram") as histogram: + session_with(FakeAdapter(200)).get("https://api.example.com/") + finally: + client.shutdown() + + assert histogram.call_args.args[0] == "outbound.duration" + + def test_a_name_function_sees_the_request_and_can_skip_it(self): + seen = [] + + def name(request): + seen.append(request) + return None if request["method"] == "DELETE" else "kept" + + client = make_client({"name": name}) + try: + with mock.patch.object(client.metrics, "histogram") as histogram: + session = session_with(FakeAdapter(200, 200)) + session.delete("https://api.example.com/a") + session.get("https://api.example.com/b?x=1") + finally: + client.shutdown() + + assert seen == [ + {"url": "https://api.example.com/a", "method": "DELETE"}, + {"url": "https://api.example.com/b?x=1", "method": "GET"}, + ] + histogram.assert_called_once() + assert histogram.call_args.args[0] == "kept" + + def test_attributes_merge_over_and_replace_the_defaults(self): + def attributes(request, response): + assert response["status"] == 200 + assert response["duration_ms"] >= 0 + return {"path": "/users/{id}", "team": "billing"} + + client = make_client({"attributes": attributes}) + try: + with mock.patch.object(client.metrics, "histogram") as histogram: + session_with(FakeAdapter(200)).get("https://api.example.com/users/7") + finally: + client.shutdown() + + assert recorded_attributes(histogram) == { + "method": "GET", + "host": "api.example.com", + "path": "/users/{id}", + "status_class": "2xx", + "team": "billing", + } + + def test_a_raising_attributes_function_is_logged_and_the_request_succeeds( + self, caplog + ): + def attributes(request, response): + raise ValueError("bad attributes") + + client = make_client({"attributes": attributes}) + try: + with caplog.at_level(logging.WARNING, logger="posthog"): + response = session_with(FakeAdapter(200)).get( + "https://api.example.com/" + ) + finally: + client.shutdown() + + assert response.status_code == 200 + assert "bad attributes" in caplog.text + + @pytest.mark.parametrize("network", ["yes", 1, {"name": 3}]) + def test_invalid_config_warns_and_uses_the_defaults(self, network, caplog): + with caplog.at_level(logging.WARNING, logger="posthog"): + client = make_client(network) + try: + with mock.patch.object(client.metrics, "histogram") as histogram: + session_with(FakeAdapter(200)).get("https://api.example.com/") + finally: + client.shutdown() + + assert "network" in caplog.text + assert histogram.call_args.args[0] == "http.client.request.duration" + + +class TestLifecycle: + @pytest.mark.parametrize("network", [None, False, {}]) + def test_nothing_is_wrapped_while_network_metrics_are_off(self, network): + client = Client(FAKE_API_KEY, sync_mode=True, metrics={"network": network}) + try: + assert requests.Session.send is ORIGINAL_REQUESTS_SEND + assert httpx.Client.send is ORIGINAL_HTTPX_SEND + assert httpx.AsyncClient.send is ORIGINAL_HTTPX_ASYNC_SEND + finally: + client.shutdown() + + def test_wrappers_install_when_the_client_is_built_and_leave_on_shutdown(self): + client = make_client() + + assert requests.Session.send is not ORIGINAL_REQUESTS_SEND + assert httpx.Client.send is not ORIGINAL_HTTPX_SEND + assert httpx.AsyncClient.send is not ORIGINAL_HTTPX_ASYNC_SEND + + client.shutdown() + + assert requests.Session.send is ORIGINAL_REQUESTS_SEND + assert httpx.Client.send is ORIGINAL_HTTPX_SEND + assert httpx.AsyncClient.send is ORIGINAL_HTTPX_ASYNC_SEND + + def test_a_wrapper_that_cannot_be_removed_passes_requests_through(self): + client = make_client() + histogram = mock.patch.object(client.metrics, "histogram").start() + ours = requests.Session.send + + def layered(self, request, **kwargs): + return ours(self, request, **kwargs) + + requests.Session.send = layered + try: + client.shutdown() + assert requests.Session.send is layered + + response = session_with(FakeAdapter(200)).get("https://api.example.com/") + + assert response.status_code == 200 + histogram.assert_not_called() + finally: + mock.patch.stopall() + requests.Session.send = ORIGINAL_REQUESTS_SEND + + def test_module_level_setup_installs_the_wrappers(self): + saved = ( + posthog.default_client, + posthog.api_key, + posthog.host, + posthog.sync_mode, + posthog.metrics, + ) + posthog.default_client = None + posthog.api_key = FAKE_API_KEY + posthog.host = "https://us.example.com" + posthog.sync_mode = True + posthog.metrics = {"network": True} + try: + posthog.setup() + + assert requests.Session.send is not ORIGINAL_REQUESTS_SEND + + posthog.shutdown() + + assert requests.Session.send is ORIGINAL_REQUESTS_SEND + finally: + ( + posthog.default_client, + posthog.api_key, + posthog.host, + posthog.sync_mode, + posthog.metrics, + ) = saved + + +@pytest.mark.parametrize( + "path,templated", + [ + ("/", "/"), + ("/users", "/users"), + ("/users/123/orders/4", "/users/:id/orders/:id"), + ( + "/items/3fa85f64-5717-4562-b3fc-2c963f66afa6", + "/items/:id", + ), + ("/items/9f8e7d6c", "/items/:id"), + ("/items/abcdefgh", "/items/abcdefgh"), + ("/orders/order-123", "/orders/order-123"), + ("/files/38217.pdf", "/files/38217.pdf"), + ], +) +def test_template_path_replaces_id_like_segments(path, templated): + assert _template_path(path) == templated diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index 6599ea487..2c4b2fc4e 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -851,6 +851,8 @@ attribute posthog.metrics = None 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.network_metrics.DEFAULT_METRIC_NAME = 'http.client.request.duration' +attribute posthog.network_metrics.log = logging.getLogger('posthog') attribute posthog.on_error = None attribute posthog.personal_api_key = None attribute posthog.poll_interval = 30 @@ -1556,6 +1558,7 @@ module posthog.mcp.tools module posthog.mcp.types module posthog.mcp.version module posthog.metrics_capture +module posthog.network_metrics module posthog.poller module posthog.request module posthog.types