feat(config): propagator plugin loading via entry points for declarative config

CHANGELOG.md

@@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- `opentelemetry-sdk`: add propagator plugin loading to declarative file configuration via the `opentelemetry_propagator` entry point group, matching the spec's PluginComponentProvider mechanism
+  ([#5070](https://github.com/open-telemetry/opentelemetry-python/pull/5070))
+- `opentelemetry-sdk`: add `load_entry_point` shared utility to declarative file configuration for loading plugins via entry points; refactor propagator loading to use it
+  ([#5093](https://github.com/open-telemetry/opentelemetry-python/pull/5093))
 - `opentelemetry-sdk`: Add `create_logger_provider`/`configure_logger_provider` to declarative file configuration, enabling LoggerProvider instantiation from config files without reading env vars
   ([#4990](https://github.com/open-telemetry/opentelemetry-python/pull/4990))
 - `opentelemetry-sdk`: Add `service` resource detector support to declarative file configuration via `detection_development.detectors[].service`

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

@@ -15,11 +15,39 @@
 from __future__ import annotations
 
 import logging
-from typing import Optional
+from typing import Optional, Type
+
+from opentelemetry.sdk._configuration._exceptions import ConfigurationError
+from opentelemetry.util._importlib_metadata import entry_points
 
 _logger = logging.getLogger(__name__)
 
 
+def load_entry_point(group: str, name: str) -> Type:
+    """Load a plugin class from an entry point group by name.
+
+    Returns the loaded class — callers are responsible for instantiation
+    with whatever arguments their config requires.
+
+    Raises:
+        ConfigurationError: If the entry point is not found or fails to load.
+    """
+    try:
+        ep = next(iter(entry_points(group=group, name=name)), None)
+        if ep is None:
+            raise ConfigurationError(
+                f"Plugin '{name}' not found in group '{group}'. "
+                "Make sure the package providing this plugin is installed."
+            )
+        return ep.load()
+    except ConfigurationError:
+        raise
+    except Exception as exc:
+        raise ConfigurationError(
+            f"Failed to load plugin '{name}' from group '{group}': {exc}"
+        ) from exc
+
+
 def _parse_headers(
     headers: Optional[list],
     headers_list: Optional[str],

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

@@ -20,53 +20,37 @@
 from opentelemetry.propagate import set_global_textmap
 from opentelemetry.propagators.composite import CompositePropagator
 from opentelemetry.propagators.textmap import TextMapPropagator
-from opentelemetry.sdk._configuration._exceptions import ConfigurationError
+from opentelemetry.sdk._configuration._common import load_entry_point
 from opentelemetry.sdk._configuration.models import (
     Propagator as PropagatorConfig,
 )
-from opentelemetry.sdk._configuration.models import (
-    TextMapPropagator as TextMapPropagatorConfig,
-)
 from opentelemetry.trace.propagation.tracecontext import (
     TraceContextTextMapPropagator,
 )
-from opentelemetry.util._importlib_metadata import entry_points
-
-
-def _load_entry_point_propagator(name: str) -> TextMapPropagator:
-    """Load a propagator by name from the opentelemetry_propagator entry point group."""
-    try:
-        ep = next(
-            iter(entry_points(group="opentelemetry_propagator", name=name)),
-            None,
-        )
-        if not ep:
-            raise ConfigurationError(
-                f"Propagator '{name}' not found. "
-                "It may not be installed or may be misspelled."
-            )
-        return ep.load()()
-    except ConfigurationError:
-        raise
-    except Exception as exc:
-        raise ConfigurationError(
-            f"Failed to load propagator '{name}': {exc}"
-        ) from exc
+
+# Propagators bundled with the SDK — no entry point lookup needed.
+_PROPAGATOR_REGISTRY: dict = {
+    "tracecontext": lambda _: TraceContextTextMapPropagator(),
+    "baggage": lambda _: W3CBaggagePropagator(),
+}
 
 
 def _propagators_from_textmap_config(
-    config: TextMapPropagatorConfig,
+    config: dict,
 ) -> list[TextMapPropagator]:
-    """Resolve a single TextMapPropagator config entry to a list of propagators."""
+    """Resolve a TextMapPropagator config dict to a list of propagators.
+
+    Each key in the dict names a propagator. Known names (tracecontext, baggage)
+    are bootstrapped directly. All other names — including b3, b3multi, and
+    custom plugin propagators — are loaded via the ``opentelemetry_propagator``
+    entry point group, matching the spec's PluginComponentProvider mechanism.
+    """
     result: list[TextMapPropagator] = []
-    if config.tracecontext is not None:
-        result.append(TraceContextTextMapPropagator())
-    if config.baggage is not None:
-        result.append(W3CBaggagePropagator())
-    if config.b3 is not None:
-        result.append(_load_entry_point_propagator("b3"))
-    if config.b3multi is not None:
-        result.append(_load_entry_point_propagator("b3multi"))
+    for name, prop_config in config.items():
+        if name in _PROPAGATOR_REGISTRY:
+            result.append(_PROPAGATOR_REGISTRY[name](prop_config))
+        else:
+            result.append(load_entry_point("opentelemetry_propagator", name)())
     return result
 
 
@@ -102,7 +86,7 @@ def create_propagator(
             name = name.strip()
             if not name or name.lower() == "none":
                 continue
-            propagator = _load_entry_point_propagator(name)
+            propagator = load_entry_point("opentelemetry_propagator", name)()
             propagators.setdefault(type(propagator), propagator)
 
     return CompositePropagator(list(propagators.values()))

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

@@ -573,12 +573,10 @@ class SpanProcessor:
     simple: SimpleSpanProcessor | None = None
 
 
-@dataclass
-class TextMapPropagator:
-    tracecontext: TraceContextPropagator | None = None
-    baggage: BaggagePropagator | None = None
-    b3: B3Propagator | None = None
-    b3multi: B3MultiPropagator | None = None
+# Diverges from codegen: TextMapPropagator is typed as dict[str, Any] rather
+# than a dataclass so that unknown propagator names (plugin/custom propagators)
+# are preserved as dict keys through the config pipeline.
+TextMapPropagator: TypeAlias = dict[str, Any]
 
 
 @dataclass

opentelemetry-sdk/tests/_configuration/test_common.py

@@ -14,8 +14,13 @@
 
 import unittest
 from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
 
-from opentelemetry.sdk._configuration._common import _parse_headers
+from opentelemetry.sdk._configuration._common import (
+    _parse_headers,
+    load_entry_point,
+)
+from opentelemetry.sdk._configuration._exceptions import ConfigurationError
 
 
 class TestParseHeaders(unittest.TestCase):
@@ -79,3 +84,37 @@ def test_struct_headers_override_headers_list(self):
 
     def test_both_empty_struct_and_none_list_returns_empty_dict(self):
         self.assertEqual(_parse_headers([], None), {})
+
+
+class TestLoadEntryPoint(unittest.TestCase):
+    def test_returns_loaded_class(self):
+        mock_class = MagicMock()
+        mock_ep = MagicMock()
+        mock_ep.load.return_value = mock_class
+        with patch(
+            "opentelemetry.sdk._configuration._common.entry_points",
+            return_value=[mock_ep],
+        ):
+            result = load_entry_point("some_group", "some_name")
+        self.assertIs(result, mock_class)
+
+    def test_raises_when_not_found(self):
+        with patch(
+            "opentelemetry.sdk._configuration._common.entry_points",
+            return_value=[],
+        ):
+            with self.assertRaises(ConfigurationError) as ctx:
+                load_entry_point("some_group", "missing")
+        self.assertIn("missing", str(ctx.exception))
+        self.assertIn("some_group", str(ctx.exception))
+
+    def test_wraps_load_exception_in_configuration_error(self):
+        mock_ep = MagicMock()
+        mock_ep.load.side_effect = ImportError("bad import")
+        with patch(
+            "opentelemetry.sdk._configuration._common.entry_points",
+            return_value=[mock_ep],
+        ):
+            with self.assertRaises(ConfigurationError) as ctx:
+                load_entry_point("some_group", "some_name")
+        self.assertIn("bad import", str(ctx.exception))