diff --git a/.changelog/13.fixed b/.changelog/13.fixed new file mode 100644 index 00000000000..1b1b9b0760a --- /dev/null +++ b/.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. diff --git a/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py b/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py index f7fa0b8697d..fb23429b0b4 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py +++ b/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] diff --git a/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py b/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py index 50996264a09..58fcb273c2b 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py +++ b/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.", diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/__init__.py b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/__init__.py index 57bd7ca065a..8386419a123 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_common/__init__.py +++ b/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: diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py index a133874fcfa..e3f25685906 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py +++ b/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 diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py index 6cf13deae17..d43c3adf127 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py +++ b/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 diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py b/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py index 2be240103c0..7e4b8d5daf8 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py +++ b/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 diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/tests/metrics/test_otlp_metrics_exporter.py b/exporter/opentelemetry-exporter-otlp-proto-http/tests/metrics/test_otlp_metrics_exporter.py index eb9839d07c9..1ad29d1e2f9 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/tests/metrics/test_otlp_metrics_exporter.py +++ b/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( diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_log_exporter.py b/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_log_exporter.py index d7f3592e288..c08d65bfbe6 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_log_exporter.py +++ b/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 "} ) diff --git a/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_span_exporter.py b/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_span_exporter.py index 1580e5a1802..dc546f687fd 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_span_exporter.py +++ b/exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_span_exporter.py @@ -1,6 +1,7 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +import logging import threading import time import unittest @@ -21,6 +22,9 @@ OTLPSpanExporter, ) from opentelemetry.exporter.otlp.proto.http.version import __version__ +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, @@ -277,6 +281,43 @@ def test_2xx_status_code(self, mock_otlp_metric_exporter): OTLPSpanExporter().export(MagicMock()), SpanExportResult.SUCCESS ) + @patch.object(Session, "post") + def test_partial_success_logs_warning(self, mock_post): + """A partial-success response with rejected spans logs a warning.""" + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = 3 + response.partial_success.error_message = "some spans 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( + OTLPSpanExporter().export([BASIC_SPAN]), + SpanExportResult.SUCCESS, + ) + self.assertIn("3 items rejected", warning.output[0]) + self.assertIn("some spans 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 = ExportTraceServiceResponse().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( + OTLPSpanExporter().export([BASIC_SPAN]), + SpanExportResult.SUCCESS, + ) + mock_warning.assert_not_called() + @patch.dict("os.environ", {}, clear=True) @patch.object(OTLPSpanExporter, "_export", return_value=Mock(ok=True)) def test_exporter_metrics_disabled_by_default(self, _mock_export):