thrift: supported setting max requests for per downstream connection

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.

docs/root/configuration/listeners/network_filters/thrift_proxy_filter.rst

@@ -15,6 +15,14 @@ keyed by `envoy.filters.network.thrift_proxy`. The
 :ref:`ThriftProtocolOptions<envoy_v3_api_msg_extensions.filters.network.thrift_proxy.v3.ThriftProtocolOptions>`
 message describes the available options.
 
+Downstream Requests Limit
+-------------------------
+Thrift Proxy can set the
+:ref:`maximum number of requests<envoy_v3_api_field_extensions.filters.network.thrift_proxy.v3.ThriftProxy.max_requests_per_connection>`
+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
 -----------------------
 

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 <tunneling-tcp-over-http>` for details.
 * tcp_proxy: added a :ref:`use_post field <envoy_v3_api_field_extensions.filters.network.tcp_proxy.v3.TcpProxy.TunnelingConfig.use_post>` for using HTTP POST to proxy TCP streams.
 * tcp_proxy: added a :ref:`headers_to_add field <envoy_v3_api_field_extensions.filters.network.tcp_proxy.v3.TcpProxy.TunnelingConfig.headers_to_add>` for setting additional headers to the HTTP requests for TCP proxing.
+* thrift_proxy: added a :ref:`max_requests_per_connection field <envoy_v3_api_field_extensions.filters.network.thrift_proxy.v3.ThriftProxy.max_requests_per_connection>` for setting maximum requests for per downstream connection.
 
 Deprecated
 ----------

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) {

source/extensions/filters/network/thrift_proxy/config.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <cstdint>
 #include <map>
 #include <string>
 
@@ -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<ThriftFilters::FilterFactoryCb> filter_factories_;
   const bool payload_passthrough_;
+
+  uint64_t max_requests_per_connection_{};
 };
 
 } // namespace ThriftProxy

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<uint64_t>::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<Decoder>(*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;
 
@@ -55,6 +64,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 +153,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 +421,13 @@ void ConnectionManager::ActiveRpc::finalizeRequest() {
 
   parent_.stats_.request_.inc();
 
+  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;
   switch (original_msg_type_) {
   case MessageType::Call:

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,10 @@ class ConnectionManager : public Network::ReadFilter,
   bool stopped_{false};
   bool half_closed_{false};
   TimeSource& time_source_;
+
+  // The maximum number of requests remaining to be processed on the current connection.
+  uint64_t remaining_streams_{};
+  bool requests_overflow_{false};
 };
 
 } // namespace ThriftProxy

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)                                                                  \