Merge branch 'main' into fix/copy-measurement-attributes

Author: xrmx

.github/workflows/ossf-scorecard.yml

@@ -23,7 +23,7 @@ jobs:
         with:
           persist-credentials: false
 
-      - uses: ossf/scorecard-action@f49aabe0b5af0936a0987cfb85d86b75731b0186 # v2.4.1
+      - uses: ossf/scorecard-action@05b42c624433fc40578a4040d5cf5e36ddca8cde # v2.4.2
         with:
           results_file: results.sarif
           results_format: sarif

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Update OTLP gRPC/HTTP exporters: the export timeout is now inclusive of all retries and backoffs.
+  A +/-20% jitter was added to all backoffs. A pointless 32 second sleep that occurred after all retries
+  had completed/failed was removed.
+  ([#4564](https://github.com/open-telemetry/opentelemetry-python/pull/4564)).
 - Update ConsoleLogExporter.export to handle LogRecord's containing bytes type
   in the body ([#4614](https://github.com/open-telemetry/opentelemetry-python/pull/4614/)).
 - opentelemetry-sdk: Fix invalid `type: ignore` that causes mypy to ignore the whole file
@@ -17,14 +21,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4625](https://github.com/open-telemetry/opentelemetry-python/pull/4625))
 - opentelemetry-sdk: `Measurement`s `Attributes` are now copied when instantiating a `Measurement`. This stops the accidental modification of `Attibutes` after the `Measurement` is created.
   ([#4627](https://github.com/open-telemetry/opentelemetry-python/pull/4627))
+- Update logger level to NOTSET in logs example
+  ([#4637](https://github.com/open-telemetry/opentelemetry-python/pull/4637))
+- Logging API accepts optional `context`; deprecates `trace_id`, `span_id`, `trace_flags`.
+  ([#4597](https://github.com/open-telemetry/opentelemetry-python/pull/4597))
 
 ## Version 1.34.0/0.55b0 (2025-06-04)
 
 - typecheck: add sdk/resources and drop mypy
   ([#4578](https://github.com/open-telemetry/opentelemetry-python/pull/4578))
-- Refactor `BatchLogRecordProcessor` to simplify code and make the control flow more
-  clear ([#4562](https://github.com/open-telemetry/opentelemetry-python/pull/4562/)
-  and [#4535](https://github.com/open-telemetry/opentelemetry-python/pull/4535)).
 - Use PEP702 for marking deprecations
   ([#4522](https://github.com/open-telemetry/opentelemetry-python/pull/4522))
 - Refactor `BatchLogRecordProcessor` and `BatchSpanProcessor` to simplify code

docs/examples/logs/example.py

@@ -33,6 +33,9 @@
 logger_provider.add_log_record_processor(BatchLogRecordProcessor(exporter))
 handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider)
 
+# Set the root logger level to NOTSET to ensure all messages are captured
+logging.getLogger().setLevel(logging.NOTSET)
+
 # Attach OTLP handler to root logger
 logging.getLogger().addHandler(handler)
 

exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/__init__.py

@@ -17,12 +17,10 @@
 
 import logging
 from collections.abc import Sequence
-from itertools import count
 from typing import (
     Any,
     Callable,
     Dict,
-    Iterator,
     List,
     Mapping,
     Optional,
@@ -177,38 +175,3 @@ def _get_resource_data(
             )
         )
     return resource_data
-
-
-def _create_exp_backoff_generator(max_value: int = 0) -> Iterator[int]:
-    """
-    Generates an infinite sequence of exponential backoff values. The sequence starts
-    from 1 (2^0) and doubles each time (2^1, 2^2, 2^3, ...). If a max_value is specified
-    and non-zero, the generated values will not exceed this maximum, capping at max_value
-    instead of growing indefinitely.
-
-    Parameters:
-    - max_value (int, optional): The maximum value to yield. If 0 or not provided, the
-      sequence grows without bound.
-
-    Returns:
-    Iterator[int]: An iterator that yields the exponential backoff values, either uncapped or
-    capped at max_value.
-
-    Example:
-    ```
-    gen = _create_exp_backoff_generator(max_value=10)
-    for _ in range(5):
-        print(next(gen))
-    ```
-    This will print:
-    1
-    2
-    4
-    8
-    10
-
-    Note: this functionality used to be handled by the 'backoff' package.
-    """
-    for i in count(0):
-        out = 2**i
-        yield min(out, max_value) if max_value else out

exporter/opentelemetry-exporter-otlp-proto-common/tests/test_backoff.py

@@ -58,7 +58,7 @@ def __init__(
         headers: Optional[
             Union[TypingSequence[Tuple[str, str]], Dict[str, str], str]
         ] = None,
-        timeout: Optional[int] = None,
+        timeout: Optional[float] = None,
         compression: Optional[Compression] = None,
     ):
         if insecure is None:
@@ -79,7 +79,7 @@ def __init__(
 
         environ_timeout = environ.get(OTEL_EXPORTER_OTLP_LOGS_TIMEOUT)
         environ_timeout = (
-            int(environ_timeout) if environ_timeout is not None else None
+            float(environ_timeout) if environ_timeout is not None else None
         )
 
         compression = (

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/_log_exporter/__init__.py

@@ -14,12 +14,13 @@
 
 """OTLP Exporter"""
 
+import random
 import threading
 from abc import ABC, abstractmethod
 from collections.abc import Sequence  # noqa: F401
 from logging import getLogger
 from os import environ
-from time import sleep
+from time import sleep, time
 from typing import (  # noqa: F401
     Any,
     Callable,
@@ -47,7 +48,6 @@
     ssl_channel_credentials,
 )
 from opentelemetry.exporter.otlp.proto.common._internal import (
-    _create_exp_backoff_generator,
     _get_resource_data,
 )
 from opentelemetry.exporter.otlp.proto.grpc import (
@@ -74,6 +74,18 @@
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.util.re import parse_env_headers
 
+_RETRYABLE_ERROR_CODES = frozenset(
+    [
+        StatusCode.CANCELLED,
+        StatusCode.DEADLINE_EXCEEDED,
+        StatusCode.RESOURCE_EXHAUSTED,
+        StatusCode.ABORTED,
+        StatusCode.OUT_OF_RANGE,
+        StatusCode.UNAVAILABLE,
+        StatusCode.DATA_LOSS,
+    ]
+)
+_MAX_RETRYS = 6
 logger = getLogger(__name__)
 SDKDataT = TypeVar("SDKDataT")
 ResourceDataT = TypeVar("ResourceDataT")
@@ -186,8 +198,6 @@ class OTLPExporterMixin(
         compression: gRPC compression method to use
     """
 
-    _MAX_RETRY_TIMEOUT = 64
-
     def __init__(
         self,
         endpoint: Optional[str] = None,
@@ -196,7 +206,7 @@ def __init__(
         headers: Optional[
             Union[TypingSequence[Tuple[str, str]], Dict[str, str], str]
         ] = None,
-        timeout: Optional[int] = None,
+        timeout: Optional[float] = None,
         compression: Optional[Compression] = None,
     ):
         super().__init__()
@@ -233,7 +243,7 @@ def __init__(
         else:
             self._headers = tuple(self._headers) + tuple(_OTLP_GRPC_HEADERS)
 
-        self._timeout = timeout or int(
+        self._timeout = timeout or float(
             environ.get(OTEL_EXPORTER_OTLP_TIMEOUT, 10)
         )
         self._collector_kwargs = None
@@ -246,7 +256,8 @@ def __init__(
 
         if insecure:
             self._channel = insecure_channel(
-                self._endpoint, compression=compression
+                self._endpoint,
+                compression=compression,
             )
         else:
             credentials = _get_credentials(
@@ -256,7 +267,9 @@ def __init__(
                 OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE,
             )
             self._channel = secure_channel(
-                self._endpoint, credentials, compression=compression
+                self._endpoint,
+                credentials,
+                compression=compression,
             )
         self._client = self._stub(self._channel)
 
@@ -270,89 +283,61 @@ def _translate_data(
         pass
 
     def _export(
-        self, data: Union[TypingSequence[ReadableSpan], MetricsData]
+        self,
+        data: Union[TypingSequence[ReadableSpan], MetricsData],
     ) -> ExportResultT:
-        # After the call to shutdown, subsequent calls to Export are
-        # not allowed and should return a Failure result.
         if self._shutdown:
             logger.warning("Exporter already shutdown, ignoring batch")
             return self._result.FAILURE
 
         # FIXME remove this check if the export type for traces
         # gets updated to a class that represents the proto
         # TracesData and use the code below instead.
-        # logger.warning(
-        #     "Transient error %s encountered while exporting %s, retrying in %ss.",
-        #     error.code(),
-        #     data.__class__.__name__,
-        #     delay,
-        # )
-        # expo returns a generator that yields delay values which grow
-        # exponentially. Once delay is greater than max_value, the yielded
-        # value will remain constant.
-        for delay in _create_exp_backoff_generator(
-            max_value=self._MAX_RETRY_TIMEOUT
-        ):
-            if delay == self._MAX_RETRY_TIMEOUT or self._shutdown:
-                return self._result.FAILURE
-
-            with self._export_lock:
+        with self._export_lock:
+            deadline_sec = time() + self._timeout
+            for retry_num in range(_MAX_RETRYS):
                 try:
                     self._client.Export(
                         request=self._translate_data(data),
                         metadata=self._headers,
-                        timeout=self._timeout,
+                        timeout=deadline_sec - time(),
                     )
-
                     return self._result.SUCCESS
-
                 except RpcError as error:
-                    if error.code() in [
-                        StatusCode.CANCELLED,
-                        StatusCode.DEADLINE_EXCEEDED,
-                        StatusCode.RESOURCE_EXHAUSTED,
-                        StatusCode.ABORTED,
-                        StatusCode.OUT_OF_RANGE,
-                        StatusCode.UNAVAILABLE,
-                        StatusCode.DATA_LOSS,
-                    ]:
-                        retry_info_bin = dict(error.trailing_metadata()).get(
-                            "google.rpc.retryinfo-bin"
-                        )
-                        if retry_info_bin is not None:
-                            retry_info = RetryInfo()
-                            retry_info.ParseFromString(retry_info_bin)
-                            delay = (
-                                retry_info.retry_delay.seconds
-                                + retry_info.retry_delay.nanos / 1.0e9
-                            )
-
-                        logger.warning(
-                            (
-                                "Transient error %s encountered while exporting "
-                                "%s to %s, retrying in %ss."
-                            ),
-                            error.code(),
-                            self._exporting,
-                            self._endpoint,
-                            delay,
+                    retry_info_bin = dict(error.trailing_metadata()).get(
+                        "google.rpc.retryinfo-bin"
+                    )
+                    # 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)
+                    if retry_info_bin is not None:
+                        retry_info = RetryInfo()
+                        retry_info.ParseFromString(retry_info_bin)
+                        backoff_seconds = (
+                            retry_info.retry_delay.seconds
+                            + retry_info.retry_delay.nanos / 1.0e9
                         )
-                        sleep(delay)
-                        continue
-                    else:
+                    if (
+                        error.code() not in _RETRYABLE_ERROR_CODES
+                        or retry_num + 1 == _MAX_RETRYS
+                        or backoff_seconds > (deadline_sec - time())
+                    ):
                         logger.error(
                             "Failed to export %s to %s, error code: %s",
                             self._exporting,
                             self._endpoint,
                             error.code(),
                             exc_info=error.code() == StatusCode.UNKNOWN,
                         )
-
-                    if error.code() == StatusCode.OK:
-                        return self._result.SUCCESS
-
-                    return self._result.FAILURE
-
+                        return self._result.FAILURE
+                    logger.warning(
+                        "Transient error %s encountered while exporting %s to %s, retrying in %.2fs.",
+                        error.code(),
+                        self._exporting,
+                        self._endpoint,
+                        backoff_seconds,
+                    )
+                    sleep(backoff_seconds)
+        # Not possible to reach here but the linter is complaining.
         return self._result.FAILURE
 
     def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None: