fix(config): allow deflate for OTLP HTTP exporters

CHANGELOG.md

@@ -20,7 +20,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4907](https://github.com/open-telemetry/opentelemetry-python/issues/4907))
 - 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

@@ -17,6 +17,8 @@
 import logging
 from typing import Optional
 
+from opentelemetry.sdk._configuration._exceptions import ConfigurationError
+
 _logger = logging.getLogger(__name__)
 
 
@@ -47,3 +49,33 @@ def _parse_headers(
         for pair in headers:
             result[pair.name] = pair.value or ""
     return result
+
+
+def _map_compression(
+    value: str | None,
+    compression_enum: type,
+    *,
+    allow_deflate: bool = False,
+) -> object | None:
+    """Map a compression string to the given Compression enum value."""
+    if value is None:
+        return None
+
+    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 supports_deflate:
+        return compression_enum.Deflate  # type: ignore[attr-defined]
+
+    supported_values = ["'gzip'", "'none'"]
+    if supports_deflate:
+        supported_values.insert(1, "'deflate'")
+
+    raise ConfigurationError(
+        f"Unsupported compression value '{value}'. Supported values: "
+        f"{', '.join(supported_values)}."
+    )

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/src/opentelemetry/sdk/_configuration/_meter_provider.py

@@ -18,7 +18,10 @@
 from typing import Optional, Set, Type
 
 from opentelemetry import metrics
-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 (
     Aggregation as AggregationConfig,
@@ -265,19 +268,6 @@ def _create_console_metric_exporter(
     )
 
 
-def _map_compression_metric(
-    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_otlp_http_metric_exporter(
     config: OtlpHttpMetricExporterConfig,
 ) -> MetricExporter:
@@ -296,7 +286,9 @@ def _create_otlp_http_metric_exporter(
             "Install it with: pip install opentelemetry-exporter-otlp-proto-http"
         ) from exc
 
-    compression = _map_compression_metric(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
     preferred_temporality = _map_temporality(config.temporality_preference)
@@ -331,7 +323,7 @@ def _create_otlp_grpc_metric_exporter(
             "Install it with: pip install opentelemetry-exporter-otlp-proto-grpc"
         ) from exc
 
-    compression = _map_compression_metric(config.compression, grpc.Compression)
+    compression = _map_compression(config.compression, grpc.Compression)
     headers = _parse_headers(config.headers, config.headers_list)
     timeout = (config.timeout / 1000.0) if config.timeout is not None else None
     preferred_temporality = _map_temporality(config.temporality_preference)

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

@@ -18,7 +18,10 @@
 from typing import Optional
 
 from opentelemetry import trace
-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 (
     OtlpGrpcExporter as OtlpGrpcExporterConfig,
@@ -92,7 +95,9 @@ def _create_otlp_http_span_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
 
@@ -103,20 +108,6 @@ def _create_otlp_http_span_exporter(
         compression=compression,  # type: ignore[arg-type]
     )
 
-
-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_otlp_grpc_span_exporter(
     config: OtlpGrpcExporterConfig,
 ) -> SpanExporter:

opentelemetry-sdk/tests/_configuration/test_common.py

@@ -15,7 +15,11 @@
 import unittest
 from types import SimpleNamespace
 
-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
 
 
 class TestParseHeaders(unittest.TestCase):
@@ -79,3 +83,69 @@ def test_struct_headers_override_headers_list(self):
 
     def test_both_empty_struct_and_none_list_returns_empty_dict(self):
         self.assertEqual(_parse_headers([], None), {})
+
+
+class _CompressionWithDeflate:
+    Gzip = "gzip"
+    Deflate = "deflate"
+
+
+class _CompressionWithoutDeflate:
+    Gzip = "gzip"
+
+
+class TestMapCompression(unittest.TestCase):
+    def test_none_returns_none(self):
+        self.assertIsNone(_map_compression(None, _CompressionWithDeflate))
+
+    def test_none_string_returns_none(self):
+        self.assertIsNone(
+            _map_compression("none", _CompressionWithDeflate)
+        )
+
+    def test_gzip_maps_to_gzip(self):
+        self.assertEqual(
+            _map_compression("gzip", _CompressionWithDeflate), "gzip"
+        )
+
+    def test_deflate_maps_when_enabled(self):
+        self.assertEqual(
+            _map_compression(
+                "deflate", _CompressionWithDeflate, allow_deflate=True
+            ),
+            "deflate",
+        )
+
+    def test_deflate_raises_by_default(self):
+        with self.assertRaises(ConfigurationError) as ctx:
+            _map_compression("deflate", _CompressionWithDeflate)
+
+        self.assertEqual(
+            str(ctx.exception),
+            "Unsupported compression value 'deflate'. Supported values: "
+            "'gzip', 'none'.",
+        )
+
+    def test_deflate_raises_when_http_enum_lacks_support(self):
+        with self.assertRaises(ConfigurationError) as ctx:
+            _map_compression(
+                "deflate", _CompressionWithoutDeflate, allow_deflate=True
+            )
+
+        self.assertEqual(
+            str(ctx.exception),
+            "Unsupported compression value 'deflate'. Supported values: "
+            "'gzip', 'none'.",
+        )
+
+    def test_http_error_message_includes_deflate(self):
+        with self.assertRaises(ConfigurationError) as ctx:
+            _map_compression(
+                "brotli", _CompressionWithDeflate, allow_deflate=True
+            )
+
+        self.assertEqual(
+            str(ctx.exception),
+            "Unsupported compression value 'brotli'. Supported values: "
+            "'gzip', 'deflate', '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()

opentelemetry-sdk/tests/_configuration/test_meter_provider.py

@@ -274,6 +274,34 @@ def test_otlp_http_created_with_endpoint(self):
         self.assertIsNone(kwargs["timeout"])
         self.assertIsNone(kwargs["compression"])
 
+    def test_otlp_http_created_with_deflate_compression(self):
+        mock_exporter_cls = MagicMock()
+        mock_compression_cls = MagicMock()
+        mock_compression_cls.Deflate = "deflate_val"
+        mock_http_module = MagicMock()
+        mock_http_module.Compression = mock_compression_cls
+        mock_module = MagicMock()
+        mock_module.OTLPMetricExporter = mock_exporter_cls
+
+        with patch.dict(
+            sys.modules,
+            {
+                "opentelemetry.exporter.otlp.proto.http.metric_exporter": mock_module,
+                "opentelemetry.exporter.otlp.proto.http": mock_http_module,
+            },
+        ):
+            config = self._make_periodic_config(
+                PushMetricExporterConfig(
+                    otlp_http=OtlpHttpMetricExporterConfig(
+                        compression="deflate"
+                    )
+                )
+            )
+            create_meter_provider(config)
+
+        _, kwargs = mock_exporter_cls.call_args
+        self.assertEqual(kwargs["compression"], "deflate_val")
+
     def test_otlp_grpc_missing_package_raises(self):
         config = self._make_periodic_config(
             PushMetricExporterConfig(otlp_grpc=OtlpGrpcMetricExporterConfig())

opentelemetry-sdk/tests/_configuration/test_tracer_provider.py

@@ -309,6 +309,32 @@ def test_otlp_http_created_with_endpoint(self):
             compression=None,
         )
 
+    def test_otlp_http_created_with_deflate_compression(self):
+        mock_exporter_cls = MagicMock()
+        mock_compression_cls = MagicMock()
+        mock_compression_cls.Deflate = "deflate_val"
+        mock_module = MagicMock()
+        mock_module.OTLPSpanExporter = mock_exporter_cls
+        mock_http_module = MagicMock()
+        mock_http_module.Compression = mock_compression_cls
+
+        with patch.dict(
+            sys.modules,
+            {
+                "opentelemetry.exporter.otlp.proto.http.trace_exporter": mock_module,
+                "opentelemetry.exporter.otlp.proto.http": mock_http_module,
+            },
+        ):
+            config = self._make_batch_config(
+                SpanExporterConfig(
+                    otlp_http=OtlpHttpExporterConfig(compression="deflate")
+                )
+            )
+            create_tracer_provider(config)
+
+        _, kwargs = mock_exporter_cls.call_args
+        self.assertEqual(kwargs["compression"], "deflate_val")
+
     def test_otlp_http_headers_list(self):
         mock_exporter_cls = MagicMock()
         mock_http_module = MagicMock()