diff --git a/.changelog/5472.fixed b/.changelog/5472.fixed new file mode 100644 index 00000000000..915535e82d1 --- /dev/null +++ b/.changelog/5472.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: for both the simple and batch span/log processors, count `otel.sdk.processor.{span,log}.processed` when the processor submits records to the exporter instead of after export completes, and stop stamping exporter failures onto this metric as `error.type` diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py index b9763d92be9..8759f2dfae9 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py @@ -230,7 +230,6 @@ def on_emit(self, log_record: ReadWriteLogRecord): set_value(_ON_EMIT_RECURSION_COUNT_KEY, cnt + 1), # pyright: ignore[reportOperatorIssue] ) ) - error: Exception | None = None try: if self._shutdown: _logger.warning("Processor is already shutdown, ignoring call") @@ -248,12 +247,12 @@ def on_emit(self, log_record: ReadWriteLogRecord): instrumentation_scope=log_record.instrumentation_scope, limits=log_record.limits, ) + # Record on submission to the exporter. + self._metrics.finish_items(1) self._exporter.export((readable_log_record,)) - except Exception as err: # pylint: disable=broad-exception-caught - error = err + except Exception: # pylint: disable=broad-exception-caught _logger.exception("Exception while exporting logs.") finally: - self._metrics.finish_items(1, error) detach(token) def shutdown(self): diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 3e2b8a263a4..77c6027f517 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py @@ -172,27 +172,20 @@ def _export(self, batch_strategy: BatchExportStrategy) -> None: while self._should_export_batch(batch_strategy, iteration): iteration += 1 token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True)) - error: Exception | None = None - count = 0 + count = min( + self._max_export_batch_size, + len(self._queue), + ) + # Oldest records are at the back, so pop from there. + batch = [self._queue.pop() for _ in range(count)] + # Record on submission to the exporter. + self._metrics.finish_items(count) try: - count = min( - self._max_export_batch_size, - len(self._queue), - ) - self._exporter.export( - [ - # Oldest records are at the back, so pop from there. - self._queue.pop() - for _ in range(count) - ] - ) - except Exception as err: # pylint: disable=broad-exception-caught - error = err + self._exporter.export(batch) + except Exception: # pylint: disable=broad-exception-caught _logger.exception( "Exception while exporting %s.", self._exporting ) - finally: - self._metrics.finish_items(count, error) detach(token) def emit(self, data: Telemetry) -> None: diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/_processor_metrics.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/_processor_metrics.py index 9f7e7d7c068..59dfc2f65bd 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/_processor_metrics.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/_processor_metrics.py @@ -33,7 +33,7 @@ def register_queue_size( def drop_items(self, count: int) -> None: ... - def finish_items(self, count: int, error: Exception | None) -> None: ... + def finish_items(self, count: int) -> None: ... class NoOpProcessorMetrics: @@ -43,7 +43,7 @@ def register_queue_size(self, get_queue_size: Callable[[], int]) -> None: def drop_items(self, count: int) -> None: pass - def finish_items(self, count: int, error: Exception | None) -> None: + def finish_items(self, count: int) -> None: pass @@ -115,15 +115,8 @@ def record_queue_size( def drop_items(self, count: int) -> None: self._processed.add(count, self._dropped_attrs) - def finish_items(self, count: int, error: Exception | None) -> None: - if not error: - self._processed.add(count, self._standard_attrs) - return - attrs = { - **self._standard_attrs, - ERROR_TYPE: type(error).__name__, - } - self._processed.add(count, attrs) + def finish_items(self, count: int) -> None: + self._processed.add(count, self._standard_attrs) def create_processor_metrics( diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py index ad8f57840a0..7daaa638d78 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py @@ -124,16 +124,15 @@ def on_end(self, span: ReadableSpan) -> None: if not (span.context and span.context.trace_flags.sampled): return token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True)) - error: Exception | None = None + # Record on submission to the exporter. + self._metrics.finish_items(1) try: self.span_exporter.export((span,)) # pylint: disable=broad-exception-caught - except Exception as err: - error = err + except Exception: logger.exception("Exception while exporting Span.") finally: - self._metrics.finish_items(1, error) - detach(token) + detach(token) def shutdown(self) -> None: self.span_exporter.shutdown() diff --git a/opentelemetry-sdk/tests/logs/test_export.py b/opentelemetry-sdk/tests/logs/test_export.py index 5d8b4328cea..7fd99b228da 100644 --- a/opentelemetry-sdk/tests/logs/test_export.py +++ b/opentelemetry-sdk/tests/logs/test_export.py @@ -433,13 +433,12 @@ def export_logs(_logs): metrics = sorted(scope_metrics.metrics, key=lambda m: m.name) self.assertEqual(len(metrics), 1) self.assertEqual(metrics[0].name, "otel.sdk.processor.log.processed") - processed_data_points = sorted( - metrics[0].data.data_points, - key=lambda dp: dp.attributes.get("error.type", ""), - ) - self.assertEqual(len(processed_data_points), 2) + processed_data_points = metrics[0].data.data_points + self.assertEqual(len(processed_data_points), 1) processed_data_point0 = processed_data_points[0] - self.assertEqual(processed_data_point0.value, 2) + # All 3 logs are counted as processed when submitted to the exporter, + # independent of the export outcome (the 3rd export fails). + self.assertEqual(processed_data_point0.value, 3) self.assertEqual( processed_data_point0.attributes["otel.component.type"], "simple_log_processor", @@ -450,20 +449,39 @@ def export_logs(_logs): ) ) self.assertIsNone(processed_data_point0.attributes.get("error.type")) - processed_data_point1 = processed_data_points[1] - self.assertEqual(processed_data_point1.value, 1) - self.assertEqual( - processed_data_point1.attributes["otel.component.type"], - "simple_log_processor", - ) - self.assertTrue( - processed_data_point1.attributes["otel.component.name"].startswith( - "simple_log_processor/" - ) + + @patch.dict( + "os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"} + ) + def test_metrics_not_counted_after_shutdown(self): + metric_reader = InMemoryMetricReader() + meter_provider = MeterProvider(metric_readers=[metric_reader]) + + exporter = mock.MagicMock() + exporter.export.return_value = LogRecordExportResult.SUCCESS + processor = SimpleLogRecordProcessor( + exporter, meter_provider=meter_provider ) - self.assertEqual( - processed_data_point1.attributes["error.type"], "RuntimeError" + + processor.on_emit(EMPTY_LOG) + + # Shut only the processor down; the record emitted afterwards hits the + # already-shutdown early return and must not be counted as processed. + processor.shutdown() + processor.on_emit(EMPTY_LOG) + + metrics_data = metric_reader.get_metrics_data() + scope_metrics = metrics_data.resource_metrics[0].scope_metrics[0] + metrics = scope_metrics.metrics + self.assertEqual(len(metrics), 1) + self.assertEqual(metrics[0].name, "otel.sdk.processor.log.processed") + processed_data_points = metrics[0].data.data_points + self.assertEqual(len(processed_data_points), 1) + self.assertEqual(processed_data_points[0].value, 1) + self.assertIsNone( + processed_data_points[0].attributes.get("error.type") ) + self.assertEqual(exporter.export.call_count, 1) # Many more test cases for the BatchLogRecordProcessor exist under @@ -746,7 +764,9 @@ def export_logs(_logs): metrics[0].data.data_points, key=lambda dp: dp.attributes.get("error.type", ""), ) - self.assertEqual(len(processed_data_points), 1) + # "foo" is counted as processed when submitted to the exporter (before + # its export call blocks); "baz" is dropped due to a full queue. + self.assertEqual(len(processed_data_points), 2) processed_data_point0 = processed_data_points[0] self.assertEqual(processed_data_point0.value, 1) self.assertEqual( @@ -758,8 +778,12 @@ def export_logs(_logs): "batching_log_processor/" ) ) + self.assertIsNone(processed_data_point0.attributes.get("error.type")) + processed_data_point_queue_full = processed_data_points[1] + self.assertEqual(processed_data_point_queue_full.value, 1) self.assertEqual( - processed_data_point0.attributes.get("error.type"), "queue_full" + processed_data_point_queue_full.attributes.get("error.type"), + "queue_full", ) self.assertEqual( metrics[1].name, "otel.sdk.processor.log.queue.capacity" @@ -806,9 +830,12 @@ def export_logs(_logs): metrics[0].data.data_points, key=lambda dp: dp.attributes.get("error.type", ""), ) - self.assertEqual(len(processed_data_points), 3) + # "foo", "bar" and "failed" are all counted as processed when submitted + # to the exporter, independent of the export outcome ("failed" raises). + # "baz" remains a queue_full drop. + self.assertEqual(len(processed_data_points), 2) processed_data_point0 = processed_data_points[0] - self.assertEqual(processed_data_point0.value, 2) + self.assertEqual(processed_data_point0.value, 3) self.assertEqual( processed_data_point0.attributes["otel.component.type"], "batching_log_processor", @@ -831,22 +858,7 @@ def export_logs(_logs): ) ) self.assertEqual( - processed_data_point1.attributes.get("error.type"), - "BrokenPipeError", - ) - processed_data_point2 = processed_data_points[2] - self.assertEqual(processed_data_point2.value, 1) - self.assertEqual( - processed_data_point2.attributes["otel.component.type"], - "batching_log_processor", - ) - self.assertTrue( - processed_data_point2.attributes["otel.component.name"].startswith( - "batching_log_processor/" - ) - ) - self.assertEqual( - processed_data_point2.attributes.get("error.type"), "queue_full" + processed_data_point1.attributes.get("error.type"), "queue_full" ) self.assertEqual( metrics[1].name, "otel.sdk.processor.log.queue.capacity" diff --git a/opentelemetry-sdk/tests/trace/export/test_export.py b/opentelemetry-sdk/tests/trace/export/test_export.py index d67aceb19f8..d08bdbadcdb 100644 --- a/opentelemetry-sdk/tests/trace/export/test_export.py +++ b/opentelemetry-sdk/tests/trace/export/test_export.py @@ -172,13 +172,12 @@ def export_spans(_spans): metrics = sorted(scope_metrics.metrics, key=lambda m: m.name) self.assertEqual(len(metrics), 1) self.assertEqual(metrics[0].name, "otel.sdk.processor.span.processed") - processed_data_points = sorted( - metrics[0].data.data_points, - key=lambda dp: dp.attributes.get("error.type", ""), - ) - self.assertEqual(len(processed_data_points), 2) + processed_data_points = metrics[0].data.data_points + self.assertEqual(len(processed_data_points), 1) processed_data_point0 = processed_data_points[0] - self.assertEqual(processed_data_point0.value, 2) + # All 3 spans are counted as processed when submitted to the exporter, + # independent of the export outcome (the 3rd export fails). + self.assertEqual(processed_data_point0.value, 3) self.assertEqual( processed_data_point0.attributes["otel.component.type"], "simple_span_processor", @@ -189,20 +188,6 @@ def export_spans(_spans): ) ) self.assertIsNone(processed_data_point0.attributes.get("error.type")) - processed_data_point1 = processed_data_points[1] - self.assertEqual(processed_data_point1.value, 1) - self.assertEqual( - processed_data_point1.attributes["otel.component.type"], - "simple_span_processor", - ) - self.assertTrue( - processed_data_point1.attributes["otel.component.name"].startswith( - "simple_span_processor/" - ) - ) - self.assertEqual( - processed_data_point1.attributes["error.type"], "RuntimeError" - ) # Many more test cases for the BatchSpanProcessor exist under @@ -447,7 +432,9 @@ def export_spans(_spans): metrics[0].data.data_points, key=lambda dp: dp.attributes.get("error.type", ""), ) - self.assertEqual(len(processed_data_points), 1) + # "foo" is counted as processed when submitted to the exporter (before + # its export call blocks); "baz" is dropped due to a full queue. + self.assertEqual(len(processed_data_points), 2) processed_data_point0 = processed_data_points[0] self.assertEqual(processed_data_point0.value, 1) self.assertEqual( @@ -459,8 +446,12 @@ def export_spans(_spans): "batching_span_processor/" ) ) + self.assertIsNone(processed_data_point0.attributes.get("error.type")) + processed_data_point_queue_full = processed_data_points[1] + self.assertEqual(processed_data_point_queue_full.value, 1) self.assertEqual( - processed_data_point0.attributes.get("error.type"), "queue_full" + processed_data_point_queue_full.attributes.get("error.type"), + "queue_full", ) self.assertEqual( metrics[1].name, "otel.sdk.processor.span.queue.capacity" @@ -508,9 +499,12 @@ def export_spans(_spans): metrics[0].data.data_points, key=lambda dp: dp.attributes.get("error.type", ""), ) - self.assertEqual(len(processed_data_points), 3) + # "foo", "bar" and "failed" are all counted as processed when submitted + # to the exporter, independent of the export outcome ("failed" raises). + # "baz" remains a queue_full drop. + self.assertEqual(len(processed_data_points), 2) processed_data_point0 = processed_data_points[0] - self.assertEqual(processed_data_point0.value, 2) + self.assertEqual(processed_data_point0.value, 3) self.assertEqual( processed_data_point0.attributes["otel.component.type"], "batching_span_processor", @@ -533,21 +527,7 @@ def export_spans(_spans): ) ) self.assertEqual( - processed_data_point1.attributes.get("error.type"), "ValueError" - ) - processed_data_point2 = processed_data_points[2] - self.assertEqual(processed_data_point2.value, 1) - self.assertEqual( - processed_data_point2.attributes["otel.component.type"], - "batching_span_processor", - ) - self.assertTrue( - processed_data_point2.attributes["otel.component.name"].startswith( - "batching_span_processor/" - ) - ) - self.assertEqual( - processed_data_point2.attributes.get("error.type"), "queue_full" + processed_data_point1.attributes.get("error.type"), "queue_full" ) self.assertEqual( metrics[1].name, "otel.sdk.processor.span.queue.capacity"