feat(config): add shared load_entry_point utility for declarative config plugin loading

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- `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,7 +20,7 @@
 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,
 )
@@ -30,28 +30,11 @@
 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
+    """Load and instantiate a propagator by name."""
+    return load_entry_point("opentelemetry_propagator", name)()
 
 
 def _propagators_from_textmap_config(

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))

opentelemetry-sdk/tests/_configuration/test_propagator.py

@@ -89,7 +89,7 @@ def test_b3_via_entry_point(self):
         mock_ep.load.return_value = lambda: mock_propagator
 
         with patch(
-            "opentelemetry.sdk._configuration._propagator.entry_points",
+            "opentelemetry.sdk._configuration._common.entry_points",
             return_value=[mock_ep],
         ):
             config = PropagatorConfig(
@@ -106,7 +106,7 @@ def test_b3multi_via_entry_point(self):
         mock_ep.load.return_value = lambda: mock_propagator
 
         with patch(
-            "opentelemetry.sdk._configuration._propagator.entry_points",
+            "opentelemetry.sdk._configuration._common.entry_points",
             return_value=[mock_ep],
         ):
             config = PropagatorConfig(
@@ -118,7 +118,7 @@ def test_b3multi_via_entry_point(self):
 
     def test_b3_not_installed_raises_configuration_error(self):
         with patch(
-            "opentelemetry.sdk._configuration._propagator.entry_points",
+            "opentelemetry.sdk._configuration._common.entry_points",
             return_value=[],
         ):
             config = PropagatorConfig(
@@ -135,7 +135,7 @@ def test_composite_list_tracecontext(self):
         mock_ep.load.return_value = lambda: mock_tc
 
         with patch(
-            "opentelemetry.sdk._configuration._propagator.entry_points",
+            "opentelemetry.sdk._configuration._common.entry_points",
             return_value=[mock_ep],
         ):
             result = create_propagator(config)
@@ -158,7 +158,7 @@ def fake_entry_points(group, name):
             return []
 
         with patch(
-            "opentelemetry.sdk._configuration._propagator.entry_points",
+            "opentelemetry.sdk._configuration._common.entry_points",
             side_effect=fake_entry_points,
         ):
             config = PropagatorConfig(composite_list="tracecontext,baggage")
@@ -182,7 +182,7 @@ def test_composite_list_whitespace_around_names(self):
         mock_ep.load.return_value = lambda: mock_tc
 
         with patch(
-            "opentelemetry.sdk._configuration._propagator.entry_points",
+            "opentelemetry.sdk._configuration._common.entry_points",
             return_value=[mock_ep],
         ):
             config = PropagatorConfig(composite_list=" tracecontext ")
@@ -195,7 +195,7 @@ def test_entry_point_load_exception_raises_configuration_error(self):
         mock_ep.load.side_effect = RuntimeError("package broken")
 
         with patch(
-            "opentelemetry.sdk._configuration._propagator.entry_points",
+            "opentelemetry.sdk._configuration._common.entry_points",
             return_value=[mock_ep],
         ):
             config = PropagatorConfig(composite_list="broken-prop")
@@ -210,7 +210,7 @@ def test_deduplication_across_composite_and_composite_list(self):
         mock_ep.load.return_value = lambda: mock_tc
 
         with patch(
-            "opentelemetry.sdk._configuration._propagator.entry_points",
+            "opentelemetry.sdk._configuration._common.entry_points",
             return_value=[mock_ep],
         ):
             config = PropagatorConfig(
@@ -229,7 +229,7 @@ def test_deduplication_across_composite_and_composite_list(self):
 
     def test_unknown_composite_list_propagator_raises(self):
         with patch(
-            "opentelemetry.sdk._configuration._propagator.entry_points",
+            "opentelemetry.sdk._configuration._common.entry_points",
             return_value=[],
         ):
             config = PropagatorConfig(composite_list="nonexistent")