overload: tcp connection refusal overload action (envoyproxy#13311)

This will allow creating an overload action that rejects (accepts the
connection and then closes it) some fraction of connections in response
to overload conditions.

Signed-off-by: Alex Konradi <[email protected]>

Author: akonradi

docs/root/configuration/listeners/stats.rst

@@ -17,6 +17,7 @@ Every listener has a statistics tree rooted at *listener.<address>.* with the fo
    downstream_cx_active, Gauge, Total active connections
    downstream_cx_length_ms, Histogram, Connection length milliseconds
    downstream_cx_overflow, Counter, Total connections rejected due to enforcement of listener connection limit
+   downstream_cx_overload_reject, Counter, Total connections rejected due to configured overload actions
    downstream_pre_cx_timeout, Counter, Sockets that timed out during listener filter processing
    downstream_pre_cx_active, Gauge, Sockets currently undergoing listener filter processing
    global_cx_overflow, Counter, Total connections rejected due to enforecement of the global connection limit

docs/root/configuration/operations/overload_manager/overload_manager.rst

@@ -68,14 +68,27 @@ Overload actions
 
 The following overload actions are supported:
 
-.. csv-table::
-  :header: Name, Description
+.. list-table::
+  :header-rows: 1
   :widths: 1, 2
 
-  envoy.overload_actions.stop_accepting_requests, Envoy will immediately respond with a 503 response code to new requests
-  envoy.overload_actions.disable_http_keepalive, Envoy will stop accepting streams on incoming HTTP connections
-  envoy.overload_actions.stop_accepting_connections, Envoy will stop accepting new network connections on its configured listeners
-  envoy.overload_actions.shrink_heap, Envoy will periodically try to shrink the heap by releasing free memory to the system
+  * - Name
+    - Description
+
+  * - envoy.overload_actions.stop_accepting_requests
+    - Envoy will immediately respond with a 503 response code to new requests
+
+  * - envoy.overload_actions.disable_http_keepalive
+    - Envoy will stop accepting streams on incoming HTTP connections
+
+  * - envoy.overload_actions.stop_accepting_connections
+    - Envoy will stop accepting new network connections on its configured listeners
+
+  * - envoy.overload_actions.reject_incoming_connections
+    - Envoy will reject incoming connections on its configured listeners without processing any data
+
+  * - envoy.overload_actions.shrink_heap
+    - Envoy will periodically try to shrink the heap by releasing free memory to the system
 
 Limiting Active Connections
 ---------------------------

docs/root/version_history/current.rst

@@ -27,6 +27,7 @@ New Features
 * grpc: implemented header value syntax support when defining :ref:`initial metadata <envoy_v3_api_field_config.core.v3.GrpcService.initial_metadata>` for gRPC-based `ext_authz` :ref:`HTTP <envoy_v3_api_field_extensions.filters.http.ext_authz.v3.ExtAuthz.grpc_service>` and :ref:`network <envoy_v3_api_field_extensions.filters.network.ext_authz.v3.ExtAuthz.grpc_service>` filters, and :ref:`ratelimit <envoy_v3_api_field_config.ratelimit.v3.RateLimitServiceConfig.grpc_service>` filters.
 * health_check: added option to use :ref:`no_traffic_healthy_interval <envoy_v3_api_field_config.core.v3.HealthCheck.no_traffic_healthy_interval>` which allows a different no traffic interval when the host is healthy.
 * mongo_proxy: the list of commands to produce metrics for is now :ref:`configurable <envoy_v3_api_field_extensions.filters.network.mongo_proxy.v3.MongoProxy.commands>`.
+* tcp: added a new :ref:`envoy.overload_actions.reject_incoming_connections <config_overload_manager_overload_actions>` action to reject incoming TCP connections.
 
 Deprecated
 ----------

include/envoy/network/connection_handler.h

@@ -94,6 +94,12 @@ class ConnectionHandler {
    */
   virtual void enableListeners() PURE;
 
+  /**
+   * Set the fraction of connections the listeners should reject.
+   * @param reject_fraction a value between 0 (reject none) and 1 (reject all).
+   */
+  virtual void setListenerRejectFraction(float reject_fraction) PURE;
+
   /**
    * @return the stat prefix used for per-handler stats.
    */

include/envoy/network/listener.h

@@ -197,10 +197,14 @@ class TcpListenerCallbacks {
    */
   virtual void onAccept(ConnectionSocketPtr&& socket) PURE;
 
+  enum class RejectCause {
+    GlobalCxLimit,
+    OverloadAction,
+  };
   /**
    * Called when a new connection is rejected.
    */
-  virtual void onReject() PURE;
+  virtual void onReject(RejectCause cause) PURE;
 };
 
 /**
@@ -324,6 +328,12 @@ class Listener {
    * Enable accepting new connections.
    */
   virtual void enable() PURE;
+
+  /**
+   * Set the fraction of incoming connections that will be closed immediately
+   * after being opened.
+   */
+  virtual void setRejectFraction(float reject_fraction) PURE;
 };
 
 using ListenerPtr = std::unique_ptr<Listener>;

include/envoy/server/overload_manager.h

@@ -62,6 +62,10 @@ class OverloadActionNameValues {
   // Overload action to stop accepting new connections.
   const std::string StopAcceptingConnections = "envoy.overload_actions.stop_accepting_connections";
 
+  // Overload action to reject (accept and then close) new connections.
+  const std::string RejectIncomingConnections =
+      "envoy.overload_actions.reject_incoming_connections";
+
   // Overload action to try to shrink the heap by releasing free memory.
   const std::string ShrinkHeap = "envoy.overload_actions.shrink_heap";
 };

source/common/event/dispatcher_impl.cc

@@ -162,8 +162,8 @@ Network::ListenerPtr DispatcherImpl::createListener(Network::SocketSharedPtr&& s
                                                     Network::TcpListenerCallbacks& cb,
                                                     bool bind_to_port, uint32_t backlog_size) {
   ASSERT(isThreadSafe());
-  return std::make_unique<Network::TcpListenerImpl>(*this, std::move(socket), cb, bind_to_port,
-                                                    backlog_size);
+  return std::make_unique<Network::TcpListenerImpl>(
+      *this, api_.randomGenerator(), std::move(socket), cb, bind_to_port, backlog_size);
 }
 
 Network::UdpListenerPtr DispatcherImpl::createUdpListener(Network::SocketSharedPtr socket,

source/common/network/tcp_listener_impl.cc

@@ -62,7 +62,11 @@ void TcpListenerImpl::onSocketEvent(short flags) {
     if (rejectCxOverGlobalLimit()) {
       // The global connection limit has been reached.
       io_handle->close();
-      cb_.onReject();
+      cb_.onReject(TcpListenerCallbacks::RejectCause::GlobalCxLimit);
+      continue;
+    } else if (random_.bernoulli(reject_fraction_)) {
+      io_handle->close();
+      cb_.onReject(TcpListenerCallbacks::RejectCause::OverloadAction);
       continue;
     }
 
@@ -106,9 +110,11 @@ void TcpListenerImpl::setupServerSocket(Event::DispatcherImpl& dispatcher, Socke
   }
 }
 
-TcpListenerImpl::TcpListenerImpl(Event::DispatcherImpl& dispatcher, SocketSharedPtr socket,
-                                 TcpListenerCallbacks& cb, bool bind_to_port, uint32_t backlog_size)
-    : BaseListenerImpl(dispatcher, std::move(socket)), cb_(cb), backlog_size_(backlog_size) {
+TcpListenerImpl::TcpListenerImpl(Event::DispatcherImpl& dispatcher, Random::RandomGenerator& random,
+                                 SocketSharedPtr socket, TcpListenerCallbacks& cb,
+                                 bool bind_to_port, uint32_t backlog_size)
+    : BaseListenerImpl(dispatcher, std::move(socket)), cb_(cb), backlog_size_(backlog_size),
+      random_(random), reject_fraction_(0.0) {
   if (bind_to_port) {
     setupServerSocket(dispatcher, *socket_);
   }
@@ -118,5 +124,10 @@ void TcpListenerImpl::enable() { file_event_->setEnabled(Event::FileReadyType::R
 
 void TcpListenerImpl::disable() { file_event_->setEnabled(0); }
 
+void TcpListenerImpl::setRejectFraction(const float reject_fraction) {
+  ASSERT(0 <= reject_fraction && reject_fraction <= 1);
+  reject_fraction_ = reject_fraction;
+}
+
 } // namespace Network
 } // namespace Envoy

source/common/network/tcp_listener_impl.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "envoy/common/random_generator.h"
 #include "envoy/runtime/runtime.h"
 
 #include "absl/strings/string_view.h"
@@ -13,10 +14,12 @@ namespace Network {
  */
 class TcpListenerImpl : public BaseListenerImpl {
 public:
-  TcpListenerImpl(Event::DispatcherImpl& dispatcher, SocketSharedPtr socket,
-                  TcpListenerCallbacks& cb, bool bind_to_port, uint32_t backlog_size);
+  TcpListenerImpl(Event::DispatcherImpl& dispatcher, Random::RandomGenerator& random,
+                  SocketSharedPtr socket, TcpListenerCallbacks& cb, bool bind_to_port,
+                  uint32_t backlog_size);
   void disable() override;
   void enable() override;
+  void setRejectFraction(float reject_fraction) override;
 
   static const absl::string_view GlobalMaxCxRuntimeKey;
 
@@ -33,7 +36,9 @@ class TcpListenerImpl : public BaseListenerImpl {
   // rejected/closed. If the accepted socket is to be admitted, false is returned.
   static bool rejectCxOverGlobalLimit();
 
+  Random::RandomGenerator& random_;
   Event::FileEventPtr file_event_;
+  float reject_fraction_;
 };
 
 } // namespace Network

source/common/network/udp_listener_impl.h

@@ -30,6 +30,7 @@ class UdpListenerImpl : public BaseListenerImpl,
   // Network::Listener Interface
   void disable() override;
   void enable() override;
+  void setRejectFraction(float) override {}
 
   // Network::UdpListener Interface
   Event::Dispatcher& dispatcher() override;

source/server/connection_handler_impl.cc

@@ -66,6 +66,9 @@ void ConnectionHandlerImpl::addListener(absl::optional<uint64_t> overridden_list
   if (disable_listeners_) {
     details.listener_->pauseListening();
   }
+  if (auto* listener = details.listener_->listener(); listener != nullptr) {
+    listener->setRejectFraction(listener_reject_fraction_);
+  }
   listeners_.emplace_back(config.listenSocketFactory().localAddress(), std::move(details));
 }
 
@@ -148,6 +151,13 @@ void ConnectionHandlerImpl::enableListeners() {
   }
 }
 
+void ConnectionHandlerImpl::setListenerRejectFraction(float reject_fraction) {
+  listener_reject_fraction_ = reject_fraction;
+  for (auto& listener : listeners_) {
+    listener.second.listener_->listener()->setRejectFraction(reject_fraction);
+  }
+}
+
 void ConnectionHandlerImpl::ActiveTcpListener::removeConnection(ActiveTcpConnection& connection) {
   ENVOY_CONN_LOG(debug, "adding to cleanup list", *connection.connection_);
   ActiveConnections& active_connections = connection.active_connections_;
@@ -391,6 +401,17 @@ void ConnectionHandlerImpl::ActiveTcpListener::onAccept(Network::ConnectionSocke
   onAcceptWorker(std::move(socket), config_->handOffRestoredDestinationConnections(), false);
 }
 
+void ConnectionHandlerImpl::ActiveTcpListener::onReject(RejectCause cause) {
+  switch (cause) {
+  case RejectCause::GlobalCxLimit:
+    stats_.downstream_global_cx_overflow_.inc();
+    break;
+  case RejectCause::OverloadAction:
+    stats_.downstream_cx_overload_reject_.inc();
+    break;
+  }
+}
+
 void ConnectionHandlerImpl::ActiveTcpListener::onAcceptWorker(
     Network::ConnectionSocketPtr&& socket, bool hand_off_restored_destination_connections,
     bool rebalanced) {

source/server/connection_handler_impl.h

@@ -30,6 +30,7 @@ namespace Server {
   COUNTER(downstream_cx_destroy)                                                                   \
   COUNTER(downstream_cx_overflow)                                                                  \
   COUNTER(downstream_cx_total)                                                                     \
+  COUNTER(downstream_cx_overload_reject)                                                           \
   COUNTER(downstream_global_cx_overflow)                                                           \
   COUNTER(downstream_pre_cx_timeout)                                                               \
   COUNTER(no_filter_chain_match)                                                                   \
@@ -82,6 +83,7 @@ class ConnectionHandlerImpl : public Network::ConnectionHandler,
   void stopListeners() override;
   void disableListeners() override;
   void enableListeners() override;
+  void setListenerRejectFraction(float reject_fraction) override;
   const std::string& statPrefix() const override { return per_handler_stat_prefix_; }
 
   /**
@@ -133,7 +135,7 @@ class ConnectionHandlerImpl : public Network::ConnectionHandler,
 
     // Network::TcpListenerCallbacks
     void onAccept(Network::ConnectionSocketPtr&& socket) override;
-    void onReject() override { stats_.downstream_global_cx_overflow_.inc(); }
+    void onReject(RejectCause) override;
 
     // ActiveListenerImplBase
     Network::Listener* listener() override { return listener_.get(); }
@@ -361,6 +363,7 @@ class ConnectionHandlerImpl : public Network::ConnectionHandler,
   std::list<std::pair<Network::Address::InstanceConstSharedPtr, ActiveListenerDetails>> listeners_;
   std::atomic<uint64_t> num_handler_connections_{};
   bool disable_listeners_;
+  float listener_reject_fraction_{0};
 };
 
 class ActiveUdpListenerBase : public ConnectionHandlerImpl::ActiveListenerImplBase,

source/server/worker_impl.cc

@@ -31,6 +31,9 @@ WorkerImpl::WorkerImpl(ThreadLocal::Instance& tls, ListenerHooks& hooks,
   overload_manager.registerForAction(
       OverloadActionNames::get().StopAcceptingConnections, *dispatcher_,
       [this](OverloadActionState state) { stopAcceptingConnectionsCb(state); });
+  overload_manager.registerForAction(
+      OverloadActionNames::get().RejectIncomingConnections, *dispatcher_,
+      [this](OverloadActionState state) { rejectIncomingConnectionsCb(state); });
 }
 
 void WorkerImpl::addListener(absl::optional<uint64_t> overridden_listener,
@@ -149,5 +152,9 @@ void WorkerImpl::stopAcceptingConnectionsCb(OverloadActionState state) {
   }
 }
 
+void WorkerImpl::rejectIncomingConnectionsCb(OverloadActionState state) {
+  handler_->setListenerRejectFraction(static_cast<float>(state.value()));
+}
+
 } // namespace Server
 } // namespace Envoy

source/server/worker_impl.h

@@ -58,6 +58,7 @@ class WorkerImpl : public Worker, Logger::Loggable<Logger::Id::main> {
 private:
   void threadRoutine(GuardDog& guard_dog);
   void stopAcceptingConnectionsCb(OverloadActionState state);
+  void rejectIncomingConnectionsCb(OverloadActionState state);
 
   ThreadLocal::Instance& tls_;
   ListenerHooks& hooks_;

test/common/network/dns_impl_test.cc

@@ -281,7 +281,7 @@ class TestDnsServer : public TcpListenerCallbacks {
     queries_.emplace_back(query);
   }
 
-  void onReject() override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
+  void onReject(RejectCause) override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
 
   void addHosts(const std::string& hostname, const IpList& ip, const RecordType& type) {
     if (type == RecordType::A) {