add sampler plugin loading to declarative config via entry points

Extends _create_sampler() to accept raw dicts (from the YAML path) in
addition to typed dataclasses (from the direct API path). Unknown sampler
names fall back to load_entry_point("opentelemetry_sampler", name),
matching the spec's PluginComponentProvider mechanism and Java SDK
behaviour.

_create_parent_based_sampler() gets the same dict-path treatment so
custom samplers can be used as delegate samplers inside parent_based.

Assisted-by: Claude Sonnet 4.6

Author: MikeGoldsmith

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- `opentelemetry-sdk`: add sampler plugin loading to declarative file configuration via the `opentelemetry_sampler` entry point group, matching the spec's PluginComponentProvider mechanism
+  ([#5071](https://github.com/open-telemetry/opentelemetry-python/pull/5071))
 - `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

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

@@ -18,20 +18,17 @@
 from typing import Optional
 
 from opentelemetry import trace
-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
 from opentelemetry.sdk._configuration.models import (
     OtlpGrpcExporter as OtlpGrpcExporterConfig,
 )
 from opentelemetry.sdk._configuration.models import (
     OtlpHttpExporter as OtlpHttpExporterConfig,
 )
-from opentelemetry.sdk._configuration.models import (
-    ParentBasedSampler as ParentBasedSamplerConfig,
-)
-from opentelemetry.sdk._configuration.models import (
-    Sampler as SamplerConfig,
-)
 from opentelemetry.sdk._configuration.models import (
     SpanExporter as SpanExporterConfig,
 )
@@ -183,8 +180,33 @@ def _create_span_processor(
     )
 
 
-def _create_sampler(config: SamplerConfig) -> Sampler:
-    """Create a sampler from config."""
+def _create_sampler(config) -> Sampler:
+    """Create a sampler from config.
+
+    Accepts either a SamplerConfig dataclass (direct/test usage) or a raw dict
+    (from the YAML integration path). For unknown sampler names, falls back to
+    entry point loading via the ``opentelemetry_sampler`` group — matching the
+    spec's PluginComponentProvider mechanism and Java SDK behaviour.
+    """
+    if isinstance(config, dict):
+        if len(config) != 1:
+            raise ConfigurationError(
+                f"Sampler config must have exactly one key, got: {list(config.keys())}"
+            )
+        name, plugin_config = next(iter(config.items()))
+        known = {
+            "always_on": lambda _: ALWAYS_ON,
+            "always_off": lambda _: ALWAYS_OFF,
+            "trace_id_ratio_based": lambda c: TraceIdRatioBased(
+                (c or {}).get("ratio", 1.0)
+            ),
+            "parent_based": lambda c: _create_parent_based_sampler(c or {}),
+        }
+        if name in known:
+            return known[name](plugin_config)
+        return load_entry_point("opentelemetry_sampler", name)()
+
+    # Dataclass path (direct API / unit tests)
     if config.always_on is not None:
         return ALWAYS_ON
     if config.always_off is not None:
@@ -200,12 +222,30 @@ def _create_sampler(config: SamplerConfig) -> Sampler:
     )
 
 
-def _create_parent_based_sampler(config: ParentBasedSamplerConfig) -> Sampler:
-    """Create a ParentBased sampler from config, applying SDK defaults for absent delegates."""
+def _create_parent_based_sampler(config) -> Sampler:
+    """Create a ParentBased sampler from config, applying SDK defaults for absent delegates.
+
+    Accepts either a ParentBasedSamplerConfig dataclass or a raw dict.
+    """
+    if isinstance(config, dict):
+        root = (
+            _create_sampler(config["root"]) if "root" in config else ALWAYS_ON
+        )
+        kwargs: dict = {"root": root}
+        for key in (
+            "remote_parent_sampled",
+            "remote_parent_not_sampled",
+            "local_parent_sampled",
+            "local_parent_not_sampled",
+        ):
+            if key in config:
+                kwargs[key] = _create_sampler(config[key])
+        return ParentBased(**kwargs)
+
     root = (
         _create_sampler(config.root) if config.root is not None else ALWAYS_ON
     )
-    kwargs: dict = {"root": root}
+    kwargs = {"root": root}
     if config.remote_parent_sampled is not None:
         kwargs["remote_parent_sampled"] = _create_sampler(
             config.remote_parent_sampled

opentelemetry-sdk/tests/_configuration/test_tracer_provider.py

@@ -69,6 +69,7 @@
     ALWAYS_OFF,
     ALWAYS_ON,
     ParentBased,
+    Sampler,
     TraceIdRatioBased,
 )
 
@@ -223,6 +224,48 @@ def test_unknown_sampler_raises_configuration_error(self):
                 TracerProviderConfig(processors=[], sampler=SamplerConfig())
             )
 
+    # --- dict path (YAML integration) ---
+
+    def test_dict_always_on(self):
+        provider = self._make_provider({"always_on": {}})
+        self.assertIs(provider.sampler, ALWAYS_ON)
+
+    def test_dict_always_off(self):
+        provider = self._make_provider({"always_off": {}})
+        self.assertIs(provider.sampler, ALWAYS_OFF)
+
+    def test_dict_trace_id_ratio_based(self):
+        provider = self._make_provider(
+            {"trace_id_ratio_based": {"ratio": 0.25}}
+        )
+        self.assertIsInstance(provider.sampler, TraceIdRatioBased)
+        self.assertAlmostEqual(provider.sampler._rate, 0.25)
+
+    def test_dict_parent_based(self):
+        provider = self._make_provider(
+            {"parent_based": {"root": {"always_off": {}}}}
+        )
+        self.assertIsInstance(provider.sampler, ParentBased)
+        self.assertIs(provider.sampler._root, ALWAYS_OFF)
+
+    def test_dict_plugin_sampler_loaded_via_entry_point(self):
+        mock_sampler = MagicMock(spec=Sampler)
+        mock_class = MagicMock(return_value=mock_sampler)
+        with patch(
+            "opentelemetry.sdk._configuration._common.entry_points",
+            return_value=[MagicMock(**{"load.return_value": mock_class})],
+        ):
+            provider = self._make_provider({"my_custom_sampler": {}})
+        self.assertIs(provider.sampler, mock_sampler)
+
+    def test_dict_unknown_plugin_raises_configuration_error(self):
+        with patch(
+            "opentelemetry.sdk._configuration._common.entry_points",
+            return_value=[],
+        ):
+            with self.assertRaises(ConfigurationError):
+                self._make_provider({"no_such_sampler": {}})
+
 
 class TestCreateSpanExporterAndProcessor(unittest.TestCase):
     # pylint: disable=no-self-use