opentelemetry-sdk: make it possible to customize processors configuration

Make it easier for distributions to override the processors set up by
the sdk configurator by specifying a span processor and a log record
processor.

Author: xrmx

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

@@ -38,7 +38,11 @@
 )
 from opentelemetry.metrics import set_meter_provider
 from opentelemetry.sdk._events import EventLoggerProvider
-from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
+from opentelemetry.sdk._logs import (
+    LoggerProvider,
+    LoggingHandler,
+    LogRecordProcessor,
+)
 from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, LogExporter
 from opentelemetry.sdk.environment_variables import (
     _OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED,
@@ -56,7 +60,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
@@ -208,6 +212,7 @@ def _init_tracing(
     sampler: Sampler | None = None,
     resource: Resource | None = None,
     exporter_args_map: ExporterArgsMap | None = None,
+    processor: Type[SpanProcessor] | None = None,
 ):
     provider = TracerProvider(
         id_generator=id_generator,
@@ -217,10 +222,11 @@ def _init_tracing(
     set_tracer_provider(provider)
 
     exporter_args_map = exporter_args_map or {}
+    span_processor = processor or BatchSpanProcessor
     for _, exporter_class in exporters.items():
         exporter_args = exporter_args_map.get(exporter_class, {})
         provider.add_span_processor(
-            BatchSpanProcessor(exporter_class(**exporter_args))
+            span_processor(exporter_class(**exporter_args))
         )
 
 
@@ -254,15 +260,17 @@ def _init_logging(
     resource: Resource | None = None,
     setup_logging_handler: bool = True,
     exporter_args_map: ExporterArgsMap | None = None,
+    processor: Type[LogRecordProcessor] | None = None,
 ):
     provider = LoggerProvider(resource=resource)
     set_logger_provider(provider)
 
     exporter_args_map = exporter_args_map or {}
+    log_record_processor = processor or BatchLogRecordProcessor
     for _, exporter_class in exporters.items():
         exporter_args = exporter_args_map.get(exporter_class, {})
         provider.add_log_record_processor(
-            BatchLogRecordProcessor(exporter_class(**exporter_args))
+            log_record_processor(exporter_class(**exporter_args))
         )
 
     event_logger_provider = EventLoggerProvider(logger_provider=provider)
@@ -416,6 +424,8 @@ def _initialize_components(
     id_generator: IdGenerator | None = None,
     setup_logging_handler: bool | None = None,
     exporter_args_map: ExporterArgsMap | None = None,
+    span_processor: Type[SpanProcessor] | None = None,
+    log_record_processor: Type[LogRecordProcessor] | None = None,
 ):
     if trace_exporter_names is None:
         trace_exporter_names = []
@@ -451,6 +461,7 @@ def _initialize_components(
         sampler=sampler,
         resource=resource,
         exporter_args_map=exporter_args_map,
+        processor=span_processor,
     )
     _init_metrics(
         metric_exporters, resource, exporter_args_map=exporter_args_map
@@ -469,6 +480,7 @@ def _initialize_components(
         resource,
         setup_logging_handler,
         exporter_args_map=exporter_args_map,
+        processor=log_record_processor,
     )
 
 

opentelemetry-sdk/tests/test_configurator.py

@@ -47,7 +47,10 @@
 )
 from opentelemetry.sdk._logs import LoggingHandler
 from opentelemetry.sdk._logs._internal.export import LogExporter
-from opentelemetry.sdk._logs.export import ConsoleLogExporter
+from opentelemetry.sdk._logs.export import (
+    ConsoleLogExporter,
+    SimpleLogRecordProcessor,
+)
 from opentelemetry.sdk.environment_variables import (
     OTEL_TRACES_SAMPLER,
     OTEL_TRACES_SAMPLER_ARG,
@@ -62,7 +65,10 @@
 )
 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.export import (
+    ConsoleSpanExporter,
+    SimpleSpanProcessor,
+)
 from opentelemetry.sdk.trace.id_generator import IdGenerator, RandomIdGenerator
 from opentelemetry.sdk.trace.sampling import (
     ALWAYS_ON,
@@ -396,6 +402,16 @@ def test_trace_init_exporter_uses_exporter_args_map(self):
         exporter = provider.processor.exporter
         self.assertEqual(exporter.compression, "gzip")
 
+    def test_trace_init_custom_span_processor(self):
+        _init_tracing(
+            {"otlp": OTLPSpanExporter},
+            id_generator=RandomIdGenerator(),
+            processor=SimpleSpanProcessor,
+        )
+
+        provider = self.set_provider_mock.call_args[0][0]
+        self.assertTrue(isinstance(provider.processor, SimpleSpanProcessor))
+
     @patch.dict(environ, {OTEL_PYTHON_ID_GENERATOR: "custom_id_generator"})
     @patch("opentelemetry.sdk._configuration.IdGenerator", new=IdGenerator)
     @patch("opentelemetry.sdk._configuration.entry_points")
@@ -706,6 +722,17 @@ def test_logging_init_exporter_uses_exporter_args_map(self):
             provider = self.set_provider_mock.call_args[0][0]
             self.assertEqual(provider.processor.exporter.compression, "gzip")
 
+    def test_logging_init_custom_log_record_processor(self):
+        with ResetGlobalLoggingState():
+            resource = Resource.create({})
+            _init_logging(
+                {"otlp": DummyOTLPLogExporter},
+                resource=resource,
+                processor=SimpleLogRecordProcessor,
+            )
+            provider = self.set_provider_mock.call_args[0][0]
+            self.assertIsInstance(provider.processor, SimpleLogRecordProcessor)
+
     @patch.dict(
         environ,
         {"OTEL_RESOURCE_ATTRIBUTES": "service.name=otlp-service"},
@@ -742,7 +769,7 @@ 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, processor=None
         )
 
     @patch.dict(
@@ -758,7 +785,11 @@ 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,
+            processor=None,
         )
         self.assertEqual(tracing_mock.call_count, 1)
 
@@ -843,6 +874,8 @@ def test_initialize_components_kwargs(
             "id_generator": "TEST_GENERATOR",
             "setup_logging_handler": True,
             "exporter_args_map": {1: {"compression": "gzip"}},
+            "log_record_processor": SimpleLogRecordProcessor,
+            "span_processor": SimpleSpanProcessor,
         }
         _initialize_components(**kwargs)
 
@@ -877,6 +910,7 @@ def test_initialize_components_kwargs(
             sampler="TEST_SAMPLER",
             resource="TEST_RESOURCE",
             exporter_args_map={1: {"compression": "gzip"}},
+            processor=SimpleSpanProcessor,
         )
         metrics_mock.assert_called_once_with(
             "TEST_METRICS_EXPORTERS_DICT",
@@ -888,6 +922,7 @@ def test_initialize_components_kwargs(
             "TEST_RESOURCE",
             True,
             exporter_args_map={1: {"compression": "gzip"}},
+            processor=SimpleLogRecordProcessor,
         )
 
     def test_basicConfig_works_with_otel_handler(self):