fix(config): cover OTLP HTTP deflate for logs

Signed-off-by: Asish Kumar <officialasishkumar@gmail.com>

Author: officialasishkumar

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Drop Python 3.9 support
   ([#5076](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/5076))
 - `opentelemetry-sdk`: Allow declarative OTLP HTTP exporters to map `compression: deflate` instead of rejecting it as unsupported
+  ([#5075](https://github.com/open-telemetry/opentelemetry-python/pull/5075))
 
 ## Version 1.41.0/0.62b0 (2026-04-09)
 

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

@@ -52,25 +52,27 @@ def _parse_headers(
 
 
 def _map_compression(
-    value: Optional[str],
+    value: str | None,
     compression_enum: type,
     *,
     allow_deflate: bool = False,
-) -> Optional[object]:
+) -> object | None:
     """Map a compression string to the given Compression enum value."""
-    if value is None or value.lower() == "none":
+    if value is None:
         return None
-    if value.lower() == "gzip":
+
+    value_lower = value.lower()
+    supports_deflate = allow_deflate and hasattr(compression_enum, "Deflate")
+
+    if value_lower == "none":
+        return None
+    if value_lower == "gzip":
         return compression_enum.Gzip  # type: ignore[attr-defined]
-    if (
-        value.lower() == "deflate"
-        and allow_deflate
-        and hasattr(compression_enum, "Deflate")
-    ):
+    if value_lower == "deflate" and supports_deflate:
         return compression_enum.Deflate  # type: ignore[attr-defined]
 
     supported_values = ["'gzip'", "'none'"]
-    if allow_deflate and hasattr(compression_enum, "Deflate"):
+    if supports_deflate:
         supported_values.insert(1, "'deflate'")
 
     raise ConfigurationError(

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

@@ -15,10 +15,11 @@
 from __future__ import annotations
 
 import logging
-from typing import Optional
-
 from opentelemetry._logs import set_logger_provider
-from opentelemetry.sdk._configuration._common import _parse_headers
+from opentelemetry.sdk._configuration._common import (
+    _map_compression,
+    _parse_headers,
+)
 from opentelemetry.sdk._configuration._exceptions import ConfigurationError
 from opentelemetry.sdk._configuration.models import (
     BatchLogRecordProcessor as BatchLogRecordProcessorConfig,
@@ -58,20 +59,6 @@
 _DEFAULT_MAX_QUEUE_SIZE = 2048
 _DEFAULT_MAX_EXPORT_BATCH_SIZE = 512
 
-
-def _map_compression(
-    value: Optional[str], compression_enum: type
-) -> Optional[object]:
-    """Map a compression string to the given Compression enum value."""
-    if value is None or value.lower() == "none":
-        return None
-    if value.lower() == "gzip":
-        return compression_enum.Gzip  # type: ignore[attr-defined]
-    raise ConfigurationError(
-        f"Unsupported compression value '{value}'. Supported values: 'gzip', 'none'."
-    )
-
-
 def _create_console_log_exporter() -> ConsoleLogRecordExporter:
     """Create a ConsoleLogRecordExporter."""
     return ConsoleLogRecordExporter()
@@ -95,7 +82,9 @@ def _create_otlp_http_log_exporter(
             "Install it with: pip install opentelemetry-exporter-otlp-proto-http"
         ) from exc
 
-    compression = _map_compression(config.compression, Compression)
+    compression = _map_compression(
+        config.compression, Compression, allow_deflate=True
+    )
     headers = _parse_headers(config.headers, config.headers_list)
     timeout = (config.timeout / 1000.0) if config.timeout is not None else None
 

opentelemetry-sdk/tests/_configuration/test_logger_provider.py

@@ -323,6 +323,30 @@ def test_otlp_http_exporter_headers(self):
         call_kwargs = mock_exporter_cls.call_args.kwargs
         self.assertEqual(call_kwargs["headers"], {"x-api-key": "secret"})
 
+    def test_otlp_http_exporter_deflate_compression(self):
+        mock_exporter_cls = MagicMock()
+        mock_compression_cls = MagicMock()
+        mock_compression_cls.Deflate = "deflate"
+        mock_module = MagicMock()
+        mock_module.Compression = mock_compression_cls
+        mock_log_module = MagicMock()
+        mock_log_module.OTLPLogExporter = mock_exporter_cls
+
+        with patch.dict(
+            sys.modules,
+            {
+                "opentelemetry.exporter.otlp.proto.http": mock_module,
+                "opentelemetry.exporter.otlp.proto.http._log_exporter": mock_log_module,
+            },
+        ):
+            config = LogRecordExporterConfig(
+                otlp_http=OtlpHttpExporterConfig(compression="deflate")
+            )
+            _create_log_record_exporter(config)
+
+        call_kwargs = mock_exporter_cls.call_args.kwargs
+        self.assertEqual(call_kwargs["compression"], "deflate")
+
     def test_otlp_grpc_exporter_endpoint(self):
         mock_exporter_cls = MagicMock()
         mock_grpc = MagicMock()