From f8538bf66727b2c92b5ecf098465d3d98b7cf550 Mon Sep 17 00:00:00 2001 From: wbpcode Date: Sat, 20 Feb 2021 19:48:12 +0800 Subject: [PATCH 1/7] thrift: supported setting max requests for per downstream connection Signed-off-by: wbpcode --- .../thrift_proxy/v3/thrift_proxy.proto | 6 ++- .../thrift_proxy/v4alpha/thrift_proxy.proto | 6 ++- docs/root/version_history/current.rst | 1 + .../thrift_proxy/v3/thrift_proxy.proto | 6 ++- .../thrift_proxy/v4alpha/thrift_proxy.proto | 6 ++- .../filters/network/thrift_proxy/config.cc | 5 ++ .../filters/network/thrift_proxy/config.h | 4 ++ .../network/thrift_proxy/conn_manager.cc | 15 ++++++ .../network/thrift_proxy/conn_manager.h | 6 +++ .../network/thrift_proxy/conn_manager_test.cc | 47 +++++++++++++++++++ 10 files changed, 98 insertions(+), 4 deletions(-) diff --git a/api/envoy/extensions/filters/network/thrift_proxy/v3/thrift_proxy.proto b/api/envoy/extensions/filters/network/thrift_proxy/v3/thrift_proxy.proto index 659ee14604401..512f810e41398 100644 --- a/api/envoy/extensions/filters/network/thrift_proxy/v3/thrift_proxy.proto +++ b/api/envoy/extensions/filters/network/thrift_proxy/v3/thrift_proxy.proto @@ -5,6 +5,7 @@ package envoy.extensions.filters.network.thrift_proxy.v3; import "envoy/extensions/filters/network/thrift_proxy/v3/route.proto"; import "google/protobuf/any.proto"; +import "google/protobuf/wrappers.proto"; import "udpa/annotations/status.proto"; import "udpa/annotations/versioning.proto"; @@ -57,7 +58,7 @@ enum ProtocolType { TWITTER = 4; } -// [#next-free-field: 7] +// [#next-free-field: 8] message ThriftProxy { option (udpa.annotations.versioning).previous_message_type = "envoy.config.filter.network.thrift_proxy.v2alpha1.ThriftProxy"; @@ -88,6 +89,9 @@ message ThriftProxy { // is the same, the transport type is framed and the protocol is not Twitter. Otherwise Envoy will // fallback to decode the data. bool payload_passthrough = 6; + + // Optional maximum requests for a single downstream connection. If not specified, there is no limit. + google.protobuf.UInt32Value max_requests_per_connection = 7; } // ThriftFilter configures a Thrift filter. diff --git a/api/envoy/extensions/filters/network/thrift_proxy/v4alpha/thrift_proxy.proto b/api/envoy/extensions/filters/network/thrift_proxy/v4alpha/thrift_proxy.proto index 1834202272362..4ae4e26e4dbfe 100644 --- a/api/envoy/extensions/filters/network/thrift_proxy/v4alpha/thrift_proxy.proto +++ b/api/envoy/extensions/filters/network/thrift_proxy/v4alpha/thrift_proxy.proto @@ -5,6 +5,7 @@ package envoy.extensions.filters.network.thrift_proxy.v4alpha; import "envoy/extensions/filters/network/thrift_proxy/v4alpha/route.proto"; import "google/protobuf/any.proto"; +import "google/protobuf/wrappers.proto"; import "udpa/annotations/status.proto"; import "udpa/annotations/versioning.proto"; @@ -57,7 +58,7 @@ enum ProtocolType { TWITTER = 4; } -// [#next-free-field: 7] +// [#next-free-field: 8] message ThriftProxy { option (udpa.annotations.versioning).previous_message_type = "envoy.extensions.filters.network.thrift_proxy.v3.ThriftProxy"; @@ -88,6 +89,9 @@ message ThriftProxy { // is the same, the transport type is framed and the protocol is not Twitter. Otherwise Envoy will // fallback to decode the data. bool payload_passthrough = 6; + + // Optional maximum requests for a single downstream connection. If not specified, there is no limit. + google.protobuf.UInt32Value max_requests_per_connection = 7; } // ThriftFilter configures a Thrift filter. diff --git a/docs/root/version_history/current.rst b/docs/root/version_history/current.rst index 8bf947e256f05..62f65abb1150c 100644 --- a/docs/root/version_history/current.rst +++ b/docs/root/version_history/current.rst @@ -101,6 +101,7 @@ New Features * tcp_proxy: add support for converting raw TCP streams into HTTP/1.1 CONNECT requests. See :ref:`upgrade documentation ` for details. * tcp_proxy: added a :ref:`use_post field ` for using HTTP POST to proxy TCP streams. * tcp_proxy: added a :ref:`headers_to_add field ` for setting additional headers to the HTTP requests for TCP proxing. +* thrift_proxy: added a :ref:`max_requests_per_connection field ` for setting maximum requests for per downstream connection. Deprecated ---------- diff --git a/generated_api_shadow/envoy/extensions/filters/network/thrift_proxy/v3/thrift_proxy.proto b/generated_api_shadow/envoy/extensions/filters/network/thrift_proxy/v3/thrift_proxy.proto index 8485693caaf9e..3083973f3da20 100644 --- a/generated_api_shadow/envoy/extensions/filters/network/thrift_proxy/v3/thrift_proxy.proto +++ b/generated_api_shadow/envoy/extensions/filters/network/thrift_proxy/v3/thrift_proxy.proto @@ -6,6 +6,7 @@ import "envoy/extensions/filters/network/thrift_proxy/v3/route.proto"; import "google/protobuf/any.proto"; import "google/protobuf/struct.proto"; +import "google/protobuf/wrappers.proto"; import "udpa/annotations/status.proto"; import "udpa/annotations/versioning.proto"; @@ -58,7 +59,7 @@ enum ProtocolType { TWITTER = 4; } -// [#next-free-field: 7] +// [#next-free-field: 8] message ThriftProxy { option (udpa.annotations.versioning).previous_message_type = "envoy.config.filter.network.thrift_proxy.v2alpha1.ThriftProxy"; @@ -89,6 +90,9 @@ message ThriftProxy { // is the same, the transport type is framed and the protocol is not Twitter. Otherwise Envoy will // fallback to decode the data. bool payload_passthrough = 6; + + // Optional maximum requests for a single downstream connection. If not specified, there is no limit. + google.protobuf.UInt32Value max_requests_per_connection = 7; } // ThriftFilter configures a Thrift filter. diff --git a/generated_api_shadow/envoy/extensions/filters/network/thrift_proxy/v4alpha/thrift_proxy.proto b/generated_api_shadow/envoy/extensions/filters/network/thrift_proxy/v4alpha/thrift_proxy.proto index 1834202272362..4ae4e26e4dbfe 100644 --- a/generated_api_shadow/envoy/extensions/filters/network/thrift_proxy/v4alpha/thrift_proxy.proto +++ b/generated_api_shadow/envoy/extensions/filters/network/thrift_proxy/v4alpha/thrift_proxy.proto @@ -5,6 +5,7 @@ package envoy.extensions.filters.network.thrift_proxy.v4alpha; import "envoy/extensions/filters/network/thrift_proxy/v4alpha/route.proto"; import "google/protobuf/any.proto"; +import "google/protobuf/wrappers.proto"; import "udpa/annotations/status.proto"; import "udpa/annotations/versioning.proto"; @@ -57,7 +58,7 @@ enum ProtocolType { TWITTER = 4; } -// [#next-free-field: 7] +// [#next-free-field: 8] message ThriftProxy { option (udpa.annotations.versioning).previous_message_type = "envoy.extensions.filters.network.thrift_proxy.v3.ThriftProxy"; @@ -88,6 +89,9 @@ message ThriftProxy { // is the same, the transport type is framed and the protocol is not Twitter. Otherwise Envoy will // fallback to decode the data. bool payload_passthrough = 6; + + // Optional maximum requests for a single downstream connection. If not specified, there is no limit. + google.protobuf.UInt32Value max_requests_per_connection = 7; } // ThriftFilter configures a Thrift filter. diff --git a/source/extensions/filters/network/thrift_proxy/config.cc b/source/extensions/filters/network/thrift_proxy/config.cc index 8c7a967597c48..4f570ac9f3779 100644 --- a/source/extensions/filters/network/thrift_proxy/config.cc +++ b/source/extensions/filters/network/thrift_proxy/config.cc @@ -135,6 +135,11 @@ ConfigImpl::ConfigImpl( processFilter(filter); } } + + if (config.has_max_requests_per_connection() && + config.max_requests_per_connection().value() > 0) { + max_requests_per_connection_ = config.max_requests_per_connection().value(); + } } void ConfigImpl::createFilterChain(ThriftFilters::FilterChainFactoryCallbacks& callbacks) { diff --git a/source/extensions/filters/network/thrift_proxy/config.h b/source/extensions/filters/network/thrift_proxy/config.h index 02c1fcf4d13a4..8e6c7a5098cc3 100644 --- a/source/extensions/filters/network/thrift_proxy/config.h +++ b/source/extensions/filters/network/thrift_proxy/config.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include @@ -82,6 +83,7 @@ class ConfigImpl : public Config, ProtocolPtr createProtocol() override; Router::Config& routerConfig() override { return *this; } bool payloadPassthrough() const override { return payload_passthrough_; } + uint64_t maxRequestsPerConnection() const override { return max_requests_per_connection_; } private: void processFilter( @@ -96,6 +98,8 @@ class ConfigImpl : public Config, std::list filter_factories_; const bool payload_passthrough_; + + uint64_t max_requests_per_connection_ = UINT64_MAX; }; } // namespace ThriftProxy diff --git a/source/extensions/filters/network/thrift_proxy/conn_manager.cc b/source/extensions/filters/network/thrift_proxy/conn_manager.cc index d3e23b5d29ccf..1adda8d9fa1dd 100644 --- a/source/extensions/filters/network/thrift_proxy/conn_manager.cc +++ b/source/extensions/filters/network/thrift_proxy/conn_manager.cc @@ -55,6 +55,11 @@ void ConnectionManager::dispatch() { return; } + if (requests_overflow_) { + ENVOY_CONN_LOG(debug, "thrift filter requests overflow", read_callbacks_->connection()); + return; + } + try { bool underflow = false; while (!underflow) { @@ -139,6 +144,9 @@ void ConnectionManager::continueDecoding() { void ConnectionManager::doDeferredRpcDestroy(ConnectionManager::ActiveRpc& rpc) { read_callbacks_->connection().dispatcher().deferredDelete(rpc.removeFromList(rpcs_)); + if (requests_overflow_ && rpcs_.empty()) { + read_callbacks_->connection().close(Network::ConnectionCloseType::FlushWrite); + } } void ConnectionManager::resetAllRpcs(bool local_reset) { @@ -404,6 +412,13 @@ void ConnectionManager::ActiveRpc::finalizeRequest() { parent_.stats_.request_.inc(); + parent_.accumulated_requests_++; + + if (parent_.accumulated_requests_ >= parent_.config_.maxRequestsPerConnection()) { + parent_.read_callbacks_->connection().readDisable(true); + parent_.requests_overflow_ = true; + } + bool destroy_rpc = false; switch (original_msg_type_) { case MessageType::Call: diff --git a/source/extensions/filters/network/thrift_proxy/conn_manager.h b/source/extensions/filters/network/thrift_proxy/conn_manager.h index 12bc08347150d..9f3dd59468c44 100644 --- a/source/extensions/filters/network/thrift_proxy/conn_manager.h +++ b/source/extensions/filters/network/thrift_proxy/conn_manager.h @@ -40,6 +40,7 @@ class Config { virtual ProtocolPtr createProtocol() PURE; virtual Router::Config& routerConfig() PURE; virtual bool payloadPassthrough() const PURE; + virtual uint64_t maxRequestsPerConnection() const PURE; }; /** @@ -272,6 +273,11 @@ class ConnectionManager : public Network::ReadFilter, bool stopped_{false}; bool half_closed_{false}; TimeSource& time_source_; + + // The number of requests accumulated on the current connection. A connection is processed by only + // one thread, so there is no need to consider the thread safety of the count. + uint32_t accumulated_requests_{0}; + bool requests_overflow_{false}; }; } // namespace ThriftProxy diff --git a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc index 3764bbd44573f..ea4ed8536802d 100644 --- a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc +++ b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc @@ -1058,6 +1058,53 @@ TEST_F(ThriftConnectionManagerTest, ResetDownstreamConnection) { EXPECT_EQ(0U, stats_.request_active_.value()); } +TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsPerConnection) { + const std::string yaml = R"EOF( +stat_prefix: test +route_config: + name: local_route +max_requests_per_connection: 1 +)EOF"; + + initializeFilter(yaml); + EXPECT_EQ(1, config_->maxRequestsPerConnection()); + + writeComplexFramedBinaryMessage(buffer_, MessageType::Call, 0x0F); + + ThriftFilters::DecoderFilterCallbacks* callbacks{}; + EXPECT_CALL(*decoder_filter_, setDecoderFilterCallbacks(_)) + .WillOnce( + Invoke([&](ThriftFilters::DecoderFilterCallbacks& cb) -> void { callbacks = &cb; })); + + EXPECT_EQ(filter_->onData(buffer_, false), Network::FilterStatus::StopIteration); + EXPECT_EQ(1U, store_.counter("test.request_call").value()); + + writeComplexFramedBinaryMessage(write_buffer_, MessageType::Reply, 0x0F); + + FramedTransportImpl transport; + BinaryProtocolImpl proto; + callbacks->startUpstreamResponse(transport, proto); + + EXPECT_CALL(filter_callbacks_.connection_.dispatcher_, deferredDelete_(_)); + // Since max requests per connection is set to 1, the connection will be disconnected after the + // first request is completed. + EXPECT_CALL(filter_callbacks_.connection_, close(Network::ConnectionCloseType::FlushWrite)); + + EXPECT_EQ(ThriftFilters::ResponseStatus::Complete, callbacks->upstreamData(write_buffer_)); + + filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); + + EXPECT_EQ(1U, store_.counter("test.request").value()); + EXPECT_EQ(1U, store_.counter("test.request_call").value()); + EXPECT_EQ(0U, stats_.request_active_.value()); + EXPECT_EQ(1U, store_.counter("test.response").value()); + EXPECT_EQ(1U, store_.counter("test.response_reply").value()); + EXPECT_EQ(0U, store_.counter("test.response_exception").value()); + EXPECT_EQ(0U, store_.counter("test.response_invalid_type").value()); + EXPECT_EQ(1U, store_.counter("test.response_success").value()); + EXPECT_EQ(0U, store_.counter("test.response_error").value()); +} + TEST_F(ThriftConnectionManagerTest, DownstreamProtocolUpgrade) { custom_transport_ = new NiceMock(); custom_protocol_ = new NiceMock(); From cb664b0581affc0ea2c905caa314eb2bb0b75289 Mon Sep 17 00:00:00 2001 From: wbpcode Date: Tue, 23 Feb 2021 20:35:13 +0800 Subject: [PATCH 2/7] update some docs and some test Signed-off-by: wbpcode --- .../network_filters/thrift_proxy_filter.rst | 8 + .../filters/network/thrift_proxy/config.h | 2 +- .../network/thrift_proxy/conn_manager.cc | 17 +- .../network/thrift_proxy/conn_manager.h | 5 +- .../filters/network/thrift_proxy/stats.h | 1 + .../network/thrift_proxy/conn_manager_test.cc | 206 +++++++++++++++--- 6 files changed, 203 insertions(+), 36 deletions(-) diff --git a/docs/root/configuration/listeners/network_filters/thrift_proxy_filter.rst b/docs/root/configuration/listeners/network_filters/thrift_proxy_filter.rst index 504f6f8737529..480f12d69dd86 100644 --- a/docs/root/configuration/listeners/network_filters/thrift_proxy_filter.rst +++ b/docs/root/configuration/listeners/network_filters/thrift_proxy_filter.rst @@ -15,6 +15,14 @@ keyed by `envoy.filters.network.thrift_proxy`. The :ref:`ThriftProtocolOptions` message describes the available options. +Downstream Requests Limit +------------------------- +Thrift Proxy can set the +:ref:`maximum number of requests` +that each downstream connection can handle. When the number of requests exceeds the connection limit, Thrift Proxy will +actively disconnect from the Thrfit client. + + Thrift Request Metadata ----------------------- diff --git a/source/extensions/filters/network/thrift_proxy/config.h b/source/extensions/filters/network/thrift_proxy/config.h index 8e6c7a5098cc3..637be14f79f39 100644 --- a/source/extensions/filters/network/thrift_proxy/config.h +++ b/source/extensions/filters/network/thrift_proxy/config.h @@ -99,7 +99,7 @@ class ConfigImpl : public Config, std::list filter_factories_; const bool payload_passthrough_; - uint64_t max_requests_per_connection_ = UINT64_MAX; + uint64_t max_requests_per_connection_{}; }; } // namespace ThriftProxy diff --git a/source/extensions/filters/network/thrift_proxy/conn_manager.cc b/source/extensions/filters/network/thrift_proxy/conn_manager.cc index 1adda8d9fa1dd..2da10f4bb7d2a 100644 --- a/source/extensions/filters/network/thrift_proxy/conn_manager.cc +++ b/source/extensions/filters/network/thrift_proxy/conn_manager.cc @@ -12,12 +12,21 @@ namespace Extensions { namespace NetworkFilters { namespace ThriftProxy { +namespace { +// Translate zero to UINT64_MAX so that the zero/unlimited case doesn't +// have to be handled specially. +uint64_t translateZeroToUnlimited(uint64_t limit) { + return (limit != 0) ? limit : std::numeric_limits::max(); +} +} // namespace + ConnectionManager::ConnectionManager(Config& config, Random::RandomGenerator& random_generator, TimeSource& time_source) : config_(config), stats_(config_.stats()), transport_(config.createTransport()), protocol_(config.createProtocol()), decoder_(std::make_unique(*transport_, *protocol_, *this)), - random_generator_(random_generator), time_source_(time_source) {} + random_generator_(random_generator), time_source_(time_source), + remaining_streams_(translateZeroToUnlimited(config.maxRequestsPerConnection())) {} ConnectionManager::~ConnectionManager() = default; @@ -412,11 +421,11 @@ void ConnectionManager::ActiveRpc::finalizeRequest() { parent_.stats_.request_.inc(); - parent_.accumulated_requests_++; - - if (parent_.accumulated_requests_ >= parent_.config_.maxRequestsPerConnection()) { + parent_.remaining_streams_--; + if (parent_.remaining_streams_ == 0) { parent_.read_callbacks_->connection().readDisable(true); parent_.requests_overflow_ = true; + parent_.stats_.downstream_cx_max_requests_.inc(); } bool destroy_rpc = false; diff --git a/source/extensions/filters/network/thrift_proxy/conn_manager.h b/source/extensions/filters/network/thrift_proxy/conn_manager.h index 9f3dd59468c44..03898801cf7d2 100644 --- a/source/extensions/filters/network/thrift_proxy/conn_manager.h +++ b/source/extensions/filters/network/thrift_proxy/conn_manager.h @@ -274,9 +274,8 @@ class ConnectionManager : public Network::ReadFilter, bool half_closed_{false}; TimeSource& time_source_; - // The number of requests accumulated on the current connection. A connection is processed by only - // one thread, so there is no need to consider the thread safety of the count. - uint32_t accumulated_requests_{0}; + // The maximum number of requests remaining to be processed on the current connection. + uint64_t remaining_streams_{}; bool requests_overflow_{false}; }; diff --git a/source/extensions/filters/network/thrift_proxy/stats.h b/source/extensions/filters/network/thrift_proxy/stats.h index 9166f37be6ca3..7e57db76f803b 100644 --- a/source/extensions/filters/network/thrift_proxy/stats.h +++ b/source/extensions/filters/network/thrift_proxy/stats.h @@ -16,6 +16,7 @@ namespace ThriftProxy { #define ALL_THRIFT_FILTER_STATS(COUNTER, GAUGE, HISTOGRAM) \ COUNTER(cx_destroy_local_with_active_rq) \ COUNTER(cx_destroy_remote_with_active_rq) \ + COUNTER(downstream_cx_max_requests) \ COUNTER(request) \ COUNTER(request_call) \ COUNTER(request_decoding_error) \ diff --git a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc index ea4ed8536802d..b5ed33f0dbb3a 100644 --- a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc +++ b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc @@ -120,6 +120,33 @@ class ThriftConnectionManagerTest : public testing::Test { filter_->onBelowWriteBufferLowWatermark(); } + // Return the number of requests actually sent. + uint32_t sendSomeThriftRequest(uint32_t request_number) { + for (uint32_t i = 0; i < request_number; i++) { + writeComplexFramedBinaryMessage(buffer_, MessageType::Call, 0x0F); + writeComplexFramedBinaryMessage(write_buffer_, MessageType::Reply, 0x0F); + + ThriftFilters::DecoderFilterCallbacks* callbacks{}; + ON_CALL(*decoder_filter_, setDecoderFilterCallbacks(_)) + .WillByDefault( + Invoke([&](ThriftFilters::DecoderFilterCallbacks& cb) -> void { callbacks = &cb; })); + + EXPECT_EQ(filter_->onData(buffer_, false), Network::FilterStatus::StopIteration); + + if (!callbacks) { + return i; + } + + FramedTransportImpl transport; + BinaryProtocolImpl proto; + callbacks->startUpstreamResponse(transport, proto); + + EXPECT_CALL(filter_callbacks_.connection_.dispatcher_, deferredDelete_(_)); + EXPECT_EQ(ThriftFilters::ResponseStatus::Complete, callbacks->upstreamData(write_buffer_)); + } + return request_number; + } + void writeMessage(Buffer::Instance& buffer, TransportType transport_type, ProtocolType protocol_type, MessageType msg_type, int32_t seq_id) { Buffer::OwnedImpl msg; @@ -1058,50 +1085,172 @@ TEST_F(ThriftConnectionManagerTest, ResetDownstreamConnection) { EXPECT_EQ(0U, stats_.request_active_.value()); } -TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsPerConnection) { +// Test the base case where there is no limit on the number of requests. +TEST_F(ThriftConnectionManagerTest, RequestWithNoMaxRequestsLimit) { + initializeFilter(""); + EXPECT_EQ(0, config_->maxRequestsPerConnection()); + + EXPECT_EQ(50, sendSomeThriftRequest(50)); + + filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); + + EXPECT_EQ(0U, store_.counter("test.downstream_cx_max_requests").value()); + EXPECT_EQ(50U, store_.counter("test.request").value()); + EXPECT_EQ(50U, store_.counter("test.request_call").value()); + EXPECT_EQ(0U, stats_.request_active_.value()); + EXPECT_EQ(50U, store_.counter("test.response").value()); + EXPECT_EQ(50U, store_.counter("test.response_reply").value()); + EXPECT_EQ(0U, store_.counter("test.response_exception").value()); + EXPECT_EQ(0U, store_.counter("test.response_invalid_type").value()); + EXPECT_EQ(50U, store_.counter("test.response_success").value()); + EXPECT_EQ(0U, store_.counter("test.response_error").value()); +} + +// Test the case where there is a limit on the number of requests but the actual number of requests +// does not reach the limit. +TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitButNotReach) { const std::string yaml = R"EOF( -stat_prefix: test -route_config: - name: local_route -max_requests_per_connection: 1 -)EOF"; + stat_prefix: test + route_config: + name: local_route + max_requests_per_connection: 50 + )EOF"; initializeFilter(yaml); - EXPECT_EQ(1, config_->maxRequestsPerConnection()); + EXPECT_EQ(50, config_->maxRequestsPerConnection()); - writeComplexFramedBinaryMessage(buffer_, MessageType::Call, 0x0F); + EXPECT_EQ(49, sendSomeThriftRequest(49)); - ThriftFilters::DecoderFilterCallbacks* callbacks{}; - EXPECT_CALL(*decoder_filter_, setDecoderFilterCallbacks(_)) - .WillOnce( - Invoke([&](ThriftFilters::DecoderFilterCallbacks& cb) -> void { callbacks = &cb; })); + filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); - EXPECT_EQ(filter_->onData(buffer_, false), Network::FilterStatus::StopIteration); - EXPECT_EQ(1U, store_.counter("test.request_call").value()); + EXPECT_EQ(0U, store_.counter("test.downstream_cx_max_requests").value()); + EXPECT_EQ(49U, store_.counter("test.request").value()); + EXPECT_EQ(49U, store_.counter("test.request_call").value()); + EXPECT_EQ(0U, stats_.request_active_.value()); + EXPECT_EQ(49U, store_.counter("test.response").value()); + EXPECT_EQ(49U, store_.counter("test.response_reply").value()); + EXPECT_EQ(0U, store_.counter("test.response_exception").value()); + EXPECT_EQ(0U, store_.counter("test.response_invalid_type").value()); + EXPECT_EQ(49U, store_.counter("test.response_success").value()); + EXPECT_EQ(0U, store_.counter("test.response_error").value()); +} - writeComplexFramedBinaryMessage(write_buffer_, MessageType::Reply, 0x0F); +// Test the case where there is a limit on the number of requests and the actual number of requests +// happens to reach the limit. +TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitAndReached) { + const std::string yaml = R"EOF( + stat_prefix: test + route_config: + name: local_route + max_requests_per_connection: 50 + )EOF"; - FramedTransportImpl transport; - BinaryProtocolImpl proto; - callbacks->startUpstreamResponse(transport, proto); + initializeFilter(yaml); + EXPECT_EQ(50, config_->maxRequestsPerConnection()); - EXPECT_CALL(filter_callbacks_.connection_.dispatcher_, deferredDelete_(_)); - // Since max requests per connection is set to 1, the connection will be disconnected after the - // first request is completed. + // Since max requests per connection is set to 50, the connection will be disconnected after + // all 50 requests is completed. EXPECT_CALL(filter_callbacks_.connection_, close(Network::ConnectionCloseType::FlushWrite)); - EXPECT_EQ(ThriftFilters::ResponseStatus::Complete, callbacks->upstreamData(write_buffer_)); + EXPECT_EQ(50, sendSomeThriftRequest(50)); filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); - EXPECT_EQ(1U, store_.counter("test.request").value()); - EXPECT_EQ(1U, store_.counter("test.request_call").value()); + EXPECT_EQ(1U, store_.counter("test.downstream_cx_max_requests").value()); + EXPECT_EQ(50U, store_.counter("test.request").value()); + EXPECT_EQ(50U, store_.counter("test.request_call").value()); EXPECT_EQ(0U, stats_.request_active_.value()); - EXPECT_EQ(1U, store_.counter("test.response").value()); - EXPECT_EQ(1U, store_.counter("test.response_reply").value()); + EXPECT_EQ(50U, store_.counter("test.response").value()); + EXPECT_EQ(50U, store_.counter("test.response_reply").value()); EXPECT_EQ(0U, store_.counter("test.response_exception").value()); EXPECT_EQ(0U, store_.counter("test.response_invalid_type").value()); - EXPECT_EQ(1U, store_.counter("test.response_success").value()); + EXPECT_EQ(50U, store_.counter("test.response_success").value()); + EXPECT_EQ(0U, store_.counter("test.response_error").value()); +} + +// Test the case where there is a limit on the number of requests and the actual number of requests +// exceeds the limit. +TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitAndReachedWithMoreRequests) { + const std::string yaml = R"EOF( + stat_prefix: test + route_config: + name: local_route + max_requests_per_connection: 50 + )EOF"; + + initializeFilter(yaml); + EXPECT_EQ(50, config_->maxRequestsPerConnection()); + + // Since max requests per connection is set to 50, the connection will be disconnected after + // all 50 requests is completed. + EXPECT_CALL(filter_callbacks_.connection_, close(Network::ConnectionCloseType::FlushWrite)); + + EXPECT_EQ(50, sendSomeThriftRequest(55)); + + filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); + + EXPECT_EQ(1U, store_.counter("test.downstream_cx_max_requests").value()); + EXPECT_EQ(50U, store_.counter("test.request").value()); + EXPECT_EQ(50U, store_.counter("test.request_call").value()); + EXPECT_EQ(0U, stats_.request_active_.value()); + EXPECT_EQ(50U, store_.counter("test.response").value()); + EXPECT_EQ(50U, store_.counter("test.response_reply").value()); + EXPECT_EQ(0U, store_.counter("test.response_exception").value()); + EXPECT_EQ(0U, store_.counter("test.response_invalid_type").value()); + EXPECT_EQ(50U, store_.counter("test.response_success").value()); + EXPECT_EQ(0U, store_.counter("test.response_error").value()); +} + +// Test cases where the number of requests is limited and the actual number of requests exceeds the +// limit several times. +TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitAndReachedRepeatedly) { + const std::string yaml = R"EOF( + stat_prefix: test + route_config: + name: local_route + max_requests_per_connection: 5 + )EOF"; + + initializeFilter(yaml); + EXPECT_EQ(5, config_->maxRequestsPerConnection()); + + EXPECT_CALL(filter_callbacks_.connection_, close(Network::ConnectionCloseType::FlushWrite)) + .Times(5); + + auto mock_new_connection = [this]() { + filter_ = nullptr; + + filter_callbacks_.connection_.read_enabled_ = true; + filter_callbacks_.connection_.state_ = Network::Connection::State::Open; + + ON_CALL(random_, random()).WillByDefault(Return(42)); + filter_ = std::make_unique( + *config_, random_, filter_callbacks_.connection_.dispatcher_.timeSource()); + filter_->initializeReadFilterCallbacks(filter_callbacks_); + filter_->onNewConnection(); + + filter_->onAboveWriteBufferHighWatermark(); + filter_->onBelowWriteBufferLowWatermark(); + }; + + EXPECT_EQ(5, sendSomeThriftRequest(6)); + + for (size_t i = 0; i < 4; i++) { + mock_new_connection(); + EXPECT_EQ(5, sendSomeThriftRequest(6)); + } + + filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); + + EXPECT_EQ(5U, store_.counter("test.downstream_cx_max_requests").value()); + EXPECT_EQ(25U, store_.counter("test.request").value()); + EXPECT_EQ(25U, store_.counter("test.request_call").value()); + EXPECT_EQ(0U, stats_.request_active_.value()); + EXPECT_EQ(25U, store_.counter("test.response").value()); + EXPECT_EQ(25U, store_.counter("test.response_reply").value()); + EXPECT_EQ(0U, store_.counter("test.response_exception").value()); + EXPECT_EQ(0U, store_.counter("test.response_invalid_type").value()); + EXPECT_EQ(25U, store_.counter("test.response_success").value()); EXPECT_EQ(0U, store_.counter("test.response_error").value()); } @@ -1226,7 +1375,8 @@ TEST_F(ThriftConnectionManagerTest, OnDataResumesWithNextFilter) { EXPECT_EQ(1U, stats_.request_active_.value()); } -// Tests stop iteration/resume with multiple filters when iteration is stopped during transportEnd. +// Tests stop iteration/resume with multiple filters when iteration is stopped during +// transportEnd. TEST_F(ThriftConnectionManagerTest, OnDataResumesWithNextFilterOnTransportEnd) { auto* filter = new NiceMock(); custom_filter_.reset(filter); From 3446082d481f5586a316a0e39c412b2b6440a6ed Mon Sep 17 00:00:00 2001 From: wbpcode Date: Wed, 24 Feb 2021 10:53:28 +0800 Subject: [PATCH 3/7] fix asan error Signed-off-by: wbpcode --- .../extensions/filters/network/thrift_proxy/conn_manager_test.cc | 1 + 1 file changed, 1 insertion(+) diff --git a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc index b5ed33f0dbb3a..392661a45bde8 100644 --- a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc +++ b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc @@ -1222,6 +1222,7 @@ TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitAndReachedRepeate filter_callbacks_.connection_.read_enabled_ = true; filter_callbacks_.connection_.state_ = Network::Connection::State::Open; + filter_callbacks_.connection_.callbacks_.clear(); ON_CALL(random_, random()).WillByDefault(Return(42)); filter_ = std::make_unique( From aa038d455517fc43fd1b3e60116bbc40adb1e016 Mon Sep 17 00:00:00 2001 From: wbpcode Date: Wed, 24 Feb 2021 16:40:32 +0800 Subject: [PATCH 4/7] fix asan error Signed-off-by: wbpcode --- .../filters/network/thrift_proxy/conn_manager_test.cc | 2 ++ 1 file changed, 2 insertions(+) diff --git a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc index 392661a45bde8..6c0ff9302f376 100644 --- a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc +++ b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc @@ -1218,6 +1218,8 @@ TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitAndReachedRepeate .Times(5); auto mock_new_connection = [this]() { + filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); + filter_ = nullptr; filter_callbacks_.connection_.read_enabled_ = true; From 7f5b510b972f27779bf7ace7270c85e795467748 Mon Sep 17 00:00:00 2001 From: wbpcode Date: Thu, 25 Feb 2021 05:22:54 +0000 Subject: [PATCH 5/7] fix docs Signed-off-by: wbpcode --- .../listeners/network_filters/thrift_proxy_filter.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/root/configuration/listeners/network_filters/thrift_proxy_filter.rst b/docs/root/configuration/listeners/network_filters/thrift_proxy_filter.rst index 480f12d69dd86..94972ebe3112e 100644 --- a/docs/root/configuration/listeners/network_filters/thrift_proxy_filter.rst +++ b/docs/root/configuration/listeners/network_filters/thrift_proxy_filter.rst @@ -20,7 +20,7 @@ Downstream Requests Limit Thrift Proxy can set the :ref:`maximum number of requests` that each downstream connection can handle. When the number of requests exceeds the connection limit, Thrift Proxy will -actively disconnect from the Thrfit client. +actively disconnect from the Thrift client. Thrift Request Metadata From 1c99671ccf196e49c81293ec15334ee1e563b049 Mon Sep 17 00:00:00 2001 From: wbpcode Date: Fri, 26 Feb 2021 10:55:12 +0800 Subject: [PATCH 6/7] minor update Signed-off-by: wbpcode --- .../filters/network/thrift_proxy/config.cc | 8 ++------ .../filters/network/thrift_proxy/config.h | 3 +-- .../filters/network/thrift_proxy/conn_manager.cc | 16 ++++------------ .../filters/network/thrift_proxy/conn_manager.h | 4 ++-- 4 files changed, 9 insertions(+), 22 deletions(-) diff --git a/source/extensions/filters/network/thrift_proxy/config.cc b/source/extensions/filters/network/thrift_proxy/config.cc index 4f570ac9f3779..dd835c7cf48d2 100644 --- a/source/extensions/filters/network/thrift_proxy/config.cc +++ b/source/extensions/filters/network/thrift_proxy/config.cc @@ -122,7 +122,8 @@ ConfigImpl::ConfigImpl( stats_(ThriftFilterStats::generateStats(stats_prefix_, context_.scope())), transport_(lookupTransport(config.transport())), proto_(lookupProtocol(config.protocol())), route_matcher_(new Router::RouteMatcher(config.route_config())), - payload_passthrough_(config.payload_passthrough()) { + payload_passthrough_(config.payload_passthrough()), + max_requests_per_connection_(config.max_requests_per_connection().value()) { if (config.thrift_filters().empty()) { ENVOY_LOG(debug, "using default router filter"); @@ -135,11 +136,6 @@ ConfigImpl::ConfigImpl( processFilter(filter); } } - - if (config.has_max_requests_per_connection() && - config.max_requests_per_connection().value() > 0) { - max_requests_per_connection_ = config.max_requests_per_connection().value(); - } } void ConfigImpl::createFilterChain(ThriftFilters::FilterChainFactoryCallbacks& callbacks) { diff --git a/source/extensions/filters/network/thrift_proxy/config.h b/source/extensions/filters/network/thrift_proxy/config.h index 637be14f79f39..de2cd2bb4e292 100644 --- a/source/extensions/filters/network/thrift_proxy/config.h +++ b/source/extensions/filters/network/thrift_proxy/config.h @@ -1,6 +1,5 @@ #pragma once -#include #include #include @@ -99,7 +98,7 @@ class ConfigImpl : public Config, std::list filter_factories_; const bool payload_passthrough_; - uint64_t max_requests_per_connection_{}; + const uint64_t max_requests_per_connection_{}; }; } // namespace ThriftProxy diff --git a/source/extensions/filters/network/thrift_proxy/conn_manager.cc b/source/extensions/filters/network/thrift_proxy/conn_manager.cc index 2da10f4bb7d2a..122eaabbdedfa 100644 --- a/source/extensions/filters/network/thrift_proxy/conn_manager.cc +++ b/source/extensions/filters/network/thrift_proxy/conn_manager.cc @@ -12,21 +12,12 @@ namespace Extensions { namespace NetworkFilters { namespace ThriftProxy { -namespace { -// Translate zero to UINT64_MAX so that the zero/unlimited case doesn't -// have to be handled specially. -uint64_t translateZeroToUnlimited(uint64_t limit) { - return (limit != 0) ? limit : std::numeric_limits::max(); -} -} // namespace - ConnectionManager::ConnectionManager(Config& config, Random::RandomGenerator& random_generator, TimeSource& time_source) : config_(config), stats_(config_.stats()), transport_(config.createTransport()), protocol_(config.createProtocol()), decoder_(std::make_unique(*transport_, *protocol_, *this)), - random_generator_(random_generator), time_source_(time_source), - remaining_streams_(translateZeroToUnlimited(config.maxRequestsPerConnection())) {} + random_generator_(random_generator), time_source_(time_source) {} ConnectionManager::~ConnectionManager() = default; @@ -421,8 +412,9 @@ void ConnectionManager::ActiveRpc::finalizeRequest() { parent_.stats_.request_.inc(); - parent_.remaining_streams_--; - if (parent_.remaining_streams_ == 0) { + parent_.accumulated_requests_++; + if (parent_.config_.maxRequestsPerConnection() > 0 && + parent_.accumulated_requests_ >= parent_.config_.maxRequestsPerConnection()) { parent_.read_callbacks_->connection().readDisable(true); parent_.requests_overflow_ = true; parent_.stats_.downstream_cx_max_requests_.inc(); diff --git a/source/extensions/filters/network/thrift_proxy/conn_manager.h b/source/extensions/filters/network/thrift_proxy/conn_manager.h index 03898801cf7d2..af41b6c4d6c72 100644 --- a/source/extensions/filters/network/thrift_proxy/conn_manager.h +++ b/source/extensions/filters/network/thrift_proxy/conn_manager.h @@ -274,8 +274,8 @@ class ConnectionManager : public Network::ReadFilter, bool half_closed_{false}; TimeSource& time_source_; - // The maximum number of requests remaining to be processed on the current connection. - uint64_t remaining_streams_{}; + // The number of requests accumulated on the current connection. + uint64_t accumulated_requests_{}; bool requests_overflow_{false}; }; From d3e44e39e6f08b726fc2281490949b62d313001e Mon Sep 17 00:00:00 2001 From: wbpcode Date: Mon, 1 Mar 2021 13:47:55 +0800 Subject: [PATCH 7/7] update method name Signed-off-by: wbpcode --- .../network/thrift_proxy/conn_manager_test.cc | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc index 6c0ff9302f376..d02028fd90e03 100644 --- a/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc +++ b/test/extensions/filters/network/thrift_proxy/conn_manager_test.cc @@ -121,7 +121,7 @@ class ThriftConnectionManagerTest : public testing::Test { } // Return the number of requests actually sent. - uint32_t sendSomeThriftRequest(uint32_t request_number) { + uint32_t sendRequests(uint32_t request_number) { for (uint32_t i = 0; i < request_number; i++) { writeComplexFramedBinaryMessage(buffer_, MessageType::Call, 0x0F); writeComplexFramedBinaryMessage(write_buffer_, MessageType::Reply, 0x0F); @@ -1090,7 +1090,7 @@ TEST_F(ThriftConnectionManagerTest, RequestWithNoMaxRequestsLimit) { initializeFilter(""); EXPECT_EQ(0, config_->maxRequestsPerConnection()); - EXPECT_EQ(50, sendSomeThriftRequest(50)); + EXPECT_EQ(50, sendRequests(50)); filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); @@ -1119,7 +1119,7 @@ TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitButNotReach) { initializeFilter(yaml); EXPECT_EQ(50, config_->maxRequestsPerConnection()); - EXPECT_EQ(49, sendSomeThriftRequest(49)); + EXPECT_EQ(49, sendRequests(49)); filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); @@ -1152,7 +1152,7 @@ TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitAndReached) { // all 50 requests is completed. EXPECT_CALL(filter_callbacks_.connection_, close(Network::ConnectionCloseType::FlushWrite)); - EXPECT_EQ(50, sendSomeThriftRequest(50)); + EXPECT_EQ(50, sendRequests(50)); filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); @@ -1185,7 +1185,7 @@ TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitAndReachedWithMor // all 50 requests is completed. EXPECT_CALL(filter_callbacks_.connection_, close(Network::ConnectionCloseType::FlushWrite)); - EXPECT_EQ(50, sendSomeThriftRequest(55)); + EXPECT_EQ(50, sendRequests(55)); filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList(); @@ -1236,11 +1236,11 @@ TEST_F(ThriftConnectionManagerTest, RequestWithMaxRequestsLimitAndReachedRepeate filter_->onBelowWriteBufferLowWatermark(); }; - EXPECT_EQ(5, sendSomeThriftRequest(6)); + EXPECT_EQ(5, sendRequests(6)); for (size_t i = 0; i < 4; i++) { mock_new_connection(); - EXPECT_EQ(5, sendSomeThriftRequest(6)); + EXPECT_EQ(5, sendRequests(6)); } filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList();