HTTPX integration for OpenTelemetry (Proof of Concept)

async_example.py

from opentelemetry.instrumentation import httpx_client as httpx_opentelemetry

async def main():
    opentelemetry_on_request, opentelemetry_on_response = httpx_opentelemetry.create_async_hooks()

    event_hooks = {
        "request": [opentelemetry_on_request],
        "response": [opentelemetry_on_response]
    }

    async with httpx.AsyncClient(event_hooks=event_hooks) as client:
        ...

example.py

from opentelemetry.instrumentation import httpx_client as httpx_opentelemetry

opentelemetry_on_request, opentelemetry_on_response = httpx_opentelemetry.create_hooks()

event_hooks = {
    "request": [opentelemetry_on_request],
    "response": [opentelemetry_on_response]
}

with httpx.Client(event_hooks=event_hooks) as client:
    ...

httpx_opentelemetry.py

from typing import Callable, Tuple

import httpx

from opentelemetry import context as context_api, propagators, trace
from opentelemetry.instrumentation.utils import http_status_to_canonical_code


def create_hooks() -> Tuple[Callable, Callable]:
    tracer = trace.get_tracer_provider().get_tracer(__name__, "0.0.1")

    def on_request(request: httpx.Request) -> None:
        # Drop query parameters (could contain sensitive data).
        url = request.url.copy_with(query="")

        span_name = str(url)
        span = tracer.start_span(
            span_name,
            kind=trace.SpanKind.CLIENT,
            attributes={
                "component": "http",
                "http.method": request.method.upper(),
                "http.url": str(url),
            },
        )

        token = context_api.attach(trace.set_span_in_context(span))
        propagators.inject(type(request.headers).__setitem__, request.headers)

        context = {
            "span_name": span_name,
            "span": span,
            "token": token,
        }

        # Store for access in response hook.
        # XXX: This may not work if the `request` object gets swapped with another
        # (e.g. due to authentication or redirects).
        request.__opentelemetry_context = context  # type: ignore

    def on_response(response: httpx.Response, request: httpx.Request) -> None:
        context = getattr(request, "__opentelemetry_context", None)

        if context is None:
            return

        span: trace.Span = context["span"]
        span.set_status(
            trace.status.Status(http_status_to_canonical_code(response.status_code))
        )
        span.set_attribute("http.status_code", response.status_code)
        span.set_attribute("http.status_text", response.reason_phrase)

        # End span.
        token: str = context["token"]
        context_api.detach(token)
        span.end()

    return on_request, on_response


def create_async_request_hook() -> Tuple[Callable, Callable]:
    sync_on_request, sync_on_response = create_hooks()

    async def on_request(request: httpx.Request) -> None:
        sync_on_request(request)

    async def on_response(response: httpx.Response, request: httpx.Request) -> None:
        sync_on_response(response, request)

    return on_request, on_response