[WIP] gzip: record compression latencies

Helps with #envoyproxy#8448.
Related to envoyproxy#10530.

Signed-off-by: Raul Gutierrez Segales <rgs@pinterest.com>

source/extensions/filters/http/common/compressor/BUILD

@@ -20,6 +20,7 @@ envoy_cc_library(
         "//source/common/http:header_map_lib",
         "//source/common/protobuf",
         "//source/common/runtime:runtime_lib",
+        "//source/common/stats:symbol_table_lib",
         "//source/extensions/filters/http/common:pass_through_filter_lib",
         "@envoy_api//envoy/extensions/filters/http/compressor/v3:pkg_cc_proto",
     ],

source/extensions/filters/http/common/compressor/compressor.cc

@@ -37,13 +37,16 @@ const std::string& compressorRegistryKey() { CONSTRUCT_ON_FIRST_USE(std::string,
 CompressorFilterConfig::CompressorFilterConfig(
     const envoy::extensions::filters::http::compressor::v3::Compressor& compressor,
     const std::string& stats_prefix, Stats::Scope& scope, Runtime::Loader& runtime,
-    const std::string& content_encoding)
+    const std::string& content_encoding, TimeSource& time_source)
     : content_length_(contentLengthUint(compressor.content_length().value())),
       content_type_values_(contentTypeSet(compressor.content_type())),
       disable_on_etag_header_(compressor.disable_on_etag_header()),
       remove_accept_encoding_header_(compressor.remove_accept_encoding_header()),
       stats_(generateStats(stats_prefix, scope)), enabled_(compressor.runtime_enabled(), runtime),
-      content_encoding_(content_encoding) {}
+      content_encoding_(content_encoding), time_source_(time_source), scope_(scope),
+      stat_name_set_(scope.symbolTable().makeSet("Gzip")),
+      compression_latency_(stat_name_set_->add("compression_latency")),
+      stats_prefix_(stat_name_set_->add(stats_prefix)) {}
 
 StringUtil::CaseUnorderedSet
 CompressorFilterConfig::contentTypeSet(const Protobuf::RepeatedPtrField<std::string>& types) {
@@ -118,19 +121,35 @@ Http::FilterHeadersStatus CompressorFilter::encodeHeaders(Http::ResponseHeaderMa
   return Http::FilterHeadersStatus::Continue;
 }
 
+void CompressorFilter::compress(Buffer::Instance& data, Compressor::State state) {
+  const auto start_time = config_->timeSource().monotonicTime();
+  compressor_->compress(data, state);
+  const auto latency = std::chrono::duration_cast<std::chrono::milliseconds>(
+      config_->timeSource().monotonicTime() - start_time);
+  auto& scope = config_->scope();
+  Stats::SymbolTable::StoragePtr storage =
+      scope.symbolTable().join({config_->statsPrefix(), config_->compressionLatencyStatName()});
+  scope.histogramFromStatName(Stats::StatName(storage.get()), Stats::Histogram::Unit::Milliseconds)
+      .recordValue(latency.count());
+
+  if (state == Compressor::State::NoFlush) {
+    config_->stats().compressed_no_flush_.inc();
+  } else if (state == Compressor::State::Finish) {
+    config_->stats().compressed_finished_.inc();
+  }
+}
+
 Http::FilterDataStatus CompressorFilter::encodeData(Buffer::Instance& data, bool end_stream) {
   if (!skip_compression_) {
     config_->stats().total_uncompressed_bytes_.add(data.length());
     if (end_stream) {
-      compressor_->compress(data, Compressor::State::Finish);
-      config_->stats().compressed_finished_.inc();
+      compress(data, Compressor::State::Finish);
     } else {
       // Note: it's fine that we don't flush here, it'll either happen in a future encodeData
       // call or in encodeTrailers(). This will not cause a regression for
       // https://github.com/envoyproxy/envoy/pull/3025, given that before encodeTrailers wasn't
       // handled.
-      compressor_->compress(data, Compressor::State::NoFlush);
-      config_->stats().compressed_no_flush_.inc();
+      compress(data, Compressor::State::NoFlush);
     }
     config_->stats().total_compressed_bytes_.add(data.length());
   }
@@ -140,8 +159,7 @@ Http::FilterDataStatus CompressorFilter::encodeData(Buffer::Instance& data, bool
 Http::FilterTrailersStatus CompressorFilter::encodeTrailers(Http::ResponseTrailerMap&) {
   if (!skip_compression_) {
     Buffer::OwnedImpl empty_buffer;
-    compressor_->compress(empty_buffer, Compressor::State::Finish);
-    config_->stats().compressed_finished_.inc();
+    compress(empty_buffer, Compressor::State::Finish);
     config_->stats().total_compressed_bytes_.add(empty_buffer.length());
     encoder_callbacks_->addEncodedData(empty_buffer, true);
   }

source/extensions/filters/http/common/compressor/compressor.h

@@ -8,6 +8,7 @@
 
 #include "common/protobuf/protobuf.h"
 #include "common/runtime/runtime_protos.h"
+#include "common/stats/symbol_table_impl.h"
 
 #include "extensions/filters/http/common/pass_through_filter.h"
 
@@ -72,12 +73,16 @@ class CompressorFilterConfig {
   uint32_t minimumLength() const { return content_length_; }
   const std::string contentEncoding() const { return content_encoding_; };
   const std::map<std::string, uint32_t> registeredCompressors() const;
+  TimeSource& timeSource() { return time_source_; }
+  Stats::Scope& scope() { return scope_; }
+  Stats::StatName compressionLatencyStatName() { return compression_latency_; }
+  Stats::StatName statsPrefix() { return stats_prefix_; }
 
 protected:
   CompressorFilterConfig(
       const envoy::extensions::filters::http::compressor::v3::Compressor& compressor,
       const std::string& stats_prefix, Stats::Scope& scope, Runtime::Loader& runtime,
-      const std::string& content_encoding);
+      const std::string& content_encoding, TimeSource& time_source);
 
 private:
   static StringUtil::CaseUnorderedSet
@@ -97,6 +102,11 @@ class CompressorFilterConfig {
   const CompressorStats stats_;
   Runtime::FeatureFlag enabled_;
   const std::string content_encoding_;
+  TimeSource& time_source_;
+  Stats::Scope& scope_;
+  Stats::StatNameSetPtr stat_name_set_;
+  const Stats::StatName compression_latency_;
+  const Stats::StatName stats_prefix_;
 };
 using CompressorFilterConfigSharedPtr = std::shared_ptr<CompressorFilterConfig>;
 
@@ -148,6 +158,7 @@ class CompressorFilter : public Http::PassThroughFilter {
 
   std::unique_ptr<EncodingDecision> chooseEncoding(const Http::ResponseHeaderMap& headers) const;
   bool shouldCompress(const EncodingDecision& decision) const;
+  void compress(Buffer::Instance& data, Compressor::State state);
 
   bool skip_compression_;
   std::unique_ptr<Compressor::Compressor> compressor_;

source/extensions/filters/http/gzip/config.cc

@@ -10,8 +10,9 @@ namespace Gzip {
 Http::FilterFactoryCb GzipFilterFactory::createFilterFactoryFromProtoTyped(
     const envoy::extensions::filters::http::gzip::v3::Gzip& proto_config,
     const std::string& stats_prefix, Server::Configuration::FactoryContext& context) {
-  Common::Compressors::CompressorFilterConfigSharedPtr config = std::make_shared<GzipFilterConfig>(
-      proto_config, stats_prefix, context.scope(), context.runtime());
+  Common::Compressors::CompressorFilterConfigSharedPtr config =
+      std::make_shared<GzipFilterConfig>(proto_config, stats_prefix, context.scope(),
+                                         context.runtime(), context.dispatcher().timeSource());
   return [config](Http::FilterChainFactoryCallbacks& callbacks) -> void {
     callbacks.addStreamFilter(std::make_shared<Common::Compressors::CompressorFilter>(config));
   };

source/extensions/filters/http/gzip/gzip_filter.cc

@@ -22,9 +22,9 @@ const uint64_t GzipHeaderValue = 16;
 
 GzipFilterConfig::GzipFilterConfig(const envoy::extensions::filters::http::gzip::v3::Gzip& gzip,
                                    const std::string& stats_prefix, Stats::Scope& scope,
-                                   Runtime::Loader& runtime)
+                                   Runtime::Loader& runtime, TimeSource& time_source)
     : CompressorFilterConfig(compressorConfig(gzip), stats_prefix + "gzip.", scope, runtime,
-                             Http::Headers::get().ContentEncodingValues.Gzip),
+                             Http::Headers::get().ContentEncodingValues.Gzip, time_source),
       compression_level_(compressionLevelEnum(gzip.compression_level())),
       compression_strategy_(compressionStrategyEnum(gzip.compression_strategy())),
       memory_level_(memoryLevelUint(gzip.memory_level().value())),

source/extensions/filters/http/gzip/gzip_filter.h

@@ -18,7 +18,8 @@ class GzipFilterConfig : public Common::Compressors::CompressorFilterConfig {
 
 public:
   GzipFilterConfig(const envoy::extensions::filters::http::gzip::v3::Gzip& gzip,
-                   const std::string& stats_prefix, Stats::Scope& scope, Runtime::Loader& runtime);
+                   const std::string& stats_prefix, Stats::Scope& scope, Runtime::Loader& runtime,
+                   TimeSource& time_source);
 
   std::unique_ptr<Compressor::Compressor> makeCompressor() override;
 

test/extensions/filters/http/common/compressor/BUILD

@@ -17,6 +17,7 @@ envoy_cc_test(
         "//test/mocks/http:http_mocks",
         "//test/mocks/protobuf:protobuf_mocks",
         "//test/mocks/runtime:runtime_mocks",
+        "//test/test_common:simulated_time_system_lib",
         "//test/test_common:utility_lib",
         "@envoy_api//envoy/extensions/filters/http/compressor/v3:pkg_cc_proto",
     ],

test/extensions/filters/http/common/compressor/compressor_filter_test.cc

@@ -9,6 +9,7 @@
 #include "test/mocks/runtime/mocks.h"
 #include "test/mocks/stats/mocks.h"
 #include "test/test_common/utility.h"
+#include "test/test_common/simulated_time_system.h"
 
 #include "gtest/gtest.h"
 
@@ -29,9 +30,9 @@ class MockCompressorFilterConfig : public CompressorFilterConfig {
   MockCompressorFilterConfig(
       const envoy::extensions::filters::http::compressor::v3::Compressor& compressor,
       const std::string& stats_prefix, Stats::Scope& scope, Runtime::Loader& runtime,
-      const std::string& compressor_name)
+      const std::string& compressor_name, TimeSource& time_source)
       : CompressorFilterConfig(compressor, stats_prefix + compressor_name + ".", scope, runtime,
-                               compressor_name) {}
+                               compressor_name, time_source) {}
 
   std::unique_ptr<Compressor::Compressor> makeCompressor() override {
     return std::make_unique<MockCompressor>();
@@ -90,7 +91,8 @@ class CompressorFilterTest : public testing::Test {
   void setUpFilter(std::string&& json) {
     envoy::extensions::filters::http::compressor::v3::Compressor compressor;
     TestUtility::loadFromJson(json, compressor);
-    config_.reset(new MockCompressorFilterConfig(compressor, "test.", stats_, runtime_, "test"));
+    config_.reset(new MockCompressorFilterConfig(compressor, "test.", stats_, runtime_, "test",
+                                                 time_system_));
     filter_ = std::make_unique<CompressorFilter>(config_);
     filter_->setEncoderFilterCallbacks(encoder_callbacks_);
   }
@@ -156,11 +158,12 @@ class CompressorFilterTest : public testing::Test {
     EXPECT_EQ(1, stats_.counter("test.test.not_compressed").value());
   }
 
+  Stats::IsolatedStoreImpl stats_;
+  Event::SimulatedTimeSystem time_system_;
   CompressorFilterConfigSharedPtr config_;
   std::unique_ptr<CompressorFilter> filter_;
   Buffer::OwnedImpl data_;
   std::string expected_str_;
-  Stats::IsolatedStoreImpl stats_;
   NiceMock<Runtime::MockLoader> runtime_;
   NiceMock<Http::MockStreamEncoderFilterCallbacks> encoder_callbacks_;
 };
@@ -348,10 +351,12 @@ TEST_F(CompressorFilterTest, IsAcceptEncodingAllowed) {
     // The independence is simulated with a new instance DecoderFilterCallbacks set for "test2".
     Stats::IsolatedStoreImpl stats;
     NiceMock<Runtime::MockLoader> runtime;
+    Event::SimulatedTimeSystem time_system;
     envoy::extensions::filters::http::compressor::v3::Compressor compressor;
     TestUtility::loadFromJson("{}", compressor);
     CompressorFilterConfigSharedPtr config2;
-    config2.reset(new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test2"));
+    config2.reset(
+        new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test2", time_system));
     std::unique_ptr<CompressorFilter> filter2 = std::make_unique<CompressorFilter>(config2);
     NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks;
     filter2->setDecoderFilterCallbacks(decoder_callbacks);
@@ -370,10 +375,12 @@ TEST_F(CompressorFilterTest, IsAcceptEncodingAllowed) {
     // check if the legacy "header_gzip" counter is incremented for gzip compression filter
     Stats::IsolatedStoreImpl stats;
     NiceMock<Runtime::MockLoader> runtime;
+    Event::SimulatedTimeSystem time_system;
     envoy::extensions::filters::http::compressor::v3::Compressor compressor;
     TestUtility::loadFromJson("{}", compressor);
     CompressorFilterConfigSharedPtr config2;
-    config2.reset(new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "gzip"));
+    config2.reset(
+        new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "gzip", time_system));
     std::unique_ptr<CompressorFilter> gzip_filter = std::make_unique<CompressorFilter>(config2);
     NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks;
     gzip_filter->setDecoderFilterCallbacks(decoder_callbacks);
@@ -388,10 +395,12 @@ TEST_F(CompressorFilterTest, IsAcceptEncodingAllowed) {
     // check if identity stat is increased twice (the second time via the cached path).
     Stats::IsolatedStoreImpl stats;
     NiceMock<Runtime::MockLoader> runtime;
+    Event::SimulatedTimeSystem time_system;
     envoy::extensions::filters::http::compressor::v3::Compressor compressor;
     TestUtility::loadFromJson("{}", compressor);
     CompressorFilterConfigSharedPtr config2;
-    config2.reset(new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test"));
+    config2.reset(
+        new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test", time_system));
     std::unique_ptr<CompressorFilter> filter2 = std::make_unique<CompressorFilter>(config2);
     NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks;
     filter2->setDecoderFilterCallbacks(decoder_callbacks);
@@ -406,10 +415,12 @@ TEST_F(CompressorFilterTest, IsAcceptEncodingAllowed) {
     // check if not_valid stat is increased twice (the second time via the cached path).
     Stats::IsolatedStoreImpl stats;
     NiceMock<Runtime::MockLoader> runtime;
+    Event::SimulatedTimeSystem time_system;
     envoy::extensions::filters::http::compressor::v3::Compressor compressor;
     TestUtility::loadFromJson("{}", compressor);
     CompressorFilterConfigSharedPtr config2;
-    config2.reset(new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test"));
+    config2.reset(
+        new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test", time_system));
     std::unique_ptr<CompressorFilter> filter2 = std::make_unique<CompressorFilter>(config2);
     NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks;
     filter2->setDecoderFilterCallbacks(decoder_callbacks);
@@ -424,13 +435,16 @@ TEST_F(CompressorFilterTest, IsAcceptEncodingAllowed) {
     // Test that encoding decision is cached when used by multiple filters.
     Stats::IsolatedStoreImpl stats;
     NiceMock<Runtime::MockLoader> runtime;
+    Event::SimulatedTimeSystem time_system;
     envoy::extensions::filters::http::compressor::v3::Compressor compressor;
     TestUtility::loadFromJson("{}", compressor);
     CompressorFilterConfigSharedPtr config1;
-    config1.reset(new MockCompressorFilterConfig(compressor, "test1.", stats, runtime, "test1"));
+    config1.reset(
+        new MockCompressorFilterConfig(compressor, "test1.", stats, runtime, "test1", time_system));
     std::unique_ptr<CompressorFilter> filter1 = std::make_unique<CompressorFilter>(config1);
     CompressorFilterConfigSharedPtr config2;
-    config2.reset(new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test2"));
+    config2.reset(
+        new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test2", time_system));
     std::unique_ptr<CompressorFilter> filter2 = std::make_unique<CompressorFilter>(config2);
     NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks;
     filter1->setDecoderFilterCallbacks(decoder_callbacks);
@@ -451,13 +465,16 @@ TEST_F(CompressorFilterTest, IsAcceptEncodingAllowed) {
     // Test that first registered filter is used when handling wildcard.
     Stats::IsolatedStoreImpl stats;
     NiceMock<Runtime::MockLoader> runtime;
+    Event::SimulatedTimeSystem time_system;
     envoy::extensions::filters::http::compressor::v3::Compressor compressor;
     TestUtility::loadFromJson("{}", compressor);
     CompressorFilterConfigSharedPtr config1;
-    config1.reset(new MockCompressorFilterConfig(compressor, "test1.", stats, runtime, "test1"));
+    config1.reset(
+        new MockCompressorFilterConfig(compressor, "test1.", stats, runtime, "test1", time_system));
     std::unique_ptr<CompressorFilter> filter1 = std::make_unique<CompressorFilter>(config1);
     CompressorFilterConfigSharedPtr config2;
-    config2.reset(new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test2"));
+    config2.reset(
+        new MockCompressorFilterConfig(compressor, "test2.", stats, runtime, "test2", time_system));
     std::unique_ptr<CompressorFilter> filter2 = std::make_unique<CompressorFilter>(config2);
     NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks;
     filter1->setDecoderFilterCallbacks(decoder_callbacks);

test/extensions/filters/http/gzip/BUILD

@@ -23,6 +23,7 @@ envoy_extension_cc_test(
         "//source/extensions/filters/http/gzip:gzip_filter_lib",
         "//test/mocks/http:http_mocks",
         "//test/mocks/runtime:runtime_mocks",
+        "//test/test_common:simulated_time_system_lib",
         "//test/test_common:utility_lib",
         "@envoy_api//envoy/extensions/filters/http/gzip/v3:pkg_cc_proto",
     ],

test/extensions/filters/http/gzip/gzip_filter_test.cc

@@ -12,6 +12,7 @@
 #include "test/mocks/http/mocks.h"
 #include "test/mocks/runtime/mocks.h"
 #include "test/mocks/stats/mocks.h"
+#include "test/test_common/simulated_time_system.h"
 #include "test/test_common/utility.h"
 
 #include "absl/container/fixed_array.h"
@@ -41,7 +42,7 @@ class GzipFilterTest : public testing::Test {
     Json::ObjectSharedPtr config = Json::Factory::loadFromString(json);
     envoy::extensions::filters::http::gzip::v3::Gzip gzip;
     TestUtility::loadFromJson(json, gzip);
-    config_.reset(new GzipFilterConfig(gzip, "test.", stats_, runtime_));
+    config_.reset(new GzipFilterConfig(gzip, "test.", stats_, runtime_, time_system_));
     filter_ = std::make_unique<Common::Compressors::CompressorFilter>(config_);
     filter_->setEncoderFilterCallbacks(encoder_callbacks_);
     filter_->setDecoderFilterCallbacks(decoder_callbacks_);
@@ -151,13 +152,14 @@ class GzipFilterTest : public testing::Test {
     EXPECT_EQ(1, stats_.counter("test.gzip.not_compressed").value());
   }
 
+  Stats::IsolatedStoreImpl stats_;
+  Event::SimulatedTimeSystem time_system_;
   std::shared_ptr<GzipFilterConfig> config_;
   std::unique_ptr<Common::Compressors::CompressorFilter> filter_;
   Buffer::OwnedImpl data_;
   Decompressor::ZlibDecompressorImpl decompressor_;
   Buffer::OwnedImpl decompressed_data_;
   std::string expected_str_;
-  Stats::IsolatedStoreImpl stats_;
   NiceMock<Runtime::MockLoader> runtime_;
   NiceMock<Http::MockStreamEncoderFilterCallbacks> encoder_callbacks_;
   NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks_;