generalize rule based configurator

Author: herin049

opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py

@@ -60,9 +60,9 @@
 )
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.util import ns_to_iso_str
+from opentelemetry.sdk.util._configurator import RuleBasedConfigurator
 from opentelemetry.sdk.util.instrumentation import (
     InstrumentationScope,
-    _InstrumentationScopePredicateT,
 )
 from opentelemetry.semconv._incubating.attributes import code_attributes
 from opentelemetry.semconv.attributes import exception_attributes
@@ -654,7 +654,7 @@ class LoggerConfig:
     is_enabled: bool = True
 
     @classmethod
-    def default(cls) -> "LoggerConfig":
+    def default(cls) -> LoggerConfig:
         return LoggerConfig()
 
 
@@ -751,9 +751,7 @@ def emit(
 
 
 LoggerConfiguratorT = Callable[[InstrumentationScope], LoggerConfig]
-LoggerConfiguratorRulesT = Sequence[
-    tuple[_InstrumentationScopePredicateT, LoggerConfig]
-]
+RuleBasedLoggerConfigurator = RuleBasedConfigurator[LoggerConfig]
 
 
 def default_logger_configurator(
@@ -768,24 +766,6 @@ def disable_logger_configurator(
     return LoggerConfig(is_enabled=False)
 
 
-class RuleBasedLoggerConfigurator:
-    def __init__(
-        self,
-        *,
-        rules: LoggerConfiguratorRulesT,
-        default_config: LoggerConfig,
-    ):
-        self._rules = rules
-        self._default_config = default_config
-
-    def __call__(self, logger_scope: InstrumentationScope) -> LoggerConfig:
-        for predicate, logger_config in self._rules:
-            if predicate(logger_scope):
-                return logger_config
-        # by default return default config
-        return self._default_config
-
-
 class LoggerProvider(APILoggerProvider):
     def __init__(
         self,

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

@@ -814,3 +814,27 @@ def channel_credential_provider() -> grpc.ChannelCredentials:
 This is an experimental environment variable and the name of this variable and its behavior can
 change in a non-backwards compatible way.
 """
+
+OTEL_PYTHON_METER_CONFIGURATOR = "OTEL_PYTHON_METER_CONFIGURATOR"
+"""
+.. envvar:: OTEL_PYTHON_METER_CONFIGURATOR
+
+The :envvar:`OTEL_PYTHON_METER_CONFIGURATOR` environment variable allows users to set a
+custom Meter Configurator function.
+Default: opentelemetry.sdk.metrics._internal._default_meter_configurator
+
+This is an experimental environment variable and the name of this variable and its behavior can
+change in a non-backwards compatible way.
+"""
+
+OTEL_PYTHON_LOGGER_CONFIGURATOR = "OTEL_PYTHON_LOGGER_CONFIGURATOR"
+"""
+.. envvar:: OTEL_PYTHON_LOGGER_CONFIGURATOR
+
+The :envvar:`OTEL_PYTHON_LOGGER_CONFIGURATOR` environment variable allows users to set a
+custom Logger Configurator function.
+Default: opentelemetry.sdk._logs._internal._default_logger_configurator
+
+This is an experimental environment variable and the name of this variable and its behavior can
+change in a non-backwards compatible way.
+"""

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/__init__.py

@@ -63,6 +63,7 @@
     SdkConfiguration,
 )
 from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.util._configurator import RuleBasedConfigurator
 from opentelemetry.sdk.util.instrumentation import (
     InstrumentationScope,
     _InstrumentationScopePredicateT,
@@ -414,9 +415,7 @@ def _get_exemplar_filter(exemplar_filter: str) -> ExemplarFilter:
 
 
 _MeterConfiguratorT = Callable[[InstrumentationScope], _MeterConfig]
-_MeterConfiguratorRulesT = Sequence[
-    tuple[_InstrumentationScopePredicateT, _MeterConfig]
-]
+_RuleBasedMeterConfigurator = RuleBasedConfigurator[_MeterConfig]
 
 
 def _default_meter_configurator(
@@ -431,27 +430,6 @@ def _disable_meter_configurator(
     return _MeterConfig(is_enabled=False)
 
 
-class _RuleBasedMeterConfigurator:
-    def __init__(
-        self,
-        *,
-        rules: _MeterConfiguratorRulesT,
-        default_config: _MeterConfig,
-    ):
-        self._rules = rules
-        self._default_config = default_config
-
-    def __call__(self, meter_scope: InstrumentationScope) -> _MeterConfig:
-        for predicate, meter_config in self._rules:
-            if predicate(meter_scope):
-                return meter_config
-        # by default return default config
-        return self._default_config
-
-    def update_rules(self, rules: _MeterConfiguratorRulesT) -> None:
-        self._rules = rules
-
-
 class MeterProvider(APIMeterProvider):
     r"""See `opentelemetry.metrics.MeterProvider`.
 

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

@@ -65,10 +65,10 @@
 from opentelemetry.sdk.trace._tracer_metrics import TracerMetrics
 from opentelemetry.sdk.trace.id_generator import IdGenerator, RandomIdGenerator
 from opentelemetry.sdk.util import BoundedList
+from opentelemetry.sdk.util._configurator import RuleBasedConfigurator
 from opentelemetry.sdk.util.instrumentation import (
     InstrumentationInfo,
     InstrumentationScope,
-    _InstrumentationScopePredicateT,
 )
 from opentelemetry.semconv.attributes.exception_attributes import (
     EXCEPTION_ESCAPED,
@@ -1264,28 +1264,7 @@ def start_span(  # pylint: disable=too-many-locals
 
 
 _TracerConfiguratorT = Callable[[InstrumentationScope], _TracerConfig]
-_TracerConfiguratorRulesT = Sequence[
-    tuple[_InstrumentationScopePredicateT, _TracerConfig]
-]
-
-
-class _RuleBasedTracerConfigurator:
-    def __init__(
-        self,
-        *,
-        rules: _TracerConfiguratorRulesT,
-        default_config: _TracerConfig,
-    ):
-        self._rules = rules
-        self._default_config = default_config
-
-    def __call__(self, tracer_scope: InstrumentationScope) -> _TracerConfig:
-        for predicate, tracer_config in self._rules:
-            if predicate(tracer_scope):
-                return tracer_config
-
-        # if no rule matched return the default config
-        return self._default_config
+_RuleBasedTracerConfigurator = RuleBasedConfigurator[_TracerConfig]
 
 
 def _default_tracer_configurator(

opentelemetry-sdk/src/opentelemetry/sdk/util/_configurator.py

@@ -0,0 +1,25 @@
+from typing import Generic, Sequence, TypeVar
+
+from opentelemetry.sdk.util.instrumentation import (
+    InstrumentationScope,
+    _InstrumentationScopePredicateT,
+)
+
+ConfigT = TypeVar("ConfigT")
+ConfiguratorRulesT = Sequence[tuple[_InstrumentationScopePredicateT, ConfigT]]
+
+
+class RuleBasedConfigurator(Generic[ConfigT]):
+    def __init__(self, *, rules: ConfiguratorRulesT, default_config: ConfigT):
+        self._rules = rules
+        self._default_config = default_config
+
+    def __call__(self, scope: InstrumentationScope) -> ConfigT:
+        for predicate, config in self._rules:
+            if predicate(scope):
+                return config
+        # by default return default config
+        return self._default_config
+
+    def update_rules(self, rules: ConfiguratorRulesT) -> None:
+        self._rules = rules