feat(logs): add OTEL_LOG_LEVEL support

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- `opentelemetry-sdk`: Honour `OTEL_LOG_LEVEL` environment variable to configure the SDK's internal diagnostic logger; accepted values: `debug`, `info`, `warn`, `warning`, `error`, `critical`; invalid values emit a warning and default to `info`
+  ([#1059](https://github.com/open-telemetry/opentelemetry-python/issues/1059))
 - `opentelemetry-sdk`: fix YAML structure injection via environment variable substitution in declarative file configuration; values containing newlines are now emitted as quoted YAML scalars per spec requirement
   ([#5091](https://github.com/open-telemetry/opentelemetry-python/pull/5091))
 - `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/_logs/_internal/__init__.py

@@ -61,6 +61,7 @@
 from opentelemetry.sdk.environment_variables import (
     OTEL_ATTRIBUTE_COUNT_LIMIT,
     OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT,
+    OTEL_LOG_LEVEL,
     OTEL_SDK_DISABLED,
 )
 from opentelemetry.sdk.resources import Resource
@@ -80,8 +81,34 @@
 _DEFAULT_OTEL_ATTRIBUTE_COUNT_LIMIT = 128
 _ENV_VALUE_UNSET = ""
 
+# "warn" is included alongside "warning" because the OTel spec default is
+# "info" (lowercase OTel style) and OTel canonical short names use "WARN",
+# so users following OTel documentation will naturally try "warn".
+_OTEL_LOG_LEVEL_TO_PYTHON = {
+    "debug": logging.DEBUG,
+    "info": logging.INFO,
+    "warn": logging.WARNING,
+    "warning": logging.WARNING,
+    "error": logging.ERROR,
+    "critical": logging.CRITICAL,
+}
+
 _logger = logging.getLogger(__name__)
 
+# Target opentelemetry.sdk (not the module-level _logger) so the level
+# propagates to all SDK sub-modules: trace, metrics, logs, exporters.
+_otel_log_level_raw = environ.get(OTEL_LOG_LEVEL)
+_otel_log_level = (_otel_log_level_raw or "info").lower()
+if _otel_log_level_raw and _otel_log_level not in _OTEL_LOG_LEVEL_TO_PYTHON:
+    _logger.warning(
+        "Invalid value for OTEL_LOG_LEVEL: %r. "
+        "Valid values: debug, info, warn, warning, error, critical. "
+        "Defaulting to INFO.",
+        _otel_log_level_raw,
+    )
+_python_level = _OTEL_LOG_LEVEL_TO_PYTHON.get(_otel_log_level, logging.INFO)
+logging.getLogger("opentelemetry.sdk").setLevel(_python_level)
+
 
 class BytesEncoder(json.JSONEncoder):
     def default(self, o):
@@ -724,6 +751,7 @@ def emit(
         """
         if not self._is_enabled():
             return
+
         # If a record is provided, use it directly
         if record is not None:
             if not isinstance(record, ReadWriteLogRecord):

opentelemetry-sdk/tests/logs/test_logs.py

@@ -14,9 +14,12 @@
 
 # pylint: disable=protected-access
 
+import importlib
+import logging
 import unittest
 from unittest.mock import Mock, patch
 
+import opentelemetry.sdk._logs._internal as _logs_internal
 from opentelemetry._logs import LogRecord, SeverityNumber
 from opentelemetry.attributes import BoundedAttributes
 from opentelemetry.context import get_current
@@ -28,14 +31,18 @@
     ReadWriteLogRecord,
 )
 from opentelemetry.sdk._logs._internal import (
+    _OTEL_LOG_LEVEL_TO_PYTHON,
     LoggerMetrics,
     NoOpLogger,
     SynchronousMultiLogRecordProcessor,
     _disable_logger_configurator,
     _LoggerConfig,
     _RuleBasedLoggerConfigurator,
 )
-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,
@@ -463,3 +470,84 @@ 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):
+        importlib.reload(_logs_internal)
+
+    def test_otel_log_level_to_python_mapping_accepted_values(self):
+        expected_keys = {
+            "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_default_level_is_info(self):
+        importlib.reload(_logs_internal)
+        self.assertEqual(self._sdk_logger.level, logging.INFO)
+
+    def test_invalid_value_warns_and_defaults_to_info(self):
+        # "trace", "verbose", "none" are valid in other SDKs but not accepted here
+        for invalid in ("INVALID", "trace", "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,
+                    ):
+                        importlib.reload(_logs_internal)
+                self.assertEqual(self._sdk_logger.level, logging.INFO)
+
+    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}):
+                    importlib.reload(_logs_internal)
+                self.assertEqual(self._sdk_logger.level, expected_level)
+
+    @patch.dict("os.environ", {OTEL_LOG_LEVEL: "critical"})
+    def test_level_propagates_to_child_loggers(self):
+        importlib.reload(_logs_internal)
+        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 = [
+            ("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}):
+                    importlib.reload(_logs_internal)
+                self.assertEqual(self._sdk_logger.level, expected_level)