diff --git a/.changelog/5871.fixed b/.changelog/5871.fixed new file mode 100644 index 00000000000..343f3214379 --- /dev/null +++ b/.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 diff --git a/exporter/opentelemetry-exporter-otlp-common/src/opentelemetry/exporter/otlp/common/http.py b/exporter/opentelemetry-exporter-otlp-common/src/opentelemetry/exporter/otlp/common/http.py index aadfb24eb10..89afc2bc235 100644 --- a/exporter/opentelemetry-exporter-otlp-common/src/opentelemetry/exporter/otlp/common/http.py +++ b/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) diff --git a/exporter/opentelemetry-exporter-otlp-common/tests/test_http_client.py b/exporter/opentelemetry-exporter-otlp-common/tests/test_http_client.py index 76d79b29fcb..be0ab7ce244 100644 --- a/exporter/opentelemetry-exporter-otlp-common/tests/test_http_client.py +++ b/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), diff --git a/exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/_partial_success.py b/exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/_partial_success.py new file mode 100644 index 00000000000..6998f825804 --- /dev/null +++ b/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 ``ExportServiceResponse`` 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 ``ExportServiceResponse`` 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) diff --git a/exporter/opentelemetry-exporter-otlp-proto-common/tests/test_partial_success.py b/exporter/opentelemetry-exporter-otlp-proto-common/tests/test_partial_success.py new file mode 100644 index 00000000000..fd4f30c14a8 --- /dev/null +++ b/exporter/opentelemetry-exporter-otlp-proto-common/tests/test_partial_success.py @@ -0,0 +1,225 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +import unittest +from logging import DEBUG, WARNING, getLogger +from unittest.mock import patch + +from opentelemetry.exporter.otlp.proto.common._internal._partial_success import ( + _REJECTED_FIELDS, + _log_partial_success, + _partial_success_detail, + _warning_format, +) +from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import ( + ExportLogsServiceResponse, +) +from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( + ExportMetricsServiceResponse, +) +from opentelemetry.proto.collector.profiles.v1development.profiles_service_pb2 import ( + ExportProfilesServiceResponse, +) +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( + ExportTraceServiceRequest, + ExportTraceServiceResponse, +) + +_logger = getLogger(__name__) +_ENDPOINT = "http://collector.test/v1/traces" + + +class TestPartialSuccessWarning(unittest.TestCase): + def test_absent_partial_success_is_not_reported(self): + self.assertIsNone(_partial_success_detail(ExportTraceServiceResponse())) + + def test_absent_response_is_not_reported(self): + self.assertIsNone(_partial_success_detail(None)) + + def test_a_message_type_without_the_field_is_not_reported(self): + self.assertIsNone(_partial_success_detail(ExportTraceServiceRequest())) + + def test_zeroed_partial_success_is_not_reported(self): + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = 0 + response.partial_success.error_message = "" + + self.assertIsNone(_partial_success_detail(response)) + + def test_rejected_spans_are_reported_with_the_error_message(self): + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = 3 + response.partial_success.error_message = "some spans were invalid" + + self.assertEqual( + _partial_success_detail(response), + "the server rejected 3 spans: some spans were invalid", + ) + + def test_rejected_spans_are_reported_without_an_error_message(self): + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = 5 + + self.assertEqual( + _partial_success_detail(response), + "the server rejected 5 spans", + ) + + def test_error_message_alone_is_reported(self): + response = ExportTraceServiceResponse() + response.partial_success.error_message = "deprecated attribute in use" + + self.assertEqual( + _partial_success_detail(response), + "deprecated attribute in use", + ) + + def test_rejected_data_points_are_reported(self): + response = ExportMetricsServiceResponse() + response.partial_success.rejected_data_points = 7 + + self.assertEqual( + _partial_success_detail(response), + "the server rejected 7 data points", + ) + + def test_rejected_log_records_are_reported(self): + response = ExportLogsServiceResponse() + response.partial_success.rejected_log_records = 2 + + self.assertEqual( + _partial_success_detail(response), + "the server rejected 2 log records", + ) + + def test_rejected_profiles_are_reported(self): + response = ExportProfilesServiceResponse() + response.partial_success.rejected_profiles = 6 + + self.assertEqual( + _partial_success_detail(response), + "the server rejected 6 profiles", + ) + + def test_a_single_rejected_item_is_reported_in_the_singular(self): + responses = ( + (ExportTraceServiceResponse(), "rejected_spans", "span"), + (ExportMetricsServiceResponse(), "rejected_data_points", "data point"), + (ExportLogsServiceResponse(), "rejected_log_records", "log record"), + (ExportProfilesServiceResponse(), "rejected_profiles", "profile"), + ) + + for response, field_name, kind in responses: + with self.subTest(kind=kind): + setattr(response.partial_success, field_name, 1) + + self.assertEqual( + _partial_success_detail(response), + f"the server rejected 1 {kind}", + ) + + def test_every_rejected_field_the_protos_define_is_covered(self): + responses = ( + ExportTraceServiceResponse(), + ExportMetricsServiceResponse(), + ExportLogsServiceResponse(), + ExportProfilesServiceResponse(), + ) + + for response in responses: + with self.subTest(response=type(response).__name__): + names = {field.name for field in response.partial_success.DESCRIPTOR.fields} + rejected = {name for name in names if name.startswith("rejected_")} + self.assertTrue(rejected.issubset(set(_REJECTED_FIELDS))) + + +class TestLogPartialSuccess(unittest.TestCase): + def test_empty_body_logs_nothing(self): + with self.assertNoLogs(_logger, WARNING): + _log_partial_success(b"", ExportTraceServiceResponse, _logger, "spans", _ENDPOINT) + + def test_body_carrying_a_zeroed_partial_success_logs_nothing(self): + zeroed = ExportTraceServiceResponse() + zeroed.partial_success.rejected_spans = 0 + serialized = zeroed.SerializeToString() + self.assertTrue(serialized) + + with self.assertNoLogs(_logger, WARNING): + _log_partial_success(serialized, ExportTraceServiceResponse, _logger, "spans", _ENDPOINT) + + def test_body_with_partial_success_is_logged(self): + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = 4 + response.partial_success.error_message = "quota exceeded" + + with self.assertLogs(_logger, WARNING) as logs: + _log_partial_success(response.SerializeToString(), ExportTraceServiceResponse, _logger, "spans", _ENDPOINT) + + self.assertEqual( + logs.records[0].getMessage(), + "OTLP partial success exporting spans to http://collector.test/v1/traces: " + "the server rejected 4 spans: quota exceeded", + ) + + def test_the_logged_message_stays_constant_so_duplicates_collapse(self): + records = [] + for rejected in (1, 2, 3): + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = rejected + + with self.assertLogs(_logger, WARNING) as logs: + _log_partial_success( + response.SerializeToString(), ExportTraceServiceResponse, _logger, "spans", _ENDPOINT + ) + records.append(logs.records[0]) + + self.assertEqual({record.msg for record in records}, {_warning_format("spans")}) + self.assertEqual(len({record.getMessage() for record in records}), 3) + + def test_the_logged_message_differs_between_signals_so_they_do_not_collapse(self): + records = [] + for signal in ("spans", "metrics", "logs"): + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = 1 + + with self.assertLogs(_logger, WARNING) as logs: + _log_partial_success( + response.SerializeToString(), ExportTraceServiceResponse, _logger, signal, _ENDPOINT + ) + records.append(logs.records[0]) + + self.assertEqual(len({record.msg for record in records}), 3) + + def test_unparsable_body_is_logged_at_debug_and_ignored(self): + with self.assertLogs(_logger, DEBUG) as logs: + _log_partial_success(b"\xff\xff\xff\xff", ExportTraceServiceResponse, _logger, "spans", _ENDPOINT) + + self.assertEqual([record.levelno for record in logs.records], [DEBUG]) + self.assertIn("Failed to parse the OTLP export response", logs.records[0].getMessage()) + + def test_non_protobuf_body_is_not_warned_about(self): + for body in (b"ok", b'{"partialSuccess":{}}'): + with self.subTest(body=body), self.assertNoLogs(_logger, WARNING): + _log_partial_success(body, ExportTraceServiceResponse, _logger, "spans", _ENDPOINT) + + def test_invalid_utf8_in_the_error_message_is_ignored(self): + body = b"\x0a\x04\x12\x02\xff\xfe" + + with self.assertLogs(_logger, DEBUG) as logs: + _log_partial_success(body, ExportTraceServiceResponse, _logger, "spans", _ENDPOINT) + + self.assertEqual([record.levelno for record in logs.records], [DEBUG]) + + def test_a_parser_raising_unicode_decode_error_is_ignored(self): + error = UnicodeDecodeError("utf-8", b"\xff\xfe", 0, 1, "invalid start byte") + + with patch.object(ExportTraceServiceResponse, "ParseFromString", side_effect=error): + with self.assertLogs(_logger, DEBUG) as logs: + _log_partial_success(b"irrelevant", ExportTraceServiceResponse, _logger, "spans", _ENDPOINT) + + self.assertEqual([record.levelno for record in logs.records], [DEBUG]) + self.assertIn("Failed to parse the OTLP export response", logs.records[0].getMessage()) + + +if __name__ == "__main__": + unittest.main() diff --git a/exporter/opentelemetry-exporter-otlp-proto-grpc/benchmarks/test_benchmark_trace_exporter.py b/exporter/opentelemetry-exporter-otlp-proto-grpc/benchmarks/test_benchmark_trace_exporter.py index b8ae543ce45..67fa698e4de 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-grpc/benchmarks/test_benchmark_trace_exporter.py +++ b/exporter/opentelemetry-exporter-otlp-proto-grpc/benchmarks/test_benchmark_trace_exporter.py @@ -8,6 +8,9 @@ from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( OTLPSpanExporter, ) +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( + ExportTraceServiceResponse, +) from opentelemetry.sdk.trace import TracerProvider, sampling from opentelemetry.sdk.trace.export import ( BatchSpanProcessor, @@ -26,7 +29,7 @@ def get_tracer_with_processor(span_processor_class): class MockTraceServiceStub: def __init__(self, channel): - self.Export = lambda *args, **kwargs: None + self.Export = lambda *args, **kwargs: ExportTraceServiceResponse() @patch( 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 4aea624602b..78e79390d06 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 @@ -50,6 +50,10 @@ from opentelemetry.exporter.otlp.proto.common._internal import ( _get_resource_data, ) +from opentelemetry.exporter.otlp.proto.common._internal._partial_success import ( + _partial_success_detail, + _warning_format, +) from opentelemetry.exporter.otlp.proto.grpc import ( _OTLP_GRPC_CHANNEL_OPTIONS, _timeout_from_env, @@ -430,11 +434,19 @@ 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(), ) + try: + detail = _partial_success_detail(response) + # pylint: disable-next=broad-exception-caught + except Exception as error: + logger.debug("Failed to read the OTLP export response: %s", error) + detail = None + if detail is not None: + logger.warning(_warning_format(self._exporting), self._endpoint, detail) 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 1d8a47be84e..b118ee4275e 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 @@ -8,7 +8,7 @@ from concurrent.futures import ( # pylint: disable=no-name-in-module ThreadPoolExecutor, ) -from logging import WARNING, getLogger +from logging import DEBUG, WARNING, getLogger from platform import system from typing import Any from unittest import TestCase @@ -32,6 +32,9 @@ OTLPExporterMixin, environ_to_compression, ) +from opentelemetry.exporter.otlp.proto.grpc.exporter import ( + logger as _exporter_logger, +) from opentelemetry.exporter.otlp.proto.grpc.version import __version__ from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( ExportTraceServiceRequest, @@ -42,6 +45,7 @@ TraceServiceStub, add_TraceServiceServicer_to_server, ) +from opentelemetry.sdk import _shared_internal from opentelemetry.sdk.environment_variables import ( _OTEL_PYTHON_EXPORTER_OTLP_GRPC_CREDENTIAL_PROVIDER, OTEL_EXPORTER_OTLP_COMPRESSION, @@ -109,12 +113,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_success_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_success_message = optional_partial_success_message # pylint: disable=invalid-name,unused-argument def Export(self, request, context): @@ -134,7 +142,12 @@ 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_success_message is not None: + response.partial_success.error_message = self.optional_partial_success_message + return response class ThreadWithReturnValue(threading.Thread): @@ -180,6 +193,9 @@ def setUp(self): trace_id=67545097771067222548457157018666467027, ), ) + for log_filter in _exporter_logger.filters: + if isinstance(log_filter, _shared_internal.DuplicateFilter): + log_filter.__dict__.pop("last_log", None) def tearDown(self): self.server.stop(None) @@ -409,6 +425,114 @@ def test_shutdown_interrupts_export_retry_backoff(self): "Shutdown in progress, aborting retry.", ) + def test_partial_success_is_logged(self): + add_TraceServiceServicer_to_server( + TraceServiceServicerWithExportParams( + StatusCode.OK, + optional_rejected_spans=2, + optional_partial_success_message="quota exceeded", + ), + self.server, + ) + + with self.assertLogs(level=WARNING) as warning: + self.assertEqual(self.exporter.export([self.span]), SpanExportResult.SUCCESS) + + self.assertEqual( + warning.records[0].getMessage(), + "OTLP partial success exporting traces to localhost:4317: the server rejected 2 spans: quota exceeded", + ) + + def test_partial_success_warning_without_rejected_spans_is_logged(self): + add_TraceServiceServicer_to_server( + TraceServiceServicerWithExportParams( + StatusCode.OK, + optional_partial_success_message="deprecated attribute in use", + ), + self.server, + ) + + with self.assertLogs(level=WARNING) as warning: + self.assertEqual(self.exporter.export([self.span]), SpanExportResult.SUCCESS) + + self.assertEqual( + warning.records[0].getMessage(), + "OTLP partial success exporting traces to localhost:4317: deprecated attribute in use", + ) + + def test_partial_success_of_a_second_signal_survives_the_duplicate_filter(self): + add_TraceServiceServicer_to_server( + TraceServiceServicerWithExportParams( + StatusCode.OK, + optional_partial_success_message="quota exceeded", + ), + self.server, + ) + + with patch.object(_shared_internal, "time", Mock(time=lambda: 0.0)): + with self.assertLogs(level=WARNING) as warning: + self.assertEqual(self.exporter.export([self.span]), SpanExportResult.SUCCESS) + with patch.object(OTLPSpanExporterForTesting, "_exporting", property(lambda _: "metrics")): + self.assertEqual(self.exporter.export([self.span]), SpanExportResult.SUCCESS) + + self.assertEqual( + [record.getMessage() for record in warning.records], + [ + "OTLP partial success exporting traces to localhost:4317: quota exceeded", + "OTLP partial success exporting metrics to localhost:4317: quota exceeded", + ], + ) + + def test_repeated_partial_success_of_one_signal_is_filtered_as_a_duplicate(self): + add_TraceServiceServicer_to_server( + TraceServiceServicerWithExportParams( + StatusCode.OK, + optional_rejected_spans=2, + ), + self.server, + ) + + with patch.object(_shared_internal, "time", Mock(time=lambda: 0.0)): + with self.assertLogs(level=WARNING) as warning: + self.assertEqual(self.exporter.export([self.span]), SpanExportResult.SUCCESS) + self.assertEqual(self.exporter.export([self.span]), SpanExportResult.SUCCESS) + + self.assertEqual( + [record.getMessage() for record in warning.records], + ["OTLP partial success exporting traces to localhost:4317: the server rejected 2 spans"], + ) + + def test_a_response_the_helper_cannot_read_does_not_fail_the_export(self): + add_TraceServiceServicer_to_server( + TraceServiceServicerWithExportParams(StatusCode.OK), + self.server, + ) + + with patch( + "opentelemetry.exporter.otlp.proto.grpc.exporter._partial_success_detail", + side_effect=AttributeError("error_message"), + ): + with self.assertLogs(_exporter_logger, DEBUG) as logs: + self.assertEqual(self.exporter.export([self.span]), SpanExportResult.SUCCESS) + + self.assertEqual([record.levelno for record in logs.records], [DEBUG]) + self.assertIn("Failed to read the OTLP export response", logs.records[0].getMessage()) + + def test_a_stub_returning_a_test_double_does_not_fail_the_export(self): + # pylint: disable=protected-access + self.exporter._client = Mock() + + self.assertEqual(self.exporter.export([self.span]), SpanExportResult.SUCCESS) + + def test_full_success_logs_nothing(self): + add_TraceServiceServicer_to_server( + TraceServiceServicerWithExportParams(StatusCode.OK), + self.server, + ) + + with self.assertNoLogs(level=WARNING): + self.assertEqual(self.exporter.export([self.span]), SpanExportResult.SUCCESS) + def test_export_over_closed_grpc_channel(self): # pylint: disable=protected-access 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 69e7d434983..50a80359407 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 @@ -12,6 +12,9 @@ from opentelemetry.exporter.otlp.common._exporter_metrics import ( create_exporter_metrics, ) +from opentelemetry.exporter.otlp.proto.common._internal._partial_success import ( + _log_partial_success, +) from opentelemetry.exporter.otlp.proto.common._log_encoder import encode_logs from opentelemetry.exporter.otlp.proto.http import Compression from opentelemetry.exporter.otlp.proto.http._common import ( @@ -27,6 +30,9 @@ _resolve_timeout, ) 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, @@ -209,6 +215,7 @@ def export(self, batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult: else None ) return LogRecordExportResult.FAILURE + _log_partial_success(export_result.content, ExportLogsServiceResponse, _logger, "logs", self._endpoint) return LogRecordExportResult.SUCCESS def force_flush(self, timeout_millis: int = 10_000) -> bool: 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 0509696e349..e4cfc04a63d 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 @@ -25,6 +25,9 @@ from opentelemetry.exporter.otlp.proto.common._internal import ( _get_resource_data, ) +from opentelemetry.exporter.otlp.proto.common._internal._partial_success import ( + _log_partial_success, +) from opentelemetry.exporter.otlp.proto.common.metrics_encoder import ( encode_metrics, ) @@ -44,6 +47,7 @@ from opentelemetry.metrics import MeterProvider from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( ExportMetricsServiceRequest, + ExportMetricsServiceResponse, ) from opentelemetry.proto.common.v1.common_pb2 import ( # noqa: F401 AnyValue, @@ -257,6 +261,9 @@ def _export_batch( else None ) return MetricExportResult.FAILURE + _log_partial_success( + export_result.content, ExportMetricsServiceResponse, _logger, "metrics", self._endpoint + ) return MetricExportResult.SUCCESS def export( 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 adb6787aaf4..3e53566f5ce 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 @@ -12,6 +12,9 @@ from opentelemetry.exporter.otlp.common._exporter_metrics import ( create_exporter_metrics, ) +from opentelemetry.exporter.otlp.proto.common._internal._partial_success import ( + _log_partial_success, +) from opentelemetry.exporter.otlp.proto.common.trace_encoder import ( encode_spans, ) @@ -29,6 +32,9 @@ _resolve_timeout, ) 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_TRACES_CERTIFICATE, @@ -205,6 +211,7 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: else None ) return SpanExportResult.FAILURE + _log_partial_success(export_result.content, ExportTraceServiceResponse, _logger, "spans", self._endpoint) return SpanExportResult.SUCCESS def shutdown(self): 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 4d4099ddbe4..cb6285b8158 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 @@ -40,6 +40,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, @@ -205,6 +206,46 @@ def test_negative_max_request_size_disables_limit(self): self.assertEqual(result, MetricExportResult.SUCCESS) self.assertEqual(len(Mocket.request_list()), 1) + @mocketize + def test_partial_success_is_logged(self): + response = ExportMetricsServiceResponse() + response.partial_success.rejected_data_points = 4 + response.partial_success.error_message = "quota exceeded" + Entry.single_register( + Entry.POST, + _TEST_ENDPOINT, + status=200, + body=response.SerializeToString(), + ) + exporter = OTLPMetricExporter(endpoint=_TEST_ENDPOINT) + + with self.assertLogs(level="WARNING") as logs: + result = exporter.export(self.metrics["sum_int"]) + + self.assertEqual(result, MetricExportResult.SUCCESS) + self.assertEqual( + logs.records[0].getMessage(), + "OTLP partial success exporting metrics to " + f"{_TEST_ENDPOINT}: the server rejected 4 data points: quota exceeded", + ) + + @mocketize + def test_full_success_logs_nothing(self): + zeroed = ExportMetricsServiceResponse() + zeroed.partial_success.rejected_data_points = 0 + Entry.single_register( + Entry.POST, + _TEST_ENDPOINT, + status=200, + body=zeroed.SerializeToString(), + ) + exporter = OTLPMetricExporter(endpoint=_TEST_ENDPOINT) + + with self.assertNoLogs(level="WARNING"): + result = exporter.export(self.metrics["sum_int"]) + + self.assertEqual(result, MetricExportResult.SUCCESS) + @mocketize def test_oversized_payload_dropped_with_batch_splitting(self): exporter = OTLPMetricExporter( 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 b6dfa260620..fdd670fac02 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 @@ -35,12 +35,17 @@ DEFAULT_LOGS_EXPORT_PATH, OTLPLogExporter, ) +from opentelemetry.exporter.otlp.proto.http._log_exporter import ( + _logger as _log_exporter_logger, +) 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 +from opentelemetry.sdk._shared_internal import DuplicateFilter from opentelemetry.sdk.environment_variables import ( _OTEL_PYTHON_EXPORTER_OTLP_HTTP_LOGS_CREDENTIAL_PROVIDER, OTEL_EXPORTER_OTLP_COMPRESSION, @@ -101,6 +106,10 @@ def setUp(self): self.metric_reader = InMemoryMetricReader() self.meter_provider = MeterProvider(metric_readers=[self.metric_reader]) + for log_filter in _log_exporter_logger.filters: + if isinstance(log_filter, DuplicateFilter): + log_filter.__dict__.pop("last_log", None) + @staticmethod def _mocked_shutdown_event() -> Mock: shutdown_event = Mock(spec=threading.Event) @@ -225,6 +234,46 @@ def test_compression_dual_enum_acceptance(self): self.assertIs(exporter._compression, expected) self.assertIs(exporter._client._compression, expected) + @mocketize + def test_partial_success_is_logged(self): + response = ExportLogsServiceResponse() + response.partial_success.rejected_log_records = 3 + response.partial_success.error_message = "quota exceeded" + Entry.single_register( + Entry.POST, + _TEST_ENDPOINT, + status=200, + body=response.SerializeToString(), + ) + exporter = OTLPLogExporter(endpoint=_TEST_ENDPOINT) + + with self.assertLogs(level="WARNING") as logs: + result = exporter.export([_make_log_record()]) + + self.assertEqual(result, LogRecordExportResult.SUCCESS) + self.assertEqual( + logs.records[0].getMessage(), + "OTLP partial success exporting logs to " + f"{_TEST_ENDPOINT}: the server rejected 3 log records: quota exceeded", + ) + + @mocketize + def test_full_success_logs_nothing(self): + zeroed = ExportLogsServiceResponse() + zeroed.partial_success.rejected_log_records = 0 + Entry.single_register( + Entry.POST, + _TEST_ENDPOINT, + status=200, + body=zeroed.SerializeToString(), + ) + exporter = OTLPLogExporter(endpoint=_TEST_ENDPOINT) + + with self.assertNoLogs(level="WARNING"): + result = exporter.export([_make_log_record()]) + + self.assertEqual(result, LogRecordExportResult.SUCCESS) + @mocketize def test_export_single_log(self): Entry.single_register(Entry.POST, _TEST_ENDPOINT, status=200) 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 b0dbb47a7cc..09939a773cf 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 @@ -41,6 +41,7 @@ from opentelemetry.exporter.otlp.proto.http.version import __version__ from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( ExportTraceServiceRequest, + ExportTraceServiceResponse, ) from opentelemetry.sdk.environment_variables import ( _OTEL_PYTHON_EXPORTER_OTLP_HTTP_TRACES_CREDENTIAL_PROVIDER, @@ -274,6 +275,45 @@ def test_export_single_span(self): sent_data = mock_request.call_args.kwargs["data"] self.assertEqual(_decode_body(sent_data), encode_spans(spans)) + @mocketize + def test_partial_success_is_logged(self): + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = 2 + response.partial_success.error_message = "quota exceeded" + Entry.single_register( + Entry.POST, + _TEST_ENDPOINT, + status=200, + body=response.SerializeToString(), + ) + exporter = OTLPSpanExporter(endpoint=_TEST_ENDPOINT) + + with self.assertLogs(level="WARNING") as logs: + result = exporter.export(self._make_span()) + + self.assertEqual(result, SpanExportResult.SUCCESS) + self.assertEqual( + logs.records[0].getMessage(), + f"OTLP partial success exporting spans to {_TEST_ENDPOINT}: the server rejected 2 spans: quota exceeded", + ) + + @mocketize + def test_full_success_logs_nothing(self): + zeroed = ExportTraceServiceResponse() + zeroed.partial_success.rejected_spans = 0 + Entry.single_register( + Entry.POST, + _TEST_ENDPOINT, + status=200, + body=zeroed.SerializeToString(), + ) + exporter = OTLPSpanExporter(endpoint=_TEST_ENDPOINT) + + with self.assertNoLogs(level="WARNING"): + result = exporter.export(self._make_span()) + + self.assertEqual(result, SpanExportResult.SUCCESS) + @mocketize def test_export_spans_different_resources(self): Entry.single_register(Entry.POST, _TEST_ENDPOINT, status=200)