From 086f58f0dd52c499185e1e7c6741a106c955c0f8 Mon Sep 17 00:00:00 2001 From: Kiell Tampubolon <93207632+glatinone@users.noreply.github.com> Date: Tue, 28 Jul 2026 14:02:41 +0800 Subject: [PATCH] feat(exporter): add HTTP/2 support and exponential backoff retries for OTLP HTTP exporter (#3833) --- .../otlp/proto/http/_common/__init__.py | 38 ++++++++- .../proto/http/_common/_transport_httpx.py | 83 +++++++++++++++++++ .../otlp/proto/http/_log_exporter/__init__.py | 44 +++++++--- .../proto/http/metric_exporter/__init__.py | 44 +++++++--- .../proto/http/trace_exporter/__init__.py | 43 +++++++--- .../tests/test_http2_transport.py | 44 ++++++++++ 6 files changed, 258 insertions(+), 38 deletions(-) create mode 100644 exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/_transport_httpx.py create mode 100644 exporter/opentelemetry-exporter-otlp-proto-http/tests/test_http2_transport.py diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/__init__.py b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/__init__.py index 46db16dd86..7a504b9cb8 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/__init__.py +++ b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/__init__.py @@ -2,9 +2,11 @@ # SPDX-License-Identifier: Apache-2.0 from os import environ -from typing import Literal +from typing import Literal, Mapping import requests +from email.utils import parsedate_to_datetime +from datetime import datetime, timezone from opentelemetry.sdk.environment_variables import ( _OTEL_PYTHON_EXPORTER_OTLP_HTTP_CREDENTIAL_PROVIDER, @@ -24,13 +26,43 @@ class RequestPayloadTooLargeError(Exception): def _is_retryable(resp: requests.Response) -> bool: - if resp.status_code == 408: + if resp.status_code in (408, 429): return True - if resp.status_code >= 500 and resp.status_code <= 599: + if 500 <= resp.status_code <= 599: return True return False +def _get_retry_after_seconds(headers: Mapping[str, str] | None) -> float | None: + """Parse Retry-After header into seconds if present. + + Supports both delta-seconds and HTTP-date formats. + """ + if not headers: + return None + value = headers.get("Retry-After") + if not value: + return None + value = value.strip() + # delta-seconds + if value.isdigit(): + try: + seconds = int(value) + return max(0, float(seconds)) + except Exception: + return None + # HTTP-date + try: + dt = parsedate_to_datetime(value) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + now = datetime.now(timezone.utc) + delta = (dt - now).total_seconds() + return max(0.0, delta) + except Exception: + return None + + def _is_request_too_large( serialized_data: bytes, max_request_size: int ) -> bool: diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/_transport_httpx.py b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/_transport_httpx.py new file mode 100644 index 0000000000..95ec2531b4 --- /dev/null +++ b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/_transport_httpx.py @@ -0,0 +1,83 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 +from __future__ import annotations + +import typing as _t + +try: + import httpx # type: ignore +except Exception: # pragma: no cover - optional dependency + httpx = None # type: ignore + +import requests + + +class _ResponseAdapter: + def __init__(self, resp: "httpx.Response") -> None: # type: ignore[name-defined] + self._resp = resp + self.ok: bool = resp.is_success + self.status_code: int = resp.status_code + # reason_phrase is available on httpx.Response + self.reason: str = getattr(resp, "reason_phrase", "") + self.headers: _t.Mapping[str, str] = resp.headers + + +class HttpxSession: + """Minimal requests-compatible session backed by httpx.Client. + + - Exposes a dict-like ``headers`` attribute for parity with requests.Session. + - Provides ``post`` and ``close`` methods used by OTLP HTTP exporters. + - Negotiates HTTP/2 when available, falls back to HTTP/1.1 on negotiation errors. + """ + + def __init__(self) -> None: + if httpx is None: # pragma: no cover - guarded by importorskip in tests + raise RuntimeError("httpx is not available") + self.headers: dict[str, str] = {} + self._client: "httpx.Client | None" = None # type: ignore[name-defined] + self._http2_enabled: bool = True + + def _ensure_client(self, verify: _t.Any, cert: _t.Any, timeout: float) -> None: + if self._client is None: + # Create client lazily to honor any header updates performed before the first request + self._client = httpx.Client( # type: ignore[attr-defined] + http2=self._http2_enabled, + headers=self.headers.copy(), + verify=verify, + cert=cert, + timeout=timeout, + ) + + def post( + self, + url: str, + data: bytes, + verify: _t.Any, + timeout: float, + cert: _t.Any, + ) -> _ResponseAdapter: + self._ensure_client(verify, cert, timeout) + try: + resp = self._client.post(url, content=data) # type: ignore[union-attr] + return _ResponseAdapter(resp) + except Exception as exc: # httpx.HTTPError and transport errors + # Fallback to HTTP/1.1 on first HTTP/2 failure, then re-raise as requests exception + if self._http2_enabled: + self._http2_enabled = False + try: + if self._client is not None: + self._client.close() + self._client = None + self._ensure_client(verify, cert, timeout) + resp = self._client.post(url, content=data) # type: ignore[union-attr] + return _ResponseAdapter(resp) + except Exception as exc2: + raise requests.exceptions.RequestException(str(exc2)) from exc2 + raise requests.exceptions.RequestException(str(exc)) from exc + + def close(self) -> None: + if self._client is not None: + try: + self._client.close() + finally: + self._client = None diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py index c25ebec675..26b226250d 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py +++ b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py @@ -27,6 +27,7 @@ from opentelemetry.exporter.otlp.proto.http._common import ( _DEFAULT_MAX_REQUEST_SIZE, RequestPayloadTooLargeError, + _get_retry_after_seconds, _is_request_too_large, _is_retryable, _load_session_from_envvar, @@ -157,13 +158,33 @@ def __init__( else max_request_size ) self._compression = compression or _compression_from_env() - self._session = ( - session - or _load_session_from_envvar( - _OTEL_PYTHON_EXPORTER_OTLP_HTTP_LOGS_CREDENTIAL_PROVIDER - ) - or requests.Session() + + session_from_env = _load_session_from_envvar( + _OTEL_PYTHON_EXPORTER_OTLP_HTTP_LOGS_CREDENTIAL_PROVIDER ) + if session is not None: + self._session = session + elif session_from_env is not None: + self._session = session_from_env + else: + use_httpx = ( + os.environ.get("OTEL_EXPORTER_OTLP_HTTP_TRANSPORT", "") + .strip() + .lower() + == "httpx" + ) + if use_httpx: + try: + from opentelemetry.exporter.otlp.proto.http._common._transport_httpx import ( + HttpxSession, + ) + + self._session = HttpxSession() + except Exception: + self._session = requests.Session() + else: + self._session = requests.Session() + self._session.headers.update(self._headers) self._session.headers.update(_OTLP_HTTP_HEADERS) # let users override our defaults @@ -200,10 +221,6 @@ def _export( if timeout_sec is None: timeout_sec = self._timeout - # By default, keep-alive is enabled in Session's request - # headers. Backends may choose to close the connection - # while a post happens which causes an unhandled - # exception. This try/except will retry the post on such exceptions try: resp = self._session.post( url=self._endpoint, @@ -246,8 +263,7 @@ def export( return LogRecordExportResult.FAILURE deadline_sec = time() + self._timeout for retry_num in range(_MAX_RETRYS): - # multiplying by a random number between .8 and 1.2 introduces a +/20% jitter to each backoff. - backoff_seconds = 2**retry_num * random.uniform(0.8, 1.2) + base_backoff = 2**retry_num * random.uniform(0.8, 1.2) export_error: Exception | None = None try: resp = self._export(serialized_data, deadline_sec - time()) @@ -258,10 +274,14 @@ def export( export_error = error retryable = isinstance(error, ConnectionError) status_code = None + retry_after = None else: reason = resp.reason retryable = _is_retryable(resp) status_code = resp.status_code + retry_after = _get_retry_after_seconds(getattr(resp, "headers", None)) + + backoff_seconds = base_backoff if retry_after is None else max(base_backoff, retry_after) if not retryable: _logger.error( diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py index 7020beb7f3..78387784c4 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py +++ b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py @@ -41,6 +41,7 @@ from opentelemetry.exporter.otlp.proto.http._common import ( _DEFAULT_MAX_REQUEST_SIZE, RequestPayloadTooLargeError, + _get_retry_after_seconds, _is_request_too_large, _is_retryable, _load_session_from_envvar, @@ -196,13 +197,33 @@ def __init__( ) ) self._compression = compression or _compression_from_env() - self._session = ( - session - or _load_session_from_envvar( - _OTEL_PYTHON_EXPORTER_OTLP_HTTP_METRICS_CREDENTIAL_PROVIDER - ) - or requests.Session() + + session_from_env = _load_session_from_envvar( + _OTEL_PYTHON_EXPORTER_OTLP_HTTP_METRICS_CREDENTIAL_PROVIDER ) + if session is not None: + self._session = session + elif session_from_env is not None: + self._session = session_from_env + else: + use_httpx = ( + os.environ.get("OTEL_EXPORTER_OTLP_HTTP_TRANSPORT", "") + .strip() + .lower() + == "httpx" + ) + if use_httpx: + try: + from opentelemetry.exporter.otlp.proto.http._common._transport_httpx import ( + HttpxSession, + ) + + self._session = HttpxSession() + except Exception: + self._session = requests.Session() + else: + self._session = requests.Session() + self._session.headers.update(self._headers) self._session.headers.update(_OTLP_HTTP_HEADERS) # let users override our defaults @@ -249,10 +270,6 @@ def _export( if timeout_sec is None: timeout_sec = self._timeout - # By default, keep-alive is enabled in Session's request - # headers. Backends may choose to close the connection - # while a post happens which causes an unhandled - # exception. This try/except will retry the post on such exceptions try: resp = self._session.post( url=self._endpoint, @@ -303,8 +320,7 @@ def _export_with_retries( return MetricExportResult.FAILURE deadline_sec = time() + self._timeout for retry_num in range(_MAX_RETRYS): - # multiplying by a random number between .8 and 1.2 introduces a +/20% jitter to each backoff. - backoff_seconds = 2**retry_num * random.uniform(0.8, 1.2) + base_backoff = 2**retry_num * random.uniform(0.8, 1.2) export_error: Exception | None = None try: resp = self._export(serialized_data, deadline_sec - time()) @@ -315,10 +331,14 @@ def _export_with_retries( export_error = error retryable = isinstance(error, ConnectionError) status_code = None + retry_after = None else: reason = resp.reason retryable = _is_retryable(resp) status_code = resp.status_code + retry_after = _get_retry_after_seconds(getattr(resp, "headers", None)) + + backoff_seconds = base_backoff if retry_after is None else max(base_backoff, retry_after) if not retryable: _logger.error( diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py index 56d0a92a9e..79f2101e59 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py +++ b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py @@ -29,6 +29,7 @@ from opentelemetry.exporter.otlp.proto.http._common import ( _DEFAULT_MAX_REQUEST_SIZE, RequestPayloadTooLargeError, + _get_retry_after_seconds, _is_request_too_large, _is_retryable, _load_session_from_envvar, @@ -152,13 +153,33 @@ def __init__( else max_request_size ) self._compression = compression or _compression_from_env() - self._session = ( - session - or _load_session_from_envvar( - _OTEL_PYTHON_EXPORTER_OTLP_HTTP_TRACES_CREDENTIAL_PROVIDER - ) - or requests.Session() + + session_from_env = _load_session_from_envvar( + _OTEL_PYTHON_EXPORTER_OTLP_HTTP_TRACES_CREDENTIAL_PROVIDER ) + if session is not None: + self._session = session + elif session_from_env is not None: + self._session = session_from_env + else: + use_httpx = ( + os.environ.get("OTEL_EXPORTER_OTLP_HTTP_TRANSPORT", "") + .strip() + .lower() + == "httpx" + ) + if use_httpx: + try: + from opentelemetry.exporter.otlp.proto.http._common._transport_httpx import ( + HttpxSession, + ) + + self._session = HttpxSession() + except Exception: + self._session = requests.Session() + else: + self._session = requests.Session() + self._session.headers.update(self._headers) self._session.headers.update(_OTLP_HTTP_HEADERS) # let users override our defaults @@ -195,10 +216,6 @@ def _export( if timeout_sec is None: timeout_sec = self._timeout - # By default, keep-alive is enabled in Session's request - # headers. Backends may choose to close the connection - # while a post happens which causes an unhandled - # exception. This try/except will retry the post on such exceptions try: resp = self._session.post( url=self._endpoint, @@ -240,7 +257,7 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: deadline_sec = time() + self._timeout for retry_num in range(_MAX_RETRYS): # multiplying by a random number between .8 and 1.2 introduces a +/20% jitter to each backoff. - backoff_seconds = 2**retry_num * random.uniform(0.8, 1.2) + base_backoff = 2**retry_num * random.uniform(0.8, 1.2) export_error: Exception | None = None try: resp = self._export(serialized_data, deadline_sec - time()) @@ -251,10 +268,14 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: export_error = error retryable = isinstance(error, ConnectionError) status_code = None + retry_after = None else: reason = resp.reason retryable = _is_retryable(resp) status_code = resp.status_code + retry_after = _get_retry_after_seconds(getattr(resp, "headers", None)) + + backoff_seconds = base_backoff if retry_after is None else max(base_backoff, retry_after) if not retryable: _logger.error( diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_http2_transport.py b/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_http2_transport.py new file mode 100644 index 0000000000..c49b9cb150 --- /dev/null +++ b/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_http2_transport.py @@ -0,0 +1,44 @@ +import os +import types +import time +import pytest + +from opentelemetry.exporter.otlp.proto.http._common import _is_retryable, _get_retry_after_seconds + + +class _FakeResp: + def __init__(self, status_code: int, headers: dict[str, str] | None = None): + self.status_code = status_code + self.headers = headers or {} + + +def test_is_retryable_includes_429(): + resp = _FakeResp(429) + assert _is_retryable(resp) is True + + +def test_parse_retry_after_seconds_delta(): + assert _get_retry_after_seconds({"Retry-After": "2"}) in (2.0, pytest.approx(2.0, rel=0.2)) + + +def test_parse_retry_after_seconds_date(): + # RFC1123 date slightly in the future + future = time.gmtime(time.time() + 1) + header = time.strftime("%a, %d %b %Y %H:%M:%S GMT", future) + val = _get_retry_after_seconds({"Retry-After": header}) + assert val is not None + assert val >= 0.0 + + +@pytest.mark.skipif("httpx" not in {m.__name__ for m in map(lambda k: types.ModuleType(k), list(__import__("sys").modules.keys()))}, reason="httpx not installed") +def test_httpx_transport_selected(monkeypatch): + from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter + + monkeypatch.setenv("OTEL_EXPORTER_OTLP_HTTP_TRANSPORT", "httpx") + exp = OTLPSpanExporter() + # Import inside test to avoid ImportError when httpx missing + try: + from opentelemetry.exporter.otlp.proto.http._common._transport_httpx import HttpxSession + except Exception: + pytest.skip("httpx transport unavailable") + assert isinstance(exp._session, HttpxSession)