oss_10_otlp_partial_success

.changelog/13.fixed

@@ -0,0 +1,4 @@
+OTLP exporters (gRPC and HTTP) now inspect the export response and log a warning
+on partial success, reporting the number of rejected spans, data points or log
+records and any error message returned by the collector. Previously
+collector-rejected telemetry was silently dropped.

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

@@ -152,6 +152,35 @@
 }
 
 
+def _log_partial_success(response) -> None:
+    """Log a warning if an OTLP export response reports a partial success.
+
+    A partial success is signalled by the collector either rejecting some
+    items (``rejected_spans`` / ``rejected_data_points`` / ``rejected_log_records``
+    != 0) or by returning a non-empty ``error_message``. The request itself was
+    still accepted, so this only logs; it does not change the export result.
+    """
+    if response is None or not response.HasField("partial_success"):
+        return
+    partial_success = response.partial_success
+    rejected = 0
+    for field_name in (
+        "rejected_spans",
+        "rejected_data_points",
+        "rejected_log_records",
+    ):
+        if hasattr(partial_success, field_name):
+            rejected = getattr(partial_success, field_name)
+            break
+    error_message = partial_success.error_message
+    if rejected != 0 or error_message:
+        logger.warning(
+            "Partial success received from collector: %s items rejected. %s",
+            rejected,
+            error_message,
+        )
+
+
 class InvalidCompressionValueException(Exception):
     def __init__(self, environ_key: str, environ_value: str):
         super().__init__(
@@ -462,11 +491,12 @@ def _export(
                 try:
                     if self._client is None:
                         return self._result.FAILURE
-                    self._client.Export(
+                    response = self._client.Export(
                         request=self._translate_data(data),
                         metadata=self._headers,
                         timeout=deadline_sec - time(),
                     )
+                    _log_partial_success(response)
                     return self._result.SUCCESS  # type: ignore [reportReturnType]
                 except RpcError as error:
                     retry_info_bin = dict(error.trailing_metadata()).get(  # type: ignore [reportAttributeAccessIssue]

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

@@ -110,12 +110,16 @@ def __init__(
         optional_retry_nanos: int | None = None,
         optional_export_sleep: float | None = None,
         optional_error_details: str | None = None,
+        optional_rejected_spans: int | None = None,
+        optional_partial_error_message: str | None = None,
     ):
         self.export_result = export_result
         self.optional_export_sleep = optional_export_sleep
         self.optional_retry_nanos = optional_retry_nanos
         self.num_requests = 0
         self.optional_error_details = optional_error_details
+        self.optional_rejected_spans = optional_rejected_spans
+        self.optional_partial_error_message = optional_partial_error_message
 
     # pylint: disable=invalid-name,unused-argument
     def Export(self, request, context):
@@ -139,7 +143,16 @@ def Export(self, request, context):
         if self.optional_error_details:
             context.set_details(self.optional_error_details)
 
-        return ExportTraceServiceResponse()
+        response = ExportTraceServiceResponse()
+        if self.optional_rejected_spans is not None:
+            response.partial_success.rejected_spans = (
+                self.optional_rejected_spans
+            )
+        if self.optional_partial_error_message is not None:
+            response.partial_success.error_message = (
+                self.optional_partial_error_message
+            )
+        return response
 
 
 class ThreadWithReturnValue(threading.Thread):
@@ -427,6 +440,44 @@ def test_shutdown(self):
                 "Exporter already shutdown, ignoring batch",
             )
 
+    def test_partial_success_logs_warning(self):
+        """A partial-success response with rejected spans logs a warning."""
+        add_TraceServiceServicer_to_server(
+            TraceServiceServicerWithExportParams(
+                StatusCode.OK,
+                optional_rejected_spans=4,
+                optional_partial_error_message="some spans were dropped",
+            ),
+            self.server,
+        )
+        exporter = OTLPSpanExporterForTesting(
+            insecure=True, meter_provider=self.meter_provider
+        )
+        with self.assertLogs(level=WARNING) as warning:
+            self.assertEqual(
+                exporter.export([self.span]), SpanExportResult.SUCCESS
+            )
+        self.assertIn("4 items rejected", warning.output[0])
+        self.assertIn("some spans were dropped", warning.output[0])
+
+    def test_full_success_logs_nothing(self):
+        """A fully successful (empty partial_success) response logs nothing."""
+        add_TraceServiceServicer_to_server(
+            TraceServiceServicerWithExportParams(StatusCode.OK),
+            self.server,
+        )
+        exporter = OTLPSpanExporterForTesting(
+            insecure=True, meter_provider=self.meter_provider
+        )
+        exporter_logger = getLogger(
+            "opentelemetry.exporter.otlp.proto.grpc.exporter"
+        )
+        with patch.object(exporter_logger, "warning") as mock_warning:
+            self.assertEqual(
+                exporter.export([self.span]), SpanExportResult.SUCCESS
+            )
+        mock_warning.assert_not_called()
+
     @unittest.skipIf(
         system() == "Windows",
         "For gRPC + windows there's some added delay in the RPCs which breaks the assertion over amount of time passed.",

exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/__init__.py

@@ -1,6 +1,7 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+import logging
 from os import environ
 from typing import Literal
 
@@ -11,6 +12,43 @@
 )
 from opentelemetry.util._importlib_metadata import entry_points
 
+_logger = logging.getLogger(__name__)
+
+
+def _log_partial_success(response_bytes: bytes, response_class) -> None:
+    """Deserialize an OTLP export response and log a warning if it reports a
+    partial success.
+
+    A partial success is signalled by the collector either rejecting some
+    items (``rejected_spans`` / ``rejected_data_points`` / ``rejected_log_records``
+    != 0) or by returning a non-empty ``error_message``. The request itself was
+    still accepted, so this only logs; it does not change the export result.
+    """
+    try:
+        response = response_class.FromString(response_bytes)
+    except Exception:  # pylint: disable=broad-except
+        # An unparseable body must not turn a successful export into a failure.
+        return
+    if not response.HasField("partial_success"):
+        return
+    partial_success = response.partial_success
+    rejected = 0
+    for field_name in (
+        "rejected_spans",
+        "rejected_data_points",
+        "rejected_log_records",
+    ):
+        if hasattr(partial_success, field_name):
+            rejected = getattr(partial_success, field_name)
+            break
+    error_message = partial_success.error_message
+    if rejected != 0 or error_message:
+        _logger.warning(
+            "Partial success received from collector: %s items rejected. %s",
+            rejected,
+            error_message,
+        )
+
 
 def _is_retryable(resp: requests.Response) -> bool:
     if resp.status_code == 408:

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

@@ -27,8 +27,12 @@
 from opentelemetry.exporter.otlp.proto.http._common import (
     _is_retryable,
     _load_session_from_envvar,
+    _log_partial_success,
 )
 from opentelemetry.metrics import MeterProvider
+from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import (
+    ExportLogsServiceResponse,
+)
 from opentelemetry.sdk._logs import ReadableLogRecord
 from opentelemetry.sdk._logs.export import (
     LogRecordExporter,
@@ -208,6 +212,9 @@ def export(
                 try:
                     resp = self._export(serialized_data, deadline_sec - time())
                     if resp.ok:
+                        _log_partial_success(
+                            resp.content, ExportLogsServiceResponse
+                        )
                         return LogRecordExportResult.SUCCESS
                 except requests.exceptions.RequestException as error:
                     reason = error

exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py

@@ -41,10 +41,12 @@
 from opentelemetry.exporter.otlp.proto.http._common import (
     _is_retryable,
     _load_session_from_envvar,
+    _log_partial_success,
 )
 from opentelemetry.metrics import MeterProvider
 from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import (  # noqa: F401
     ExportMetricsServiceRequest,
+    ExportMetricsServiceResponse,
 )
 from opentelemetry.proto.common.v1.common_pb2 import (  # noqa: F401
     AnyValue,
@@ -279,6 +281,9 @@ def _export_with_retries(
                 try:
                     resp = self._export(serialized_data, deadline_sec - time())
                     if resp.ok:
+                        _log_partial_success(
+                            resp.content, ExportMetricsServiceResponse
+                        )
                         return MetricExportResult.SUCCESS
                 except requests.exceptions.RequestException as error:
                     reason = error

exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py

@@ -29,8 +29,12 @@
 from opentelemetry.exporter.otlp.proto.http._common import (
     _is_retryable,
     _load_session_from_envvar,
+    _log_partial_success,
 )
 from opentelemetry.metrics import MeterProvider
+from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import (
+    ExportTraceServiceResponse,
+)
 from opentelemetry.sdk.environment_variables import (
     _OTEL_PYTHON_EXPORTER_OTLP_HTTP_TRACES_CREDENTIAL_PROVIDER,
     OTEL_EXPORTER_OTLP_CERTIFICATE,
@@ -201,6 +205,9 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
                 try:
                     resp = self._export(serialized_data, deadline_sec - time())
                     if resp.ok:
+                        _log_partial_success(
+                            resp.content, ExportTraceServiceResponse
+                        )
                         return SpanExportResult.SUCCESS
                 except requests.exceptions.RequestException as error:
                     reason = error

exporter/opentelemetry-exporter-otlp-proto-http/tests/metrics/test_otlp_metrics_exporter.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 # pylint: disable=too-many-lines
+import logging
 import threading
 import time
 from logging import WARNING
@@ -30,6 +31,7 @@
 from opentelemetry.exporter.otlp.proto.http.version import __version__
 from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import (
     ExportMetricsServiceRequest,
+    ExportMetricsServiceResponse,
 )
 from opentelemetry.proto.common.v1.common_pb2 import (
     InstrumentationScope,
@@ -1321,6 +1323,43 @@ def test_2xx_status_code(self, mock_otlp_metric_exporter):
             MetricExportResult.SUCCESS,
         )
 
+    @patch.object(Session, "post")
+    def test_partial_success_logs_warning(self, mock_post):
+        """A partial-success response with rejected data points logs a warning."""
+        response = ExportMetricsServiceResponse()
+        response.partial_success.rejected_data_points = 5
+        response.partial_success.error_message = "some points were dropped"
+        resp = Response()
+        resp.status_code = 200
+        resp._content = response.SerializeToString()
+        mock_post.return_value = resp
+
+        with self.assertLogs(level=WARNING) as warning:
+            self.assertEqual(
+                OTLPMetricExporter().export(self.metrics["sum_int"]),
+                MetricExportResult.SUCCESS,
+            )
+        self.assertIn("5 items rejected", warning.output[0])
+        self.assertIn("some points were dropped", warning.output[0])
+
+    @patch.object(Session, "post")
+    def test_full_success_logs_nothing(self, mock_post):
+        """A fully successful (empty partial_success) response logs nothing."""
+        resp = Response()
+        resp.status_code = 200
+        resp._content = ExportMetricsServiceResponse().SerializeToString()
+        mock_post.return_value = resp
+
+        logger = logging.getLogger(
+            "opentelemetry.exporter.otlp.proto.http._common"
+        )
+        with patch.object(logger, "warning") as mock_warning:
+            self.assertEqual(
+                OTLPMetricExporter().export(self.metrics["sum_int"]),
+                MetricExportResult.SUCCESS,
+            )
+        mock_warning.assert_not_called()
+
     @patch.dict("os.environ", {}, clear=True)
     @patch.object(OTLPMetricExporter, "_export", return_value=Mock(ok=True))
     def test_exporter_metrics_disabled_after_set_meter_provider(

exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_log_exporter.py

@@ -3,6 +3,7 @@
 
 # pylint: disable=protected-access
 
+import logging
 import threading
 import time
 import unittest
@@ -27,6 +28,7 @@
 from opentelemetry.exporter.otlp.proto.http.version import __version__
 from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import (
     ExportLogsServiceRequest,
+    ExportLogsServiceResponse,
 )
 from opentelemetry.sdk._logs import ReadWriteLogRecord
 from opentelemetry.sdk._logs.export import LogRecordExportResult
@@ -456,6 +458,43 @@ def test_2xx_status_code(self, mock_otlp_metric_exporter):
             LogRecordExportResult.SUCCESS,
         )
 
+    @patch.object(Session, "post")
+    def test_partial_success_logs_warning(self, mock_post):
+        """A partial-success response with rejected log records logs a warning."""
+        response = ExportLogsServiceResponse()
+        response.partial_success.rejected_log_records = 2
+        response.partial_success.error_message = "some logs were dropped"
+        resp = Response()
+        resp.status_code = 200
+        resp._content = response.SerializeToString()
+        mock_post.return_value = resp
+
+        with self.assertLogs(level=WARNING) as warning:
+            self.assertEqual(
+                OTLPLogExporter().export(self._get_sdk_log_data()),
+                LogRecordExportResult.SUCCESS,
+            )
+        self.assertIn("2 items rejected", warning.output[0])
+        self.assertIn("some logs were dropped", warning.output[0])
+
+    @patch.object(Session, "post")
+    def test_full_success_logs_nothing(self, mock_post):
+        """A fully successful (empty partial_success) response logs nothing."""
+        resp = Response()
+        resp.status_code = 200
+        resp._content = ExportLogsServiceResponse().SerializeToString()
+        mock_post.return_value = resp
+
+        logger = logging.getLogger(
+            "opentelemetry.exporter.otlp.proto.http._common"
+        )
+        with patch.object(logger, "warning") as mock_warning:
+            self.assertEqual(
+                OTLPLogExporter().export(self._get_sdk_log_data()),
+                LogRecordExportResult.SUCCESS,
+            )
+        mock_warning.assert_not_called()
+
     @patch.dict(
         "os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: " true "}
     )