feat(logs): add OTEL_LOG_LEVEL support

.changelog/5115.added

@@ -0,0 +1 @@
+`opentelemetry-sdk`: honour `OTEL_LOG_LEVEL` to set SDK internal log level

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

@@ -30,7 +30,10 @@
     LoggingHandler,
     LogRecordProcessor,
 )
-from opentelemetry.sdk._logs._internal import _LoggerConfiguratorT
+from opentelemetry.sdk._logs._internal import (
+    _configure_otel_log_level,
+    _LoggerConfiguratorT,
+)
 from opentelemetry.sdk._logs.export import (
     BatchLogRecordProcessor,
     LogRecordExporter,
@@ -558,6 +561,8 @@ def _initialize_components(
     logger_configurator: _LoggerConfiguratorT | None = None,
 ):
     # pylint: disable=too-many-locals
+    _configure_otel_log_level()
+
     if trace_exporter_names is None:
         trace_exporter_names = []
     if metric_exporter_names is None:

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

@@ -53,6 +53,7 @@
 from opentelemetry.sdk.environment_variables import (
     OTEL_ATTRIBUTE_COUNT_LIMIT,
     OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT,
+    OTEL_LOG_LEVEL,
     OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED,
     OTEL_SDK_DISABLED,
 )
@@ -78,9 +79,42 @@
 _DEFAULT_OTEL_ATTRIBUTE_COUNT_LIMIT = 128
 _ENV_VALUE_UNSET = ""
 
+# "warn" is accepted alongside "warning" because OTel canonical short names
+# use "WARN", so users following OTel documentation will naturally try "warn".
+# "trace" maps to DEBUG because Python has no TRACE level and the OTel
+# declarative config schema includes "trace" as a valid log level value.
+_OTEL_LOG_LEVEL_TO_PYTHON = {
+    "trace": logging.DEBUG,
+    "debug": logging.DEBUG,
+    "info": logging.INFO,
+    "warn": logging.WARNING,
+    "warning": logging.WARNING,
+    "error": logging.ERROR,
+    "critical": logging.CRITICAL,
+}
+
 _logger = logging.getLogger(__name__)
 
 
+def _configure_otel_log_level() -> None:
+    """Apply OTEL_LOG_LEVEL to the ``opentelemetry.sdk`` logger hierarchy."""
+    otel_log_level_raw = environ.get(OTEL_LOG_LEVEL)
+    if not otel_log_level_raw:
+        return
+    otel_log_level = otel_log_level_raw.lower()
+    if otel_log_level in _OTEL_LOG_LEVEL_TO_PYTHON:
+        logging.getLogger("opentelemetry.sdk").setLevel(
+            _OTEL_LOG_LEVEL_TO_PYTHON[otel_log_level]
+        )
+    else:
+        _logger.warning(
+            "Invalid value for OTEL_LOG_LEVEL: %r. "
+            "Valid values: trace, debug, info, warn, warning, error, critical. "
+            "Logger level unchanged.",
+            otel_log_level_raw,
+        )
+
+
 class BytesEncoder(json.JSONEncoder):
     def default(self, o):
         if isinstance(o, bytes):

opentelemetry-sdk/tests/logs/test_logs.py

@@ -3,6 +3,7 @@
 
 # pylint: disable=protected-access
 
+import logging
 import unittest
 from unittest.mock import Mock, patch
 
@@ -16,14 +17,19 @@
     ReadWriteLogRecord,
 )
 from opentelemetry.sdk._logs._internal import (
+    _OTEL_LOG_LEVEL_TO_PYTHON,
     NoOpLogger,
     SynchronousMultiLogRecordProcessor,
+    _configure_otel_log_level,
     _disable_logger_configurator,
     _LoggerConfig,
     _RuleBasedLoggerConfigurator,
     create_logger_metrics,
 )
-from opentelemetry.sdk.environment_variables import OTEL_SDK_DISABLED
+from opentelemetry.sdk.environment_variables import (
+    OTEL_LOG_LEVEL,
+    OTEL_SDK_DISABLED,
+)
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.util.instrumentation import (
     InstrumentationScope,
@@ -451,3 +457,87 @@ def test_emit_readwrite_logrecord_uses_exception(self):
         self.assertEqual(
             attributes[exception_attributes.EXCEPTION_TYPE], "RuntimeError"
         )
+
+
+class TestOtelLogLevelEnvVar(unittest.TestCase):
+    """Tests for OTEL_LOG_LEVEL → SDK internal logger level."""
+
+    def setUp(self):
+        self._sdk_logger = logging.getLogger("opentelemetry.sdk")
+
+    def tearDown(self):
+        self._sdk_logger.setLevel(logging.NOTSET)
+
+    def test_otel_log_level_to_python_mapping_accepted_keys(self):
+        expected_keys = {
+            "trace",
+            "debug",
+            "info",
+            "warn",
+            "warning",
+            "error",
+            "critical",
+        }
+        self.assertEqual(set(_OTEL_LOG_LEVEL_TO_PYTHON.keys()), expected_keys)
+
+    @patch.dict("os.environ", {OTEL_LOG_LEVEL: ""})
+    def test_unset_env_var_does_not_modify_logger_level(self):
+        _configure_otel_log_level()
+        self.assertEqual(self._sdk_logger.level, logging.NOTSET)
+
+    def test_invalid_value_warns_and_leaves_level_unchanged(self):
+        for invalid in ("INVALID", "verbose", "none", "0"):
+            with self.subTest(invalid=invalid):
+                with patch.dict("os.environ", {OTEL_LOG_LEVEL: invalid}):
+                    with self.assertLogs(
+                        "opentelemetry.sdk._logs._internal",
+                        level=logging.WARNING,
+                    ):
+                        _configure_otel_log_level()
+                self.assertEqual(self._sdk_logger.level, logging.NOTSET)
+
+    def test_case_insensitive(self):
+        for env_value, expected_level in (
+            ("DEBUG", logging.DEBUG),
+            ("WARN", logging.WARNING),
+            ("Warning", logging.WARNING),
+            ("cRiTiCaL", logging.CRITICAL),
+        ):
+            with self.subTest(env_value=env_value):
+                with patch.dict("os.environ", {OTEL_LOG_LEVEL: env_value}):
+                    _configure_otel_log_level()
+                self.assertEqual(self._sdk_logger.level, expected_level)
+                self._sdk_logger.setLevel(logging.NOTSET)
+
+    @patch.dict("os.environ", {OTEL_LOG_LEVEL: "critical"})
+    def test_level_propagates_to_child_loggers(self):
+        _configure_otel_log_level()
+        self.assertEqual(
+            self._sdk_logger.getChild("trace").getEffectiveLevel(),
+            logging.CRITICAL,
+        )
+        self.assertEqual(
+            self._sdk_logger.getChild("metrics").getEffectiveLevel(),
+            logging.CRITICAL,
+        )
+        self.assertEqual(
+            self._sdk_logger.getChild("logs").getEffectiveLevel(),
+            logging.CRITICAL,
+        )
+
+    def test_all_valid_values_map_to_correct_level(self):
+        cases = [
+            ("trace", logging.DEBUG),
+            ("debug", logging.DEBUG),
+            ("info", logging.INFO),
+            ("warn", logging.WARNING),
+            ("warning", logging.WARNING),
+            ("error", logging.ERROR),
+            ("critical", logging.CRITICAL),
+        ]
+        for env_value, expected_level in cases:
+            with self.subTest(env_value=env_value):
+                with patch.dict("os.environ", {OTEL_LOG_LEVEL: env_value}):
+                    _configure_otel_log_level()
+                self.assertEqual(self._sdk_logger.level, expected_level)
+                self._sdk_logger.setLevel(logging.NOTSET)