[bug] LogRecordProcessor exceptions in on_emit escape and crash the host application #5841

Describe your environment

• OS: Linux / macOS / Windows (All platforms)
• Python version: Python 3.9+ (All supported Python versions)
• SDK version: main (and all released versions of opentelemetry-sdk)

What happened?

Both SynchronousMultiLogRecordProcessor.on_emit and ConcurrentMultiLogRecordProcessor.on_emit in opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py invoke registered LogRecordProcessor instances without an exception guard (try/except).

If any LogRecordProcessor.on_emit (or an underlying exporter/queue) raises an exception, the exception is not caught or logged by the SDK. Instead:

1. Host application crashes: The unhandled exception propagates directly through logger.emit(...) or standard library logging calls (logging.Logger.info(...), logging.Logger.error(...) via LoggingHandler), crashing the user request or service thread.
2. Telemetry data loss: In SynchronousMultiLogRecordProcessor, an exception in one processor immediately halts the loop, preventing all subsequent registered log processors from ever receiving the log record.

OpenTelemetry Specification

Per the official OpenTelemetry Error Handling Specification:

"OpenTelemetry implementations MUST NOT throw unhandled exceptions at runtime."
"API methods that accept external callbacks MUST handle all errors."

Per the Logs SDK Specification (OnEmit):

"This method MUST be called synchronously... therefore it should not block or throw an exception."

Steps to Reproduce

1. Crashes standard Python logging calls:

import logging
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler, LogRecordProcessor

class FaultyLogProcessor(LogRecordProcessor):
    def on_emit(self, log_record):
        raise RuntimeError("Transient network or serialization failure in processor")

    def shutdown(self):
        pass

    def force_flush(self, timeout_millis=30000):
        return True

lp = LoggerProvider()
lp.add_log_record_processor(FaultyLogProcessor())

# Attach to Python stdlib logging
handler = LoggingHandler(logger_provider=lp)
py_logger = logging.getLogger("service")
py_logger.addHandler(handler)

# Host application calls logging in normal business logic
py_logger.error("User transaction failed, attempting rollback")

2. Crashes direct logger.emit and drops subsequent processors:

from opentelemetry._logs import LogRecord
from opentelemetry.sdk._logs import LoggerProvider, LogRecordProcessor

executed = []

class FaultyProcessor(LogRecordProcessor):
    def on_emit(self, log_record):
        executed.append("faulty")
        raise RuntimeError("Exporter queue error")
    def shutdown(self): pass
    def force_flush(self, timeout_millis=30000): return True

class SafeProcessor(LogRecordProcessor):
    def on_emit(self, log_record):
        executed.append("safe")
    def shutdown(self): pass
    def force_flush(self, timeout_millis=30000): return True

lp = LoggerProvider()
lp.add_log_record_processor(FaultyProcessor())
lp.add_log_record_processor(SafeProcessor())

logger = lp.get_logger("test")
logger.emit(LogRecord(body="important message"))

Expected Result

1. Observability code must not crash user production applications.
2. If a LogRecordProcessor encounters an error during on_emit, shutdown, or force_flush, the SDK should catch the exception and log it using _logger.exception("Exception while calling LogRecordProcessor.on_emit.").
3. Subsequent registered processors should still be invoked even if an earlier processor throws.
4. executed should be ['faulty', 'safe'], and the caller's thread continues unaffected.

Actual Result

The application immediately terminates with an uncaught RuntimeError:

Traceback (most recent call last):
  File "repro.py", line 25, in <module>
    logger.emit(LogRecord(body="important message"))
  File ".../opentelemetry/sdk/_logs/_internal/__init__.py", line 1083, in emit
    self._multi_log_record_processor.on_emit(read_write_log_record)
  File ".../opentelemetry/sdk/_logs/_internal/__init__.py", line 427, in on_emit
    lp.on_emit(log_record)
  File "repro.py", line 9, in on_emit
    raise RuntimeError("Exporter queue error")
RuntimeError: Exporter queue error

executed is only ['faulty'] ‚Äî SafeProcessor was completely bypassed and its log data dropped.

Proposed Fix

In opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py:

1. Wrap processor invocation in SynchronousMultiLogRecordProcessor.on_emit:

    def on_emit(self, log_record: ReadWriteLogRecord) -> None:
        for lp in self._log_record_processors:
            try:
                lp.on_emit(log_record)
            except Exception:
                _logger.exception("Exception while calling LogRecordProcessor.on_emit.")

2. Similarly guard shutdown() and force_flush() across both SynchronousMultiLogRecordProcessor and ConcurrentMultiLogRecordProcessor.

Would you like to implement a fix?

Yes, I would be happy to submit a PR with the fix and regression tests.