Skip to content

feat(exporter): add HTTP/2 support and exponential backoff retries for OTLP HTTP exporter (#3833) #5466

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
glatinone wants to merge 1 commit into from
Closed

feat(exporter): add HTTP/2 support and exponential backoff retries for OTLP HTTP exporter (#3833) #5466

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 35 additions & 3 deletions ...y-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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:
Expand Down
83 changes: 83 additions & 0 deletions ...er-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/_transport_httpx.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
44 changes: 32 additions & 12 deletions ...rter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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())
Expand All @@ -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(
Expand Down
44 changes: 32 additions & 12 deletions ...er-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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())
Expand All @@ -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(
Expand Down
43 changes: 32 additions & 11 deletions ...ter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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())
Expand All @@ -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(
Expand Down
Loading
Loading