add sampler plugin loading to declarative config via entry points

Sampler and ParentBasedSampler are changed from @DataClass to
TypeAlias = dict[str, Any] in models.py. The generated dataclass
representation dropped unknown keys, making plugin sampler names
unrecoverable before reaching the factory. The dict type preserves
the raw YAML key, which is the plugin name.

_create_sampler() now has a single code path: extract the single key
as the sampler name, look it up in _SAMPLER_REGISTRY (always_on,
always_off, trace_id_ratio_based, parent_based), and fall back to
load_entry_point("opentelemetry_sampler", name) for unknown names.
This matches the spec's PluginComponentProvider mechanism.

Assisted-by: Claude Sonnet 4.6

Author: MikeGoldsmith

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

@@ -15,7 +15,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Optional
+from typing import Any, Optional
 
 from opentelemetry import trace
 from opentelemetry.sdk._configuration._common import (
@@ -180,88 +180,46 @@ def _create_span_processor(
     )
 
 
-def _create_sampler(config) -> Sampler:
-    """Create a sampler from config.
+_SAMPLER_REGISTRY: dict[str, Any] = {
+    "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 {}),
+}
 
-    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:
-        return ALWAYS_OFF
-    if config.trace_id_ratio_based is not None:
-        ratio = config.trace_id_ratio_based.ratio
-        return TraceIdRatioBased(ratio if ratio is not None else 1.0)
-    if config.parent_based is not None:
-        return _create_parent_based_sampler(config.parent_based)
-    raise ConfigurationError(
-        f"Unknown or unsupported sampler type in config: {config!r}. "
-        "Supported types: always_on, always_off, trace_id_ratio_based, parent_based."
-    )
 
+def _create_sampler(config: dict) -> Sampler:
+    """Create a sampler from a config dict with a single key naming the sampler type.
 
-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.
+    Known names (always_on, always_off, trace_id_ratio_based, parent_based) are
+    bootstrapped directly. Unknown names are looked up via the
+    ``opentelemetry_sampler`` entry point group, matching the spec's
+    PluginComponentProvider mechanism.
     """
-    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 = {"root": root}
-    if config.remote_parent_sampled is not None:
-        kwargs["remote_parent_sampled"] = _create_sampler(
-            config.remote_parent_sampled
-        )
-    if config.remote_parent_not_sampled is not None:
-        kwargs["remote_parent_not_sampled"] = _create_sampler(
-            config.remote_parent_not_sampled
-        )
-    if config.local_parent_sampled is not None:
-        kwargs["local_parent_sampled"] = _create_sampler(
-            config.local_parent_sampled
-        )
-    if config.local_parent_not_sampled is not None:
-        kwargs["local_parent_not_sampled"] = _create_sampler(
-            config.local_parent_not_sampled
+    if len(config) != 1:
+        raise ConfigurationError(
+            f"Sampler config must have exactly one key, got: {list(config.keys())}"
         )
+    name, sampler_config = next(iter(config.items()))
+    if name in _SAMPLER_REGISTRY:
+        return _SAMPLER_REGISTRY[name](sampler_config)
+    return load_entry_point("opentelemetry_sampler", name)()
+
+
+def _create_parent_based_sampler(config: dict) -> Sampler:
+    """Create a ParentBased sampler from a config dict, applying SDK defaults for absent delegates."""
+    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)
 
 

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

@@ -759,24 +759,13 @@ class ExperimentalJaegerRemoteSampler:
     interval: int | None = None
 
 
-@dataclass
-class ParentBasedSampler:
-    root: Sampler | None = None
-    remote_parent_sampled: Sampler | None = None
-    remote_parent_not_sampled: Sampler | None = None
-    local_parent_sampled: Sampler | None = None
-    local_parent_not_sampled: Sampler | None = None
-
-
-@dataclass
-class Sampler:
-    always_off: AlwaysOffSampler | None = None
-    always_on: AlwaysOnSampler | None = None
-    composite_development: ExperimentalComposableSampler | None = None
-    jaeger_remote_development: ExperimentalJaegerRemoteSampler | None = None
-    parent_based: ParentBasedSampler | None = None
-    probability_development: ExperimentalProbabilitySampler | None = None
-    trace_id_ratio_based: TraceIdRatioBasedSampler | None = None
+# Diverges from codegen: Sampler and ParentBasedSampler are typed as
+# dict[str, Any] rather than dataclasses so that unknown sampler names
+# (plugin/custom samplers) are preserved as dict keys through the config
+# pipeline. The loader stores nested fields as raw dicts anyway, so the
+# typed dataclass representation would drop unknown keys silently.
+Sampler: TypeAlias = dict[str, Any]
+ParentBasedSampler: TypeAlias = dict[str, Any]
 
 
 @dataclass

opentelemetry-sdk/tests/_configuration/test_tracer_provider.py

@@ -34,12 +34,6 @@
 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 (
     SimpleSpanProcessor as SimpleSpanProcessorConfig,
 )
@@ -52,9 +46,6 @@
 from opentelemetry.sdk._configuration.models import (
     SpanProcessor as SpanProcessorConfig,
 )
-from opentelemetry.sdk._configuration.models import (
-    TraceIdRatioBasedSampler as TraceIdRatioBasedConfig,
-)
 from opentelemetry.sdk._configuration.models import (
     TracerProvider as TracerProviderConfig,
 )
@@ -159,57 +150,47 @@ def _make_provider(sampler_config):
         )
 
     def test_always_on(self):
-        provider = self._make_provider(SamplerConfig(always_on={}))
+        provider = self._make_provider({"always_on": {}})
         self.assertIs(provider.sampler, ALWAYS_ON)
 
     def test_always_off(self):
-        provider = self._make_provider(SamplerConfig(always_off={}))
+        provider = self._make_provider({"always_off": {}})
         self.assertIs(provider.sampler, ALWAYS_OFF)
 
     def test_trace_id_ratio_based(self):
         provider = self._make_provider(
-            SamplerConfig(
-                trace_id_ratio_based=TraceIdRatioBasedConfig(ratio=0.5)
-            )
+            {"trace_id_ratio_based": {"ratio": 0.5}}
         )
         self.assertIsInstance(provider.sampler, TraceIdRatioBased)
         self.assertAlmostEqual(provider.sampler._rate, 0.5)
 
     def test_trace_id_ratio_based_none_ratio_defaults_to_1(self):
-        provider = self._make_provider(
-            SamplerConfig(trace_id_ratio_based=TraceIdRatioBasedConfig())
-        )
+        provider = self._make_provider({"trace_id_ratio_based": {}})
         self.assertIsInstance(provider.sampler, TraceIdRatioBased)
         self.assertAlmostEqual(provider.sampler._rate, 1.0)
 
     def test_parent_based_with_root(self):
         provider = self._make_provider(
-            SamplerConfig(
-                parent_based=ParentBasedSamplerConfig(
-                    root=SamplerConfig(always_on={})
-                )
-            )
+            {"parent_based": {"root": {"always_on": {}}}}
         )
         self.assertIsInstance(provider.sampler, ParentBased)
 
     def test_parent_based_no_root_defaults_to_always_on(self):
-        provider = self._make_provider(
-            SamplerConfig(parent_based=ParentBasedSamplerConfig())
-        )
+        provider = self._make_provider({"parent_based": {}})
         self.assertIsInstance(provider.sampler, ParentBased)
         self.assertIs(provider.sampler._root, ALWAYS_ON)
 
     def test_parent_based_with_delegate_samplers(self):
         provider = self._make_provider(
-            SamplerConfig(
-                parent_based=ParentBasedSamplerConfig(
-                    root=SamplerConfig(always_on={}),
-                    remote_parent_sampled=SamplerConfig(always_on={}),
-                    remote_parent_not_sampled=SamplerConfig(always_off={}),
-                    local_parent_sampled=SamplerConfig(always_on={}),
-                    local_parent_not_sampled=SamplerConfig(always_off={}),
-                )
-            )
+            {
+                "parent_based": {
+                    "root": {"always_on": {}},
+                    "remote_parent_sampled": {"always_on": {}},
+                    "remote_parent_not_sampled": {"always_off": {}},
+                    "local_parent_sampled": {"always_on": {}},
+                    "local_parent_not_sampled": {"always_off": {}},
+                }
+            }
         )
         sampler = provider.sampler
         self.assertIsInstance(sampler, ParentBased)
@@ -218,37 +199,11 @@ def test_parent_based_with_delegate_samplers(self):
         self.assertIs(sampler._local_parent_sampled, ALWAYS_ON)
         self.assertIs(sampler._local_parent_not_sampled, ALWAYS_OFF)
 
-    def test_unknown_sampler_raises_configuration_error(self):
+    def test_multiple_keys_raises_configuration_error(self):
         with self.assertRaises(ConfigurationError):
-            create_tracer_provider(
-                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)
+            self._make_provider({"always_on": {}, "always_off": {}})
 
-    def test_dict_plugin_sampler_loaded_via_entry_point(self):
+    def test_plugin_sampler_loaded_via_entry_point(self):
         mock_sampler = MagicMock(spec=Sampler)
         mock_class = MagicMock(return_value=mock_sampler)
         with patch(
@@ -258,7 +213,7 @@ def test_dict_plugin_sampler_loaded_via_entry_point(self):
             provider = self._make_provider({"my_custom_sampler": {}})
         self.assertIs(provider.sampler, mock_sampler)
 
-    def test_dict_unknown_plugin_raises_configuration_error(self):
+    def test_unknown_plugin_raises_configuration_error(self):
         with patch(
             "opentelemetry.sdk._configuration._common.entry_points",
             return_value=[],