From f4f222f817a3975c821170ff08a9df9d2b05e521 Mon Sep 17 00:00:00 2001 From: RichardoMu <44485717+RichardoMrMu@users.noreply.github.com> Date: Sun, 20 Sep 2026 17:49:29 +0800 Subject: [PATCH 1/3] fix(sdk): count logs dropped by the recursion guard SimpleLogRecordProcessor.on_emit discards a record when it detects a recursive emit loop, and says so in the log, but does not record the drop on otel.sdk.processor.log.processed. The already-shutdown early return nine lines below it does, with error.type=already_shutdown, so the two discard paths in the same method disagree about whether a dropped record is worth counting. Measured on main, driving a real recursion with an exporter that logs while exporting: recursion drop: {'': 4} -- 1 record dropped, 0 counted shutdown drop: {'already_shutdown': 1} -- 1 record dropped, 1 counted So the counter under-reports: a consumer summing otel.sdk.processor.log .processed cannot tell that a record was lost, even though the SDK knew it was dropping one. Note that drop_items selects between prebuilt attribute dicts rather than building them from its argument, so passing a new error_type without adding the matching dict would silently attribute the drop to error.type=queue_full. Hence the small change to _processor_metrics.py alongside it. The recursion guard was added in #4799 (merged 2025-12) and the processor accounting in #5472 (merged 2026-08); #5472's diff on this file starts after the try: block, so this path was simply never reached by that work. Assisted-by: Doubao --- .changelog/5676.fixed | 1 + .../sdk/_logs/_internal/export/__init__.py | 1 + .../_shared_internal/_processor_metrics.py | 7 ++ opentelemetry-sdk/tests/logs/test_export.py | 68 +++++++++++++++++++ 4 files changed, 77 insertions(+) create mode 100644 .changelog/5676.fixed diff --git a/.changelog/5676.fixed b/.changelog/5676.fixed new file mode 100644 index 00000000000..dfdc467d8ac --- /dev/null +++ b/.changelog/5676.fixed @@ -0,0 +1 @@ +Count logs dropped by `SimpleLogRecordProcessor`'s recursion guard on `otel.sdk.processor.log.processed` with `error.type=recursion` 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 a56cfc622ae..0848d13f4d5 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py @@ -208,6 +208,7 @@ def on_emit(self, log_record: ReadWriteLogRecord): _propagate_false_logger.warning( "SimpleLogRecordProcessor.on_emit has entered a recursive loop. Dropping log and exiting the loop." ) + self._metrics.drop_items(1, "recursion") return token = attach( set_value( 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 699f7e1821c..aae2b14d967 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/_processor_metrics.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/_processor_metrics.py @@ -76,6 +76,11 @@ def __init__( ERROR_TYPE: "already_shutdown", } + self._recursion_attrs = { + **self._standard_attrs, + ERROR_TYPE: "recursion", + } + if signal == "traces": create_processed = create_otel_sdk_processor_span_processed create_queue_capacity = create_otel_sdk_processor_span_queue_capacity @@ -114,6 +119,8 @@ def record_queue_size( 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) + elif error_type == "recursion": + self._processed.add(count, self._recursion_attrs) else: self._processed.add(count, self._dropped_attrs) diff --git a/opentelemetry-sdk/tests/logs/test_export.py b/opentelemetry-sdk/tests/logs/test_export.py index cc0b22d29c7..0ca6122687d 100644 --- a/opentelemetry-sdk/tests/logs/test_export.py +++ b/opentelemetry-sdk/tests/logs/test_export.py @@ -95,6 +95,74 @@ def export(self, batch: Sequence[ReadableLogRecord]): finally: root_logger.removeHandler(handler) + + @patch.dict("os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"}) + @mark.skipif( + (3, 13, 0) <= sys.version_info <= (3, 13, 5), + reason="This will fail on 3.13.5 due to https://github.com/python/cpython/pull/131812 which prevents the recursion from being detected.", + ) + def test_metrics_recursive_loop(self): + metric_reader = InMemoryMetricReader() + meter_provider = MeterProvider(metric_readers=[metric_reader]) + + class Exporter(LogRecordExporter): + def shutdown(self): + pass + + def force_flush(self, timeout_millis: int = 10_000) -> bool: + return True + + def export(self, batch: Sequence[ReadableLogRecord]): + logger = logging.getLogger("any logger..") + logger.warning("Something happened.") + + exporter = Exporter() + logger_provider = LoggerProvider() + logger_provider.add_log_record_processor( + SimpleLogRecordProcessor(exporter, meter_provider=meter_provider) + ) + root_logger = logging.getLogger() + handler = LoggingHandler( + level=logging.DEBUG, logger_provider=logger_provider + ) + root_logger.addHandler(handler) + propagate_false_logger = logging.getLogger( + "opentelemetry.sdk._logs._internal.export.propagate.false" + ) + try: + with self.assertLogs(propagate_false_logger) as cm: + root_logger.warning("hello!") + assert ( + "SimpleLogRecordProcessor.on_emit has entered a recursive loop" + in cm.output[0] + ) + finally: + root_logger.removeHandler(handler) + + 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") + data_points = sorted( + metrics[0].data.data_points, + key=lambda dp: dp.attributes.get("error.type", ""), + ) + # The record the recursion guard discards is counted as processed with + # error.type=recursion, so the total stays reconcilable with the number + # of records the processor accepted. + recursion_points = [ + dp + for dp in data_points + if dp.attributes.get("error.type") == "recursion" + ] + self.assertEqual( + len(recursion_points), + 1, + "the log dropped by the recursion guard was not counted", + ) + self.assertEqual(recursion_points[0].value, 1) + def test_simple_log_record_processor_default_level(self): exporter = InMemoryLogRecordExporter() logger_provider = LoggerProvider() From 53a657f507ecc5d62481ad4286a54d1e4d20baab Mon Sep 17 00:00:00 2001 From: RichardoMu <44485717+RichardoMrMu@users.noreply.github.com> Date: Sun, 27 Sep 2026 10:26:20 +0800 Subject: [PATCH 2/3] fix(sdk): add package name to changelog and fix formatting - add opentelemetry-sdk package name to .changelog/5676.fixed - remove extra blank line in test_export.py - run pre-commit to fix formatting --- .changelog/5676.fixed | 2 +- opentelemetry-sdk/tests/logs/test_export.py | 24 +++++---------------- 2 files changed, 6 insertions(+), 20 deletions(-) diff --git a/.changelog/5676.fixed b/.changelog/5676.fixed index dfdc467d8ac..149f94505f6 100644 --- a/.changelog/5676.fixed +++ b/.changelog/5676.fixed @@ -1 +1 @@ -Count logs dropped by `SimpleLogRecordProcessor`'s recursion guard on `otel.sdk.processor.log.processed` with `error.type=recursion` +`opentelemetry-sdk`: count logs dropped by `SimpleLogRecordProcessor`'s recursion guard on `otel.sdk.processor.log.processed` with `error.type=recursion` diff --git a/opentelemetry-sdk/tests/logs/test_export.py b/opentelemetry-sdk/tests/logs/test_export.py index 0ca6122687d..a3265772eff 100644 --- a/opentelemetry-sdk/tests/logs/test_export.py +++ b/opentelemetry-sdk/tests/logs/test_export.py @@ -95,7 +95,6 @@ def export(self, batch: Sequence[ReadableLogRecord]): finally: root_logger.removeHandler(handler) - @patch.dict("os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"}) @mark.skipif( (3, 13, 0) <= sys.version_info <= (3, 13, 5), @@ -118,24 +117,15 @@ def export(self, batch: Sequence[ReadableLogRecord]): exporter = Exporter() logger_provider = LoggerProvider() - logger_provider.add_log_record_processor( - SimpleLogRecordProcessor(exporter, meter_provider=meter_provider) - ) + logger_provider.add_log_record_processor(SimpleLogRecordProcessor(exporter, meter_provider=meter_provider)) root_logger = logging.getLogger() - handler = LoggingHandler( - level=logging.DEBUG, logger_provider=logger_provider - ) + handler = LoggingHandler(level=logging.DEBUG, logger_provider=logger_provider) root_logger.addHandler(handler) - propagate_false_logger = logging.getLogger( - "opentelemetry.sdk._logs._internal.export.propagate.false" - ) + propagate_false_logger = logging.getLogger("opentelemetry.sdk._logs._internal.export.propagate.false") try: with self.assertLogs(propagate_false_logger) as cm: root_logger.warning("hello!") - assert ( - "SimpleLogRecordProcessor.on_emit has entered a recursive loop" - in cm.output[0] - ) + assert "SimpleLogRecordProcessor.on_emit has entered a recursive loop" in cm.output[0] finally: root_logger.removeHandler(handler) @@ -151,11 +141,7 @@ def export(self, batch: Sequence[ReadableLogRecord]): # The record the recursion guard discards is counted as processed with # error.type=recursion, so the total stays reconcilable with the number # of records the processor accepted. - recursion_points = [ - dp - for dp in data_points - if dp.attributes.get("error.type") == "recursion" - ] + recursion_points = [dp for dp in data_points if dp.attributes.get("error.type") == "recursion"] self.assertEqual( len(recursion_points), 1, From f6c5ad5a3113a2a3ed176d57397a4a2977db4c75 Mon Sep 17 00:00:00 2001 From: RichardoMu <44485717+RichardoMrMu@users.noreply.github.com> Date: Sun, 27 Sep 2026 13:46:45 +0800 Subject: [PATCH 3/3] chore: prefix changelog fragment with opentelemetry-sdk package name Assisted-by: Claude