diff --git a/.sampo/changesets/metrics-autocapture.md b/.sampo/changesets/metrics-autocapture.md new file mode 100644 index 000000000..e6e25dc4f --- /dev/null +++ b/.sampo/changesets/metrics-autocapture.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: minor +--- + +Add `metrics={"autocapture": True}` (alpha): HTTP, database and runtime metrics from the official OpenTelemetry instrumentations, with no instrumentation code. Install `posthog[metrics]` and run `opentelemetry-bootstrap -a install` to add the instrumentors for your libraries. Create the client before your web app (for example before `Flask(...)`, or in Django's `AppConfig.ready()`): apps created earlier are not measured. Autocapture does not start when the app already set up OpenTelemetry; export those metrics to PostHog instead. Raw URLs and paths, client addresses, user agents and captured headers are never recorded. `metrics.before_send` also applies to autocaptured measurements: it can change their value and attributes, or drop them. diff --git a/posthog/__init__.py b/posthog/__init__.py index 2e870403c..a722ea29c 100644 --- a/posthog/__init__.py +++ b/posthog/__init__.py @@ -341,7 +341,11 @@ 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. ``autocapture`` (alpha, + ``True`` or ``{"http": ..., "db": ..., "runtime": ...}``) starts HTTP, + database and runtime metrics from the OpenTelemetry instrumentations + when ``setup()`` runs with it; call ``setup()`` before the web app is + created, because apps created earlier are not measured. traces: Config dict for distributed tracing: ``service_name``, ``service_version``, ``environment``, ``resource_attributes``, ``flush_interval`` (5 s), ``max_queue_size`` (2048), @@ -1477,6 +1481,8 @@ def setup() -> Client: # already forced setup()) still applies until the metrics API is first used. if default_client._metrics is None: default_client._metrics_config = metrics + # Autocapture is not lazy, so a late `metrics` option starts it here. + default_client._start_metrics_autocapture_once(metrics) # traces=None means off, so the module option applies only where the client # has none; False is the latch of an init that failed and stays off. if traces is not None and default_client._traces_config is None: diff --git a/posthog/_metrics_autocapture.py b/posthog/_metrics_autocapture.py new file mode 100644 index 000000000..814f7cf98 --- /dev/null +++ b/posthog/_metrics_autocapture.py @@ -0,0 +1,1231 @@ +"""Metrics autocapture (``metrics={"autocapture": True}``) — alpha. + +HTTP, database and runtime metrics from the official OpenTelemetry +instrumentations, exported to PostHog's ``/i/v1/metrics`` with no +instrumentation code in the app. + +The OpenTelemetry packages come from the ``posthog[metrics]`` extra and are +imported only when autocapture starts. Library instrumentors (requests, Django, +psycopg, ...) are found through the ``opentelemetry_instrumentor`` entry points, +the same way ``opentelemetry-instrument`` finds them. An instrumentor changes the +library classes when it starts, so objects created before that, such as a Flask +or FastAPI app, are not measured: create the client before the web app. Django +is instrumented only when its settings are configured, as in ``AppConfig.ready()``. + +The meter and tracer providers are private: nothing is set on the OpenTelemetry +globals, and the app's ``OTEL_EXPORTER_OTLP_*`` settings are not used. The +library instrumentors are process-wide singletons, though, so an app cannot +instrument a library again with its own OpenTelemetry setup while autocapture +instruments it. Apps with their own OpenTelemetry instrumentation should export +their metrics to PostHog and not turn on autocapture. + +Raw URLs and paths, client addresses, user agents and captured headers are never +recorded; routes (such as ``/users/{id}``) are. Spans are only used to derive +database metrics and are never exported. The HTTP +instrumentors get no tracer, and the instrumentors that autocapture starts do not +propagate context, so no trace context or baggage goes to other services. +""" + +import importlib +import logging +import os +import re +import sys +import threading +import weakref +from contextlib import contextmanager +from importlib.metadata import entry_points +from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple +from urllib.parse import urlsplit + +from .version import VERSION + +try: + from opentelemetry.sdk.trace import SpanProcessor as _SpanProcessorBase +except ImportError: # pragma: no cover - the processor is only built once OTel imports + _SpanProcessorBase = object + +log = logging.getLogger("posthog") + +_import_module = importlib.import_module + +_INSTALL_HINT = 'pip install "posthog[metrics]" && opentelemetry-bootstrap -a install' + +_HTTP_CLIENT_INSTRUMENTORS = ( + "requests", + "urllib", + "urllib3", + "httpx", + "aiohttp-client", +) +_HTTP_SERVER_INSTRUMENTORS = ( + "django", + "flask", + "fastapi", + "starlette", + "falcon", + "pyramid", + "tornado", + "aiohttp-server", +) +_DB_DRIVER_INSTRUMENTORS = ( + "psycopg2", + "psycopg", + "asyncpg", + "pymysql", + "mysql", + "mysqlclient", + "sqlite3", + "pymongo", + "redis", + "pymssql", + "cassandra", +) +# SQLAlchemy adds connection pool metrics. Its spans wrap a driver span, so they +# are not turned into operation metrics: that would count one query twice. +_DB_INSTRUMENTORS = _DB_DRIVER_INSTRUMENTORS + ("sqlalchemy",) +_DERIVED_DB_METRIC_SCOPES = frozenset( + f"opentelemetry.instrumentation.{name}" for name in _DB_DRIVER_INSTRUMENTORS +) + +# Process-level metrics only; the host-wide system.* metrics are left out. +_RUNTIME_CONFIG: Dict[str, Optional[List[str]]] = { + "process.cpu.time": ["user", "system"], + "process.cpu.utilization": ["user", "system"], + "process.memory.usage": None, + "process.memory.virtual": None, + "process.open_file_descriptor.count": None, + "process.thread.count": None, + "process.context_switches": ["involuntary", "voluntary"], + "cpython.gc.collections": None, + "cpython.gc.collected_objects": None, + "cpython.gc.uncollectable_objects": None, +} + +_DB_OPERATION_DURATION = "db.client.operation.duration" +# OpenTelemetry semantic convention advice for `db.client.operation.duration`. +_DB_DURATION_BUCKETS = [0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10] +# The leading keyword (`SELECT`, `GET`) is bounded; the rest of a statement is not. +_STATEMENT_KEYWORD = re.compile(r"^\s*([A-Za-z]{2,20})\b") + +# The OpenTelemetry SDK keeps every attribute set it has seen, also with delta +# temporality, and has no limit of its own. Some attributes, such as the +# `server.address` of an outgoing request, can come from user input. So each +# instrument keeps at most this many attribute sets. Other measurements go to one +# overflow series, as the OpenTelemetry cardinality limit specifies. +_MAX_ATTRIBUTE_SETS = 2000 +_OVERFLOW_ATTRIBUTES = {"otel.metric.overflow": True} + +# Attributes that some instrumentors put on metrics, such as Tornado's raw +# `http.target` and `http.url`. Paths and URLs can hold tokens and emails, and +# client addresses and user agents identify users. They are never recorded. +_UNSAFE_ATTRIBUTES = frozenset( + { + "http.url", + "http.target", + "url.full", + "url.path", + "url.query", + "http.client_ip", + "client.address", + "net.peer.ip", + "net.sock.peer.addr", + "http.user_agent", + "user_agent.original", + } +) +_UNSAFE_ATTRIBUTE_PREFIXES = ("http.request.header.", "http.response.header.") +_ATTRIBUTE_VALUE_TYPES = (str, bool, int, float) + +_SEMCONV_OPT_IN = "OTEL_SEMCONV_STABILITY_OPT_IN" +# The app's OTLP settings (headers with other vendors' keys, endpoint, TLS files) +# are for its own collector. The exporter reads them when it is created. +_OTLP_ENVIRONMENT_PREFIXES = ("OTEL_EXPORTER_OTLP_", "OTEL_PYTHON_EXPORTER_OTLP_") +_EXPORT_TIMEOUT_SECONDS = 10.0 +_DEFAULT_PORTS = {"http": 80, "https": 443} + +# The `metrics.before_send` sample type of each instrument kind. +_SAMPLE_TYPES = { + "create_counter": "count", + "create_up_down_counter": "count", + "create_histogram": "histogram", + "create_gauge": "gauge", + "create_observable_counter": "count", + "create_observable_up_down_counter": "count", + "create_observable_gauge": "gauge", +} + +_lock = threading.Lock() +_active: Optional["MetricsAutocapture"] = None +_warned_about_missing_packages = False +# Flask apps created while autocapture runs. FlaskInstrumentor.uninstrument() +# restores only the Flask class, so shutdown uninstruments these apps and the +# next autocapture instruments them again. +_flask_apps: "weakref.WeakSet[Any]" = weakref.WeakSet() +# The `inject` functions that autocapture replaced in instrumentor modules, +# by module name. +_replaced_injects: Dict[str, Any] = {} + + +def _always_enabled() -> bool: + return True + + +def _apply_before_send( + before_send: Callable, + name: str, + metric_type: str, + unit: str, + value: Any, + attributes: Any, +) -> Optional[Tuple[Any, Any]]: + """Runs the ``metrics.before_send`` hook on one measurement, like the + ``client.metrics`` API does. Returns the value and attributes to record, or + ``None`` to drop it. An instrument's name, type and unit are fixed, so the + hook can change only the value and the attributes.""" + sample = { + "name": name, + "type": metric_type, + "value": value, + "unit": unit, + "attributes": dict(attributes) if attributes else {}, + } + try: + filtered = before_send(sample) + except Exception as e: + log.error("Error in metrics before_send: %s", e) + return None + if not filtered: + return None + if not isinstance(filtered, dict): + log.warning( + "Dropping metric: before_send must return the sample dict or a falsy value" + ) + return None + value = filtered.get("value") + attributes = filtered.get("attributes") + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not isinstance(attributes, (dict, type(None))) + ): + log.warning( + "Dropping metric '%s': before_send returned an invalid value or attributes", + name, + ) + return None + return value, attributes + + +def _clean_attributes(attributes: Any) -> Any: + """Removes the unsafe attributes, and keeps only valid OpenTelemetry values, + with sequences as tuples. Some OpenTelemetry versions raise for a list value, + and `before_send` can return any value.""" + if not attributes: + return attributes + try: + items = list(attributes.items()) + except (AttributeError, TypeError): + return None + cleaned: Dict[str, Any] = {} + for key, value in items: + if ( + not isinstance(key, str) + or key in _UNSAFE_ATTRIBUTES + or key.startswith(_UNSAFE_ATTRIBUTE_PREFIXES) + ): + continue + if isinstance(value, _ATTRIBUTE_VALUE_TYPES): + cleaned[key] = value + elif isinstance(value, (list, tuple)) and all( + isinstance(item, _ATTRIBUTE_VALUE_TYPES) for item in value + ): + cleaned[key] = tuple(value) + return cleaned + + +def _instrument_name_and_unit(args: tuple, kwargs: dict, unit_index: int): + name = kwargs.get("name", args[0] if args else "") + unit = kwargs.get("unit", args[unit_index] if len(args) > unit_index else "") + return name, unit + + +class _BoundedInstrument: + """A synchronous instrument that records only while autocapture is enabled, + applies ``before_send``, and sends new attribute sets to the overflow series + once it has seen ``limit`` of them.""" + + def __init__( + self, + instrument: Any, + limit: int, + is_enabled: Callable[[], bool] = _always_enabled, + before_send: Optional[Callable] = None, + name: str = "", + metric_type: str = "", + unit: str = "", + stopped: Optional[threading.Event] = None, + ): + self._instrument = instrument + self._stopped = stopped + self._limit = limit + self._is_enabled = is_enabled + self._before_send = before_send + self._name = name + self._metric_type = metric_type + self._unit = unit + # No lock: a race can only go over the limit by the number of threads. + self._seen: set = set() + + def _bounded(self, attributes: Any) -> Any: + try: + key = frozenset(attributes.items()) if attributes else frozenset() + except (AttributeError, TypeError): + return attributes + if key in self._seen: + return attributes + if len(self._seen) < self._limit: + self._seen.add(key) + return attributes + return _OVERFLOW_ATTRIBUTES + + def _record(self, method: str, amount, attributes, args, kwargs): + # Like `client.metrics`, nothing is recorded while the client is disabled, + # so that period is not exported once the client is enabled again. + # Instrumentation that shutdown leaves in place records nothing more. + if ( + self._stopped is not None and self._stopped.is_set() + ) or not self._is_enabled(): + return None + attributes = _clean_attributes(attributes) + if self._before_send is not None: + measurement = _apply_before_send( + self._before_send, + self._name, + self._metric_type, + self._unit, + amount, + attributes, + ) + if measurement is None: + return None + amount, attributes = measurement + attributes = _clean_attributes(attributes) + # This runs in the instrumented call, such as the app's HTTP request. + try: + return getattr(self._instrument, method)( + amount, self._bounded(attributes), *args, **kwargs + ) + except Exception: + log.debug( + "Metrics autocapture failed to record %s", self._name, exc_info=True + ) + return None + + def add(self, amount, attributes=None, *args, **kwargs): + return self._record("add", amount, attributes, args, kwargs) + + def record(self, amount, attributes=None, *args, **kwargs): + return self._record("record", amount, attributes, args, kwargs) + + def set(self, amount, attributes=None, *args, **kwargs): + return self._record("set", amount, attributes, args, kwargs) + + def __getattr__(self, name): + return getattr(self._instrument, name) + + +class _BoundedMeter: + def __init__( + self, + meter: Any, + limit: int, + is_enabled: Callable[[], bool] = _always_enabled, + before_send: Optional[Callable] = None, + stopped: Optional[threading.Event] = None, + ): + self._meter = meter + self._limit = limit + self._is_enabled = is_enabled + self._before_send = before_send + self._stopped = stopped + + def _bounded(self, factory: str, args, kwargs) -> _BoundedInstrument: + name, unit = _instrument_name_and_unit(args, kwargs, unit_index=1) + return _BoundedInstrument( + getattr(self._meter, factory)(*args, **kwargs), + self._limit, + self._is_enabled, + self._before_send, + name, + _SAMPLE_TYPES[factory], + unit, + self._stopped, + ) + + def _observable(self, factory: str, args, kwargs): + # Observable instruments report from bounded callbacks (runtime metrics), + # so only the enabled check and `before_send` apply to them. They still + # report in the last collection of shutdown. + name, unit = _instrument_name_and_unit(args, kwargs, unit_index=2) + args = list(args) + if len(args) > 1: + args[1] = self._gated_callbacks(args[1], name, _SAMPLE_TYPES[factory], unit) + elif "callbacks" in kwargs: + kwargs["callbacks"] = self._gated_callbacks( + kwargs["callbacks"], name, _SAMPLE_TYPES[factory], unit + ) + return getattr(self._meter, factory)(*args, **kwargs) + + def _gated_callbacks(self, callbacks, name: str, metric_type: str, unit: str): + if not callbacks: + return callbacks + is_enabled, before_send = self._is_enabled, self._before_send + observation_class = _import_module("opentelemetry.metrics").Observation + + def gated(callback): + def observe(options): + if not is_enabled(): + return [] + observations = callback(options) + if before_send is None: + return observations + kept = [] + for observation in observations or (): + measurement = _apply_before_send( + before_send, + name, + metric_type, + unit, + observation.value, + observation.attributes, + ) + if measurement is not None: + value, attributes = measurement + kept.append( + observation_class(value, _clean_attributes(attributes)) + ) + return kept + + return observe + + # Generator objects are also valid callbacks; they are passed on as they are. + return [ + gated(callback) if callable(callback) else callback + for callback in callbacks + ] + + def create_counter(self, *args, **kwargs): + return self._bounded("create_counter", args, kwargs) + + def create_up_down_counter(self, *args, **kwargs): + return self._bounded("create_up_down_counter", args, kwargs) + + def create_histogram(self, *args, **kwargs): + return self._bounded("create_histogram", args, kwargs) + + def create_gauge(self, *args, **kwargs): + return self._bounded("create_gauge", args, kwargs) + + def create_observable_counter(self, *args, **kwargs): + return self._observable("create_observable_counter", args, kwargs) + + def create_observable_up_down_counter(self, *args, **kwargs): + return self._observable("create_observable_up_down_counter", args, kwargs) + + def create_observable_gauge(self, *args, **kwargs): + return self._observable("create_observable_gauge", args, kwargs) + + def __getattr__(self, name): + return getattr(self._meter, name) + + +class _BoundedMeterProvider: + """The meter provider given to the instrumentors. Its instruments record only + while autocapture is enabled, apply ``before_send``, and limit the number of + attribute sets of each synchronous instrument.""" + + def __init__( + self, + meter_provider: Any, + limit: int, + is_enabled: Callable[[], bool] = _always_enabled, + before_send: Optional[Callable] = None, + stopped: Optional[threading.Event] = None, + ): + self._meter_provider = meter_provider + self._limit = limit + self._is_enabled = is_enabled + self._before_send = before_send + self._stopped = stopped + + def get_meter(self, *args, **kwargs): + return _BoundedMeter( + self._meter_provider.get_meter(*args, **kwargs), + self._limit, + self._is_enabled, + self._before_send, + self._stopped, + ) + + def __getattr__(self, name): + return getattr(self._meter_provider, name) + + +@contextmanager +def _stable_http_semconv() -> Iterator[None]: + """Selects the stable HTTP names (`http.server.request.duration`, seconds), + the same names posthog-node sends, while the instrumentors load. + + OpenTelemetry reads the opt-in once per process. If another instrumentor + read it first, without an opt-in, the HTTP mode is set here, so the names + do not depend on import order. The process environment is restored + afterwards, so subprocesses and other code do not see the change. + An opt-in that the app set is kept.""" + previous = os.environ.get(_SEMCONV_OPT_IN) + if previous is None: + os.environ[_SEMCONV_OPT_IN] = "http" + try: + try: + semconv = _import_module("opentelemetry.instrumentation._semconv") + stability = semconv._OpenTelemetrySemanticConventionStability + stability._initialize() + if previous is None: + stability._OTEL_SEMCONV_STABILITY_SIGNAL_MAPPING[ + semconv._OpenTelemetryStabilitySignalType.HTTP + ] = semconv._StabilityMode.HTTP + except Exception: # older instrumentation packages have no opt-in + pass + yield + finally: + if previous is None: + os.environ.pop(_SEMCONV_OPT_IN, None) + + +@contextmanager +def _patched_environ(values: Dict[str, Optional[str]]) -> Iterator[None]: + """Sets the environment variables (``None`` removes one) while OpenTelemetry + reads them, then restores the process environment.""" + previous = {key: os.environ.get(key) for key in values} + try: + for key, value in values.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + yield + finally: + for key, value in previous.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + +def _posthog_url_pattern(host: str) -> str: + """An ``excluded_urls`` pattern for the PostHog host, in every form the HTTP + instrumentors see it: in any case, and with or without the default port.""" + parts = urlsplit(host) + scheme = parts.scheme.lower() + hostname = parts.hostname + if not scheme or not hostname: + pattern = "^" + re.escape(host) + "(?:[/?#]|$)" + else: + if ":" in hostname: + hostname = f"[{hostname}]" + default_port = _DEFAULT_PORTS.get(scheme) + if parts.port is not None and parts.port != default_port: + port = f":{parts.port}" + else: + port = f"(?::{default_port})?" if default_port else "" + pattern = ( + "^(?i:" + + re.escape(f"{scheme}://{hostname}") + + ")" + + port + + re.escape(parts.path.rstrip("/")) + + "(?:[/?#]|$)" + ) + # The instrumentors split the excluded URLs at commas. + return pattern.replace(",", r"\x2c") + + +def _excluded_urls(instrumentor_name: str, posthog_pattern: str) -> Tuple[str, str]: + """The excluded URLs for an HTTP client instrumentor: the PostHog host, and + the URLs that the app excludes in ``OTEL_PYTHON_*_EXCLUDED_URLS``. Returns + the instrumentor's environment variable too, which httpx and aiohttp read + in place of the ``excluded_urls`` argument.""" + variable = ( + f"OTEL_PYTHON_{instrumentor_name.upper().replace('-', '_')}_EXCLUDED_URLS" + ) + excluded = os.environ.get(variable, os.environ.get("OTEL_PYTHON_EXCLUDED_URLS", "")) + return variable, ",".join(url for url in (posthog_pattern, excluded) if url) + + +def _track_flask_apps(instrumentor_class: Any, providers: Dict[str, Any]) -> None: + flask: Any = _import_module("flask") + instrumented_flask = flask.Flask + + class _TrackedFlask(instrumented_flask): # type: ignore[misc,valid-type] + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + # FlaskInstrumentor.uninstrument_app() needs `_teardown_request`, which + # only instrument_app() sets. The instrumented class registers that + # hook last. + teardown_funcs = self.teardown_request_funcs.get(None) + if not hasattr(self, "_teardown_request") and teardown_funcs: + self._teardown_request = teardown_funcs[-1] + self._posthog_wsgi_app = self.wsgi_app + _flask_apps.add(self) + + flask.Flask = _TrackedFlask + for app in list(_flask_apps): + if not getattr(app, "_is_instrumented_by_opentelemetry", False): + instrumentor_class.instrument_app(app, **providers) + app._posthog_wsgi_app = app.wsgi_app + + +def _uninstrument_flask_apps() -> None: + flask_instrumentation = sys.modules.get("opentelemetry.instrumentation.flask") + if flask_instrumentation is None: + return + for app in list(_flask_apps): + if not getattr(app, "_is_instrumented_by_opentelemetry", False): + continue + # uninstrument_app() restores the WSGI app saved when the app was + # instrumented, which would remove middleware that the app added later, + # such as authentication. Such an app keeps the instrumentation, which + # records nothing after shutdown, and is not measured again. + if getattr(app, "wsgi_app", None) is not getattr( + app, "_posthog_wsgi_app", None + ): + log.debug( + "Metrics autocapture left the instrumentation of %r in place", app + ) + continue + flask_instrumentation.FlaskInstrumentor.uninstrument_app(app) + + +def _django_settings_configured() -> bool: + conf = sys.modules.get("django.conf") + return bool(getattr(getattr(conf, "settings", None), "configured", False)) + + +def _no_inject(*args: Any, **kwargs: Any) -> None: + return None + + +def _disable_propagation(instrumentor_class: Any) -> None: + """Replaces ``inject`` in the instrumentor's modules, which would send the + current trace context and baggage to other services. The instrumentors call + it through their module global, and autocapture owns these instrumentors + while it runs.""" + inject = _import_module("opentelemetry.propagate").inject + package = instrumentor_class.__module__ + if package.startswith("opentelemetry.instrumentation."): + # Such as `opentelemetry.instrumentation.tornado`, with its client module. + package = ".".join(package.split(".")[:3]) + for name, module in list(sys.modules.items()): + if name != package and not name.startswith(package + "."): + continue + if getattr(module, "inject", None) is inject: + _replaced_injects[name] = inject + setattr(module, "inject", _no_inject) + + +def _restore_propagation() -> None: + while _replaced_injects: + name, inject = _replaced_injects.popitem() + module = sys.modules.get(name) + if module is not None and getattr(module, "inject", None) is _no_inject: + setattr(module, "inject", inject) + + +def _first_string(attributes: Any, keys: tuple) -> Optional[str]: + for key in keys: + value = attributes.get(key) + if isinstance(value, str) and value: + return value + return None + + +class DbSpanMetricsProcessor(_SpanProcessorBase): + """Turns driver-level database client spans into a + ``db.client.operation.duration`` histogram. + + Only bounded attributes are kept: never the statement, which holds values, + and never anything per-user. + """ + + def __init__(self, meter): + try: + self._histogram = meter.create_histogram( + _DB_OPERATION_DURATION, + unit="s", + description="Duration of database client operations.", + explicit_bucket_boundaries_advisory=_DB_DURATION_BUCKETS, + ) + except TypeError: + # opentelemetry-api before 1.23 has no bucket advice. + self._histogram = meter.create_histogram( + _DB_OPERATION_DURATION, + unit="s", + description="Duration of database client operations.", + ) + + def on_start(self, span, parent_context=None): + pass + + def on_end(self, span): + try: + self._record(span) + except Exception: # never raise into the instrumented call + log.debug( + "Metrics autocapture failed to record a database span", exc_info=True + ) + + def _record(self, span): + from opentelemetry.trace import SpanKind, StatusCode + + scope = getattr(span, "instrumentation_scope", None) + if ( + span.kind != SpanKind.CLIENT + or getattr(scope, "name", None) not in _DERIVED_DB_METRIC_SCOPES + ): + return + attributes = span.attributes or {} + system = _first_string(attributes, ("db.system.name", "db.system")) + if not system or span.start_time is None or span.end_time is None: + return + metric_attributes: Dict[str, Any] = {"db.system.name": system} + operation = _first_string(attributes, ("db.operation.name", "db.operation")) + if operation is None: + statement = _first_string(attributes, ("db.query.text", "db.statement")) + match = _STATEMENT_KEYWORD.match(statement or "") + operation = match.group(1).upper() if match else None + else: + # Only the leading keyword is bounded, as for a statement. + words = operation.split() + operation = words[0] if words else None + collection = _first_string( + attributes, + ( + "db.collection.name", + "db.mongodb.collection", + "db.sql.table", + "db.cassandra.table", + ), + ) + namespace = _first_string(attributes, ("db.namespace", "db.name")) + address = _first_string(attributes, ("server.address", "net.peer.name")) + port = attributes.get("server.port", attributes.get("net.peer.port")) + if operation: + metric_attributes["db.operation.name"] = operation + if collection: + metric_attributes["db.collection.name"] = collection + if namespace: + metric_attributes["db.namespace"] = namespace + if address: + metric_attributes["server.address"] = address + if isinstance(port, int) and not isinstance(port, bool): + metric_attributes["server.port"] = port + if span.status.status_code == StatusCode.ERROR: + metric_attributes["error.type"] = ( + _first_string(attributes, ("error.type",)) or "_OTHER" + ) + self._histogram.record( + (span.end_time - span.start_time) / 1e9, metric_attributes + ) + + def _on_ending(self, span): + pass + + def shutdown(self): + pass + + def force_flush(self, timeout_millis=30000): + return True + + +class MetricsAutocapture: + """The running autocapture. ``shutdown()`` exports the last window and + removes the instrumentation.""" + + def __init__( + self, + meter_provider, + tracer_provider, + instrumentors, + reader, + stopped: Optional[threading.Event] = None, + ): + self._meter_provider = meter_provider + self._tracer_provider = tracer_provider + self._instrumentors = instrumentors + self._reader = reader + self._stopped = stopped if stopped is not None else threading.Event() + self._reset_flush_state() + + def force_flush(self, timeout_millis: int = 10_000) -> None: + self._meter_provider.force_flush(timeout_millis) + + def _reset_flush_state(self) -> None: + self._flush_done = threading.Condition() + self._flush_thread: Optional[threading.Thread] = None + # Flush requests are numbered. An export covers the requests made + # before it starts. + self._flush_requested = 0 + self._flush_exported = 0 + + def start_flush(self) -> Optional[Callable[[Optional[float]], bool]]: + """Starts an export of the current window. Returns a function that waits + for it, at most the given seconds, and returns whether it finished; or + ``None`` when there is nothing to wait for. + + One thread exports at a time. A call made while it exports gets one more + export, so the measurements recorded before the call are also sent, and + callers with short timeouts do not start a thread each. Each call waits + only for the first export that starts after it.""" + with self._flush_done: + self._flush_requested += 1 + request = self._flush_requested + flusher = self._flush_thread + if flusher is threading.current_thread(): + # From the export itself, waiting would never end. + return None + if flusher is None: + flusher = threading.Thread( + target=self._flush_until_done, + name="posthog-metrics-autocapture-flush", + daemon=True, + ) + try: + flusher.start() + except RuntimeError: # no new threads at interpreter shutdown + self._flush_requested -= 1 + return None + self._flush_thread = flusher + + def wait(timeout: Optional[float] = None) -> bool: + with self._flush_done: + return self._flush_done.wait_for( + lambda: self._flush_exported >= request, timeout + ) + + return wait + + def _flush_until_done(self) -> None: + while True: + with self._flush_done: + covered = self._flush_requested + self._flush_quietly() + with self._flush_done: + self._flush_exported = covered + self._flush_done.notify_all() + if self._flush_requested == covered: + self._flush_thread = None + return + + def _flush_quietly(self) -> None: + try: + self._meter_provider.force_flush() + except Exception: + log.debug("Metrics autocapture failed to flush", exc_info=True) + + def _reinit_after_fork(self) -> None: + # The instrumentation and the providers stay active in the child, and the + # OpenTelemetry metric reader restarts its export thread there. The reader + # does not replace its export lock, which the parent's export thread can + # hold at fork time. + if getattr(self._reader, "_export_lock", None) is not None: + self._reader._export_lock = threading.Lock() + # The parent's flush thread does not exist in the child. + self._reset_flush_state() + try: + _replace_collection_locks(self._meter_provider) + except Exception: + log.debug( + "Metrics autocapture failed to reset locks after fork", exc_info=True + ) + # The runtime instrumentor keeps the process it was created in, so it + # would report the parent process. + for instrumentor in self._instrumentors: + try: + psutil = _import_module("psutil") + if isinstance(getattr(instrumentor, "_proc", None), psutil.Process): + instrumentor._proc = psutil.Process(os.getpid()) + except Exception: + log.debug( + "Metrics autocapture failed to update %s after fork", + instrumentor, + exc_info=True, + ) + + def shutdown(self) -> None: + global _active + with _lock: + if _active is not self: + return + self._stopped.set() + # Uninstrument before another client can start. It would skip the + # libraries that are still instrumented, and then lose them here. + _uninstrument(self._instrumentors) + _active = None + # The last export can take a while, so it runs outside the lock. + _shut_down(self._tracer_provider, self._meter_provider) + + +def _uninstrument(instrumentors: List[Any]) -> None: + for instrumentor in instrumentors: + try: + instrumentor.uninstrument() + except Exception: + log.debug( + "Metrics autocapture failed to uninstrument %s", + instrumentor, + exc_info=True, + ) + try: + _uninstrument_flask_apps() + except Exception: + log.debug( + "Metrics autocapture failed to uninstrument Flask apps", exc_info=True + ) + _restore_propagation() + + +def _fresh_lock(owner: Any, name: str = "_lock") -> None: + lock = getattr(owner, name, None) + if lock is None: + return + rlock_type = type(threading.RLock()) + setattr( + owner, + name, + threading.RLock() if isinstance(lock, rlock_type) else threading.Lock(), + ) + + +def _replace_collection_locks(meter_provider: Any) -> None: + """Replaces the locks of the metrics SDK that a parent thread can hold at + fork time, while it collects or records. The SDK replaces only the locks of + the meter provider itself. In the child, no thread holds these locks.""" + for meter in list(getattr(meter_provider, "_meters", {}).values()): + _fresh_lock(meter, "_instrument_registration_lock") + consumer = getattr(meter_provider, "_measurement_consumer", None) + if consumer is None: + return + _fresh_lock(consumer) + for storage in list(getattr(consumer, "_reader_storages", {}).values()): + _fresh_lock(storage) + matches_by_instrument = getattr( + storage, "_instrument_view_instrument_matches", {} + ) + for matches in list(matches_by_instrument.values()): + for match in matches: + _fresh_lock(match) + for aggregation in list( + getattr(match, "_attributes_aggregation", {}).values() + ): + _fresh_lock(aggregation) + _fresh_lock(getattr(aggregation, "_reservoir", None)) + + +def _shut_down(*components: Any) -> None: + for component in components: + if component is None: + continue + try: + component.shutdown() + except Exception: + log.debug( + "Metrics autocapture failed to shut down %s", + component, + exc_info=True, + ) + + +def _reinit_after_fork() -> None: + global _lock + # A parent thread can hold the lock at fork time. Replace it, never acquire it. + _lock = threading.Lock() + active = _active + if active is not None: + active._reinit_after_fork() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_reinit_after_fork) + + +def _is_default_provider(provider: Any) -> bool: + name = type(provider).__name__ + return name.startswith(("_Proxy", "Proxy", "NoOp", "_NoOp", "Default")) + + +def _instrumentor_names(areas: Dict[str, bool]) -> List[str]: + names: List[str] = [] + if areas.get("http"): + names.extend(_HTTP_CLIENT_INSTRUMENTORS + _HTTP_SERVER_INSTRUMENTORS) + if areas.get("db"): + names.extend(_DB_INSTRUMENTORS) + if areas.get("runtime"): + names.append("system_metrics") + return names + + +def start_metrics_autocapture( + *, + host: str, + api_key: str, + areas: Dict[str, bool], + service_name: Optional[str], + resource_attributes: Dict[str, Any], + is_enabled: Callable[[], bool], + export_interval_seconds: Optional[float] = None, + before_send: Optional[Callable] = None, + _metric_reader: Any = None, +) -> Optional[MetricsAutocapture]: + """Starts metrics autocapture. Returns ``None``, after one warning, when the + OpenTelemetry packages are missing, when OpenTelemetry is already set up in + the process, or when autocapture is already running. Never raises. + + ``before_send`` is the ``metrics.before_send`` hook. It gets each + measurement as a ``client.metrics`` sample and can change its value and + attributes, or drop it.""" + global _active, _warned_about_missing_packages + host = host.rstrip("/") + with _lock: + if _active is not None: + log.warning( + "Metrics autocapture is already running in this process, so this client does not start it again." + ) + return None + try: + metrics_api = _import_module("opentelemetry.metrics") + trace_api = _import_module("opentelemetry.trace") + sdk_metrics = _import_module("opentelemetry.sdk.metrics") + sdk_metrics_export = _import_module("opentelemetry.sdk.metrics.export") + sdk_trace = _import_module("opentelemetry.sdk.trace") + sampling = _import_module("opentelemetry.sdk.trace.sampling") + sdk_resources = _import_module("opentelemetry.sdk.resources") + otlp = _import_module( + "opentelemetry.exporter.otlp.proto.http.metric_exporter" + ) + _import_module("opentelemetry.instrumentation") + except ImportError as e: + if not _warned_about_missing_packages: + _warned_about_missing_packages = True + log.warning( + "Metrics autocapture needs the OpenTelemetry packages. Install them with: %s (%s)", + _INSTALL_HINT, + e, + ) + return None + + reader: Any = None + meter_provider: Any = None + tracer_provider: Any = None + instrumentors: List[Any] = [] + try: + if not _is_default_provider( + metrics_api.get_meter_provider() + ) or not _is_default_provider(trace_api.get_tracer_provider()): + log.warning( + "OpenTelemetry is already set up in this process, so metrics autocapture does not start. " + "Point your OTLP metrics exporter at PostHog instead: https://posthog.com/docs/metrics" + ) + return None + + # The same resource as `client.metrics`: the OTEL_SERVICE_NAME and + # OTEL_RESOURCE_ATTRIBUTES of the app are not read. + resource = sdk_resources.Resource( + { + **resource_attributes, + "service.name": service_name or "unknown_service", + "telemetry.sdk.language": "python", + "telemetry.sdk.name": "opentelemetry", + "telemetry.sdk.version": _import_module( + "opentelemetry.sdk.version" + ).__version__, + "telemetry.distro.name": "posthog-python", + "telemetry.distro.version": VERSION, + } + ) + + reader = _metric_reader + if reader is None: + exporter_base: Any = otlp.OTLPMetricExporter + delta = sdk_metrics_export.AggregationTemporality.DELTA + cumulative = sdk_metrics_export.AggregationTemporality.CUMULATIVE + + class _GatedExporter(exporter_base): + # The exporter still sends while the client is disabled unless it is gated here. + def export(self, metrics_data, timeout_millis=10_000, **kwargs): + if not is_enabled(): + return sdk_metrics_export.MetricExportResult.SUCCESS + return super().export( + metrics_data, timeout_millis=timeout_millis, **kwargs + ) + + with _patched_environ( + { + key: None + for key in os.environ + if key.startswith(_OTLP_ENVIRONMENT_PREFIXES) + } + ): + exporter = _GatedExporter( + endpoint=f"{host}/i/v1/metrics", + headers={"Authorization": f"Bearer {api_key}"}, + timeout=_EXPORT_TIMEOUT_SECONDS, + # Delta, like the `posthog.metrics` wire shape; values that go + # up and down stay cumulative. + preferred_temporality={ + sdk_metrics.Counter: delta, + sdk_metrics.Histogram: delta, + sdk_metrics.ObservableCounter: delta, + sdk_metrics.UpDownCounter: cumulative, + sdk_metrics.ObservableUpDownCounter: cumulative, + sdk_metrics.ObservableGauge: cumulative, + }, + ) + reader = sdk_metrics_export.PeriodicExportingMetricReader( + exporter, + export_interval_millis=(export_interval_seconds or 10.0) * 1000, + ) + # The client exports the last window at exit, within its exit budget. + # The provider's own exit hook waits up to the export timeout, also + # when the PostHog host does not answer. + meter_provider = sdk_metrics.MeterProvider( + resource=resource, metric_readers=[reader], shutdown_on_exit=False + ) + + sampler_base: Any = sampling.Sampler + + class _ClientSpansOnly(sampler_base): + # Records only client spans, the ones a database call makes. Database + # instrumentors set `db.system` after start, so the kind is all a + # sampler can check. Server spans stay no-ops. + def should_sample( + self, + parent_context, + trace_id, + name, + kind=None, + attributes=None, + links=None, + trace_state=None, + ): + if kind != trace_api.SpanKind.CLIENT: + return sampling.SamplingResult(sampling.Decision.DROP) + # A recorded span starts with these attributes only. Some + # instrumentors, such as asyncpg, pass `db.system` here. + return sampling.SamplingResult( + sampling.Decision.RECORD_ONLY, attributes + ) + + def get_description(self): + return "PostHogClientSpansOnly" + + stopped = threading.Event() + bounded_meter_provider = _BoundedMeterProvider( + meter_provider, _MAX_ATTRIBUTE_SETS, is_enabled, before_send, stopped + ) + tracer_provider = sdk_trace.TracerProvider( + resource=resource, sampler=_ClientSpansOnly(), shutdown_on_exit=False + ) + # A span of the private provider is still the current span, so an HTTP + # client would send its trace context to other services (as "not + # sampled", which makes them drop their traces). HTTP metrics do not + # need spans. + no_op_tracer_provider = trace_api.NoOpTracerProvider() + if areas.get("db"): + tracer_provider.add_span_processor( + DbSpanMetricsProcessor( + bounded_meter_provider.get_meter( + "posthog.metrics_autocapture", VERSION + ) + ) + ) + + posthog_url_pattern = _posthog_url_pattern(host) + wanted = set(_instrumentor_names(areas)) + with _stable_http_semconv(): + for entry_point in entry_points(group="opentelemetry_instrumentor"): + if entry_point.name not in wanted: + continue + if ( + entry_point.name == "django" + and not _django_settings_configured() + ): + # Without settings, the Django instrumentor configures empty + # ones, and the app's own settings.configure() then fails. + # In AppConfig.ready() the settings are configured. + log.debug( + "Metrics autocapture skipped django: its settings are not configured yet" + ) + continue + try: + kwargs: Dict[str, Any] = { + "tracer_provider": tracer_provider + if entry_point.name in _DB_INSTRUMENTORS + else no_op_tracer_provider, + "meter_provider": bounded_meter_provider, + } + environ: Dict[str, Optional[str]] = {} + if entry_point.name in _HTTP_CLIENT_INSTRUMENTORS: + # PostHog's own uploads, including this exporter's, are not app traffic. + variable, excluded = _excluded_urls( + entry_point.name, posthog_url_pattern + ) + kwargs["excluded_urls"] = excluded + environ[variable] = excluded + with _patched_environ(environ): + instrumentor_class = entry_point.load() + # Instrumentors are singletons, and creating one runs + # `__init__` again on the existing instance. That would + # reset the config of an instance that the app uses. + existing = getattr(instrumentor_class, "_instance", None) + if ( + existing is not None + and existing.is_instrumented_by_opentelemetry + ): + continue + instrumentor = ( + instrumentor_class(config=_RUNTIME_CONFIG) + if entry_point.name == "system_metrics" + else instrumentor_class() + ) + if instrumentor.is_instrumented_by_opentelemetry: + # The app instrumented this library itself. + continue + instrumentor.instrument(**kwargs) + if instrumentor.is_instrumented_by_opentelemetry: + instrumentors.append(instrumentor) + _disable_propagation(instrumentor_class) + if entry_point.name == "flask": + _track_flask_apps( + instrumentor_class, + { + "tracer_provider": no_op_tracer_provider, + "meter_provider": bounded_meter_provider, + }, + ) + except Exception: + log.debug( + "Metrics autocapture skipped %s", + entry_point.name, + exc_info=True, + ) + + _active = MetricsAutocapture( + meter_provider, tracer_provider, instrumentors, reader, stopped + ) + return _active + except Exception: + log.exception("Metrics autocapture failed to start") + # Each start attempt would otherwise leave an export thread behind. + _uninstrument(instrumentors) + _shut_down(tracer_provider, meter_provider or reader) + return None diff --git a/posthog/client.py b/posthog/client.py index bff6def5d..f9582f1ac 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -19,7 +19,12 @@ from posthog._async_utils import _BackgroundEventLoopRunner from posthog._disabled_lane_queue import _DisabledLaneQueue from posthog.args import ID_TYPES, ExceptionArg, OptionalCaptureArgs, OptionalSetArgs -from posthog.metrics_capture import PostHogMetrics +from posthog.metrics_capture import ( + _DEFAULT_FLUSH_INTERVAL_SECONDS, + PostHogMetrics, + _validated_before_send, + _validated_flush_interval, +) from posthog.tracing._config import resolve_traces_config from posthog.tracing._drops import DropLog from posthog.tracing._export import SpanExporter @@ -884,10 +889,16 @@ def __init__( self._lifecycle_callback_context: ContextVar[bool] = ContextVar( "posthog_lifecycle_callback", default=False ) + # Set while a metrics `before_send` hook runs: a flush it calls does not + # start another autocapture export, which would run the hook again. + self._skip_metrics_autocapture_flush: ContextVar[bool] = ContextVar( + "posthog_skip_metrics_autocapture_flush", default=False + ) self._deferred_flush_lock = threading.Lock() self._deferred_flush_pending = False self._deferred_flush_followup = False self._deferred_flush_followup_timeout: Optional[float] = None + self._deferred_flush_followup_metrics = False # Used for session replay URL generation - we don't want the server host here. self.raw_host = normalize_host(host) self.host = determine_server_host(host) @@ -1110,6 +1121,10 @@ def __init__( ) self._lanes = [self._analytics_lane, self._ai_lane] + self._metrics_autocapture: Optional[Any] = None + self._metrics_autocapture_tried = False + self._start_metrics_autocapture_once(metrics) + if hasattr(os, "register_at_fork"): weak_self = weakref.ref(self) os.register_at_fork( @@ -1118,6 +1133,81 @@ def __init__( self._warn_if_duplicate_async_client() + def _start_metrics_autocapture_once(self, metrics_config: Any) -> None: + """Starts metrics autocapture when the ``metrics`` option asks for it. A + client tries once, so a module-level ``setup()`` on each call does not + start it again or repeat a warning.""" + autocapture = ( + metrics_config.get("autocapture") + if isinstance(metrics_config, dict) + else None + ) + if not autocapture or self.disabled or self._metrics_autocapture_tried: + return + self._metrics_autocapture_tried = True + self._metrics_autocapture = self._start_metrics_autocapture( + metrics_config, autocapture if isinstance(autocapture, dict) else {} + ) + # Sync mode has no exit hook for events; the window needs its own. + if self._metrics_autocapture is not None and self.sync_mode and self.send: + atexit.register(self._atexit_metrics_autocapture) + + def _start_metrics_autocapture(self, metrics_config: dict, areas: dict): + # Imported here, so a client without autocapture never loads OpenTelemetry. + from ._metrics_autocapture import start_metrics_autocapture + + resource_attributes = metrics_config.get("resource_attributes") + resource_attributes = ( + dict(resource_attributes) if isinstance(resource_attributes, dict) else {} + ) + for key, attribute in ( + ("service_version", "service.version"), + ("environment", "deployment.environment"), + ): + if metrics_config.get(key) and attribute not in resource_attributes: + resource_attributes[attribute] = metrics_config[key] + weak_self = weakref.ref(self) + before_send = _validated_before_send(metrics_config.get("before_send")) + if before_send is not None: + hook = before_send + + def run_as_callback(sample): + # The hook runs while OpenTelemetry collects or records. As an + # SDK callback, a flush() or shutdown() it calls is deferred, so + # it does not wait for that same collection. + client = weak_self() + if client is None: + return hook(sample) + callback_token = client._lifecycle_callback_context.set(True) + skip_token = client._skip_metrics_autocapture_flush.set(True) + try: + return hook(sample) + finally: + client._skip_metrics_autocapture_flush.reset(skip_token) + client._lifecycle_callback_context.reset(callback_token) + + before_send = run_as_callback + + return start_metrics_autocapture( + host=self.host, + api_key=self.api_key, + areas={ + "http": areas.get("http", True), + "db": areas.get("db", True), + "runtime": areas.get("runtime", True), + }, + service_name=resource_attributes.get("service.name") + or metrics_config.get("service_name"), + resource_attributes=resource_attributes, + is_enabled=lambda: (client := weak_self()) is not None + and not client.disabled + and client.send, + export_interval_seconds=_validated_flush_interval( + metrics_config.get("flush_interval", _DEFAULT_FLUSH_INTERVAL_SECONDS) + ), + before_send=before_send, + ) + def _set_library_identity(self, library_id: str, library_version: str) -> None: """Override the SDK identity stamped on events and outbound requests.""" self._library_id = library_id @@ -2325,10 +2415,14 @@ def _reinit_after_fork(self): self._lifecycle_callback_context = ContextVar( "posthog_lifecycle_callback", default=False ) + self._skip_metrics_autocapture_flush = ContextVar( + "posthog_skip_metrics_autocapture_flush", default=False + ) self._deferred_flush_lock = threading.Lock() self._deferred_flush_pending = False self._deferred_flush_followup = False self._deferred_flush_followup_timeout = None + self._deferred_flush_followup_metrics = False # Async runner threads do not survive fork(); recreate lazily on next async cache call. self._flag_definition_cache_provider_async_runner = None @@ -2713,7 +2807,7 @@ def get_active_span(self) -> Optional[Span]: def flush(self, timeout_seconds: Optional[float] = 10) -> None: """ - Attempt immediate delivery of queued events and ended spans, keeping the client reusable. + Attempt immediate delivery of queued events, ended spans and autocaptured metrics, keeping the client reusable. For a client shared across requests or serverless invocations (including the module-level client), call ``flush()`` before a short-lived invocation @@ -2750,12 +2844,22 @@ def flush(self, timeout_seconds: Optional[float] = 10) -> None: posthog.flush(timeout_seconds=10) ``` """ - if self._defer_flush_from_callback(timeout_seconds): + include_metrics_autocapture = not self._skip_metrics_autocapture_flush.get() + if self._defer_flush_from_callback( + timeout_seconds, include_metrics_autocapture + ): return try: # Spans drain with events: serverless handlers call flush(), not # shutdown(), and leaving spans on their own timer would lose them. span_flush = self._start_span_flush(timeout_seconds) + # Autocaptured metrics would otherwise wait for their export timer. + metrics_flush = ( + self._start_metrics_autocapture_flush() + if include_metrics_autocapture + else None + ) + deadline: Optional[float] = None if timeout_seconds is None: for lane in self._lanes: lane.flush(None) @@ -2763,6 +2867,10 @@ def flush(self, timeout_seconds: Optional[float] = 10) -> None: deadline = time.monotonic() + timeout_seconds for lane in self._lanes: lane.flush(max(0.0, deadline - time.monotonic())) + if metrics_flush is not None: + metrics_flush( + None if deadline is None else max(0.0, deadline - time.monotonic()) + ) if span_flush is not None: # The last span request is bounded only by the request # timeout, so the wait is not. @@ -2771,6 +2879,12 @@ def flush(self, timeout_seconds: Optional[float] = 10) -> None: self.log.exception("error flushing queue: %s", e) return + def _start_metrics_autocapture_flush( + self, + ) -> Optional[Callable[[Optional[float]], None]]: + autocapture = self._metrics_autocapture + return autocapture.start_flush() if autocapture is not None else None + def _start_span_flush( self, timeout_seconds: Optional[float] ) -> Optional[Callable[[Optional[float]], None]]: @@ -2850,7 +2964,9 @@ def run() -> None: self._start_lifecycle_thread(run, "lifecycle") return True - def _defer_flush_from_callback(self, timeout_seconds: Optional[float]) -> bool: + def _defer_flush_from_callback( + self, timeout_seconds: Optional[float], include_metrics_autocapture: bool = True + ) -> bool: if not self._is_lifecycle_callback_thread(): return False with self._deferred_flush_lock: @@ -2858,7 +2974,12 @@ def _defer_flush_from_callback(self, timeout_seconds: Optional[float]) -> bool: if not self._deferred_flush_followup: self._deferred_flush_followup = True self._deferred_flush_followup_timeout = timeout_seconds - elif ( + self._deferred_flush_followup_metrics = include_metrics_autocapture + return True + self._deferred_flush_followup_metrics = ( + self._deferred_flush_followup_metrics or include_metrics_autocapture + ) + if ( self._deferred_flush_followup_timeout is not None and timeout_seconds is not None ): @@ -2872,13 +2993,20 @@ def _defer_flush_from_callback(self, timeout_seconds: Optional[float]) -> bool: def run() -> None: next_timeout = timeout_seconds + next_metrics = include_metrics_autocapture while True: - self.flush(next_timeout) + skip_token = self._skip_metrics_autocapture_flush.set(not next_metrics) + try: + self.flush(next_timeout) + finally: + self._skip_metrics_autocapture_flush.reset(skip_token) with self._deferred_flush_lock: if self._deferred_flush_followup: next_timeout = self._deferred_flush_followup_timeout + next_metrics = self._deferred_flush_followup_metrics self._deferred_flush_followup = False self._deferred_flush_followup_timeout = None + self._deferred_flush_followup_metrics = False continue self._deferred_flush_pending = False return @@ -2990,6 +3118,14 @@ def _shutdown_once(self, errors: list[Exception]) -> None: self._run_lifecycle_cleanup( "Failed to reset metrics on shutdown", self._metrics.reset, errors ) + if self._metrics_autocapture is not None: + # Exports the last window, then removes the instrumentation. + self._run_lifecycle_cleanup( + "Failed to shut down metrics autocapture", + self._metrics_autocapture.shutdown, + errors, + ) + self._metrics_autocapture = None with self._traces_lock: traces = self._traces if traces is not None: @@ -3091,8 +3227,11 @@ def _atexit(self) -> None: span_flush = self._start_span_flush( max(0.0, deadline - time.monotonic()) ) + metrics_flush = self._start_metrics_autocapture_flush() for lane in self._lanes: lane.flush(max(0.0, deadline - time.monotonic())) + if metrics_flush is not None: + metrics_flush(max(0.0, deadline - time.monotonic())) self._join_span_flush(span_flush, deadline) finally: # Consumers are daemon threads. Publish a non-draining stop to @@ -3105,6 +3244,12 @@ def _atexit(self) -> None: self._lifecycle_owner = None self._lifecycle_condition.notify_all() + @no_throw() + def _atexit_metrics_autocapture(self) -> None: + waiter = self._start_metrics_autocapture_flush() + if waiter is not None: + waiter(max(0.0, _get_atexit_deadline() - time.monotonic())) + @no_throw() def _atexit_spans(self) -> None: # The span timer is a daemon thread that dies at exit. diff --git a/posthog/metrics_capture.py b/posthog/metrics_capture.py index 781aa89b3..3d8bc15d4 100644 --- a/posthog/metrics_capture.py +++ b/posthog/metrics_capture.py @@ -272,6 +272,28 @@ def __init__( self.exemplar: Optional[dict] = None +def _validated_flush_interval(flush_interval: Any) -> float: + # Shared with metrics autocapture, which reads the same option. + 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, + ) + return _DEFAULT_FLUSH_INTERVAL_SECONDS + return float(flush_interval) + + +def _validated_before_send(before_send: Any) -> Optional[Callable]: + if before_send is not None and not callable(before_send): + log.warning("Ignoring metrics before_send: expected a callable") + return None + return before_send + + class PostHogMetrics: """The ``client.metrics`` API: ``count``, ``gauge``, ``histogram``, ``flush``. @@ -318,18 +340,9 @@ def __init__(self, client, config: Optional[dict] = None): "deployment.environment" ) or config.get("environment") self._resource_attributes: dict = resource_attributes - 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) + self._flush_interval: float = _validated_flush_interval( + config.get("flush_interval", _DEFAULT_FLUSH_INTERVAL_SECONDS) + ) max_series = config.get("max_series_per_flush", _DEFAULT_MAX_SERIES_PER_FLUSH) if ( not isinstance(max_series, int) @@ -342,11 +355,9 @@ def __init__(self, client, config: Optional[dict] = None): ) 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._before_send: Optional[Callable] = _validated_before_send( + config.get("before_send") + ) self._lock = threading.Lock() self._pid = os.getpid() diff --git a/posthog/test/ai/langchain/test_callbacks.py b/posthog/test/ai/langchain/test_callbacks.py index 4c4de1be2..53d673826 100644 --- a/posthog/test/ai/langchain/test_callbacks.py +++ b/posthog/test/ai/langchain/test_callbacks.py @@ -1403,9 +1403,10 @@ def test_tool_calls(mock_client): ) -async def test_async_traces(mock_client): +@pytest.mark.parametrize("delay", [0, 0.1]) +async def test_async_traces(mock_client, delay): async def sleep(x): # -> Any: - await asyncio.sleep(0.1) + await asyncio.sleep(delay) return x prompt = ChatPromptTemplate.from_messages([("user", "Foo")]) @@ -1422,18 +1423,22 @@ async def sleep(x): # -> Any: approximate_latency = math.floor(time.time() - start_time) assert mock_client.capture.call_count == 4 - first_call, second_call, third_call, fourth_call = ( - mock_client.capture.call_args_list - ) - assert first_call[1]["event"] == "$ai_span" - assert second_call[1]["event"] == "$ai_generation" - assert third_call[1]["event"] == "$ai_trace" - assert third_call[1]["properties"]["$ai_span_name"] == "RunnableSequence" - assert fourth_call[1]["event"] == "$ai_trace" - assert fourth_call[1]["properties"]["$ai_span_name"] == "sleep" + calls = [call.kwargs for call in mock_client.capture.call_args_list] + assert sorted(call["event"] for call in calls) == [ + "$ai_generation", + "$ai_span", + "$ai_trace", + "$ai_trace", + ] + traces = { + call["properties"]["$ai_span_name"]: call["properties"] + for call in calls + if call["event"] == "$ai_trace" + } + assert set(traces) == {"RunnableSequence", "sleep"} assert ( min(approximate_latency - 1, 0) - <= math.floor(third_call[1]["properties"]["$ai_latency"]) + <= math.floor(traces["RunnableSequence"]["$ai_latency"]) <= approximate_latency ) diff --git a/posthog/test/test_metrics_autocapture.py b/posthog/test/test_metrics_autocapture.py new file mode 100644 index 000000000..77d03e404 --- /dev/null +++ b/posthog/test/test_metrics_autocapture.py @@ -0,0 +1,1496 @@ +import os +import socket +import sqlite3 +import subprocess +import sys +import textwrap +import threading +import time +from contextlib import ExitStack, contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from unittest import mock + +import flask +import httpx +import opentelemetry.propagate +import pytest +import requests +from opentelemetry import baggage, context, trace +from opentelemetry.instrumentation._semconv import ( + _OpenTelemetrySemanticConventionStability, +) +from opentelemetry.instrumentation.requests import RequestsInstrumentor +from opentelemetry.instrumentation.system_metrics import SystemMetricsInstrumentor +from opentelemetry.sdk.metrics import Histogram, MeterProvider +from opentelemetry.sdk.metrics.export import ( + AggregationTemporality, + InMemoryMetricReader, +) +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.trace import SpanKind, Status, StatusCode +from opentelemetry.util.http import parse_excluded_urls + +from posthog import _metrics_autocapture as metrics_autocapture +from posthog.client import Client +from posthog._metrics_autocapture import ( + DbSpanMetricsProcessor, + _BoundedInstrument, + _BoundedMeterProvider, + start_metrics_autocapture, +) + +FAKE_API_KEY = "phc_test_key" + + +class _Server: + def __init__(self): + received = self.received = [] + + class Handler(BaseHTTPRequestHandler): + def _respond(self): + length = int(self.headers.get("content-length") or 0) + received.append( + { + "path": self.path, + "headers": dict(self.headers), + "body": self.rfile.read(length), + } + ) + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", "2") + self.end_headers() + self.wfile.write(b"{}") + + do_GET = _respond + do_POST = _respond + + def log_message(self, *args): + pass + + self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.port = self.httpd.server_address[1] + self.url = f"http://127.0.0.1:{self.port}" + threading.Thread(target=self.httpd.serve_forever, daemon=True).start() + + def close(self): + self.httpd.shutdown() + self.httpd.server_close() + + +@pytest.fixture +def posthog_host(): + server = _Server() + yield server + server.close() + + +@pytest.fixture +def app_server(): + server = _Server() + yield server + server.close() + + +@pytest.fixture +def started(): + handles = [] + yield handles + for handle in handles: + handle.shutdown() + + +@pytest.fixture(autouse=True) +def _reset_autocapture(): + # A failed test must not leave autocapture running for the tests after it. + yield + active = metrics_autocapture._active + if active is not None: + active.shutdown() + metrics_autocapture._warned_about_missing_packages = False + + +class _ObservedLock: + """A lock that sets ``waited`` when a thread has to wait for it.""" + + def __init__(self, waited: threading.Event): + self._lock = threading.Lock() + self._waited = waited + + def __enter__(self): + if not self._lock.acquire(blocking=False): + self._waited.set() + self._lock.acquire() + return self + + def __exit__(self, *exc_info): + self._lock.release() + + +@contextmanager +def _held_by_another_thread(lock): + # A forked child has no copy of this thread, so the lock stays held there. + held, release = threading.Event(), threading.Event() + + def hold(): + with lock: + held.set() + release.wait() + + thread = threading.Thread(target=hold, daemon=True) + thread.start() + held.wait() + try: + yield + finally: + release.set() + thread.join() + + +def _child_result(pid, read_fd, timeout=20): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if os.waitpid(pid, os.WNOHANG) != (0, 0): + break + time.sleep(0.05) + else: + os.kill(pid, 9) + os.waitpid(pid, 0) + result = os.read(read_fd, 16) + os.close(read_fd) + return result + + +def _metrics_exports(server): + return [r for r in server.received if r["path"].endswith("/i/v1/metrics")] + + +def _start(posthog_host, started, reader=None, **overrides): + options = dict( + host=posthog_host.url, + api_key=FAKE_API_KEY, + areas={"http": True, "db": True, "runtime": True}, + service_name="checkout-api", + resource_attributes={"deployment.environment": "test"}, + is_enabled=lambda: True, + ) + options.update(overrides) + handle = start_metrics_autocapture(**options, _metric_reader=reader) + if handle is not None: + started.append(handle) + return handle + + +def _points(reader, name): + data = reader.get_metrics_data() + points = [] + for resource_metrics in data.resource_metrics if data else []: + for scope_metrics in resource_metrics.scope_metrics: + for metric in scope_metrics.metrics: + if metric.name == name: + points.extend(metric.data.data_points) + return points + + +def _names(reader): + data = reader.get_metrics_data() + return { + metric.name + for resource_metrics in (data.resource_metrics if data else []) + for scope_metrics in resource_metrics.scope_metrics + for metric in scope_metrics.metrics + } + + +class TestStartMetricsAutocapture: + def test_records_http_client_durations_but_not_requests_to_posthog( + self, posthog_host, app_server, started + ): + reader = InMemoryMetricReader() + assert _start(posthog_host, started, reader) is not None + + requests.get(f"{app_server.url}/health") + requests.post(f"{posthog_host.url}/batch/", data="{}") + + ports = [ + point.attributes.get("server.port") + for point in _points(reader, "http.client.request.duration") + ] + assert app_server.port in ports + assert posthog_host.port not in ports + + def test_records_database_operations_without_the_statement( + self, posthog_host, started + ): + reader = InMemoryMetricReader() + _start(posthog_host, started, reader) + + # The instrumentor wraps cursors, which drivers and SQLAlchemy use. + connection = sqlite3.connect(":memory:") + cursor = connection.cursor() + cursor.execute("CREATE TABLE orders (email TEXT)") + cursor.execute("SELECT * FROM orders WHERE email = 'a@b.c'") + connection.close() + + points = _points(reader, "db.client.operation.duration") + operations = {point.attributes.get("db.operation.name") for point in points} + assert {"CREATE", "SELECT"} <= operations + for point in points: + assert point.attributes["db.system.name"] == "sqlite" + assert all("a@b.c" not in str(value) for value in point.attributes.values()) + + def test_records_runtime_metrics(self, posthog_host, started): + reader = InMemoryMetricReader() + _start( + posthog_host, + started, + reader, + areas={"http": False, "db": False, "runtime": True}, + ) + + names = _names(reader) + assert "process.cpu.time" in names + assert any(name.startswith("cpython.gc.") for name in names) + assert "http.client.request.duration" not in names + + def test_exports_otlp_to_posthog_with_the_project_token( + self, posthog_host, app_server, started + ): + handle = _start(posthog_host, started) + requests.get(f"{app_server.url}/health") + + handle.force_flush() + + exports = [r for r in posthog_host.received if r["path"] == "/i/v1/metrics"] + assert exports + headers = {k.lower(): v for k, v in exports[0]["headers"].items()} + assert headers["authorization"] == f"Bearer {FAKE_API_KEY}" + assert headers["content-type"] == "application/x-protobuf" + body = b"".join(r["body"] for r in exports) + for expected in ( + b"http.client.request.duration", + b"checkout-api", + b"telemetry.distro.name", + b"posthog-python", + b"deployment.environment", + ): + assert expected in body + + def test_sends_nothing_while_opted_out(self, posthog_host, app_server, started): + handle = _start(posthog_host, started, is_enabled=lambda: False) + requests.get(f"{app_server.url}/health") + + handle.force_flush() + + assert [r for r in posthog_host.received if r["path"] == "/i/v1/metrics"] == [] + + def test_warns_once_and_returns_none_when_packages_are_missing( + self, posthog_host, started + ): + def missing(name): + raise ImportError(f"No module named '{name}'") + + with mock.patch("posthog._metrics_autocapture._import_module", missing): + with mock.patch("posthog._metrics_autocapture.log") as log: + assert _start(posthog_host, started) is None + assert _start(posthog_host, started) is None + + assert log.warning.call_count == 1 + assert "posthog[metrics]" in str(log.warning.call_args) + + def test_does_not_start_when_opentelemetry_is_already_set_up( + self, posthog_host, started + ): + with mock.patch( + "opentelemetry.metrics.get_meter_provider", return_value=MeterProvider() + ): + with mock.patch("posthog._metrics_autocapture.log") as log: + assert _start(posthog_host, started) is None + + assert "already set up" in str(log.warning.call_args) + + def test_runs_once_per_process_and_can_start_again_after_shutdown( + self, posthog_host, started + ): + first = _start(posthog_host, started, InMemoryMetricReader()) + with mock.patch("posthog._metrics_autocapture.log") as log: + assert _start(posthog_host, started) is None + assert log.warning.call_count == 1 + + first.shutdown() + reader = InMemoryMetricReader() + assert _start(posthog_host, started, reader) is not None + + connection = sqlite3.connect(":memory:") + connection.cursor().execute("SELECT 1") + connection.close() + assert _points(reader, "db.client.operation.duration") + + def test_does_not_reconfigure_runtime_metrics_that_the_app_set_up( + self, posthog_host, started + ): + # The app's meter provider is private, so autocapture still starts. + app_reader = InMemoryMetricReader() + app_instrumentor = SystemMetricsInstrumentor( + config={"system.cpu.time": ["idle", "user"]} + ) + app_instrumentor.instrument( + meter_provider=MeterProvider(metric_readers=[app_reader]) + ) + try: + assert ( + _start( + posthog_host, + started, + InMemoryMetricReader(), + areas={"runtime": True}, + ) + is not None + ) + assert "system.cpu.time" in _names(app_reader) + finally: + app_instrumentor.uninstrument() + + def test_a_client_that_starts_during_shutdown_records_http_metrics( + self, posthog_host, app_server, started, monkeypatch + ): + # Set when the other client has started, or has to wait for this shutdown. + progress = threading.Event() + monkeypatch.setattr(metrics_autocapture, "_lock", _ObservedLock(progress)) + first = _start(posthog_host, started, areas={"http": True}) + reader = InMemoryMetricReader() + replacement = [] + + def start_replacement(): + replacement.append( + _start(posthog_host, started, reader, areas={"http": True}) + ) + progress.set() + + thread = threading.Thread(target=start_replacement) + + class StartsAnotherClientDuringShutdown(list): + def __iter__(self): + thread.start() + assert progress.wait(timeout=10) + return super().__iter__() + + first._instrumentors = StartsAnotherClientDuringShutdown(first._instrumentors) + first.shutdown() + thread.join() + + assert replacement[0] is not None + requests.get(f"{app_server.url}/health") + assert _points(reader, "http.client.request.duration") + + def test_strips_a_trailing_slash_from_the_host( + self, posthog_host, app_server, started + ): + # A reverse proxy path, as in `host="https://example.com/ingest/"`. + handle = _start(posthog_host, started, host=f"{posthog_host.url}/ingest/") + requests.get(f"{app_server.url}/health") + + handle.force_flush() + + paths = {r["path"] for r in posthog_host.received} + assert "/ingest/i/v1/metrics" in paths + assert "/ingest//i/v1/metrics" not in paths + + def test_does_not_change_the_process_environment( + self, posthog_host, started, monkeypatch + ): + monkeypatch.delenv("OTEL_SEMCONV_STABILITY_OPT_IN", raising=False) + assert _start(posthog_host, started, InMemoryMetricReader()) is not None + assert "OTEL_SEMCONV_STABILITY_OPT_IN" not in os.environ + + def test_keeps_an_opt_in_that_the_app_set(self, posthog_host, started, monkeypatch): + monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "http/dup") + assert _start(posthog_host, started, InMemoryMetricReader()) is not None + assert os.environ["OTEL_SEMCONV_STABILITY_OPT_IN"] == "http/dup" + + def test_uses_stable_http_names_when_opentelemetry_read_the_opt_in_before( + self, posthog_host, app_server, started, monkeypatch + ): + # Each OpenTelemetry instrumentor reads the opt-in, but only the first + # read in the process counts. + stability = _OpenTelemetrySemanticConventionStability + monkeypatch.delenv("OTEL_SEMCONV_STABILITY_OPT_IN", raising=False) + monkeypatch.setattr(stability, "_initialized", False) + monkeypatch.setattr(stability, "_OTEL_SEMCONV_STABILITY_SIGNAL_MAPPING", {}) + stability._initialize() + + reader = InMemoryMetricReader() + _start(posthog_host, started, reader, areas={"http": True}) + requests.get(f"{app_server.url}/health") + + assert _points(reader, "http.client.request.duration") + assert not _points(reader, "http.client.duration") + + def test_limits_http_attribute_sets(self, posthog_host, app_server, started): + reader = InMemoryMetricReader() + with mock.patch.object(metrics_autocapture, "_MAX_ATTRIBUTE_SETS", 1): + _start(posthog_host, started, reader) + + requests.get(f"{app_server.url}/a") + requests.post(f"{app_server.url}/b", data="{}") + + attributes = [ + dict(point.attributes) + for point in _points(reader, "http.client.request.duration") + ] + assert len(attributes) == 2 + assert {"otel.metric.overflow": True} in attributes + + @pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") + @pytest.mark.filterwarnings("ignore:This process .* is multi-threaded") + def test_keeps_working_in_a_forked_child(self, posthog_host, app_server, started): + reader = InMemoryMetricReader() + handle = _start(posthog_host, started, reader) + read_fd, write_fd = os.pipe() + + with _held_by_another_thread(metrics_autocapture._lock): + pid = os.fork() + if pid == 0: # pragma: no cover - runs in the child + result = b"fail" + try: + runtime = [i for i in handle._instrumentors if hasattr(i, "_proc")] + requests.get(f"{app_server.url}/child") + if ( + runtime + and all(i._proc.pid == os.getpid() for i in runtime) + and _points(reader, "http.client.request.duration") + ): + handle.shutdown() + result = b"ok" + finally: + os.write(write_fd, result) + os._exit(0) + + os.close(write_fd) + assert _child_result(pid, read_fd) == b"ok" + # The child's shutdown does not stop the parent's autocapture. + assert metrics_autocapture._active is handle + + @pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") + @pytest.mark.filterwarnings("ignore:This process .* is multi-threaded") + def test_exports_from_a_child_forked_during_an_export(self, posthog_host, started): + handle = _start(posthog_host, started, export_interval_seconds=3600) + read_fd, write_fd = os.pipe() + + # The export thread holds this lock while it sends. + with _held_by_another_thread(handle._reader._export_lock): + pid = os.fork() + if pid == 0: # pragma: no cover - runs in the child + result = b"fail" + try: + handle.force_flush() + result = b"ok" + finally: + os.write(write_fd, result) + os._exit(0) + + os.close(write_fd) + assert _child_result(pid, read_fd, timeout=15) == b"ok" + + @pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") + @pytest.mark.filterwarnings("ignore:This process .* is multi-threaded") + def test_records_in_a_child_forked_during_a_collection( + self, posthog_host, app_server, started + ): + handle = _start( + posthog_host, started, export_interval_seconds=3600, areas={"http": True} + ) + requests.get(f"{app_server.url}/health") + consumer = handle._meter_provider._measurement_consumer + locks = [consumer._lock] + for storage in consumer._reader_storages.values(): + locks.append(storage._lock) + for matches in storage._instrument_view_instrument_matches.values(): + for match in matches: + locks.append(match._lock) + locks.extend( + aggregation._lock + for aggregation in match._attributes_aggregation.values() + ) + read_fd, write_fd = os.pipe() + + # Collection and recording hold these locks in other threads. + with ExitStack() as stack: + for lock in locks: + stack.enter_context(_held_by_another_thread(lock)) + pid = os.fork() + if pid == 0: # pragma: no cover - runs in the child + result = b"fail" + try: + requests.get(f"{app_server.url}/health") + handle.force_flush() + result = b"ok" + finally: + os.write(write_fd, result) + os._exit(0) + + os.close(write_fd) + assert _child_result(pid, read_fd, timeout=15) == b"ok" + + def test_does_not_send_the_apps_otlp_settings_to_posthog( + self, posthog_host, app_server, started, monkeypatch + ): + # These configure the app's own collector, not PostHog. + monkeypatch.setenv("OTEL_EXPORTER_OTLP_HEADERS", "x-vendor-key=vendor-secret") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_COMPRESSION", "gzip") + handle = _start(posthog_host, started) + requests.get(f"{app_server.url}/health") + + handle.force_flush() + + exports = _metrics_exports(posthog_host) + assert exports + headers = {k.lower(): v for k, v in exports[0]["headers"].items()} + assert "x-vendor-key" not in headers + assert "content-encoding" not in headers + assert os.environ["OTEL_EXPORTER_OTLP_HEADERS"] == "x-vendor-key=vendor-secret" + + def test_does_not_add_trace_context_to_app_requests( + self, posthog_host, app_server, started + ): + # The spans are never exported, so a downstream service must not see them. + reader = InMemoryMetricReader() + _start(posthog_host, started, reader) + + requests.get(f"{app_server.url}/health") + + headers = {k.lower() for k in app_server.received[-1]["headers"]} + assert "traceparent" not in headers + [point] = _points(reader, "http.client.request.duration") + assert point.attributes["http.response.status_code"] == 200 + + def test_records_nothing_while_disabled(self, posthog_host, app_server, started): + # Measurements from the disabled period must not go out once it is enabled. + enabled = [False] + reader = InMemoryMetricReader() + _start(posthog_host, started, reader, is_enabled=lambda: enabled[0]) + + requests.get(f"{app_server.url}/health") + connection = sqlite3.connect(":memory:") + connection.cursor().execute("SELECT 1") + connection.close() + enabled[0] = True + + assert not _points(reader, "http.client.request.duration") + assert not _points(reader, "db.client.operation.duration") + + def test_applies_before_send_to_measurements( + self, posthog_host, app_server, started + ): + samples = [] + + def before_send(sample): + samples.append(dict(sample)) + if sample["attributes"].get("http.request.method") == "POST": + return None + sample["attributes"] = { + key: value + for key, value in sample["attributes"].items() + if key != "server.address" + } + return sample + + reader = InMemoryMetricReader() + _start( + posthog_host, started, reader, areas={"http": True}, before_send=before_send + ) + + requests.get(f"{app_server.url}/a") + requests.post(f"{app_server.url}/b", data="{}") + + points = _points(reader, "http.client.request.duration") + assert [point.attributes["http.request.method"] for point in points] == ["GET"] + assert "server.address" not in points[0].attributes + sample = next(s for s in samples if s["name"] == "http.client.request.duration") + assert sample["type"] == "histogram" + assert sample["unit"] == "s" + assert isinstance(sample["value"], float) + + def test_drops_measurements_when_before_send_raises( + self, posthog_host, app_server, started + ): + def before_send(sample): + raise ValueError("bug in the hook") + + reader = InMemoryMetricReader() + _start( + posthog_host, started, reader, areas={"http": True}, before_send=before_send + ) + + with mock.patch("posthog._metrics_autocapture.log") as log: + response = requests.get(f"{app_server.url}/health") + + assert response.status_code == 200 + assert not _points(reader, "http.client.request.duration") + assert log.error.called + + def test_applies_before_send_to_runtime_metrics(self, posthog_host, started): + reader = InMemoryMetricReader() + _start( + posthog_host, + started, + reader, + areas={"runtime": True}, + before_send=lambda s: None if s["name"] == "process.cpu.time" else s, + ) + + names = _names(reader) + assert "process.cpu.time" not in names + assert "process.memory.usage" in names + + def test_does_not_record_requests_to_posthog_whatever_the_host_case( + self, posthog_host, started + ): + reader = InMemoryMetricReader() + host = f"http://LocalHost:{posthog_host.port}" + _start(posthog_host, started, reader, host=host) + + requests.post(f"{host}/batch/", data="{}") + + assert not _points(reader, "http.client.request.duration") + + def test_keeps_the_apps_excluded_urls( + self, posthog_host, app_server, started, monkeypatch + ): + monkeypatch.setenv("OTEL_PYTHON_REQUESTS_EXCLUDED_URLS", "healthz") + reader = InMemoryMetricReader() + _start(posthog_host, started, reader, areas={"http": True}) + + requests.get(f"{app_server.url}/healthz") + requests.post(f"{app_server.url}/orders", data="{}") + + methods = { + point.attributes["http.request.method"] + for point in _points(reader, "http.client.request.duration") + } + assert methods == {"POST"} + assert os.environ["OTEL_PYTHON_REQUESTS_EXCLUDED_URLS"] == "healthz" + + def test_does_not_record_httpx_requests_to_posthog( + self, posthog_host, app_server, started + ): + reader = InMemoryMetricReader() + _start(posthog_host, started, reader, areas={"http": True}) + + httpx.post(f"{posthog_host.url}/batch/", content="{}") + httpx.get(f"{app_server.url}/health") + + ports = [ + point.attributes.get("server.port") + for point in _points(reader, "http.client.request.duration") + ] + assert app_server.port in ports + assert posthog_host.port not in ports + + def test_measures_an_existing_flask_app_again_after_a_restart( + self, posthog_host, started + ): + first_reader = InMemoryMetricReader() + first = _start(posthog_host, started, first_reader, areas={"http": True}) + app = flask.Flask(__name__) + app.add_url_rule("/health", "health", lambda: "ok") + app.test_client().get("/health") + assert _points(first_reader, "http.server.request.duration") + + first.shutdown() + reader = InMemoryMetricReader() + _start(posthog_host, started, reader, areas={"http": True}) + app.test_client().get("/health") + + [point] = _points(reader, "http.server.request.duration") + assert point.count == 1 + + def test_shutdown_keeps_middleware_that_the_app_added(self, posthog_host, started): + samples = [] + + def before_send(sample): + samples.append(sample["name"]) + return sample + + handle = _start( + posthog_host, + started, + InMemoryMetricReader(), + areas={"http": True}, + before_send=before_send, + ) + app = flask.Flask(__name__) + app.add_url_rule("/secret", "secret", lambda: "secret") + protected = app.wsgi_app + + def require_auth(environ, start_response): + if environ.get("HTTP_AUTHORIZATION") != "token": + start_response("401 Unauthorized", [("Content-Length", "0")]) + return [b""] + return protected(environ, start_response) + + app.wsgi_app = require_auth + client = app.test_client() + assert client.get("/secret").status_code == 401 + + handle.shutdown() + samples.clear() + + assert client.get("/secret").status_code == 401 + assert ( + client.get("/secret", headers={"Authorization": "token"}).status_code == 200 + ) + # The instrumentation left inside the middleware records nothing more. + assert samples == [] + + def test_does_not_propagate_context_to_other_services( + self, posthog_host, app_server, started + ): + reader = InMemoryMetricReader() + handle = _start(posthog_host, started, reader, areas={"http": True}) + remote = trace.SpanContext( + trace_id=0x1234, + span_id=0x5678, + is_remote=True, + trace_flags=trace.TraceFlags(trace.TraceFlags.SAMPLED), + ) + # Baggage and an incoming trace, as a server instrumentor attaches them. + current = baggage.set_baggage( + "user.email", + "a@b.c", + trace.set_span_in_context(trace.NonRecordingSpan(remote)), + ) + token = context.attach(current) + try: + requests.get(f"{app_server.url}/a") + httpx.get(f"{app_server.url}/b") + finally: + context.detach(token) + + for request in app_server.received: + headers = {key.lower() for key in request["headers"]} + assert not headers & {"traceparent", "tracestate", "baggage"} + assert len(_points(reader, "http.client.request.duration")) == 2 + + handle.shutdown() + requests_instrumentation = sys.modules["opentelemetry.instrumentation.requests"] + assert requests_instrumentation.inject is opentelemetry.propagate.inject + + def test_keeps_the_attributes_a_database_span_starts_with( + self, posthog_host, started + ): + # asyncpg passes `db.system` when it starts the span, not later. + reader = InMemoryMetricReader() + handle = _start(posthog_host, started, reader, areas={"db": True}) + tracer = handle._tracer_provider.get_tracer( + "opentelemetry.instrumentation.asyncpg" + ) + + tracer.start_span( + "SELECT", + kind=SpanKind.CLIENT, + attributes={"db.system": "postgresql", "db.operation": "SELECT"}, + ).end() + + [point] = _points(reader, "db.client.operation.duration") + assert dict(point.attributes) == { + "db.system.name": "postgresql", + "db.operation.name": "SELECT", + } + + def test_coalesces_concurrent_flushes(self, posthog_host, started, monkeypatch): + handle = _start(posthog_host, started, InMemoryMetricReader()) + exporting = threading.Event() + release = threading.Event() + flushes = [] + + def slow_force_flush(*args, **kwargs): + flushes.append(threading.current_thread().name) + exporting.set() + release.wait(10) + + monkeypatch.setattr(handle._meter_provider, "force_flush", slow_force_flush) + waiters = [handle.start_flush()] + assert exporting.wait(5) + waiters += [handle.start_flush() for _ in range(4)] + flushers = [ + thread + for thread in threading.enumerate() + if thread.name == "posthog-metrics-autocapture-flush" + ] + release.set() + for wait in waiters: + wait(10) + + assert len(flushers) == 1 + # The calls made during the first flush get one more, so their + # measurements are exported too. + assert len(flushes) == 2 + + def test_a_flush_waits_only_for_the_export_after_it( + self, posthog_host, started, monkeypatch + ): + handle = _start(posthog_host, started, InMemoryMetricReader()) + exporting = [threading.Event(), threading.Event()] + release = [threading.Event(), threading.Event()] + exports = [] + + def slow_force_flush(*args, **kwargs): + export = min(len(exports), 1) + exports.append(export) + exporting[export].set() + release[export].wait(10) + + monkeypatch.setattr(handle._meter_provider, "force_flush", slow_force_flush) + first = handle.start_flush() + assert exporting[0].wait(5) + second = handle.start_flush() + release[0].set() + assert exporting[1].wait(5) + + # The second export is still running, but it is not for the first call. + began = time.monotonic() + assert first(5) + assert time.monotonic() - began < 2 + assert not second(0.1) + release[1].set() + assert second(5) + + def test_normalizes_attributes_from_before_send( + self, posthog_host, app_server, started + ): + def before_send(sample): + sample["attributes"] = { + **sample["attributes"], + "deployment.tags": ["blue", "canary"], + "owner": {"team": "core"}, + } + return sample + + reader = InMemoryMetricReader() + _start( + posthog_host, started, reader, areas={"http": True}, before_send=before_send + ) + + response = requests.get(f"{app_server.url}/health") + + assert response.status_code == 200 + [point] = _points(reader, "http.client.request.duration") + assert point.attributes["deployment.tags"] == ("blue", "canary") + assert "owner" not in point.attributes + + def test_normalizes_runtime_attributes_from_before_send( + self, posthog_host, started + ): + def before_send(sample): + sample["attributes"] = { + **sample["attributes"], + "deployment.tags": ["blue"], + "owner": {"team": "core"}, + } + return sample + + reader = InMemoryMetricReader() + _start( + posthog_host, + started, + reader, + areas={"runtime": True}, + before_send=before_send, + ) + + [point, *_] = _points(reader, "process.thread.count") + assert point.attributes["deployment.tags"] == ("blue",) + assert "owner" not in point.attributes + + def test_cleans_up_when_startup_fails(self, posthog_host, started): + real_entry_points = metrics_autocapture.entry_points + + def failing_entry_points(**kwargs): + yield from real_entry_points(**kwargs) + raise RuntimeError("broken plugin") + + def reader_threads(): + return [ + thread + for thread in threading.enumerate() + if thread.name == "OtelPeriodicExportingMetricReader" + ] + + before = len(reader_threads()) + with mock.patch.object( + metrics_autocapture, "entry_points", failing_entry_points + ): + with mock.patch("posthog._metrics_autocapture.log"): + assert _start(posthog_host, started) is None + + assert len(reader_threads()) == before + assert not RequestsInstrumentor().is_instrumented_by_opentelemetry + assert metrics_autocapture._active is None + + def test_uses_the_same_resource_as_the_manual_metrics_api( + self, posthog_host, started, monkeypatch + ): + monkeypatch.setenv("OTEL_SERVICE_NAME", "from-env") + monkeypatch.setenv("OTEL_RESOURCE_ATTRIBUTES", "team=from-env") + reader = InMemoryMetricReader() + _start( + posthog_host, + started, + reader, + areas={"runtime": True}, + service_name=None, + resource_attributes={}, + ) + + resource = reader.get_metrics_data().resource_metrics[0].resource.attributes + assert resource["service.name"] == "unknown_service" + assert "team" not in resource + assert resource["telemetry.distro.name"] == "posthog-python" + + def test_shutdown_twice_is_safe(self, posthog_host, started): + handle = _start(posthog_host, started, InMemoryMetricReader()) + + handle.shutdown() + handle.shutdown() + + assert metrics_autocapture._active is None + + +class TestPostHogUrlExclusion: + def _excluded(self, host, url): + pattern = metrics_autocapture._posthog_url_pattern(host) + return parse_excluded_urls(pattern).url_disabled(url) + + def test_matches_the_host_with_or_without_its_default_port(self): + assert self._excluded( + "https://us.i.posthog.com", "https://us.i.posthog.com:443/batch/" + ) + assert self._excluded( + "https://us.i.posthog.com:443", "https://us.i.posthog.com/batch/" + ) + assert self._excluded("http://LocalHost:8000", "http://localhost:8000/batch/") + + def test_does_not_match_other_hosts_ports_or_paths(self): + assert not self._excluded( + "https://us.i.posthog.com", "https://us.i.posthog.com.evil.com/" + ) + assert not self._excluded( + "https://us.i.posthog.com", "https://us.i.posthog.com:8443/batch/" + ) + assert not self._excluded( + "https://example.com/ingest", "https://example.com/ingestion" + ) + + def test_keeps_a_comma_in_the_host_path(self): + assert self._excluded( + "https://example.com/a,b", "https://example.com/a,b/i/v1/metrics" + ) + assert not self._excluded("https://example.com/a,b", "https://unrelated.com/b") + + +class TestBoundedMeterProvider: + def test_drops_raw_url_and_client_attributes(self): + # As the Tornado instrumentor records them: raw paths can hold tokens. + reader = InMemoryMetricReader() + provider = _BoundedMeterProvider(MeterProvider(metric_readers=[reader]), 10) + histogram = provider.get_meter("test").create_histogram("http.server.duration") + + histogram.record( + 1, + { + "http.method": "GET", + "http.status_code": 200, + "http.host": "shop.example", + "http.target": "/reset/private-token", + "http.url": "https://shop.example/reset/private-token?email=a@b.c", + "url.full": "https://shop.example/reset/private-token", + "url.path": "/reset/private-token", + "url.query": "email=a@b.c", + "http.client_ip": "203.0.113.7", + "client.address": "203.0.113.7", + "net.peer.ip": "203.0.113.7", + "http.user_agent": "Mozilla/5.0", + "user_agent.original": "Mozilla/5.0", + "http.request.header.authorization": ("Bearer secret",), + "http.response.header.set_cookie": ("session=secret",), + }, + ) + + [point] = _points(reader, "http.server.duration") + assert dict(point.attributes) == { + "http.method": "GET", + "http.status_code": 200, + "http.host": "shop.example", + } + + def test_never_raises_into_the_instrumented_call(self): + class BrokenInstrument: + def record(self, *args, **kwargs): + raise TypeError("broken") + + _BoundedInstrument(BrokenInstrument(), 10).record(1.0, {"a": "b"}) + + def test_sends_new_attribute_sets_to_an_overflow_series(self): + # Delta, as autocapture exports: the SDK still keeps each attribute set. + reader = InMemoryMetricReader( + preferred_temporality={Histogram: AggregationTemporality.DELTA} + ) + provider = _BoundedMeterProvider(MeterProvider(metric_readers=[reader]), 2) + histogram = provider.get_meter("test").create_histogram("duration") + + for _ in range(2): + for host in ("a", "b", "c", "d"): + histogram.record(1, {"server.address": host}) + points = _points(reader, "duration") + assert sorted( + (str(dict(point.attributes)), point.count) for point in points + ) == [ + ("{'otel.metric.overflow': True}", 2), + ("{'server.address': 'a'}", 1), + ("{'server.address': 'b'}", 1), + ] + + +class TestDbSpanMetricsProcessor: + def _record(self, scope, kind=SpanKind.CLIENT, attributes=None, error=False): + reader = InMemoryMetricReader() + meter = MeterProvider(metric_readers=[reader]).get_meter("test") + provider = TracerProvider() + provider.add_span_processor(DbSpanMetricsProcessor(meter)) + span = provider.get_tracer(scope).start_span("query", kind=kind) + span.set_attributes(attributes or {}) + if error: + span.set_status(Status(StatusCode.ERROR)) + span.end() + return _points(reader, "db.client.operation.duration") + + def test_turns_a_driver_span_into_a_duration_histogram(self): + points = self._record( + "opentelemetry.instrumentation.psycopg2", + attributes={ + "db.system": "postgresql", + "db.statement": "select * from orders where id = 7", + "db.name": "shop", + "net.peer.name": "db.internal", + "net.peer.port": 5432, + }, + ) + + assert len(points) == 1 + assert dict(points[0].attributes) == { + "db.system.name": "postgresql", + "db.operation.name": "SELECT", + "db.namespace": "shop", + "server.address": "db.internal", + "server.port": 5432, + } + + def test_marks_failures(self): + points = self._record( + "opentelemetry.instrumentation.pymongo", + attributes={"db.system": "mongodb", "db.operation": "find"}, + error=True, + ) + + assert dict(points[0].attributes) == { + "db.system.name": "mongodb", + "db.operation.name": "find", + "error.type": "_OTHER", + } + + def test_keeps_only_the_first_word_of_the_operation(self): + points = self._record( + "opentelemetry.instrumentation.psycopg", + attributes={ + "db.system": "postgresql", + "db.operation.name": "SELECT orders", + }, + ) + + assert points[0].attributes["db.operation.name"] == "SELECT" + + def test_skips_orm_layers_and_other_spans(self): + # SQLAlchemy wraps a driver span, so counting it would count the query twice. + assert ( + self._record( + "opentelemetry.instrumentation.sqlalchemy", + attributes={"db.system": "postgresql"}, + ) + == [] + ) + assert ( + self._record( + "opentelemetry.instrumentation.psycopg2", + kind=SpanKind.SERVER, + attributes={"db.system": "postgresql"}, + ) + == [] + ) + assert ( + self._record( + "opentelemetry.instrumentation.requests", + attributes={"http.request.method": "GET"}, + ) + == [] + ) + + +class TestClientOption: + def test_starts_with_metrics_autocapture_and_stops_on_shutdown( + self, posthog_host, started + ): + client = Client( + FAKE_API_KEY, host=posthog_host.url, metrics={"autocapture": True} + ) + with mock.patch("posthog._metrics_autocapture.log"): + assert _start(posthog_host, started) is None + + client.shutdown() + assert _start(posthog_host, started) is not None + + def test_does_not_start_by_default_or_when_disabled(self, posthog_host, started): + plain = Client(FAKE_API_KEY, host=posthog_host.url) + disabled = Client( + FAKE_API_KEY, + host=posthog_host.url, + disabled=True, + metrics={"autocapture": True}, + ) + + assert _start(posthog_host, started) is not None + plain.shutdown() + disabled.shutdown() + + def test_sends_nothing_when_send_is_false(self, posthog_host, app_server): + client = Client( + FAKE_API_KEY, + host=posthog_host.url, + send=False, + metrics={"autocapture": True}, + ) + try: + assert client._metrics_autocapture is not None + requests.get(f"{app_server.url}/health") + client._metrics_autocapture.force_flush() + assert [ + r for r in posthog_host.received if r["path"] == "/i/v1/metrics" + ] == [] + finally: + client.shutdown() + + def test_passes_the_metrics_options_to_autocapture(self, posthog_host): + hook_calls = [] + + def before_send(sample): + hook_calls.append(configured._lifecycle_callback_context.get()) + return {**sample, "value": 2.0} + + with mock.patch( + "posthog._metrics_autocapture.start_metrics_autocapture", return_value=None + ) as start: + configured = Client( + FAKE_API_KEY, + host=posthog_host.url, + metrics={ + "autocapture": {"runtime": False}, + "service_name": "billing-api", + "service_version": "1.2.3", + "environment": "staging", + "resource_attributes": { + "deployment.environment": "prod", + "team": "core", + }, + "flush_interval": 30, + "before_send": before_send, + }, + ) + configured.shutdown() + Client( + FAKE_API_KEY, + host=posthog_host.url, + metrics={ + "autocapture": True, + "service_name": "billing-api", + "resource_attributes": {"service.name": "billing-worker"}, + }, + ).shutdown() + + first, second = (call.kwargs for call in start.call_args_list) + assert first["host"] == posthog_host.url + assert first["api_key"] == FAKE_API_KEY + assert first["areas"] == {"http": True, "db": True, "runtime": False} + assert first["service_name"] == "billing-api" + assert first["resource_attributes"] == { + "deployment.environment": "prod", + "team": "core", + "service.version": "1.2.3", + } + assert first["export_interval_seconds"] == 30 + # The hook runs as an SDK callback, so a flush() it calls is deferred. + assert first["before_send"]({"name": "m", "value": 1.0}) == { + "name": "m", + "value": 2.0, + } + assert hook_calls == [True] + assert second["areas"] == {"http": True, "db": True, "runtime": True} + assert second["service_name"] == "billing-worker" + assert second["export_interval_seconds"] == 10 + assert second["before_send"] is None + + @pytest.mark.parametrize("flush_interval", [-1, "5", True]) + def test_ignores_an_invalid_flush_interval(self, posthog_host, flush_interval): + client = Client( + FAKE_API_KEY, + host=posthog_host.url, + metrics={"autocapture": True, "flush_interval": flush_interval}, + ) + try: + assert client._metrics_autocapture is not None + assert client._metrics_autocapture._reader._export_interval_millis == 10_000 + finally: + client.shutdown() + + def test_flush_exports_autocaptured_metrics(self, posthog_host, app_server): + client = Client( + FAKE_API_KEY, + host=posthog_host.url, + metrics={"autocapture": True, "flush_interval": 3600}, + ) + try: + requests.get(f"{app_server.url}/health") + client.flush() + assert _metrics_exports(posthog_host) + finally: + client.shutdown() + + def test_shutdown_twice_is_safe(self, posthog_host): + client = Client( + FAKE_API_KEY, host=posthog_host.url, metrics={"autocapture": True} + ) + + client.shutdown() + client.shutdown() + + assert metrics_autocapture._active is None + + def test_exit_does_not_wait_for_an_unreachable_host(self): + # Accepts connections and never answers, like a host behind a dead proxy. + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.listen(16) + script = textwrap.dedent( + f""" + from posthog.client import Client + Client( + "{FAKE_API_KEY}", + host="http://127.0.0.1:{server.getsockname()[1]}", + metrics={{"autocapture": {{"http": False, "db": False}}}}, + ) + """ + ) + try: + began = time.monotonic() + subprocess.run( + [sys.executable, "-c", script], + check=True, + timeout=60, + cwd=Path(__file__).resolve().parents[2], + ) + elapsed = time.monotonic() - began + finally: + server.close() + + assert elapsed < 6 + + def test_sync_mode_exports_the_last_window_at_exit(self, posthog_host): + script = textwrap.dedent( + f""" + import sqlite3 + from posthog.client import Client + Client( + "{FAKE_API_KEY}", + host="{posthog_host.url}", + sync_mode=True, + metrics={{ + "autocapture": {{"http": False, "runtime": False}}, + "flush_interval": 3600, + }}, + ) + connection = sqlite3.connect(":memory:") + connection.cursor().execute("SELECT 1") + connection.close() + """ + ) + subprocess.run( + [sys.executable, "-c", script], + check=True, + timeout=60, + cwd=Path(__file__).resolve().parents[2], + ) + + exports = _metrics_exports(posthog_host) + assert exports + assert b"db.client.operation.duration" in b"".join(r["body"] for r in exports) + + def test_before_send_can_call_flush(self, posthog_host): + # Runtime callbacks run while OpenTelemetry holds its collection lock. + script = textwrap.dedent( + f""" + import os, threading + from posthog.client import Client + + def before_send(sample): + if sample["name"] == "process.cpu.time": + client.flush(timeout_seconds=None) + return sample + + client = Client( + "{FAKE_API_KEY}", + host="{posthog_host.url}", + metrics={{ + "autocapture": {{"http": False, "db": False}}, + "flush_interval": 3600, + "before_send": before_send, + }}, + ) + flush = threading.Thread( + target=client.flush, kwargs={{"timeout_seconds": None}}, daemon=True + ) + flush.start() + flush.join(10) + print("deadlock" if flush.is_alive() else "ok", flush=True) + os._exit(0) + """ + ) + result = subprocess.run( + [sys.executable, "-c", script], + capture_output=True, + text=True, + timeout=60, + cwd=Path(__file__).resolve().parents[2], + ) + + assert result.stdout.strip() == "ok" + assert _metrics_exports(posthog_host) + + def _run_django_script(self, posthog_host, body): + script = textwrap.dedent( + f""" + import os + os.environ.pop("DJANGO_SETTINGS_MODULE", None) + from django.conf import settings + from posthog.client import Client + + def start(): + return Client( + "{FAKE_API_KEY}", + host="{posthog_host.url}", + metrics={{"autocapture": {{"db": False, "runtime": False}}}}, + ) + """ + ) + textwrap.dedent(body) + return subprocess.run( + [sys.executable, "-c", script], + capture_output=True, + text=True, + timeout=60, + cwd=Path(__file__).resolve().parents[2], + ) + + def test_does_not_configure_django_settings(self, posthog_host): + # Django apps that configure settings in code do it after imports. + result = self._run_django_script( + posthog_host, + """ + client = start() + settings.configure(DEBUG=True) + print("configured", settings.DEBUG) + client.shutdown() + """, + ) + + assert result.stdout.strip() == "configured True", result.stderr[-2000:] + + def test_instruments_django_once_its_settings_are_configured(self, posthog_host): + # As in AppConfig.ready(). + result = self._run_django_script( + posthog_host, + """ + settings.configure(MIDDLEWARE=[]) + client = start() + print("middleware", settings.MIDDLEWARE) + client.shutdown() + """, + ) + + assert "_DjangoMiddleware" in result.stdout, result.stderr[-2000:] + + def test_a_flush_from_before_send_does_not_repeat(self, posthog_host): + # Each autocapture export runs the hook again. + script = textwrap.dedent( + f""" + import os, time + from posthog.client import Client + + def before_send(sample): + if sample["name"] == "process.cpu.time": + client.flush() + return sample + + client = Client( + "{FAKE_API_KEY}", + host="{posthog_host.url}", + metrics={{ + "autocapture": {{"http": False, "db": False}}, + "flush_interval": 3600, + "before_send": before_send, + }}, + ) + client.flush(timeout_seconds=5) + time.sleep(2) + os._exit(0) + """ + ) + subprocess.run( + [sys.executable, "-c", script], + check=True, + timeout=60, + cwd=Path(__file__).resolve().parents[2], + ) + + assert len(_metrics_exports(posthog_host)) == 1 + + def test_starts_on_a_later_module_setup(self, posthog_host, monkeypatch): + # A Django ready() hook can set the option after setup() already ran. + import posthog + + monkeypatch.setattr(posthog, "default_client", None) + monkeypatch.setattr(posthog, "api_key", FAKE_API_KEY) + monkeypatch.setattr(posthog, "host", posthog_host.url) + monkeypatch.setattr(posthog, "metrics", None) + client = posthog.setup() + try: + assert client._metrics_autocapture is None + monkeypatch.setattr(posthog, "metrics", {"autocapture": True}) + assert posthog.setup() is client + assert client._metrics_autocapture is not None + finally: + client.shutdown() + + def test_module_setup_tries_to_start_once(self, posthog_host, started, monkeypatch): + import posthog + + assert _start(posthog_host, started, InMemoryMetricReader()) is not None + monkeypatch.setattr(posthog, "default_client", None) + monkeypatch.setattr(posthog, "api_key", FAKE_API_KEY) + monkeypatch.setattr(posthog, "host", posthog_host.url) + monkeypatch.setattr(posthog, "metrics", {"autocapture": True}) + with mock.patch("posthog._metrics_autocapture.log") as log: + client = posthog.setup() + posthog.setup() + posthog.setup() + client.shutdown() + + assert log.warning.call_count == 1 diff --git a/pyproject.toml b/pyproject.toml index 851c242d4..11fc790cb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,18 @@ otel = [ "opentelemetry-sdk>=1.20.0", "opentelemetry-exporter-otlp-proto-http>=1.20.0", ] +# Metrics autocapture (`metrics={"autocapture": True}`). Library instrumentors +# (requests, Django, psycopg, ...) are separate packages: `opentelemetry-bootstrap +# -a install` adds the ones that match the installed libraries. +# 0.58b0 is the first release with the `process.*` and `cpython.gc.*` names that +# runtime autocapture configures, and 0.60b0 the first in which the httpx +# instrumentor (installed at the same version) excludes URLs. +metrics = [ + "opentelemetry-sdk>=1.39.0", + "opentelemetry-exporter-otlp-proto-http>=1.39.0", + "opentelemetry-instrumentation>=0.60b0", + "opentelemetry-instrumentation-system-metrics>=0.60b0", +] dev = [ "django-stubs", "griffe", @@ -100,6 +112,14 @@ test = [ "fastmcp>=2.0", "opentelemetry-sdk>=1.20.0", "opentelemetry-exporter-otlp-proto-http>=1.20.0", + "opentelemetry-instrumentation>=0.60b0", + "opentelemetry-instrumentation-system-metrics>=0.60b0", + "opentelemetry-instrumentation-requests>=0.60b0", + "opentelemetry-instrumentation-sqlite3>=0.60b0", + "opentelemetry-instrumentation-wsgi>=0.60b0", + "opentelemetry-instrumentation-flask>=0.60b0", + "opentelemetry-instrumentation-httpx>=0.60b0", + "opentelemetry-instrumentation-django>=0.60b0", "pytest-bdd>=8.1.0", "zstandard>=0.23.0", # gevent 25.4.1+ replaces queue.Queue, exercising the compatibility path. diff --git a/uv.lock b/uv.lock index 17bd3eff0..8b27827aa 100644 --- a/uv.lock +++ b/uv.lock @@ -2800,6 +2800,129 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/9e/73/b7597b2fede14cafe7122d1028bd6f648d9c1e049a61ab36481a4ed4f0df/opentelemetry_instrumentation_celery-0.66b1-py3-none-any.whl", hash = "sha256:145b5eece41331141edc8f454984238b05b107d7ead928b49e7a1916d5e48e99", size = 13480, upload-time = "2026-10-06T17:35:25.828Z" }, ] +[[package]] +name = "opentelemetry-instrumentation-dbapi" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "wrapt" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/23/b4/35c682a3a5762547cab6ac1d0f6e66a4e549cbbb852862903bb1523036fe/opentelemetry_instrumentation_dbapi-0.66b1.tar.gz", hash = "sha256:715be3080f644de2314d84388a73571acdbeca85bfbc7dc682f214c88cc18022", size = 19990, upload-time = "2026-10-06T17:36:22.281Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/77/89/7f2ccd50975b66b2977a31bb489a2f0205b7fd033fc99046dcfa001c4252/opentelemetry_instrumentation_dbapi-0.66b1-py3-none-any.whl", hash = "sha256:a38043a8a907a321ae6b838f778ad19c8e78c0bfc2f336d6d30c5fdce112e88d", size = 14706, upload-time = "2026-10-06T17:35:28.831Z" }, +] + +[[package]] +name = "opentelemetry-instrumentation-django" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-instrumentation-wsgi" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "opentelemetry-util-http" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e9/88/d0acbd7916aa3040a36486cedb60aa77c129c7029446fa6a00207345695c/opentelemetry_instrumentation_django-0.66b1.tar.gz", hash = "sha256:85b56193b6ce7b8db9c280c3ade4a4991601abc9ada4734c8911971423fd521f", size = 25233, upload-time = "2026-10-06T17:36:22.892Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5b/c3/2739c248b60b2802b9e51d2a3373269f0ca3e16059ea4f1ca7370502118c/opentelemetry_instrumentation_django-0.66b1-py3-none-any.whl", hash = "sha256:23d1052940a025ae1dac70849b998cf9ba66731fa2b7e58254ab7dac5539a989", size = 18914, upload-time = "2026-10-06T17:35:30.052Z" }, +] + +[[package]] +name = "opentelemetry-instrumentation-flask" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-instrumentation-wsgi" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "opentelemetry-util-http" }, + { name = "packaging" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c0/fc/c1a692ceb7ed56818f2f4f88044b5a0895b6666da954dbd566fdf265195b/opentelemetry_instrumentation_flask-0.66b1.tar.gz", hash = "sha256:10f788aff95236f0dbaaa066687283bf6345bb3c77bbb437f0fb67cb6222d5eb", size = 23876, upload-time = "2026-10-06T17:36:25.763Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/52/65/7aa45ca677c51afb185bcd5e7ed121c84b07b2c566a72948a9e56fd16ae7/opentelemetry_instrumentation_flask-0.66b1-py3-none-any.whl", hash = "sha256:2933e71b94cc9e8e943e648d872fd3dead9cdb61295dc6deffd5d6fa8f27a39b", size = 14961, upload-time = "2026-10-06T17:35:34.002Z" }, +] + +[[package]] +name = "opentelemetry-instrumentation-httpx" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "opentelemetry-util-http" }, + { name = "wrapt" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/de/50/41544799b043d14fdfa6fe62fa2518eaba22793fce03e9abde930b92e671/opentelemetry_instrumentation_httpx-0.66b1.tar.gz", hash = "sha256:5865a72c68098c85955a271ab8744b480a36e3ee492d35b8cadb93c7c4dbb618", size = 26637, upload-time = "2026-10-06T17:36:27.265Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/4f/c6/e5682b1bfb320b32505e88255c34ae1e99fe9fb220cc465c244e91967eac/opentelemetry_instrumentation_httpx-0.66b1-py3-none-any.whl", hash = "sha256:0342a4002c6dbc6c4bf22cc7e698f50f5c8b77f63325c6f40c94ab87e016bf4d", size = 17723, upload-time = "2026-10-06T17:35:36.501Z" }, +] + +[[package]] +name = "opentelemetry-instrumentation-requests" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "opentelemetry-util-http" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/08/62/dcca0b7a2008675056040c61b70d37b5a24439613c7822068908ad3255ae/opentelemetry_instrumentation_requests-0.66b1.tar.gz", hash = "sha256:28578f72e68e3a5be3226c618ac9570ed360ef1dd22d5d6e0721c6905a4ccc67", size = 17985, upload-time = "2026-10-06T17:36:37.478Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/88/01/98e9bb7fef12784cd19811e454cf25bce4d774fd3e46b3f5fd2bce237075/opentelemetry_instrumentation_requests-0.66b1-py3-none-any.whl", hash = "sha256:7ba17d984a2bd876b88bf0aafcca27b5e3ff4c1821e0fe07641e380a3cf3a3a2", size = 13320, upload-time = "2026-10-06T17:35:52.477Z" }, +] + +[[package]] +name = "opentelemetry-instrumentation-sqlite3" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-instrumentation-dbapi" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/25/54/742ab70c78a1589c29f5513fd5097845b2c2fda58390bbff0f6e1af2601a/opentelemetry_instrumentation_sqlite3-0.66b1.tar.gz", hash = "sha256:e5d672a347deac385a71fb52d556ea443e2f3cc51eac05cac8f0a739cac0bb38", size = 9019, upload-time = "2026-10-06T17:36:38.74Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3b/6a/c36552844c4e3545da7b2931213256c0717665c7d92246a4fe66e088fd9b/opentelemetry_instrumentation_sqlite3-0.66b1-py3-none-any.whl", hash = "sha256:f282b712f56186c36d592d4ab856d978bc7dd60f4b10500113da8c9eb45ff9bc", size = 8526, upload-time = "2026-10-06T17:35:54.362Z" }, +] + +[[package]] +name = "opentelemetry-instrumentation-system-metrics" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "psutil" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d6/81/ab8cabaf45c5248fa88a0d0d3acea4483c105d76cd6c849654ebdab16f92/opentelemetry_instrumentation_system_metrics-0.66b1.tar.gz", hash = "sha256:b0fccbc1aecf35c182431a035ec4809b8f9607eec692243cf70f23b3bda51231", size = 17207, upload-time = "2026-10-06T17:36:40.518Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a3/fd/ea072c512d65ba4054b02c8b7080cf1b362761d4727ee6c1f36dce53328d/opentelemetry_instrumentation_system_metrics-0.66b1-py3-none-any.whl", hash = "sha256:0654b41eb6db3c2f9ad9979487afcc6f461e5b35df282f0a11d0f465e906d967", size = 13913, upload-time = "2026-10-06T17:35:57.01Z" }, +] + +[[package]] +name = "opentelemetry-instrumentation-wsgi" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "opentelemetry-util-http" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/41/7a/a1ba86d95f849db8f41ffd03c7cb302559ef091d76cff78baddd4ad19205/opentelemetry_instrumentation_wsgi-0.66b1.tar.gz", hash = "sha256:ce803d8e828e75d7225c25a618b692cbd4687a11be64356308ed1feab0f70bab", size = 19527, upload-time = "2026-10-06T17:36:44.181Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a5/4f/17f06b0dc7c542c482b1d3f7f1d0925046f7e156fb39c255634b3c4f526a/opentelemetry_instrumentation_wsgi-0.66b1-py3-none-any.whl", hash = "sha256:8c86390c32fe8d0924b3541f292122d3a2ba6c3accc2bcbcc73ae985ee8e799e", size = 13752, upload-time = "2026-10-06T17:36:03.166Z" }, +] + [[package]] name = "opentelemetry-proto" version = "1.45.1" @@ -2839,6 +2962,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/bc/14/67f8aa798857f8cf686f515bf93d9bb877ce952ddc8efae0fa25b45ce0d6/opentelemetry_semantic_conventions-0.66b1-py3-none-any.whl", hash = "sha256:d4cddeb4315490b35213f55e2bdc9ac54bb1e4d318927475bed62b35545e581b", size = 206279, upload-time = "2026-10-06T17:32:56.103Z" }, ] +[[package]] +name = "opentelemetry-util-http" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/7c/b5/df4b61da899f6ebdffdbdf0c8b0f3189ee57151694ccd5b7d50ee2906241/opentelemetry_util_http-0.66b1.tar.gz", hash = "sha256:047dea1a628031f857a5a32261dc0e955bc162d39993ed1cffb8f2cff5ba8a62", size = 11586, upload-time = "2026-10-06T17:36:46.572Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/eb/9b/c77ecaea79ba0de1a11e7f06a7f5eea7043ec23f1860dcf5f03536698e4c/opentelemetry_util_http-0.66b1-py3-none-any.whl", hash = "sha256:8f443d7abcaf29c4a07b373bbd31b5b39132c0ed3c27d015a59dc0323d5b1c58", size = 8387, upload-time = "2026-10-06T17:36:06.984Z" }, +] + [[package]] name = "orjson" version = "3.11.7" @@ -3088,6 +3220,12 @@ dev = [ langchain = [ { name = "langchain" }, ] +metrics = [ + { name = "opentelemetry-exporter-otlp-proto-http" }, + { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-instrumentation-system-metrics" }, + { name = "opentelemetry-sdk" }, +] otel = [ { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-sdk" }, @@ -3116,7 +3254,15 @@ test = [ { name = "mcp" }, { name = "openai-agents" }, { name = "opentelemetry-exporter-otlp-proto-http" }, + { name = "opentelemetry-instrumentation" }, { name = "opentelemetry-instrumentation-celery" }, + { name = "opentelemetry-instrumentation-django" }, + { name = "opentelemetry-instrumentation-flask" }, + { name = "opentelemetry-instrumentation-httpx" }, + { name = "opentelemetry-instrumentation-requests" }, + { name = "opentelemetry-instrumentation-sqlite3" }, + { name = "opentelemetry-instrumentation-system-metrics" }, + { name = "opentelemetry-instrumentation-wsgi" }, { name = "opentelemetry-sdk" }, { name = "parameterized" }, { name = "pydantic" }, @@ -3171,9 +3317,21 @@ requires-dist = [ { name = "mypy", marker = "extra == 'dev'" }, { name = "mypy-baseline", marker = "extra == 'dev'" }, { name = "openai-agents", marker = "extra == 'test'", specifier = ">=0.18" }, + { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'metrics'", specifier = ">=1.39.0" }, { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'otel'", specifier = ">=1.20.0" }, { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'test'", specifier = ">=1.20.0" }, + { name = "opentelemetry-instrumentation", marker = "extra == 'metrics'", specifier = ">=0.60b0" }, + { name = "opentelemetry-instrumentation", marker = "extra == 'test'", specifier = ">=0.60b0" }, { name = "opentelemetry-instrumentation-celery", marker = "extra == 'test'", specifier = ">=0.41b0" }, + { name = "opentelemetry-instrumentation-django", marker = "extra == 'test'", specifier = ">=0.60b0" }, + { name = "opentelemetry-instrumentation-flask", marker = "extra == 'test'", specifier = ">=0.60b0" }, + { name = "opentelemetry-instrumentation-httpx", marker = "extra == 'test'", specifier = ">=0.60b0" }, + { name = "opentelemetry-instrumentation-requests", marker = "extra == 'test'", specifier = ">=0.60b0" }, + { name = "opentelemetry-instrumentation-sqlite3", marker = "extra == 'test'", specifier = ">=0.60b0" }, + { name = "opentelemetry-instrumentation-system-metrics", marker = "extra == 'metrics'", specifier = ">=0.60b0" }, + { name = "opentelemetry-instrumentation-system-metrics", marker = "extra == 'test'", specifier = ">=0.60b0" }, + { name = "opentelemetry-instrumentation-wsgi", marker = "extra == 'test'", specifier = ">=0.60b0" }, + { name = "opentelemetry-sdk", marker = "extra == 'metrics'", specifier = ">=1.39.0" }, { name = "opentelemetry-sdk", marker = "extra == 'otel'", specifier = ">=1.20.0" }, { name = "opentelemetry-sdk", marker = "extra == 'test'", specifier = ">=1.20.0" }, { name = "packaging", marker = "extra == 'dev'" }, @@ -3201,7 +3359,7 @@ requires-dist = [ { name = "zstandard", marker = "extra == 'test'", specifier = ">=0.23.0" }, { name = "zstandard", marker = "extra == 'zstd'", specifier = ">=0.23.0" }, ] -provides-extras = ["langchain", "async", "zstd", "otel", "dev", "test"] +provides-extras = ["langchain", "async", "zstd", "otel", "metrics", "dev", "test"] [package.metadata.requires-dev] dev = [{ name = "claude-agent-sdk", specifier = ">=0.1.50" }] @@ -3338,6 +3496,34 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c4/72/02445137af02769918a93807b2b7890047c32bfb9f90371cbc12688819eb/protobuf-6.33.6-py3-none-any.whl", hash = "sha256:77179e006c476e69bf8e8ce866640091ec42e1beb80b213c3900006ecfba6901", size = 170656, upload-time = "2026-03-18T19:04:59.826Z" }, ] +[[package]] +name = "psutil" +version = "7.2.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/aa/c6/d1ddf4abb55e93cebc4f2ed8b5d6dbad109ecb8d63748dd2b20ab5e57ebe/psutil-7.2.2.tar.gz", hash = "sha256:0746f5f8d406af344fd547f1c8daa5f5c33dbc293bb8d6a16d80b4bb88f59372", size = 493740, upload-time = "2026-01-28T18:14:54.428Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/51/08/510cbdb69c25a96f4ae523f733cdc963ae654904e8db864c07585ef99875/psutil-7.2.2-cp313-cp313t-macosx_10_13_x86_64.whl", hash = "sha256:2edccc433cbfa046b980b0df0171cd25bcaeb3a68fe9022db0979e7aa74a826b", size = 130595, upload-time = "2026-01-28T18:14:57.293Z" }, + { url = "https://files.pythonhosted.org/packages/d6/f5/97baea3fe7a5a9af7436301f85490905379b1c6f2dd51fe3ecf24b4c5fbf/psutil-7.2.2-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:e78c8603dcd9a04c7364f1a3e670cea95d51ee865e4efb3556a3a63adef958ea", size = 131082, upload-time = "2026-01-28T18:14:59.732Z" }, + { url = "https://files.pythonhosted.org/packages/37/d6/246513fbf9fa174af531f28412297dd05241d97a75911ac8febefa1a53c6/psutil-7.2.2-cp313-cp313t-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1a571f2330c966c62aeda00dd24620425d4b0cc86881c89861fbc04549e5dc63", size = 181476, upload-time = "2026-01-28T18:15:01.884Z" }, + { url = "https://files.pythonhosted.org/packages/b8/b5/9182c9af3836cca61696dabe4fd1304e17bc56cb62f17439e1154f225dd3/psutil-7.2.2-cp313-cp313t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:917e891983ca3c1887b4ef36447b1e0873e70c933afc831c6b6da078ba474312", size = 184062, upload-time = "2026-01-28T18:15:04.436Z" }, + { url = "https://files.pythonhosted.org/packages/16/ba/0756dca669f5a9300d0cbcbfae9a4c30e446dfc7440ffe43ded5724bfd93/psutil-7.2.2-cp313-cp313t-win_amd64.whl", hash = "sha256:ab486563df44c17f5173621c7b198955bd6b613fb87c71c161f827d3fb149a9b", size = 139893, upload-time = "2026-01-28T18:15:06.378Z" }, + { url = "https://files.pythonhosted.org/packages/1c/61/8fa0e26f33623b49949346de05ec1ddaad02ed8ba64af45f40a147dbfa97/psutil-7.2.2-cp313-cp313t-win_arm64.whl", hash = "sha256:ae0aefdd8796a7737eccea863f80f81e468a1e4cf14d926bd9b6f5f2d5f90ca9", size = 135589, upload-time = "2026-01-28T18:15:08.03Z" }, + { url = "https://files.pythonhosted.org/packages/81/69/ef179ab5ca24f32acc1dac0c247fd6a13b501fd5534dbae0e05a1c48b66d/psutil-7.2.2-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:eed63d3b4d62449571547b60578c5b2c4bcccc5387148db46e0c2313dad0ee00", size = 130664, upload-time = "2026-01-28T18:15:09.469Z" }, + { url = "https://files.pythonhosted.org/packages/7b/64/665248b557a236d3fa9efc378d60d95ef56dd0a490c2cd37dafc7660d4a9/psutil-7.2.2-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:7b6d09433a10592ce39b13d7be5a54fbac1d1228ed29abc880fb23df7cb694c9", size = 131087, upload-time = "2026-01-28T18:15:11.724Z" }, + { url = "https://files.pythonhosted.org/packages/d5/2e/e6782744700d6759ebce3043dcfa661fb61e2fb752b91cdeae9af12c2178/psutil-7.2.2-cp314-cp314t-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1fa4ecf83bcdf6e6c8f4449aff98eefb5d0604bf88cb883d7da3d8d2d909546a", size = 182383, upload-time = "2026-01-28T18:15:13.445Z" }, + { url = "https://files.pythonhosted.org/packages/57/49/0a41cefd10cb7505cdc04dab3eacf24c0c2cb158a998b8c7b1d27ee2c1f5/psutil-7.2.2-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:e452c464a02e7dc7822a05d25db4cde564444a67e58539a00f929c51eddda0cf", size = 185210, upload-time = "2026-01-28T18:15:16.002Z" }, + { url = "https://files.pythonhosted.org/packages/dd/2c/ff9bfb544f283ba5f83ba725a3c5fec6d6b10b8f27ac1dc641c473dc390d/psutil-7.2.2-cp314-cp314t-win_amd64.whl", hash = "sha256:c7663d4e37f13e884d13994247449e9f8f574bc4655d509c3b95e9ec9e2b9dc1", size = 141228, upload-time = "2026-01-28T18:15:18.385Z" }, + { url = "https://files.pythonhosted.org/packages/f2/fc/f8d9c31db14fcec13748d373e668bc3bed94d9077dbc17fb0eebc073233c/psutil-7.2.2-cp314-cp314t-win_arm64.whl", hash = "sha256:11fe5a4f613759764e79c65cf11ebdf26e33d6dd34336f8a337aa2996d71c841", size = 136284, upload-time = "2026-01-28T18:15:19.912Z" }, + { url = "https://files.pythonhosted.org/packages/e7/36/5ee6e05c9bd427237b11b3937ad82bb8ad2752d72c6969314590dd0c2f6e/psutil-7.2.2-cp36-abi3-macosx_10_9_x86_64.whl", hash = "sha256:ed0cace939114f62738d808fdcecd4c869222507e266e574799e9c0faa17d486", size = 129090, upload-time = "2026-01-28T18:15:22.168Z" }, + { url = "https://files.pythonhosted.org/packages/80/c4/f5af4c1ca8c1eeb2e92ccca14ce8effdeec651d5ab6053c589b074eda6e1/psutil-7.2.2-cp36-abi3-macosx_11_0_arm64.whl", hash = "sha256:1a7b04c10f32cc88ab39cbf606e117fd74721c831c98a27dc04578deb0c16979", size = 129859, upload-time = "2026-01-28T18:15:23.795Z" }, + { url = "https://files.pythonhosted.org/packages/b5/70/5d8df3b09e25bce090399cf48e452d25c935ab72dad19406c77f4e828045/psutil-7.2.2-cp36-abi3-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:076a2d2f923fd4821644f5ba89f059523da90dc9014e85f8e45a5774ca5bc6f9", size = 155560, upload-time = "2026-01-28T18:15:25.976Z" }, + { url = "https://files.pythonhosted.org/packages/63/65/37648c0c158dc222aba51c089eb3bdfa238e621674dc42d48706e639204f/psutil-7.2.2-cp36-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:b0726cecd84f9474419d67252add4ac0cd9811b04d61123054b9fb6f57df6e9e", size = 156997, upload-time = "2026-01-28T18:15:27.794Z" }, + { url = "https://files.pythonhosted.org/packages/8e/13/125093eadae863ce03c6ffdbae9929430d116a246ef69866dad94da3bfbc/psutil-7.2.2-cp36-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:fd04ef36b4a6d599bbdb225dd1d3f51e00105f6d48a28f006da7f9822f2606d8", size = 148972, upload-time = "2026-01-28T18:15:29.342Z" }, + { url = "https://files.pythonhosted.org/packages/04/78/0acd37ca84ce3ddffaa92ef0f571e073faa6d8ff1f0559ab1272188ea2be/psutil-7.2.2-cp36-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:b58fabe35e80b264a4e3bb23e6b96f9e45a3df7fb7eed419ac0e5947c61e47cc", size = 148266, upload-time = "2026-01-28T18:15:31.597Z" }, + { url = "https://files.pythonhosted.org/packages/b4/90/e2159492b5426be0c1fef7acba807a03511f97c5f86b3caeda6ad92351a7/psutil-7.2.2-cp37-abi3-win_amd64.whl", hash = "sha256:eb7e81434c8d223ec4a219b5fc1c47d0417b12be7ea866e24fb5ad6e84b3d988", size = 137737, upload-time = "2026-01-28T18:15:33.849Z" }, + { url = "https://files.pythonhosted.org/packages/8c/c7/7bb2e321574b10df20cbde462a94e2b71d05f9bbda251ef27d104668306a/psutil-7.2.2-cp37-abi3-win_arm64.whl", hash = "sha256:8c233660f575a5a89e6d4cb65d9f938126312bca76d8fe087b947b3a1aaac9ee", size = 134617, upload-time = "2026-01-28T18:15:36.514Z" }, +] + [[package]] name = "py-key-value-aio" version = "0.4.5"