Count post-shutdown-dropped records as already_shutdown on processor.processed (#5509)

* Count post-shutdown-dropped records as processed with error.type=already_shutdown

Batch (span+log) and simple log processors previously dropped records
silently after shutdown. Count them on otel.sdk.processor.{span,log}.processed
with error.type=already_shutdown, matching the semantic conventions and the
.NET SDK. SimpleSpanProcessor is unchanged (it has no shutdown gate).

Assisted-by: Claude Opus 4.8

* Add changelog fragment for already_shutdown processed counting

Assisted-by: Claude Opus 4.8

* Fix lint: ruff-format wrap and too-many-lines disable

Assisted-by: Claude Opus 4.8

* Re-trigger CI

Signed-off-by: cijothomas <cijo.thomas@gmail.com>

---------

Signed-off-by: cijothomas <cijo.thomas@gmail.com>

Author: cijothomas

.changelog/5509.added

@@ -0,0 +1 @@
+`opentelemetry-sdk`: count records dropped after shutdown on `otel.sdk.processor.{span,log}.processed` with `error.type=already_shutdown` (batch span/log and simple log processors), which the semantic conventions define as a valid value for this metric.

opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py

@@ -233,6 +233,7 @@ def on_emit(self, log_record: ReadWriteLogRecord):
         try:
             if self._shutdown:
                 _logger.warning("Processor is already shutdown, ignoring call")
+                self._metrics.drop_items(1, "already_shutdown")
                 return
             # Convert ReadWriteLogRecord to ReadableLogRecord before exporting
             # Note: resource should not be None at this point as it's set during Logger.emit()

opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py

@@ -191,6 +191,7 @@ def _export(self, batch_strategy: BatchExportStrategy) -> None:
     def emit(self, data: Telemetry) -> None:
         if self._shutdown:
             _logger.info("Shutdown called, ignoring %s.", self._exporting)
+            self._metrics.drop_items(1, "already_shutdown")
             return
         if self._pid != os.getpid():
             self._bsp_reset_once.do_once(self._at_fork_reinit)

opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/_processor_metrics.py

@@ -31,7 +31,9 @@ def register_queue_size(
         self, get_queue_size: Callable[[], int]
     ) -> None: ...
 
-    def drop_items(self, count: int) -> None: ...
+    def drop_items(
+        self, count: int, error_type: str = "queue_full"
+    ) -> None: ...
 
     def finish_items(self, count: int) -> None: ...
 
@@ -40,7 +42,7 @@ class NoOpProcessorMetrics:
     def register_queue_size(self, get_queue_size: Callable[[], int]) -> None:
         pass
 
-    def drop_items(self, count: int) -> None:
+    def drop_items(self, count: int, error_type: str = "queue_full") -> None:
         pass
 
     def finish_items(self, count: int) -> None:
@@ -73,6 +75,11 @@ def __init__(
             ERROR_TYPE: "queue_full",
         }
 
+        self._already_shutdown_attrs = {
+            **self._standard_attrs,
+            ERROR_TYPE: "already_shutdown",
+        }
+
         if signal == "traces":
             create_processed = create_otel_sdk_processor_span_processed
             create_queue_capacity = (
@@ -112,8 +119,11 @@ def record_queue_size(
             unit=queue_size_unit,
         )
 
-    def drop_items(self, count: int) -> None:
-        self._processed.add(count, self._dropped_attrs)
+    def drop_items(self, count: int, error_type: str = "queue_full") -> None:
+        if error_type == "already_shutdown":
+            self._processed.add(count, self._already_shutdown_attrs)
+        else:
+            self._processed.add(count, self._dropped_attrs)
 
     def finish_items(self, count: int) -> None:
         self._processed.add(count, self._standard_attrs)

opentelemetry-sdk/tests/logs/test_export.py

@@ -1,7 +1,7 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
-# pylint: disable=protected-access
+# pylint: disable=protected-access,too-many-lines
 import logging
 import os
 import sys
@@ -453,7 +453,7 @@ def export_logs(_logs):
     @patch.dict(
         "os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"}
     )
-    def test_metrics_not_counted_after_shutdown(self):
+    def test_metrics_already_shutdown(self):
         metric_reader = InMemoryMetricReader()
         meter_provider = MeterProvider(metric_readers=[metric_reader])
 
@@ -466,7 +466,8 @@ def test_metrics_not_counted_after_shutdown(self):
         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.
+        # already-shutdown early return and is counted as processed with
+        # error.type=already_shutdown (never handed to the exporter).
         processor.shutdown()
         processor.on_emit(EMPTY_LOG)
 
@@ -475,11 +476,18 @@ def test_metrics_not_counted_after_shutdown(self):
         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")
+        data_points = sorted(
+            metrics[0].data.data_points,
+            key=lambda dp: dp.attributes.get("error.type", ""),
+        )
+        self.assertEqual(len(data_points), 2)
+        # Successful submission (before shutdown), no error.type.
+        self.assertEqual(data_points[0].value, 1)
+        self.assertIsNone(data_points[0].attributes.get("error.type"))
+        # Dropped after shutdown.
+        self.assertEqual(data_points[1].value, 1)
+        self.assertEqual(
+            data_points[1].attributes.get("error.type"), "already_shutdown"
         )
         self.assertEqual(exporter.export.call_count, 1)
 
@@ -709,6 +717,46 @@ def test_validation_negative_max_queue_size(self):
             max_export_batch_size=101,
         )
 
+    @patch.dict(
+        "os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"}
+    )
+    def test_metrics_already_shutdown(self):
+        metric_reader = InMemoryMetricReader()
+        meter_provider = MeterProvider(metric_readers=[metric_reader])
+
+        exporter = mock.MagicMock()
+        exporter.export.return_value = LogRecordExportResult.SUCCESS
+        processor = BatchLogRecordProcessor(
+            exporter, meter_provider=meter_provider
+        )
+
+        # Emitted before shutdown: drained on shutdown and counted as a
+        # successful submit to the exporter.
+        processor.on_emit(EMPTY_LOG)
+        processor.shutdown()
+
+        # Emitted after shutdown: dropped and counted as already_shutdown.
+        processor.on_emit(EMPTY_LOG)
+
+        metrics_data = metric_reader.get_metrics_data()
+        scope_metrics = metrics_data.resource_metrics[0].scope_metrics[0]
+        processed = next(
+            m
+            for m in scope_metrics.metrics
+            if m.name == "otel.sdk.processor.log.processed"
+        )
+        data_points = sorted(
+            processed.data.data_points,
+            key=lambda dp: dp.attributes.get("error.type", ""),
+        )
+        self.assertEqual(len(data_points), 2)
+        self.assertEqual(data_points[0].value, 1)
+        self.assertIsNone(data_points[0].attributes.get("error.type"))
+        self.assertEqual(data_points[1].value, 1)
+        self.assertEqual(
+            data_points[1].attributes.get("error.type"), "already_shutdown"
+        )
+
     @patch.dict(
         "os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"}
     )

opentelemetry-sdk/tests/trace/export/test_export.py

@@ -375,6 +375,51 @@ def test_batch_span_processor_parameters(self):
             max_export_batch_size=512,
         )
 
+    @mock.patch.dict(
+        "os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"}
+    )
+    def test_metrics_already_shutdown(self):
+        metric_reader = InMemoryMetricReader()
+        meter_provider = MeterProvider(metric_readers=[metric_reader])
+
+        exporter = mock.MagicMock()
+        exporter.export.return_value = export.SpanExportResult.SUCCESS
+        span_processor = export.BatchSpanProcessor(
+            exporter, meter_provider=meter_provider
+        )
+        provider = trace.TracerProvider()
+        tracer = provider.get_tracer(__name__)
+        provider.add_span_processor(span_processor)
+
+        # Ended before shutdown: drained on shutdown and counted as a
+        # successful submit to the exporter.
+        with tracer.start_as_current_span("foo"):
+            pass
+        span_processor.shutdown()
+
+        # Ended after shutdown: dropped and counted as already_shutdown.
+        with tracer.start_as_current_span("bar"):
+            pass
+
+        metrics_data = metric_reader.get_metrics_data()
+        scope_metrics = metrics_data.resource_metrics[0].scope_metrics[0]
+        processed = next(
+            m
+            for m in scope_metrics.metrics
+            if m.name == "otel.sdk.processor.span.processed"
+        )
+        data_points = sorted(
+            processed.data.data_points,
+            key=lambda dp: dp.attributes.get("error.type", ""),
+        )
+        self.assertEqual(len(data_points), 2)
+        self.assertEqual(data_points[0].value, 1)
+        self.assertIsNone(data_points[0].attributes.get("error.type"))
+        self.assertEqual(data_points[1].value, 1)
+        self.assertEqual(
+            data_points[1].attributes.get("error.type"), "already_shutdown"
+        )
+
     @mock.patch.dict(
         "os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"}
     )