diff --git a/api/envoy/extensions/stat_sinks/open_telemetry/v3/open_telemetry.proto b/api/envoy/extensions/stat_sinks/open_telemetry/v3/open_telemetry.proto index d2a852ee6dee9..9c9c2779be786 100644 --- a/api/envoy/extensions/stat_sinks/open_telemetry/v3/open_telemetry.proto +++ b/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 `_. + 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}]; } diff --git a/changelogs/current/new_features/stat_sinks__added-otlp-http-compression.rst b/changelogs/current/new_features/stat_sinks__added-otlp-http-compression.rst new file mode 100644 index 0000000000000..c9e3c34b0e38c --- /dev/null +++ b/changelogs/current/new_features/stat_sinks__added-otlp-http-compression.rst @@ -0,0 +1,6 @@ +Added :ref:`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). diff --git a/source/extensions/stat_sinks/open_telemetry/BUILD b/source/extensions/stat_sinks/open_telemetry/BUILD index b00de5785f7ac..c1b52434ccf8d 100644 --- a/source/extensions/stat_sinks/open_telemetry/BUILD +++ b/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", ], ) diff --git a/source/extensions/stat_sinks/open_telemetry/config.cc b/source/extensions/stat_sinks/open_telemetry/config.cc index 07a242940d4b7..874122b8227fa 100644 --- a/source/extensions/stat_sinks/open_telemetry/config.cc +++ b/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 http_metrics_exporter = - std::make_shared(server.clusterManager(), - sink_config.http_service(), server); + std::make_shared( + server.clusterManager(), sink_config.http_service(), server, sink_config.compression()); return std::make_unique( otlp_metrics_flusher, http_metrics_exporter, diff --git a/source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.cc b/source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.cc index 9bd3ad532ef42..bc8f8236aece6 100644 --- a/source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.cc +++ b/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); diff --git a/source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.h b/source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.h index 8ee97cac44afb..1e1b9fe8ba9c7 100644 --- a/source/extensions/stat_sinks/open_telemetry/open_telemetry_http_impl.h +++ b/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 { 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 headers_applicator_; diff --git a/test/extensions/stats_sinks/open_telemetry/BUILD b/test/extensions/stats_sinks/open_telemetry/BUILD index b26dd2725b565..f0db5a241d679 100644 --- a/test/extensions/stats_sinks/open_telemetry/BUILD +++ b/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", diff --git a/test/extensions/stats_sinks/open_telemetry/open_telemetry_http_impl_test.cc b/test/extensions/stats_sinks/open_telemetry/open_telemetry_http_impl_test.cc index 866c1331ce1af..4fd1efbf6cbb9 100644 --- a/test/extensions/stats_sinks/open_telemetry/open_telemetry_http_impl_test.cc +++ b/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,7 +24,10 @@ 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()) @@ -29,7 +35,20 @@ class OpenTelemetryHttpMetricsExporterTest : public testing::Test { cluster_manager_.initializeClusters({"my_o11y_backend"}, {}); http_metrics_exporter_ = std::make_unique( - 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_context_; NiceMock cluster_manager_; + Stats::IsolatedStoreImpl stats_store_; std::unique_ptr 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(