Merge remote-tracking branch 'origin/main' into shutdown_refactor

Author: DylanRussell

.github/workflows/benchmarks.yml

@@ -6,7 +6,7 @@ on:
 
 jobs:
   sdk-benchmarks:
-    runs-on: self-hosted
+    runs-on: equinix-bare-metal
     steps:
     - name: Checkout Core Repo @ SHA - ${{ github.sha }}
       uses: actions/checkout@v4

CHANGELOG.md

@@ -12,6 +12,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   had completed/failed was removed.
   ([#4564](https://github.com/open-telemetry/opentelemetry-python/pull/4564)).
 
+- 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
+  ([#4618](https://github.com/open-telemetry/opentelemetry-python/pull/4618))
+- Add `span_exporter` property back to `BatchSpanProcessor` class
+  ([#4621](https://github.com/open-telemetry/opentelemetry-python/pull/4621))
+- Fix license field in pyproject.toml files
+  ([#4625](https://github.com/open-telemetry/opentelemetry-python/pull/4625))
+
 ## Version 1.34.0/0.55b0 (2025-06-04)
 
 - typecheck: add sdk/resources and drop mypy

docs/examples/error_handler/error_handler_0/pyproject.toml

@@ -7,15 +7,14 @@ name = "error-handler-0"
 dynamic = ["version"]
 description = "This is just an error handler example package"
 readme = "README.rst"
-license = {text = "Apache-2.0"}
+license = "Apache-2.0"
 requires-python = ">=3.9"
 authors = [
   { name = "OpenTelemetry Authors", email = "[email protected]" },
 ]
 classifiers = [
   "Development Status :: 4 - Beta",
   "Intended Audience :: Developers",
-  "License :: OSI Approved :: Apache Software License",
   "Programming Language :: Python",
   "Programming Language :: Python :: 3",
   "Programming Language :: Python :: 3.9",

docs/examples/error_handler/error_handler_1/pyproject.toml

@@ -7,15 +7,14 @@ name = "error-handler-1"
 dynamic = ["version"]
 description = "This is just an error handler example package"
 readme = "README.rst"
-license = {text = "Apache-2.0"}
+license = "Apache-2.0"
 requires-python = ">=3.9"
 authors = [
   { name = "OpenTelemetry Authors", email = "[email protected]" },
 ]
 classifiers = [
   "Development Status :: 4 - Beta",
   "Intended Audience :: Developers",
-  "License :: OSI Approved :: Apache Software License",
   "Programming Language :: Python",
   "Programming Language :: Python :: 3",
   "Programming Language :: Python :: 3.9",

exporter/opentelemetry-exporter-opencensus/pyproject.toml

@@ -7,7 +7,7 @@ name = "opentelemetry-exporter-opencensus"
 dynamic = ["version"]
 description = "OpenCensus Exporter"
 readme = "README.rst"
-license = {text = "Apache-2.0"}
+license = "Apache-2.0"
 requires-python = ">=3.9"
 authors = [
   { name = "OpenTelemetry Authors", email = "[email protected]" },
@@ -17,7 +17,6 @@ classifiers = [
   "Framework :: OpenTelemetry",
   "Framework :: OpenTelemetry :: Exporters",
   "Intended Audience :: Developers",
-  "License :: OSI Approved :: Apache Software License",
   "Programming Language :: Python",
   "Programming Language :: Python :: 3",
   "Programming Language :: Python :: 3.9",

exporter/opentelemetry-exporter-otlp-proto-common/pyproject.toml

@@ -7,7 +7,7 @@ name = "opentelemetry-exporter-otlp-proto-common"
 dynamic = ["version"]
 description = "OpenTelemetry Protobuf encoding"
 readme = "README.rst"
-license = {text = "Apache-2.0"}
+license = "Apache-2.0"
 requires-python = ">=3.9"
 authors = [
   { name = "OpenTelemetry Authors", email = "[email protected]" },
@@ -17,7 +17,6 @@ classifiers = [
   "Framework :: OpenTelemetry",
   "Framework :: OpenTelemetry :: Exporters",
   "Intended Audience :: Developers",
-  "License :: OSI Approved :: Apache Software License",
   "Programming Language :: Python",
   "Programming Language :: Python :: 3",
   "Programming Language :: Python :: 3.9",

exporter/opentelemetry-exporter-otlp-proto-grpc/pyproject.toml

@@ -7,7 +7,7 @@ name = "opentelemetry-exporter-otlp-proto-grpc"
 dynamic = ["version"]
 description = "OpenTelemetry Collector Protobuf over gRPC Exporter"
 readme = "README.rst"
-license = {text = "Apache-2.0"}
+license = "Apache-2.0"
 requires-python = ">=3.9"
 authors = [
   { name = "OpenTelemetry Authors", email = "[email protected]" },
@@ -17,7 +17,6 @@ classifiers = [
   "Framework :: OpenTelemetry",
   "Framework :: OpenTelemetry :: Exporters",
   "Intended Audience :: Developers",
-  "License :: OSI Approved :: Apache Software License",
   "Programming Language :: Python",
   "Programming Language :: Python :: 3",
   "Programming Language :: Python :: 3.9",

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

@@ -293,52 +293,55 @@ def _export(
         # 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.
-        retry_info = RetryInfo()
-        deadline_sec = time() + self._timeout
-        for retry_num in range(1, _MAX_RETRYS + 1):
-            backoff_seconds = 2 ** (retry_num - 1) * random.uniform(
-                0.8, 1.2
-            )
-            try:
-                self._client.Export(
-                    request=self._translate_data(data),
-                    metadata=self._headers,
-                    timeout=self._timeout,
-                )
-                return self._result.SUCCESS
-            except RpcError as error:
-                retry_info_bin = dict(error.trailing_metadata()).get(
-                    "google.rpc.retryinfo-bin"
-                )
-                if retry_info_bin is not None:
-                    retry_info.ParseFromString(retry_info_bin)
-                    backoff_seconds = (
-                        retry_info.retry_delay.seconds
-                        + retry_info.retry_delay.nanos / 1.0e9
+        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=deadline_sec - time(),
+                    )
+                    return self._result.SUCCESS
+                except RpcError as error:
+                    retry_info_bin = dict(error.trailing_metadata()).get(
+                        "google.rpc.retryinfo-bin"
                     )
-                if (
-                    error.code() not in _RETRYABLE_ERROR_CODES
-                    or retry_num == _MAX_RETRYS
-                    or backoff_seconds > (deadline_sec - time())
-                    or self._shutdown
-                ):
-                    logger.error(
-                        "Failed to export %s to %s, error code: %s",
+                    # 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
+                        )
+                    if (
+                        error.code() not in _RETRYABLE_ERROR_CODES
+                        or retry_num + 1 == _MAX_RETRYS
+                        or backoff_seconds > (deadline_sec - time())
+                        or self._shutdown
+                    ):
+                        logger.error(
+                            "Failed to export %s to %s, error code: %s",
+                            self._exporting,
+                            self._endpoint,
+                            error.code(),
+                            exc_info=error.code() == StatusCode.UNKNOWN,
+                        )
+                        return self._result.FAILURE
+                    logger.warning(
+                        "Transient error %s encountered while exporting %s to %s, retrying in %.2fs.",
+                        error.code(),
                         self._exporting,
                         self._endpoint,
-                        error.code(),
-                        exc_info=error.code() == StatusCode.UNKNOWN,
+                        backoff_seconds,
                     )
-                    return self._result.FAILURE
-                logger.warning(
-                    "Transient error %s encountered while exporting logs batch, retrying in %.2fs.",
-                    error.code(),
-                    backoff_seconds,
-                )
                 shutdown = self._shutdown_is_occuring.wait(backoff_seconds)
                 if shutdown:
                     logger.warning("Shutdown in progress, aborting retry.")
                     break
+        # Not possible to reach here but the linter is complaining.
         return self._result.FAILURE
 
     def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:

exporter/opentelemetry-exporter-otlp-proto-grpc/test-requirements.txt

@@ -11,6 +11,7 @@ tomli==2.0.1
 typing_extensions==4.10.0
 wrapt==1.16.0
 zipp==3.19.2
+googleapis-common-protos==1.63.2
 -e opentelemetry-api
 -e tests/opentelemetry-test-utils
 -e exporter/opentelemetry-exporter-otlp-proto-common

exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py

@@ -14,8 +14,10 @@
 
 import threading
 import time
+import unittest
 from concurrent.futures import ThreadPoolExecutor
 from logging import WARNING, getLogger
+from platform import system
 from typing import Any, Optional, Sequence
 from unittest import TestCase
 from unittest.mock import Mock, patch
@@ -90,33 +92,32 @@ class TraceServiceServicerWithExportParams(TraceServiceServicer):
     def __init__(
         self,
         export_result: StatusCode,
-        optional_retry_millis: Optional[int] = None,
+        optional_retry_nanos: Optional[int] = None,
         optional_export_sleep: Optional[float] = None,
     ):
         self.export_result = export_result
         self.optional_export_sleep = optional_export_sleep
-        self.optional_retry_millis = optional_retry_millis
+        self.optional_retry_nanos = optional_retry_nanos
         self.num_requests = 0
 
     # pylint: disable=invalid-name,unused-argument
     def Export(self, request, context):
         self.num_requests += 1
         if self.optional_export_sleep:
             time.sleep(self.optional_export_sleep)
-        if self.export_result != StatusCode.OK:
-            if self.optional_retry_millis:
-                context.set_trailing_metadata(
+        if self.export_result != StatusCode.OK and self.optional_retry_nanos:
+            context.set_trailing_metadata(
+                (
                     (
-                        (
-                            "google.rpc.retryinfo-bin",
-                            RetryInfo(
-                                retry_delay=Duration(
-                                    nanos=self.optional_retry_millis
-                                )
-                            ).SerializeToString(),
-                        ),
-                    )
+                        "google.rpc.retryinfo-bin",
+                        RetryInfo(
+                            retry_delay=Duration(
+                                nanos=self.optional_retry_nanos
+                            )
+                        ).SerializeToString(),
+                    ),
                 )
+            )
         context.set_code(self.export_result)
 
         return ExportTraceServiceResponse()
@@ -360,9 +361,14 @@ def test_export_over_closed_grpc_channel(self):
             str(err.exception), "Cannot invoke RPC on closed channel!"
         )
 
+    @unittest.skipIf(
+        system() == "Windows",
+        "For gRPC + windows there's some added delay in the RPCs which breaks the assertion over amount of time passed.",
+    )
     def test_retry_info_is_respected(self):
         mock_trace_service = TraceServiceServicerWithExportParams(
-            StatusCode.UNAVAILABLE, optional_retry_millis=200
+            StatusCode.UNAVAILABLE,
+            optional_retry_nanos=200000000,  # .2 seconds
         )
         add_TraceServiceServicer_to_server(
             mock_trace_service,
@@ -376,9 +382,13 @@ def test_retry_info_is_respected(self):
         )
         after = time.time()
         self.assertEqual(mock_trace_service.num_requests, 6)
-        # 200 * 5 = 1 second, plus some wiggle room so the test passes consistently.
-        self.assertTrue(after - before < 1.1)
+        # 1 second plus wiggle room so the test passes consistently.
+        self.assertAlmostEqual(after - before, 1, 1)
 
+    @unittest.skipIf(
+        system() == "Windows",
+        "For gRPC + windows there's some added delay in the RPCs which breaks the assertion over amount of time passed.",
+    )
     def test_retry_not_made_if_would_exceed_timeout(self):
         mock_trace_service = TraceServiceServicerWithExportParams(
             StatusCode.UNAVAILABLE
@@ -398,34 +408,39 @@ def test_retry_not_made_if_would_exceed_timeout(self):
         # First call at time 0, second at time 1, third at time 3, fourth would exceed timeout.
         self.assertEqual(mock_trace_service.num_requests, 3)
         # There's a +/-20% jitter on each backoff.
-        self.assertTrue(after - before < 3.7)
+        self.assertTrue(2.35 < after - before < 3.65)
 
+    @unittest.skipIf(
+        system() == "Windows",
+        "For gRPC + windows there's some added delay in the RPCs which breaks the assertion over amount of time passed.",
+    )
     def test_timeout_set_correctly(self):
         mock_trace_service = TraceServiceServicerWithExportParams(
-            StatusCode.OK, optional_export_sleep=0.5
+            StatusCode.UNAVAILABLE, optional_export_sleep=0.25
         )
         add_TraceServiceServicer_to_server(
             mock_trace_service,
             self.server,
         )
-        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=0.4)
-        # Should timeout. Deadline should be set to now + timeout.
-        # That is 400 millis from now, and export sleeps for 500 millis.
+        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=1.4)
+        # Should timeout after 1.4 seconds. First attempt takes .25 seconds
+        # Then a 1 second sleep, then deadline exceeded after .15 seconds,
+        # mid way through second call.
         with self.assertLogs(level=WARNING) as warning:
-            self.assertEqual(
-                exporter.export([self.span]),
-                SpanExportResult.FAILURE,
-            )
+            before = time.time()
+            # Eliminate the jitter.
+            with patch("random.uniform", return_value=1):
+                self.assertEqual(
+                    exporter.export([self.span]),
+                    SpanExportResult.FAILURE,
+                )
+            after = time.time()
             self.assertEqual(
                 "Failed to export traces to localhost:4317, error code: StatusCode.DEADLINE_EXCEEDED",
                 warning.records[-1].message,
             )
-            self.assertEqual(mock_trace_service.num_requests, 1)
-        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=0.8)
-        self.assertEqual(
-            exporter.export([self.span]),
-            SpanExportResult.SUCCESS,
-        )
+            self.assertEqual(mock_trace_service.num_requests, 2)
+            self.assertAlmostEqual(after - before, 1.4, 1)
 
     def test_otlp_headers_from_env(self):
         # pylint: disable=protected-access