feat(http): add client instrumentation and metrics

nemoguardrails/http/__init__.py

@@ -23,6 +23,7 @@
     HTTPStatusError,
     HTTPTimeoutError,
 )
+from nemoguardrails.http.instrumented import InstrumentedHTTPClient, instrument_http_client
 from nemoguardrails.http.request import http_call
 from nemoguardrails.http.retry import RetryingHTTPClient, RetryPolicy
 from nemoguardrails.http.runtime import create_http_client
@@ -40,8 +41,10 @@
     "HTTPStatusError",
     "HTTPTimeoutError",
     "HttpxHTTPClient",
+    "InstrumentedHTTPClient",
     "RetryPolicy",
     "RetryingHTTPClient",
     "create_http_client",
     "http_call",
+    "instrument_http_client",
 ]

nemoguardrails/http/instrumented.py

@@ -0,0 +1,141 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import warnings
+from contextlib import nullcontext
+from typing import TYPE_CHECKING, Any, Mapping, overload
+
+from nemoguardrails.http.client import ClosableHTTPClient, HTTPClient
+from nemoguardrails.http.telemetry import (
+    http_call_span,
+    http_request_duration,
+    set_http_response_attributes,
+)
+from nemoguardrails.http.types import HTTPResponse
+
+if TYPE_CHECKING:
+    from opentelemetry.trace import Tracer
+
+
+class InstrumentedHTTPClient:
+    def __new__(cls, client: HTTPClient, *args: Any, **kwargs: Any):
+        if isinstance(client, cls):
+            return client
+        return super().__new__(cls)
+
+    def __init__(
+        self,
+        client: HTTPClient,
+        tracer: "Tracer | None",
+        *,
+        metrics_enabled: bool = False,
+    ):
+        if client is self:
+            if tracer is not self._tracer or metrics_enabled != self._metrics_enabled:
+                warnings.warn(
+                    "InstrumentedHTTPClient is already instrumented; new instrumentation "
+                    "settings are ignored. Re-instrument the underlying wrapped_client instead.",
+                    stacklevel=2,
+                )
+            return
+        self._client = client
+        self._tracer = tracer
+        self._metrics_enabled = metrics_enabled
+        self._closed = False
+
+    @property
+    def wrapped_client(self) -> HTTPClient:
+        return self._client
+
+    async def request(
+        self,
+        method: str,
+        url: str,
+        *,
+        headers: Mapping[str, str] | None = None,
+        params: Mapping[str, Any] | None = None,
+        json: Any = None,
+        content: bytes | str | None = None,
+        timeout: float | None = None,
+    ) -> HTTPResponse:
+        if self._tracer is None and not self._metrics_enabled:
+            return await self._client.request(
+                method,
+                url,
+                headers=headers,
+                params=params,
+                json=json,
+                content=content,
+                timeout=timeout,
+            )
+
+        normalized_method = method.upper()
+        with http_call_span(self._tracer, normalized_method, url, content) as span:
+            duration = http_request_duration(normalized_method, url) if self._metrics_enabled else nullcontext(None)
+            with duration as metric_state:
+                response = await self._client.request(
+                    method,
+                    url,
+                    headers=headers,
+                    params=params,
+                    json=json,
+                    content=content,
+                    timeout=timeout,
+                )
+                if metric_state is not None:
+                    metric_state.response_status_code = response.status_code
+            set_http_response_attributes(span, response)
+            return response
+
+    async def close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if isinstance(self._client, ClosableHTTPClient):
+            await self._client.close()
+
+
+@overload
+def instrument_http_client(
+    client: ClosableHTTPClient,
+    *,
+    tracer: "Tracer | None" = None,
+    metrics_enabled: bool = False,
+) -> ClosableHTTPClient: ...
+
+
+@overload
+def instrument_http_client(
+    client: HTTPClient,
+    *,
+    tracer: "Tracer | None" = None,
+    metrics_enabled: bool = False,
+) -> HTTPClient: ...
+
+
+def instrument_http_client(
+    client: HTTPClient,
+    *,
+    tracer: "Tracer | None" = None,
+    metrics_enabled: bool = False,
+) -> HTTPClient:
+    if isinstance(client, InstrumentedHTTPClient):
+        return client
+    if tracer is None and not metrics_enabled:
+        return client
+    return InstrumentedHTTPClient(client, tracer, metrics_enabled=metrics_enabled)
+
+
+__all__ = ["InstrumentedHTTPClient", "instrument_http_client"]

nemoguardrails/http/telemetry.py

@@ -0,0 +1,182 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import time
+from contextlib import contextmanager, suppress
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Generator
+
+from nemoguardrails.http._url import sanitize_url, split_url
+from nemoguardrails.http.types import HTTPResponse
+from nemoguardrails.tracing.constants import HTTPAttributes, _ensure_http_instruments
+
+if TYPE_CHECKING:
+    from opentelemetry.trace import Span, Tracer
+
+log = logging.getLogger(__name__)
+
+
+@dataclass(slots=True)
+class _HTTPRequestDurationState:
+    response_status_code: int | None = None
+
+
+def _http_metric_attributes(method: str, url: str) -> dict[str, str | int] | None:
+    parts = split_url(url)
+    if parts.hostname is None:
+        return None
+    port = parts.port
+    if port is None:
+        port = {"http": 80, "https": 443}.get(parts.scheme)
+    if port is None:
+        return None
+    return {
+        HTTPAttributes.REQUEST_METHOD: method,
+        HTTPAttributes.SERVER_ADDRESS: parts.hostname,
+        HTTPAttributes.SERVER_PORT: port,
+    }
+
+
+@contextmanager
+def http_request_duration(
+    method: str,
+    url: str,
+) -> Generator[_HTTPRequestDurationState, None, None]:
+    state = _HTTPRequestDurationState()
+    try:
+        instruments = _ensure_http_instruments()
+    except Exception:
+        instruments = None
+    if instruments is None:
+        yield state
+        return
+
+    started_at = time.monotonic()
+    error_type: str | None = None
+    try:
+        yield state
+    except BaseException as error:
+        error_type = type(error).__name__
+        raise
+    finally:
+        with suppress(Exception):
+            attributes = _http_metric_attributes(method, url)
+            if attributes is not None:
+                if state.response_status_code is not None:
+                    attributes[HTTPAttributes.RESPONSE_STATUS_CODE] = state.response_status_code
+                    if state.response_status_code >= 400:
+                        error_type = str(state.response_status_code)
+                if error_type is not None:
+                    attributes[HTTPAttributes.ERROR_TYPE] = error_type
+                instruments.request_duration.record(
+                    time.monotonic() - started_at,
+                    attributes=attributes,
+                )
+
+
+def set_http_request_attributes(
+    span: "Span | None",
+    method: str,
+    url: str,
+    content: bytes | str | None,
+) -> None:
+    if span is None:
+        return
+    with suppress(Exception):
+        parts = split_url(url)
+        span.set_attribute(HTTPAttributes.REQUEST_METHOD, method)
+        span.set_attribute(HTTPAttributes.URL_FULL, sanitize_url(url))
+        if parts.scheme:
+            span.set_attribute(HTTPAttributes.URL_SCHEME, parts.scheme)
+        if parts.hostname:
+            span.set_attribute(HTTPAttributes.SERVER_ADDRESS, parts.hostname)
+        with suppress(ValueError):
+            if parts.port is not None:
+                span.set_attribute(HTTPAttributes.SERVER_PORT, parts.port)
+        if content is not None:
+            size = len(content) if isinstance(content, bytes) else len(content.encode())
+            span.set_attribute(HTTPAttributes.REQUEST_BODY_SIZE, size)
+
+
+def set_http_response_attributes(span: "Span | None", response: HTTPResponse) -> None:
+    if span is None:
+        return
+    with suppress(Exception):
+        from opentelemetry.trace import StatusCode
+
+        span.set_attribute(HTTPAttributes.RESPONSE_STATUS_CODE, response.status_code)
+        span.set_attribute(HTTPAttributes.RESPONSE_BODY_SIZE, len(response.content))
+        retry_count = response.extensions.get("retry_count")
+        if isinstance(retry_count, int) and retry_count > 0:
+            span.set_attribute(HTTPAttributes.REQUEST_RESEND_COUNT, retry_count)
+        if response.status_code >= 400:
+            span.set_attribute(HTTPAttributes.ERROR_TYPE, str(response.status_code))
+            span.set_status(StatusCode.ERROR)
+
+
+def record_http_error(span: "Span | None", error: BaseException) -> None:
+    if span is None:
+        return
+    from opentelemetry.trace import StatusCode
+
+    try:
+        span.set_attribute(HTTPAttributes.ERROR_TYPE, type(error).__name__)
+        retry_count = getattr(error, "retry_count", 0)
+        if isinstance(retry_count, int) and retry_count > 0:
+            span.set_attribute(HTTPAttributes.REQUEST_RESEND_COUNT, retry_count)
+        span.add_event("exception", {HTTPAttributes.EXCEPTION_TYPE: type(error).__name__})
+        span.set_status(StatusCode.ERROR)
+    except Exception as telemetry_error:
+        log.warning(
+            "Failed to record HTTP error telemetry: %s",
+            type(telemetry_error).__name__,
+        )
+
+
+@contextmanager
+def http_call_span(
+    tracer: "Tracer | None",
+    method: str,
+    url: str,
+    content: bytes | str | None,
+) -> Generator["Span | None", None, None]:
+    if tracer is None:
+        yield None
+        return
+
+    from opentelemetry.trace import SpanKind
+
+    with tracer.start_as_current_span(
+        f"HTTP {method}",
+        kind=SpanKind.CLIENT,
+        record_exception=False,
+        set_status_on_exception=False,
+    ) as span:
+        set_http_request_attributes(span, method, url, content)
+        try:
+            yield span
+        except BaseException as error:
+            record_http_error(span, error)
+            raise
+
+
+__all__ = [
+    "http_call_span",
+    "http_request_duration",
+    "record_http_error",
+    "set_http_request_attributes",
+    "set_http_response_attributes",
+]