Report OTLP partial success responses in all exporters

.changelog/5871.fixed

@@ -0,0 +1 @@
+`opentelemetry-exporter-otlp-proto-grpc`, `opentelemetry-exporter-otlp-proto-http`: report the `partial_success` field of OTLP export responses

exporter/opentelemetry-exporter-otlp-common/src/opentelemetry/exporter/otlp/common/http.py

@@ -95,12 +95,18 @@ def from_str(value: str) -> Compression:
 
 @dataclass(slots=True, frozen=True)
 class _ExportResult:
-    """Outcome of an OTLP export attempt, including retry exhaustion."""
+    """Outcome of an OTLP export attempt, including retry exhaustion.
+
+    ``content`` holds the raw response body of a successful export, which
+    carries the ``partial_success`` the server reports. It is empty when the
+    export failed or when the body could not be read.
+    """
 
     success: bool
     status_code: int | None
     reason: str | None
     error: Exception | None
+    content: bytes = b""
 
 
 class _OTLPHTTPClient:
@@ -201,7 +207,17 @@ def export(self, data: bytes) -> _ExportResult:
                 status_code = result.status_code
                 reason = result.reason
                 if status_code is not None and 200 <= status_code < 400:
-                    return _ExportResult(True, status_code, reason, None)
+                    try:
+                        content = result.content()
+                    # pylint: disable-next=broad-exception-caught
+                    except Exception as error:
+                        self._logger.debug(
+                            "Failed to read the %s export response body: %s",
+                            self._kind,
+                            error,
+                        )
+                        content = b""
+                    return _ExportResult(True, status_code, reason, None, content)
                 export_error = result.error
                 retryable = (
                     _is_retryable(status_code) if status_code else self._transport.is_connection_error(result.error)

exporter/opentelemetry-exporter-otlp-common/tests/test_http_client.py

@@ -12,6 +12,7 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from email.utils import format_datetime
+from logging import DEBUG
 from unittest.mock import Mock, patch
 
 # pylint: disable-next=import-error
@@ -58,10 +59,13 @@ def _wait(duration: float) -> bool:
 @dataclass(frozen=True, slots=True)
 class _TestHTTPResult(BaseHTTPResult):
     response_headers: dict = field(default_factory=dict)
+    body: bytes = b""
+    content_error: Exception | None = None
 
-    # pylint: disable-next=no-self-use
     def content(self) -> bytes:
-        return b""
+        if self.content_error is not None:
+            raise self.content_error
+        return self.body
 
     def headers(self):
         return self.response_headers
@@ -106,6 +110,7 @@ def close(self):
         self.closed = True
 
 
+# pylint: disable-next=too-many-public-methods
 class TestOTLPHTTPClient(unittest.TestCase):
     @staticmethod
     def _client(
@@ -144,6 +149,38 @@ def test_export_success_status_codes(self):
                 self.assertEqual(result.reason, reason)
                 self.assertIsNone(result.error)
 
+    def test_export_success_exposes_the_response_body(self):
+        transport = _TestHTTPTransport(_TestHTTPResult(status_code=200, reason="OK", body=b"partial-success"))
+        client = self._client(transport)
+
+        result = client.export(b"payload")
+
+        self.assertTrue(result.success)
+        self.assertEqual(result.content, b"partial-success")
+
+    def test_export_success_survives_an_unreadable_response_body(self):
+        for error in (ValueError("truncated"), OSError("connection broken"), RuntimeError("stream closed")):
+            with self.subTest(error=type(error).__name__):
+                transport = _TestHTTPTransport(_TestHTTPResult(status_code=200, reason="OK", content_error=error))
+                client = self._client(transport)
+
+                with self.assertLogs(level="DEBUG") as logs:
+                    result = client.export(b"payload")
+
+                self.assertTrue(result.success)
+                self.assertEqual(result.content, b"")
+                self.assertEqual([record.levelno for record in logs.records], [DEBUG])
+                self.assertIn("Failed to read the spans export response body", logs.records[0].getMessage())
+
+    def test_export_failure_has_no_response_body(self):
+        transport = _TestHTTPTransport(_TestHTTPResult(status_code=400, reason="Bad Request", body=b"ignored"))
+        client = self._client(transport)
+
+        result = client.export(b"payload")
+
+        self.assertFalse(result.success)
+        self.assertEqual(result.content, b"")
+
     @patch(
         "opentelemetry.exporter.otlp.common.http.time.time",
         side_effect=(100.0, 100.0, 100.0),

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

@@ -0,0 +1,131 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Final
+
+from google.protobuf.message import DecodeError
+
+if TYPE_CHECKING:
+    from logging import Logger
+
+    from google.protobuf.message import Message
+
+_REJECTED_FIELDS: Final[tuple[str, ...]] = (
+    "rejected_spans",
+    "rejected_data_points",
+    "rejected_log_records",
+    "rejected_profiles",
+)
+
+
+def _warning_format(signal: str) -> str:
+    """Build the format string a partial success warning is logged with.
+
+    :param signal: The signal being exported, as the caller names it.
+    :returns: A format string taking the endpoint and the detail as arguments.
+
+    The ``DuplicateFilter`` installed on the gRPC exporter logger and on the
+    OTLP/HTTP log exporter logger keys on ``LogRecord.msg``, so the signal goes
+    in the string and the endpoint and the detail stay out: repeats of one
+    signal collapse, and the three gRPC signals, which share one logger, do not
+    collapse into each other.
+    """
+    return f"OTLP partial success exporting {signal} to %s: %s"
+
+
+def _partial_success_detail(response: Message | None) -> str | None:
+    """Describe the partial success an OTLP export response reports.
+
+    :param response: A parsed ``Export<signal>ServiceResponse`` message, or
+        ``None``.
+    :returns: A description of the rejected items and the error message the
+        server sent, to be logged as the last argument of the format string
+        ``_warning_format`` builds, or ``None`` when there is nothing to report.
+
+    The OTLP specification treats a ``partial_success`` field holding a zero
+    rejected count and an empty error message as equivalent to the field being
+    absent, so both cases return ``None``. A ``response`` of ``None`` returns
+    ``None`` too, because a gRPC stub is free to return nothing and reporting
+    is not worth failing an export the server accepted.
+
+    A message whose type does not define ``partial_success`` at all returns
+    ``None`` as well. ``HasField`` raises ``ValueError`` for a field the type
+    does not define, and ``OTLPExporterMixin`` is public, so a subclass may
+    export through a service of its own whose response carries no such field.
+    """
+    if response is None or "partial_success" not in response.DESCRIPTOR.fields_by_name:
+        return None
+
+    if not response.HasField("partial_success"):
+        return None
+
+    message: Any = response
+    partial_success = message.partial_success
+    rejected = 0
+    rejected_field = ""
+    for field_name in _REJECTED_FIELDS:
+        if hasattr(partial_success, field_name):
+            rejected = getattr(partial_success, field_name)
+            rejected_field = field_name
+            break
+
+    error_message = partial_success.error_message
+    if not rejected and not error_message:
+        return None
+
+    if not rejected:
+        return error_message
+
+    kind = rejected_field.removeprefix("rejected_").replace("_", " ")
+    if rejected == 1:
+        kind = kind.removesuffix("s")
+    if not error_message:
+        return f"the server rejected {rejected} {kind}"
+    return f"the server rejected {rejected} {kind}: {error_message}"
+
+
+def _log_partial_success(
+    serialized_response: bytes,
+    response_type: type[Message],
+    logger: Logger,
+    signal: str,
+    endpoint: str,
+) -> None:
+    """Log the partial success reported by a serialized OTLP export response.
+
+    :param serialized_response: The raw response body returned by the server.
+    :param response_type: The ``Export<signal>ServiceResponse`` class to parse
+        ``serialized_response`` with.
+    :param logger: The logger that receives the warning.
+    :param signal: The signal being exported, named as the surrounding failure
+        messages name it, so a process running several exporters can tell which
+        one is dropping data.
+    :param endpoint: The endpoint the export was sent to, for the same reason.
+
+    A response that cannot be parsed is logged at debug level and otherwise
+    ignored, so that a malformed body never turns a successful export into a
+    failed one. Debug rather than warning, because the server accepted the
+    request and a body the exporter cannot read tells the user nothing they can
+    act on. Any status from 200 to 399 counts as a successful export, so an
+    unparsable body is what a redirect page or a proxy error page looks like
+    here, and warning on those would fire once per export.
+
+    The pure-Python protobuf runtime raises ``UnicodeDecodeError`` for a string
+    field holding invalid UTF-8, where the default runtime raises
+    ``DecodeError``, so both are caught.
+    """
+    if not serialized_response:
+        return
+
+    response = response_type()
+    try:
+        response.ParseFromString(serialized_response)
+    except (DecodeError, UnicodeDecodeError) as error:
+        logger.debug("Failed to parse the OTLP export response: %s", error)
+        return
+
+    detail = _partial_success_detail(response)
+    if detail is not None:
+        logger.warning(_warning_format(signal), endpoint, detail)