http: add modifyBuffer filter callback

docs/root/intro/version_history.rst

@@ -26,6 +26,8 @@ Version history
 * outlier_detection: added support for :ref:`outlier detection event protobuf-based logging <arch_overview_outlier_detection_logging>`.
 * mysql: added a MySQL proxy filter that is capable of parsing SQL queries over MySQL wire protocol. Refer to ::ref:`MySQL proxy<config_network_filters_mysql_proxy>` for more details.
 * http: added :ref:`max request headers size <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.max_request_headers_kb>`. The default behaviour is unchanged.
+* http: added modifyDecodingBuffer/modifyEncodingBuffer to allow modifying the buffered request/response data.
+* redis: added :ref:`success and error stats <config_network_filters_redis_proxy_per_command_stats>` for commands.
 * redis: added :ref:`hashtagging <envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.ConnPoolSettings.enable_hashtagging>` to guarantee a given key's upstream.
 * redis: added :ref:`latency stats <config_network_filters_redis_proxy_per_command_stats>` for commands.
 * redis: added :ref:`success and error stats <config_network_filters_redis_proxy_per_command_stats>` for commands.

include/envoy/http/filter.h

@@ -189,6 +189,12 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
    */
   virtual const Buffer::Instance* decodingBuffer() PURE;
 
+  /**
+   * Allows modifying the decoding buffer. May only be called before any data has been continued
+   * past the calling filter.
+   */
+  virtual void modifyDecodingBuffer(std::function<void(Buffer::Instance&)> callback) PURE;
+
   /**
    * Add buffered body data. This method is used in advanced cases where returning
    * StopIterationAndBuffer from decodeData() is not sufficient.
@@ -430,6 +436,12 @@ class StreamEncoderFilterCallbacks : public virtual StreamFilterCallbacks {
    */
   virtual const Buffer::Instance* encodingBuffer() PURE;
 
+  /**
+   * Allows modifying the encoding buffer. May only be called before any data has been continued
+   * past the calling filter.
+   */
+  virtual void modifyEncodingBuffer(std::function<void(Buffer::Instance&)> callback) PURE;
+
   /**
    * Add buffered body data. This method is used in advanced cases where returning
    * StopIterationAndBuffer from encodeData() is not sufficient.

source/common/http/async_client_impl.h

@@ -299,6 +299,9 @@ class AsyncStreamImpl : public AsyncClient::Stream,
     ASSERT(buffered_body_ != nullptr);
   }
   const Buffer::Instance* decodingBuffer() override { return buffered_body_.get(); }
+  void modifyDecodingBuffer(std::function<void(Buffer::Instance&)> callback) override {
+    callback(*buffered_body_.get());
+  }
   void sendLocalReply(Code code, absl::string_view body,
                       std::function<void(HeaderMap& headers)> modify_headers,
                       const absl::optional<Grpc::Status::GrpcStatus> grpc_status) override {
@@ -370,6 +373,9 @@ class AsyncRequestImpl final : public AsyncClient::Request,
     // internal use of the router filter which uses this function for buffering.
   }
   const Buffer::Instance* decodingBuffer() override { return request_->body().get(); }
+  void modifyDecodingBuffer(std::function<void(Buffer::Instance&)> callback) override {
+    callback(*request_->body().get());
+  }
 
   MessagePtr request_;
   AsyncClient::Callbacks& callbacks_;

source/common/http/conn_manager_impl.cc

@@ -37,6 +37,31 @@
 namespace Envoy {
 namespace Http {
 
+namespace {
+
+template <class T> using FilterList = std::list<std::unique_ptr<T>>;
+
+// Shared helper for recording the latest filter used.
+template <class T>
+void recordLatestDataFilter(const typename FilterList<T>::iterator current_filter,
+                            T** latest_filter, const FilterList<T>& filters) {
+  // If this is the first time we're calling onData, just record the current filter.
+  if (*latest_filter == nullptr) {
+    *latest_filter = current_filter->get();
+    return;
+  }
+
+  // We want to keep this pointing at the latest filter in the filter list that has received the
+  // onData callback. To do so, we compare the current latest with the *previous* filter. If they
+  // match, then we must be processing a new filter for the first time. We omit this check if we're
+  // the first filter, since the above check handles that case.
+  if (current_filter != filters.begin() && *latest_filter == std::prev(current_filter)->get()) {
+    *latest_filter = current_filter->get();
+  }
+}
+
+} // namespace
+
 ConnectionManagerStats ConnectionManagerImpl::generateStats(const std::string& prefix,
                                                             Stats::Scope& scope) {
   return {
@@ -874,6 +899,9 @@ void ConnectionManagerImpl::ActiveStream::decodeData(ActiveStreamDecoderFilter*
     if (end_stream) {
       state_.filter_call_state_ |= FilterCallState::LastDataFrame;
     }
+
+    recordLatestDataFilter(entry, &state_.latest_data_decoding_filter_, decoder_filters_);
+
     state_.filter_call_state_ |= FilterCallState::DecodeData;
     (*entry)->end_stream_ = end_stream && !request_trailers_;
     FilterDataStatus status = (*entry)->handle_->decodeData(data, (*entry)->end_stream_);
@@ -1325,6 +1353,9 @@ void ConnectionManagerImpl::ActiveStream::encodeData(ActiveStreamEncoderFilter*
     if (end_stream) {
       state_.filter_call_state_ |= FilterCallState::LastDataFrame;
     }
+
+    recordLatestDataFilter(entry, &state_.latest_data_encoding_filter_, encoder_filters_);
+
     (*entry)->end_stream_ = end_stream && !response_trailers_;
     FilterDataStatus status = (*entry)->handle_->encodeData(data, (*entry)->end_stream_);
     state_.filter_call_state_ &= ~FilterCallState::EncodeData;

source/common/http/conn_manager_impl.h

@@ -176,6 +176,12 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     const Buffer::Instance* decodingBuffer() override {
       return parent_.buffered_request_data_.get();
     }
+
+    void modifyDecodingBuffer(std::function<void(Buffer::Instance&)> callback) override {
+      ASSERT(parent_.state_.latest_data_decoding_filter_ == this);
+      callback(*parent_.buffered_request_data_.get());
+    }
+
     void sendLocalReply(Code code, absl::string_view body,
                         std::function<void(HeaderMap& headers)> modify_headers,
                         const absl::optional<Grpc::Status::GrpcStatus> grpc_status) override {
@@ -252,6 +258,10 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     const Buffer::Instance* encodingBuffer() override {
       return parent_.buffered_response_data_.get();
     }
+    void modifyEncodingBuffer(std::function<void(Buffer::Instance&)> callback) override {
+      ASSERT(parent_.state_.latest_data_encoding_filter_ == this);
+      callback(*parent_.buffered_response_data_.get());
+    }
 
     void responseDataTooLarge();
     void responseDataDrained();
@@ -384,6 +394,10 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
       // True if this stream is internally created. Currently only used for
       // internal redirects or other streams created via recreateStream().
       bool is_internally_created_ : 1;
+
+      // Used to track which filter is the latest filter that has received data.
+      ActiveStreamEncoderFilter* latest_data_encoding_filter_{};
+      ActiveStreamDecoderFilter* latest_data_decoding_filter_{};
     };
 
     // Possibly increases buffer_limit_ to the value of limit.

test/integration/BUILD

@@ -220,6 +220,7 @@ envoy_cc_test_library(
         "//test/common/upstream:utility_lib",
         "//test/integration/filters:add_trailers_filter_config_lib",
         "//test/integration/filters:headers_only_filter_config_lib",
+        "//test/integration/filters:modify_buffer_filter_config_lib",
         "//test/integration/filters:passthrough_filter_config_lib",
         "//test/integration/filters:pause_filter_lib",
         "//test/integration/filters:response_metadata_filter_config_lib",

test/integration/filters/BUILD

@@ -22,6 +22,20 @@ envoy_cc_test_library(
     ],
 )
 
+envoy_cc_test_library(
+    name = "modify_buffer_filter_config_lib",
+    srcs = [
+        "modify_buffer_filter.cc",
+    ],
+    deps = [
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/registry",
+        "//include/envoy/server:filter_config_interface",
+        "//source/extensions/filters/http/common:empty_http_filter_config_lib",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+    ],
+)
+
 envoy_cc_test_library(
     name = "passthrough_filter_config_lib",
     srcs = [

test/integration/filters/modify_buffer_filter.cc

@@ -0,0 +1,57 @@
+#include <string>
+
+#include "envoy/http/filter.h"
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+
+#include "extensions/filters/http/common/empty_http_filter_config.h"
+#include "extensions/filters/http/common/pass_through_filter.h"
+
+namespace Envoy {
+
+// A filter that buffers the entire request/response, then doubles
+// the content of the filter buffer.
+class ModifyBufferStreamFilter : public Http::PassThroughFilter {
+public:
+  Http::FilterDataStatus decodeData(Buffer::Instance&, bool end_stream) {
+    if (end_stream) {
+      decoder_callbacks_->modifyDecodingBuffer([](auto& buffer) {
+        // Append the buffer with itself.
+        buffer.add(buffer);
+      });
+      return Http::FilterDataStatus::Continue;
+    }
+
+    return Http::FilterDataStatus::StopIterationAndBuffer;
+  }
+
+  Http::FilterDataStatus encodeData(Buffer::Instance&, bool end_stream) {
+    if (end_stream) {
+      encoder_callbacks_->modifyEncodingBuffer([](auto& buffer) {
+        // Append the buffer with itself.
+        buffer.add(buffer);
+      });
+      return Http::FilterDataStatus::Continue;
+    }
+
+    return Http::FilterDataStatus::StopIterationAndBuffer;
+  }
+};
+
+class ModifyBuffferFilterConfig : public Extensions::HttpFilters::Common::EmptyHttpFilterConfig {
+public:
+  ModifyBuffferFilterConfig() : EmptyHttpFilterConfig("modify-buffer-filter") {}
+
+  Http::FilterFactoryCb createFilter(const std::string&, Server::Configuration::FactoryContext&) {
+    return [](Http::FilterChainFactoryCallbacks& callbacks) -> void {
+      callbacks.addStreamFilter(std::make_shared<::Envoy::ModifyBufferStreamFilter>());
+    };
+  }
+};
+
+// perform static registration
+static Registry::RegisterFactory<ModifyBuffferFilterConfig,
+                                 Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+
+} // namespace Envoy

test/integration/http2_integration_test.cc

@@ -41,6 +41,8 @@ TEST_P(Http2IntegrationTest, ComputedHealthCheck) { testComputedHealthCheck(); }
 
 TEST_P(Http2IntegrationTest, AddEncodedTrailers) { testAddEncodedTrailers(); }
 
+TEST_P(Http2IntegrationTest, ModifyBuffers) { testModifyBuffer(); }
+
 TEST_P(Http2IntegrationTest, DrainClose) { testDrainClose(); }
 
 TEST_P(Http2IntegrationTest, RouterRequestAndResponseWithBodyNoBuffer) {

test/integration/http2_upstream_integration_test.cc

@@ -21,6 +21,8 @@ TEST_P(Http2UpstreamIntegrationTest, RouterRedirect) { testRouterRedirect(); }
 
 TEST_P(Http2UpstreamIntegrationTest, ComputedHealthCheck) { testComputedHealthCheck(); }
 
+TEST_P(Http2UpstreamIntegrationTest, ModifyBuffer) { testModifyBuffer(); }
+
 TEST_P(Http2UpstreamIntegrationTest, AddEncodedTrailers) { testAddEncodedTrailers(); }
 
 TEST_P(Http2UpstreamIntegrationTest, DrainClose) { testDrainClose(); }

test/integration/http_integration.cc

@@ -467,6 +467,43 @@ config: {}
   }
 }
 
+void HttpIntegrationTest::testModifyBuffer() {
+  config_helper_.addFilter(R"EOF(
+name: modify-buffer-filter
+)EOF");
+  initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto encoder_decoder = codec_client_->startRequest(default_request_headers_);
+  request_encoder_ = &encoder_decoder.first;
+  auto response = std::move(encoder_decoder.second);
+  codec_client_->sendData(*request_encoder_, 64, false);
+  codec_client_->sendData(*request_encoder_, 64, true);
+
+  waitForNextUpstreamRequest();
+  upstream_request_->encodeHeaders(Http::TestHeaderMapImpl{{":status", "503"}}, false);
+  upstream_request_->encodeData(64, false);
+  upstream_request_->encodeData(64, true);
+  response->waitForEndStream();
+
+  // When using HTTP2 we get end_stream=true with the second onData callback, so the buffered
+  // data is only 64 bytes. With HTTP1 we get end_stream=true on a third onData callback,
+  // so the buffered data is 128 bytes. Since we double the buffered data when we get
+  // end_stream=true, the resulting data therefore vary based on whether we're using HTTP1 or HTTP2.
+  if (downstream_protocol_ == Http::CodecClient::Type::HTTP2) {
+    EXPECT_EQ(64 + 2 * 64, upstream_request_->body().length());
+  } else {
+    EXPECT_EQ(256, upstream_request_->body().length());
+  }
+  EXPECT_TRUE(response->complete());
+  EXPECT_STREQ("503", response->headers().Status()->value().c_str());
+  if (fake_upstreams_[0]->httpType() == FakeHttpConnection::Type::HTTP2) {
+    EXPECT_EQ(64 + 2 * 64, response->body().length());
+  } else {
+    EXPECT_EQ(256, response->body().length());
+  }
+}
+
 // Add a health check filter and verify correct behavior when draining.
 void HttpIntegrationTest::testDrainClose() {
   config_helper_.addFilter(ConfigHelper::DEFAULT_HEALTH_CHECK_FILTER);

test/integration/http_integration.h

@@ -172,6 +172,7 @@ class HttpIntegrationTest : public BaseIntegrationTest {
   void testMultipleContentLengths();
   void testComputedHealthCheck();
   void testAddEncodedTrailers();
+  void testModifyBuffer();
   void testDrainClose();
   void testRetry();
   void testRetryHittingBufferLimit();

test/integration/integration_test.cc

@@ -107,6 +107,8 @@ TEST_P(IntegrationTest, ComputedHealthCheck) { testComputedHealthCheck(); }
 
 TEST_P(IntegrationTest, AddEncodedTrailers) { testAddEncodedTrailers(); }
 
+TEST_P(IntegrationTest, ModifyBuffer) { testModifyBuffer(); }
+
 TEST_P(IntegrationTest, DrainClose) { testDrainClose(); }
 
 TEST_P(IntegrationTest, ConnectionClose) {

test/mocks/http/mocks.h

@@ -179,6 +179,7 @@ class MockStreamDecoderFilterCallbacks : public StreamDecoderFilterCallbacks,
   MOCK_METHOD2(addDecodedData, void(Buffer::Instance& data, bool streaming));
   MOCK_METHOD0(addDecodedTrailers, HeaderMap&());
   MOCK_METHOD0(decodingBuffer, const Buffer::Instance*());
+  MOCK_METHOD1(modifyDecodingBuffer, void(std::function<void(Buffer::Instance&)>));
   MOCK_METHOD1(encode100ContinueHeaders_, void(HeaderMap& headers));
   MOCK_METHOD2(encodeHeaders_, void(HeaderMap& headers, bool end_stream));
   MOCK_METHOD2(encodeData, void(Buffer::Instance& data, bool end_stream));
@@ -221,6 +222,7 @@ class MockStreamEncoderFilterCallbacks : public StreamEncoderFilterCallbacks,
   MOCK_METHOD0(addEncodedTrailers, HeaderMap&());
   MOCK_METHOD0(continueEncoding, void());
   MOCK_METHOD0(encodingBuffer, const Buffer::Instance*());
+  MOCK_METHOD1(modifyEncodingBuffer, void(std::function<void(Buffer::Instance&)>));
 
   Buffer::InstancePtr buffer_;
   testing::NiceMock<Tracing::MockSpan> active_span_;