stats: add optional gzip compression to OTLP/HTTP stats sink

api/envoy/extensions/stat_sinks/open_telemetry/v3/open_telemetry.proto

@@ -24,8 +24,22 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // Stats configuration proto schema for ``envoy.stat_sinks.open_telemetry`` sink.
 // [#extension: envoy.stat_sinks.open_telemetry]
 
-// [#next-free-field: 11]
+// [#next-free-field: 12]
 message SinkConfig {
+  // Compression algorithm applied to the serialized OTLP payload before it is sent
+  // to the collector. Only applies to the ``http_service`` (OTLP/HTTP) export path;
+  // for the ``grpc_service`` path, compression is configured via the gRPC service.
+  enum Compression {
+    // No compression is applied. This is the default for backward compatibility.
+    NONE = 0;
+
+    // Gzip compression is applied to the request body and ``Content-Encoding: gzip``
+    // is set on the outbound request. The OTLP/HTTP collector must support gzip
+    // decoding, as required by the
+    // `OTLP/HTTP specification <https://opentelemetry.io/docs/specs/otlp/#otlphttp>`_.
+    GZIP = 1;
+  }
+
   // ConversionAction is used to convert a stat to a metric. If a stat matches,
   // the metric_name and static_metric_labels will be
   // used to create the metric. This can be used to rename a
@@ -98,4 +112,9 @@ message SinkConfig {
   // Maximum number of data points per request. If explicitly set to 0, there is no limit. If unset, it currently defaults to no limit.
   // When the maximum number of data points is reached, the remaining data points will be sent in subsequent requests.
   uint32 max_data_points_per_request = 10;
+
+  // Compression applied to the serialized OTLP payload before it is exported over
+  // OTLP/HTTP. Defaults to ``NONE`` (no compression) for backward compatibility.
+  // This field is only honored when ``http_service`` is configured.
+  Compression compression = 11 [(validate.rules).enum = {defined_only: true}];
 }

changelogs/current/new_features/stat_sinks__added-otlp-http-compression.rst

@@ -0,0 +1,6 @@
+Added :ref:`compression
+<envoy_v3_api_field_extensions.stat_sinks.open_telemetry.v3.SinkConfig.compression>`
+configuration to the OpenTelemetry stat sink. When set to ``GZIP``, the serialized OTLP payload is
+gzip-compressed and ``Content-Encoding: gzip`` is set on the outbound OTLP/HTTP request to reduce
+network bandwidth. Defaults to ``NONE`` for backward compatibility, and is only honored on the
+``http_service`` export path (the OTLP/HTTP collector must support gzip decoding).

source/extensions/stat_sinks/open_telemetry/BUILD

@@ -48,7 +48,9 @@ envoy_cc_library(
     hdrs = ["open_telemetry_http_impl.h"],
     deps = [
         ":open_telemetry_lib",
+        "//envoy/compression/compressor:compressor_interface",
         "//envoy/upstream:cluster_manager_interface",
+        "//source/common/buffer:buffer_lib",
         "//source/common/http:async_client_lib",
         "//source/common/http:async_client_utility_lib",
         "//source/common/http:header_map_lib",
@@ -57,7 +59,9 @@ envoy_cc_library(
         "//source/common/http:utility_lib",
         "//source/common/protobuf",
         "//source/extensions/access_loggers/open_telemetry:otlp_log_utils_lib",
+        "//source/extensions/compression/gzip/compressor:compressor_lib",
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/stat_sinks/open_telemetry/v3:pkg_cc_proto",
     ],
 )
 

source/extensions/stat_sinks/open_telemetry/config.cc

@@ -50,8 +50,8 @@ OpenTelemetrySinkFactory::createStatsSink(const Protobuf::Message& config,
 
   case SinkConfig::ProtocolSpecifierCase::kHttpService: {
     std::shared_ptr<OtlpMetricsExporter> http_metrics_exporter =
-        std::make_shared<OpenTelemetryHttpMetricsExporter>(server.clusterManager(),
-                                                           sink_config.http_service(), server);
+        std::make_shared<OpenTelemetryHttpMetricsExporter>(
+            server.clusterManager(), sink_config.http_service(), server, sink_config.compression());
 
     return std::make_unique<OpenTelemetrySink>(
         otlp_metrics_flusher, http_metrics_exporter,

source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.cc

@@ -1,11 +1,13 @@
 #include "source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.h"
 
+#include "source/common/buffer/buffer_impl.h"
 #include "source/common/common/enum_to_int.h"
 #include "source/common/http/headers.h"
 #include "source/common/http/message_impl.h"
 #include "source/common/http/utility.h"
 #include "source/common/protobuf/protobuf.h"
 #include "source/extensions/access_loggers/open_telemetry/otlp_log_utils.h"
+#include "source/extensions/compression/gzip/compressor/zlib_compressor_impl.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -15,11 +17,37 @@ namespace OpenTelemetry {
 OpenTelemetryHttpMetricsExporter::OpenTelemetryHttpMetricsExporter(
     Upstream::ClusterManager& cluster_manager,
     const envoy::config::core::v3::HttpService& http_service,
-    Server::Configuration::ServerFactoryContext& server_context)
-    : cluster_manager_(cluster_manager), http_service_(http_service),
+    Server::Configuration::ServerFactoryContext& server_context,
+    envoy::extensions::stat_sinks::open_telemetry::v3::SinkConfig::Compression compression)
+    : cluster_manager_(cluster_manager), http_service_(http_service), compression_(compression),
       headers_applicator_(
           Http::HttpServiceHeadersApplicator::createOrThrow(http_service, server_context)) {}
 
+namespace {
+// Gzip compression parameters. These mirror the defaults used by the gzip compressor extension.
+// The gzip header value (16) is OR'ed into the window bits so the deflate stream is wrapped with a
+// gzip header and trailer, which is what ``Content-Encoding: gzip`` requires.
+constexpr int64_t GzipWindowBits = 12;
+constexpr int64_t GzipHeaderValue = 16;
+constexpr uint64_t GzipMemoryLevel = 5;
+} // namespace
+
+bool OpenTelemetryHttpMetricsExporter::compressBody(std::string& body) const {
+  // Gzip is the only compression scheme required by the OTLP/HTTP specification and the only
+  // value currently supported by this sink.
+  Extensions::Compression::Gzip::Compressor::ZlibCompressorImpl compressor;
+  compressor.init(
+      Extensions::Compression::Gzip::Compressor::ZlibCompressorImpl::CompressionLevel::Standard,
+      Extensions::Compression::Gzip::Compressor::ZlibCompressorImpl::CompressionStrategy::Standard,
+      GzipWindowBits | GzipHeaderValue, GzipMemoryLevel);
+
+  Buffer::OwnedImpl buffer;
+  buffer.add(body);
+  compressor.compress(buffer, Envoy::Compression::Compressor::State::Finish);
+  body = buffer.toString();
+  return true;
+}
+
 void OpenTelemetryHttpMetricsExporter::send(MetricsExportRequestPtr&& metrics) {
   std::string request_body;
   const auto ok = metrics->SerializeToString(&request_body);
@@ -46,6 +74,14 @@ void OpenTelemetryHttpMetricsExporter::send(MetricsExportRequestPtr&& metrics) {
   // User-Agent header follows the OTLP specification.
   message->headers().setReferenceUserAgent(AccessLoggers::OpenTelemetry::getOtlpUserAgentHeader());
 
+  // Optionally compress the serialized payload before sending. When enabled, set the
+  // Content-Encoding header so the OTLP/HTTP collector knows how to decode the body.
+  if (compression_ == envoy::extensions::stat_sinks::open_telemetry::v3::SinkConfig::GZIP &&
+      compressBody(request_body)) {
+    message->headers().setReference(Http::CustomHeaders::get().ContentEncoding,
+                                    Http::CustomHeaders::get().ContentEncodingValues.Gzip);
+  }
+
   // Add custom headers from config.
   headers_applicator_->apply(message->headers());
   message->body().add(request_body);

source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "envoy/config/core/v3/http_service.pb.h"
+#include "envoy/extensions/stat_sinks/open_telemetry/v3/open_telemetry.pb.h"
 #include "envoy/server/factory_context.h"
 #include "envoy/upstream/cluster_manager.h"
 
@@ -22,9 +23,12 @@ class OpenTelemetryHttpMetricsExporter : public OtlpMetricsExporter,
                                          public Http::AsyncClient::Callbacks,
                                          public Logger::Loggable<Logger::Id::stats> {
 public:
-  OpenTelemetryHttpMetricsExporter(Upstream::ClusterManager& cluster_manager,
-                                   const envoy::config::core::v3::HttpService& http_service,
-                                   Server::Configuration::ServerFactoryContext& server_context);
+  OpenTelemetryHttpMetricsExporter(
+      Upstream::ClusterManager& cluster_manager,
+      const envoy::config::core::v3::HttpService& http_service,
+      Server::Configuration::ServerFactoryContext& server_context,
+      envoy::extensions::stat_sinks::open_telemetry::v3::SinkConfig::Compression compression =
+          envoy::extensions::stat_sinks::open_telemetry::v3::SinkConfig::NONE);
 
   // OtlpMetricsExporter
   void send(MetricsExportRequestPtr&& metrics) override;
@@ -35,8 +39,12 @@ class OpenTelemetryHttpMetricsExporter : public OtlpMetricsExporter,
   void onBeforeFinalizeUpstreamSpan(Tracing::Span&, const Http::ResponseHeaderMap*) override {}
 
 private:
+  // Compresses the given body in place using gzip. Returns true on success.
+  bool compressBody(std::string& body) const;
+
   Upstream::ClusterManager& cluster_manager_;
   envoy::config::core::v3::HttpService http_service_;
+  const envoy::extensions::stat_sinks::open_telemetry::v3::SinkConfig::Compression compression_;
   // Track active HTTP requests to cancel them on destruction.
   Http::AsyncClientRequestTracker active_requests_;
   std::unique_ptr<Http::HttpServiceHeadersApplicator> headers_applicator_;

test/extensions/stats_sinks/open_telemetry/BUILD

@@ -44,6 +44,9 @@ envoy_extension_cc_test(
     srcs = ["open_telemetry_http_impl_test.cc"],
     extension_names = ["envoy.stat_sinks.open_telemetry"],
     deps = [
+        "//source/common/buffer:buffer_lib",
+        "//source/common/stats:isolated_store_lib",
+        "//source/extensions/compression/gzip/decompressor:zlib_decompressor_impl_lib",
         "//source/extensions/stat_sinks/open_telemetry:open_telemetry_http_lib",
         "//test/mocks/server:server_factory_context_mocks",
         "//test/mocks/upstream:cluster_manager_mocks",

test/extensions/stats_sinks/open_telemetry/open_telemetry_http_impl_test.cc

@@ -1,4 +1,7 @@
+#include "source/common/buffer/buffer_impl.h"
+#include "source/common/stats/isolated_store_impl.h"
 #include "source/common/tracing/null_span_impl.h"
+#include "source/extensions/compression/gzip/decompressor/zlib_decompressor_impl.h"
 #include "source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.h"
 
 #include "test/mocks/server/server_factory_context.h"
@@ -21,15 +24,31 @@ using testing::ReturnRef;
 
 class OpenTelemetryHttpMetricsExporterTest : public testing::Test {
 public:
-  void setup(envoy::config::core::v3::HttpService http_service) {
+  void
+  setup(envoy::config::core::v3::HttpService http_service,
+        envoy::extensions::stat_sinks::open_telemetry::v3::SinkConfig::Compression compression =
+            envoy::extensions::stat_sinks::open_telemetry::v3::SinkConfig::NONE) {
     cluster_manager_.thread_local_cluster_.cluster_.info_->name_ = "my_o11y_backend";
     cluster_manager_.initializeThreadLocalClusters({"my_o11y_backend"});
     ON_CALL(cluster_manager_.thread_local_cluster_, httpAsyncClient())
         .WillByDefault(ReturnRef(cluster_manager_.thread_local_cluster_.async_client_));
     cluster_manager_.initializeClusters({"my_o11y_backend"}, {});
 
     http_metrics_exporter_ = std::make_unique<OpenTelemetryHttpMetricsExporter>(
-        cluster_manager_, http_service, server_context_);
+        cluster_manager_, http_service, server_context_, compression);
+  }
+
+  // Gzip-decompresses the given payload for round-trip verification in tests.
+  std::string gzipDecompress(const std::string& compressed) {
+    Buffer::OwnedImpl input;
+    input.add(compressed);
+    Buffer::OwnedImpl output;
+    Extensions::Compression::Gzip::Decompressor::ZlibDecompressorImpl decompressor{
+        *stats_store_.rootScope(), "test.", 4096, 100};
+    // 15 (max window) + 16 selects gzip decoding to match the compressor's gzip header.
+    decompressor.init(15 + 16);
+    decompressor.decompress(input, output);
+    return output.toString();
   }
 
   MetricsExportRequestPtr createTestMetricsRequest() {
@@ -47,6 +66,7 @@ class OpenTelemetryHttpMetricsExporterTest : public testing::Test {
 protected:
   NiceMock<Server::Configuration::MockServerFactoryContext> server_context_;
   NiceMock<Upstream::MockClusterManager> cluster_manager_;
+  Stats::IsolatedStoreImpl stats_store_;
   std::unique_ptr<OpenTelemetryHttpMetricsExporter> http_metrics_exporter_;
 };
 
@@ -120,6 +140,99 @@ TEST_F(OpenTelemetryHttpMetricsExporterTest, ExportMetricsWithCustomHeaders) {
   callback->onSuccess(request, std::move(msg));
 }
 
+// Verifies that with the default (NONE) compression, no Content-Encoding header is set and the
+// body is the uncompressed serialized proto.
+TEST_F(OpenTelemetryHttpMetricsExporterTest, ExportMetricsUncompressed) {
+  std::string yaml_string = R"EOF(
+  http_uri:
+    uri: "https://some-o11y.com/v1/metrics"
+    cluster: "my_o11y_backend"
+    timeout: 0.250s
+  )EOF";
+
+  envoy::config::core::v3::HttpService http_service;
+  TestUtility::loadFromYaml(yaml_string, http_service);
+  setup(http_service);
+
+  auto request_proto = createTestMetricsRequest();
+  std::string expected_body;
+  ASSERT_TRUE(request_proto->SerializeToString(&expected_body));
+
+  Http::MockAsyncClientRequest request(&cluster_manager_.thread_local_cluster_.async_client_);
+  Http::AsyncClient::Callbacks* callback;
+
+  EXPECT_CALL(cluster_manager_.thread_local_cluster_.async_client_, send_(_, _, _))
+      .WillOnce(
+          Invoke([&](Http::RequestMessagePtr& message, Http::AsyncClient::Callbacks& callbacks,
+                     const Http::AsyncClient::RequestOptions&) -> Http::AsyncClient::Request* {
+            callback = &callbacks;
+            // No Content-Encoding header should be present.
+            EXPECT_TRUE(message->headers().get(Http::CustomHeaders::get().ContentEncoding).empty());
+            // Body is the raw serialized proto.
+            EXPECT_EQ(expected_body, message->body().toString());
+            return &request;
+          }));
+
+  http_metrics_exporter_->send(std::move(request_proto));
+
+  Http::ResponseMessagePtr msg(new Http::ResponseMessageImpl(
+      Http::ResponseHeaderMapPtr{new Http::TestResponseHeaderMapImpl{{":status", "200"}}}));
+  callback->onSuccess(request, std::move(msg));
+}
+
+// Verifies that with GZIP compression, the Content-Encoding header is set to gzip and the body is
+// a valid gzip stream that decompresses back to the original serialized proto.
+TEST_F(OpenTelemetryHttpMetricsExporterTest, ExportMetricsGzipCompressed) {
+  std::string yaml_string = R"EOF(
+  http_uri:
+    uri: "https://some-o11y.com/v1/metrics"
+    cluster: "my_o11y_backend"
+    timeout: 0.250s
+  )EOF";
+
+  envoy::config::core::v3::HttpService http_service;
+  TestUtility::loadFromYaml(yaml_string, http_service);
+  setup(http_service, envoy::extensions::stat_sinks::open_telemetry::v3::SinkConfig::GZIP);
+
+  auto request_proto = createTestMetricsRequest();
+  std::string uncompressed_body;
+  ASSERT_TRUE(request_proto->SerializeToString(&uncompressed_body));
+
+  Http::MockAsyncClientRequest request(&cluster_manager_.thread_local_cluster_.async_client_);
+  Http::AsyncClient::Callbacks* callback;
+
+  EXPECT_CALL(cluster_manager_.thread_local_cluster_.async_client_, send_(_, _, _))
+      .WillOnce(
+          Invoke([&](Http::RequestMessagePtr& message, Http::AsyncClient::Callbacks& callbacks,
+                     const Http::AsyncClient::RequestOptions&) -> Http::AsyncClient::Request* {
+            callback = &callbacks;
+            // Content-Encoding: gzip must be set.
+            const auto encoding =
+                message->headers().get(Http::CustomHeaders::get().ContentEncoding);
+            EXPECT_FALSE(encoding.empty());
+            if (!encoding.empty()) {
+              EXPECT_EQ(Http::CustomHeaders::get().ContentEncodingValues.Gzip,
+                        encoding[0]->value().getStringView());
+            }
+
+            const std::string compressed_body = message->body().toString();
+            // The compressed body should start with the gzip magic bytes (0x1f 0x8b).
+            EXPECT_GE(compressed_body.size(), 2);
+            EXPECT_EQ('\x1f', compressed_body[0]);
+            EXPECT_EQ('\x8b', compressed_body[1]);
+
+            // Round-trip: decompressing the body yields the original serialized proto.
+            EXPECT_EQ(uncompressed_body, gzipDecompress(compressed_body));
+            return &request;
+          }));
+
+  http_metrics_exporter_->send(std::move(request_proto));
+
+  Http::ResponseMessagePtr msg(new Http::ResponseMessageImpl(
+      Http::ResponseHeaderMapPtr{new Http::TestResponseHeaderMapImpl{{":status", "200"}}}));
+  callback->onSuccess(request, std::move(msg));
+}
+
 // Verifies that export is aborted gracefully when the cluster is not found.
 TEST_F(OpenTelemetryHttpMetricsExporterTest, UnsuccessfulExportWithoutThreadLocalCluster) {
   std::string yaml_string = R"EOF(