opentelemetry-sdk: make it possible to customize processors configuration

CHANGELOG.md

@@ -20,6 +20,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4798](https://github.com/open-telemetry/opentelemetry-python/pull/4798))
 - Silence events API warnings for internal users
   ([#4847](https://github.com/open-telemetry/opentelemetry-python/pull/4847))
+- opentelemetry-sdk: make it possible to override the default processors in the SDK configurator
+  ([#4806](https://github.com/open-telemetry/opentelemetry-python/pull/4806))
 
 ## Version 1.39.0/0.60b0 (2025-12-03)
 

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

@@ -37,7 +37,11 @@
     OTEL_TRACES_EXPORTER,
 )
 from opentelemetry.metrics import set_meter_provider
-from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
+from opentelemetry.sdk._logs import (
+    LoggerProvider,
+    LoggingHandler,
+    LogRecordProcessor,
+)
 from opentelemetry.sdk._logs.export import (
     BatchLogRecordProcessor,
     LogRecordExporter,
@@ -58,7 +62,7 @@
     PeriodicExportingMetricReader,
 )
 from opentelemetry.sdk.resources import Attributes, Resource
-from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace import SpanProcessor, TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter
 from opentelemetry.sdk.trace.id_generator import IdGenerator
 from opentelemetry.sdk.trace.sampling import Sampler
@@ -105,6 +109,14 @@
 ]
 
 
+class ExporterSpanProcessor(SpanProcessor):
+    def __init__(self, span_exporter: SpanExporter, *args, **kwargs): ...
+
+
+class ExporterLogRecordProcessor(LogRecordProcessor):
+    def __init__(self, exporter: LogRecordExporter, *args, **kwargs): ...
+
+
 def _import_config_components(
     selected_components: Sequence[str], entry_point_name: str
 ) -> list[tuple[str, Type]]:
@@ -210,6 +222,8 @@ def _init_tracing(
     sampler: Sampler | None = None,
     resource: Resource | None = None,
     exporter_args_map: ExporterArgsMap | None = None,
+    span_processors: Sequence[SpanProcessor] | None = None,
+    export_span_processor: Type[ExporterSpanProcessor] | None = None,
 ):
     provider = TracerProvider(
         id_generator=id_generator,
@@ -219,10 +233,16 @@ def _init_tracing(
     set_tracer_provider(provider)
 
     exporter_args_map = exporter_args_map or {}
+    export_processor = export_span_processor or BatchSpanProcessor
+
+    span_processors = span_processors or []
+    for span_processor in span_processors:
+        provider.add_span_processor(span_processor)
+
     for _, exporter_class in exporters.items():
         exporter_args = exporter_args_map.get(exporter_class, {})
         provider.add_span_processor(
-            BatchSpanProcessor(exporter_class(**exporter_args))
+            export_processor(exporter_class(**exporter_args))
         )
 
 
@@ -256,15 +276,24 @@ def _init_logging(
     resource: Resource | None = None,
     setup_logging_handler: bool = True,
     exporter_args_map: ExporterArgsMap | None = None,
+    log_record_processors: Sequence[LogRecordProcessor] | None = None,
+    export_log_record_processor: Type[ExporterLogRecordProcessor]
+    | None = None,
 ):
     provider = LoggerProvider(resource=resource)
     set_logger_provider(provider)
 
     exporter_args_map = exporter_args_map or {}
+    export_processor = export_log_record_processor or BatchLogRecordProcessor
+
+    log_record_processors = log_record_processors or []
+    for log_record_processor in log_record_processors:
+        provider.add_log_record_processor(log_record_processor)
+
     for _, exporter_class in exporters.items():
         exporter_args = exporter_args_map.get(exporter_class, {})
         provider.add_log_record_processor(
-            BatchLogRecordProcessor(exporter_class(**exporter_args))
+            export_processor(exporter_class(**exporter_args))
         )
 
     # silence warnings from internal users until we drop the deprecated Events API
@@ -429,7 +458,13 @@ def _initialize_components(
     id_generator: IdGenerator | None = None,
     setup_logging_handler: bool | None = None,
     exporter_args_map: ExporterArgsMap | None = None,
+    span_processors: Sequence[SpanProcessor] | None = None,
+    export_span_processor: Type[ExporterSpanProcessor] | None = None,
+    log_record_processors: Sequence[LogRecordProcessor] | None = None,
+    export_log_record_processor: Type[ExporterLogRecordProcessor]
+    | None = None,
 ):
+    # pylint: disable=too-many-locals
     if trace_exporter_names is None:
         trace_exporter_names = []
     if metric_exporter_names is None:
@@ -464,6 +499,8 @@ def _initialize_components(
         sampler=sampler,
         resource=resource,
         exporter_args_map=exporter_args_map,
+        span_processors=span_processors,
+        export_span_processor=export_span_processor,
     )
     _init_metrics(
         metric_exporters, resource, exporter_args_map=exporter_args_map
@@ -482,6 +519,8 @@ def _initialize_components(
         resource,
         setup_logging_handler,
         exporter_args_map=exporter_args_map,
+        log_record_processors=log_record_processors,
+        export_log_record_processor=export_log_record_processor,
     )
 
 

opentelemetry-sdk/tests/test_configurator.py

@@ -45,9 +45,12 @@
     _initialize_components,
     _OTelSDKConfigurator,
 )
-from opentelemetry.sdk._logs import LoggingHandler
+from opentelemetry.sdk._logs import LoggingHandler, LogRecordProcessor
 from opentelemetry.sdk._logs._internal.export import LogRecordExporter
-from opentelemetry.sdk._logs.export import ConsoleLogRecordExporter
+from opentelemetry.sdk._logs.export import (
+    ConsoleLogRecordExporter,
+    SimpleLogRecordProcessor,
+)
 from opentelemetry.sdk.environment_variables import (
     OTEL_TRACES_SAMPLER,
     OTEL_TRACES_SAMPLER_ARG,
@@ -62,7 +65,11 @@
 )
 from opentelemetry.sdk.metrics.view import Aggregation
 from opentelemetry.sdk.resources import SERVICE_NAME, Resource
-from opentelemetry.sdk.trace.export import ConsoleSpanExporter
+from opentelemetry.sdk.trace import SpanProcessor
+from opentelemetry.sdk.trace.export import (
+    ConsoleSpanExporter,
+    SimpleSpanProcessor,
+)
 from opentelemetry.sdk.trace.id_generator import IdGenerator, RandomIdGenerator
 from opentelemetry.sdk.trace.sampling import (
     ALWAYS_ON,
@@ -82,23 +89,23 @@ class Provider:
     def __init__(self, resource=None, sampler=None, id_generator=None):
         self.sampler = sampler
         self.id_generator = id_generator
-        self.processor = None
+        self.processors = []
         self.resource = resource or Resource.create({})
 
     def add_span_processor(self, processor):
-        self.processor = processor
+        self.processors.append(processor)
 
 
 class DummyLoggerProvider:
     def __init__(self, resource=None):
         self.resource = resource
-        self.processor = DummyLogRecordProcessor(DummyOTLPLogExporter())
+        self.processors = []
 
     def add_log_record_processor(self, processor):
-        self.processor = processor
+        self.processors.append(processor)
 
     def get_logger(self, name, *args, **kwargs):
-        return DummyLogger(name, self.resource, self.processor)
+        return DummyLogger(name, self.resource, self.processors)
 
     def force_flush(self, *args, **kwargs):
         pass
@@ -109,10 +116,10 @@ class DummyMeterProvider(MeterProvider):
 
 
 class DummyLogger:
-    def __init__(self, name, resource, processor):
+    def __init__(self, name, resource, processors):
         self.name = name
         self.resource = resource
-        self.processor = processor
+        self.processors = processors
 
     def emit(
         self,
@@ -127,7 +134,8 @@ def emit(
         attributes=None,
         event_name=None,
     ):
-        self.processor.emit(record)
+        for processor in self.processors:
+            processor.emit(record)
 
 
 class DummyLogRecordProcessor:
@@ -351,10 +359,11 @@ def test_trace_init_default(self):
         provider = self.set_provider_mock.call_args[0][0]
         self.assertIsInstance(provider, Provider)
         self.assertIsInstance(provider.id_generator, RandomIdGenerator)
-        self.assertIsInstance(provider.processor, Processor)
-        self.assertIsInstance(provider.processor.exporter, Exporter)
+        self.assertEqual(len(provider.processors), 1)
+        self.assertIsInstance(provider.processors[0], Processor)
+        self.assertIsInstance(provider.processors[0].exporter, Exporter)
         self.assertEqual(
-            provider.processor.exporter.service_name, "my-test-service"
+            provider.processors[0].exporter.service_name, "my-test-service"
         )
         self.assertEqual(
             provider.resource.attributes.get("telemetry.auto.version"),
@@ -374,8 +383,11 @@ def test_trace_init_otlp(self):
         provider = self.set_provider_mock.call_args[0][0]
         self.assertIsInstance(provider, Provider)
         self.assertIsInstance(provider.id_generator, RandomIdGenerator)
-        self.assertIsInstance(provider.processor, Processor)
-        self.assertIsInstance(provider.processor.exporter, OTLPSpanExporter)
+        self.assertEqual(len(provider.processors), 1)
+        self.assertIsInstance(provider.processors[0], Processor)
+        self.assertIsInstance(
+            provider.processors[0].exporter, OTLPSpanExporter
+        )
         self.assertIsInstance(provider.resource, Resource)
         self.assertEqual(
             provider.resource.attributes.get("service.name"),
@@ -393,9 +405,26 @@ def test_trace_init_exporter_uses_exporter_args_map(self):
         )
 
         provider = self.set_provider_mock.call_args[0][0]
-        exporter = provider.processor.exporter
+        self.assertEqual(len(provider.processors), 1)
+        exporter = provider.processors[0].exporter
         self.assertEqual(exporter.compression, "gzip")
 
+    def test_trace_init_custom_span_processors(self):
+        span_processor = mock.Mock(spec=SpanProcessor)
+        _init_tracing(
+            {"otlp": OTLPSpanExporter},
+            id_generator=RandomIdGenerator(),
+            span_processors=[span_processor],
+            export_span_processor=SimpleSpanProcessor,
+        )
+
+        provider = self.set_provider_mock.call_args[0][0]
+        self.assertEqual(len(provider.processors), 2)
+        self.assertEqual(provider.processors[0], span_processor)
+        self.assertTrue(
+            isinstance(provider.processors[1], SimpleSpanProcessor)
+        )
+
     @patch.dict(environ, {OTEL_PYTHON_ID_GENERATOR: "custom_id_generator"})
     @patch("opentelemetry.sdk._configuration.IdGenerator", new=IdGenerator)
     @patch("opentelemetry.sdk._configuration.entry_points")
@@ -684,12 +713,16 @@ def test_logging_init_exporter(self):
                 provider.resource.attributes.get("service.name"),
                 "otlp-service",
             )
-            self.assertIsInstance(provider.processor, DummyLogRecordProcessor)
+            self.assertEqual(len(provider.processors), 1)
+            self.assertIsInstance(
+                provider.processors[0], DummyLogRecordProcessor
+            )
             self.assertIsInstance(
-                provider.processor.exporter, DummyOTLPLogExporter
+                provider.processors[0].exporter, DummyOTLPLogExporter
             )
             getLogger(__name__).error("hello")
-            self.assertTrue(provider.processor.exporter.export_called)
+            self.assertEqual(len(provider.processors), 1)
+            self.assertTrue(provider.processors[0].exporter.export_called)
 
     def test_logging_init_exporter_uses_exporter_args_map(self):
         with ResetGlobalLoggingState():
@@ -704,7 +737,27 @@ def test_logging_init_exporter_uses_exporter_args_map(self):
             )
             self.assertEqual(self.set_provider_mock.call_count, 1)
             provider = self.set_provider_mock.call_args[0][0]
-            self.assertEqual(provider.processor.exporter.compression, "gzip")
+            self.assertEqual(len(provider.processors), 1)
+            self.assertEqual(
+                provider.processors[0].exporter.compression, "gzip"
+            )
+
+    def test_logging_init_custom_log_record_processors(self):
+        log_record_processor = mock.Mock(spec=LogRecordProcessor)
+        with ResetGlobalLoggingState():
+            resource = Resource.create({})
+            _init_logging(
+                {"otlp": DummyOTLPLogExporter},
+                resource=resource,
+                log_record_processors=[log_record_processor],
+                export_log_record_processor=SimpleLogRecordProcessor,
+            )
+            provider = self.set_provider_mock.call_args[0][0]
+            self.assertEqual(len(provider.processors), 2)
+            self.assertEqual(provider.processors[0], log_record_processor)
+            self.assertIsInstance(
+                provider.processors[1], SimpleLogRecordProcessor
+            )
 
     @patch.dict(
         environ,
@@ -725,12 +778,13 @@ def test_logging_init_exporter_without_handler_setup(self):
             provider.resource.attributes.get("service.name"),
             "otlp-service",
         )
-        self.assertIsInstance(provider.processor, DummyLogRecordProcessor)
+        self.assertEqual(len(provider.processors), 1)
+        self.assertIsInstance(provider.processors[0], DummyLogRecordProcessor)
         self.assertIsInstance(
-            provider.processor.exporter, DummyOTLPLogExporter
+            provider.processors[0].exporter, DummyOTLPLogExporter
         )
         getLogger(__name__).error("hello")
-        self.assertFalse(provider.processor.exporter.export_called)
+        self.assertFalse(provider.processors[0].exporter.export_called)
 
     @patch.dict(
         environ,
@@ -742,7 +796,12 @@ def test_logging_init_disable_default(self, logging_mock, tracing_mock):
         _initialize_components(auto_instrumentation_version="auto-version")
         self.assertEqual(tracing_mock.call_count, 1)
         logging_mock.assert_called_once_with(
-            mock.ANY, mock.ANY, False, exporter_args_map=None
+            mock.ANY,
+            mock.ANY,
+            False,
+            exporter_args_map=None,
+            log_record_processors=None,
+            export_log_record_processor=None,
         )
 
     @patch.dict(
@@ -758,7 +817,12 @@ def test_logging_init_enable_env(self, logging_mock, tracing_mock):
         with self.assertLogs(level=WARNING):
             _initialize_components(auto_instrumentation_version="auto-version")
         logging_mock.assert_called_once_with(
-            mock.ANY, mock.ANY, True, exporter_args_map=None
+            mock.ANY,
+            mock.ANY,
+            True,
+            exporter_args_map=None,
+            log_record_processors=None,
+            export_log_record_processor=None,
         )
         self.assertEqual(tracing_mock.call_count, 1)
 
@@ -843,6 +907,10 @@ def test_initialize_components_kwargs(
             "id_generator": "TEST_GENERATOR",
             "setup_logging_handler": True,
             "exporter_args_map": {1: {"compression": "gzip"}},
+            "export_log_record_processor": SimpleLogRecordProcessor,
+            "export_span_processor": SimpleSpanProcessor,
+            "log_record_processors": [],
+            "span_processors": [],
         }
         _initialize_components(**kwargs)
 
@@ -877,6 +945,8 @@ def test_initialize_components_kwargs(
             sampler="TEST_SAMPLER",
             resource="TEST_RESOURCE",
             exporter_args_map={1: {"compression": "gzip"}},
+            span_processors=[],
+            export_span_processor=SimpleSpanProcessor,
         )
         metrics_mock.assert_called_once_with(
             "TEST_METRICS_EXPORTERS_DICT",
@@ -888,6 +958,8 @@ def test_initialize_components_kwargs(
             "TEST_RESOURCE",
             True,
             exporter_args_map={1: {"compression": "gzip"}},
+            log_record_processors=[],
+            export_log_record_processor=SimpleLogRecordProcessor,
         )
 
     def test_basicConfig_works_with_otel_handler(self):