network: delayed conn close

.circleci/config.yml

@@ -10,7 +10,6 @@ jobs:
      resource_class: xlarge
      working_directory: /source
      environment:
-       BAZEL_REMOTE_CACHE: https://storage.googleapis.com/envoy-circleci-bazel-cache/
      steps:
        - run: rm -rf /home/circleci/project/.git # CircleCI git caching is likely broken
        - checkout
@@ -24,7 +23,6 @@ jobs:
      resource_class: xlarge
      working_directory: /source
      environment:
-       BAZEL_REMOTE_CACHE: https://storage.googleapis.com/envoy-circleci-bazel-cache/
      steps:
        - run: rm -rf /home/circleci/project/.git # CircleCI git caching is likely broken
        - run: echo $CIRCLE_SHA1
@@ -38,7 +36,6 @@ jobs:
      resource_class: xlarge
      working_directory: /source
      environment:
-       BAZEL_REMOTE_CACHE: https://storage.googleapis.com/envoy-circleci-bazel-cache/
      steps:
        - run: rm -rf /home/circleci/project/.git # CircleCI git caching is likely broken
        - checkout
@@ -71,7 +68,6 @@ jobs:
    ipv6_tests:
      machine: true
      environment:
-       BAZEL_REMOTE_CACHE: https://storage.googleapis.com/envoy-circleci-bazel-cache/
      steps:
      - run: rm -rf /home/circleci/project/.git # CircleCI git caching is likely broken
      - checkout
@@ -145,7 +141,6 @@ jobs:
      macos:
        xcode: "9.3.0"
      environment:
-       BAZEL_REMOTE_CACHE: https://storage.googleapis.com/envoy-circleci-bazel-cache/
      steps:
        - run: sudo ntpdate -vu time.apple.com
        - run: rm -rf /home/circleci/project/.git # CircleCI git caching is likely broken

api/envoy/config/filter/network/http_connection_manager/v2/http_connection_manager.proto

@@ -19,7 +19,7 @@ import "gogoproto/gogo.proto";
 // [#protodoc-title: HTTP connection manager]
 // HTTP connection manager :ref:`configuration overview <config_http_conn_man>`.
 
-// [#comment:next free field: 25]
+// [#comment:next free field: 26]
 message HttpConnectionManager {
   enum CodecType {
     option (gogoproto.goproto_enum_prefix) = false;
@@ -175,6 +175,19 @@ message HttpConnectionManager {
   // option is not specified.
   google.protobuf.Duration drain_timeout = 12 [(gogoproto.stdduration) = true];
 
+  // The delayed close timeout is for downstream connections managed by the HTTP connection manager.
+  // It is defined as a grace period after connection close processing has been locally initiated
+  // during which Envoy will flush the write buffers for the connection and await the peer to close
+  // the connection (i.e., a TCP FIN/RST is received by Envoy from the downstream connection).
+  //
+  // If the timeout triggers, Envoy will close the connection's socket.
+  //
+  // The default timeout is 1000 ms if this option is not specified.
+  //
+  // A value of 0 will completely disable delayed close processing, and the downstream connection's
+  // socket will be closed immediately after the write flush is completed.
+  google.protobuf.Duration delayed_close_timeout = 25 [(gogoproto.stdduration) = true];
+
   // Configuration for :ref:`HTTP access logs <arch_overview_access_logs>`
   // emitted by the connection manager.
   repeated envoy.config.filter.accesslog.v2.AccessLog access_log = 13;

include/envoy/network/connection.h

@@ -63,7 +63,9 @@ class ConnectionCallbacks {
  */
 enum class ConnectionCloseType {
   FlushWrite, // Flush pending write data before raising ConnectionEvent::LocalClose
-  NoFlush     // Do not flush any pending data and immediately raise ConnectionEvent::LocalClose
+  NoFlush,    // Do not flush any pending data and immediately raise ConnectionEvent::LocalClose
+  FlushWriteAndDelay // Flush pending write data and delay raising a ConnectionEvent::LocalClose
+                     // until the delayed_close_timeout expires
 };
 
 /**
@@ -86,6 +88,8 @@ class Connection : public Event::DeferredDeletable, public FilterManager {
     Stats::Gauge& write_current_;
     // Counter* as this is an optional counter. Bind errors will not be tracked if this is nullptr.
     Stats::Counter* bind_errors_;
+    // Optional counter. Delayed close semantics are only used by HTTP connections.
+    Stats::Counter* delayed_close_timeouts_;
   };
 
   virtual ~Connection() {}
@@ -233,6 +237,18 @@ class Connection : public Event::DeferredDeletable, public FilterManager {
    * Get the socket options set on this connection.
    */
   virtual const ConnectionSocket::OptionsSharedPtr& socketOptions() const PURE;
+
+  /**
+   * Set the timeout for delayed connection close()s.
+   * This is only used for downstream connections processing HTTP/1 and HTTP/2.
+   * @param timeout The timeout value in milliseconds
+   */
+  virtual void setDelayedCloseTimeout(std::chrono::milliseconds timeout) PURE;
+
+  /**
+   * @return std::chrono::milliseconds The delayed close timeout value.
+   */
+  virtual std::chrono::milliseconds delayedCloseTimeout() const PURE;
 };
 
 typedef std::unique_ptr<Connection> ConnectionPtr;

source/common/http/conn_manager_config.h

@@ -37,6 +37,7 @@ namespace Http {
   GAUGE    (downstream_cx_tx_bytes_buffered)                                                       \
   COUNTER  (downstream_cx_drain_close)                                                             \
   COUNTER  (downstream_cx_idle_timeout)                                                            \
+  COUNTER  (downstream_cx_delayed_close_timeout)                                                   \
   COUNTER  (downstream_flow_control_paused_reading_total)                                          \
   COUNTER  (downstream_flow_control_resumed_reading_total)                                         \
   COUNTER  (downstream_rq_total)                                                                   \

source/common/http/conn_manager_impl.cc

@@ -90,7 +90,7 @@ void ConnectionManagerImpl::initializeReadFilterCallbacks(Network::ReadFilterCal
   read_callbacks_->connection().setConnectionStats(
       {stats_.named_.downstream_cx_rx_bytes_total_, stats_.named_.downstream_cx_rx_bytes_buffered_,
        stats_.named_.downstream_cx_tx_bytes_total_, stats_.named_.downstream_cx_tx_bytes_buffered_,
-       nullptr});
+       nullptr, &stats_.named_.downstream_cx_delayed_close_timeout_});
 }
 
 ConnectionManagerImpl::~ConnectionManagerImpl() {
@@ -118,7 +118,7 @@ ConnectionManagerImpl::~ConnectionManagerImpl() {
 
 void ConnectionManagerImpl::checkForDeferredClose() {
   if (drain_state_ == DrainState::Closing && streams_.empty() && !codec_->wantsToWrite()) {
-    read_callbacks_->connection().close(Network::ConnectionCloseType::FlushWrite);
+    read_callbacks_->connection().close(Network::ConnectionCloseType::FlushWriteAndDelay);
   }
 }
 
@@ -233,12 +233,12 @@ Network::FilterStatus ConnectionManagerImpl::onData(Buffer::Instance& data, bool
       ENVOY_CONN_LOG(debug, "dispatch error: {}", read_callbacks_->connection(), e.what());
       stats_.named_.downstream_cx_protocol_error_.inc();
 
-      // In the protocol error case, we need to reset all streams now. Since we do a flush write,
-      // the connection might stick around long enough for a pending stream to come back and try
-      // to encode.
+      // In the protocol error case, we need to reset all streams now. Since we do a flush write and
+      // delayed close, the connection might stick around long enough for a pending stream to come
+      // back and try to encode.
       resetAllStreams();
 
-      read_callbacks_->connection().close(Network::ConnectionCloseType::FlushWrite);
+      read_callbacks_->connection().close(Network::ConnectionCloseType::FlushWriteAndDelay);
       return Network::FilterStatus::StopIteration;
     }
 
@@ -317,6 +317,8 @@ void ConnectionManagerImpl::onIdleTimeout() {
   ENVOY_CONN_LOG(debug, "idle timeout", read_callbacks_->connection());
   stats_.named_.downstream_cx_idle_timeout_.inc();
   if (!codec_) {
+    // No need to delay close after flushing since an idle timeout has already fired. Attempt to
+    // write out buffered data one last time and issue a local close if successful.
     read_callbacks_->connection().close(Network::ConnectionCloseType::FlushWrite);
   } else if (drain_state_ == DrainState::NotDraining) {
     startDrainSequence();

source/common/http/http1/codec_impl.cc

@@ -450,8 +450,11 @@ void ConnectionImpl::onResetStreamBase(StreamResetReason reason) {
 
 ServerConnectionImpl::ServerConnectionImpl(Network::Connection& connection,
                                            ServerConnectionCallbacks& callbacks,
-                                           Http1Settings settings)
-    : ConnectionImpl(connection, HTTP_REQUEST), callbacks_(callbacks), codec_settings_(settings) {}
+                                           Http1Settings settings,
+                                           std::chrono::milliseconds delayed_close_timeout)
+    : ConnectionImpl(connection, HTTP_REQUEST), callbacks_(callbacks), codec_settings_(settings) {
+  connection_.setDelayedCloseTimeout(delayed_close_timeout);
+}
 
 void ServerConnectionImpl::onEncodeComplete() {
   ASSERT(active_request_);

source/common/http/http1/codec_impl.h

@@ -277,8 +277,9 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
  */
 class ServerConnectionImpl : public ServerConnection, public ConnectionImpl {
 public:
-  ServerConnectionImpl(Network::Connection& connection, ServerConnectionCallbacks& callbacks,
-                       Http1Settings settings);
+  ServerConnectionImpl(
+      Network::Connection& connection, ServerConnectionCallbacks& callbacks, Http1Settings settings,
+      std::chrono::milliseconds delayed_close_timeout = std::chrono::milliseconds(0));
 
   virtual bool supports_http_10() override { return codec_settings_.accept_http_10_; }
 

source/common/http/http1/conn_pool.cc

@@ -334,7 +334,7 @@ ConnPoolImpl::ActiveClient::ActiveClient(ConnPoolImpl& parent)
        parent_.host_->cluster().stats().upstream_cx_rx_bytes_buffered_,
        parent_.host_->cluster().stats().upstream_cx_tx_bytes_total_,
        parent_.host_->cluster().stats().upstream_cx_tx_bytes_buffered_,
-       &parent_.host_->cluster().stats().bind_errors_});
+       &parent_.host_->cluster().stats().bind_errors_, nullptr});
 }
 
 ConnPoolImpl::ActiveClient::~ActiveClient() {

source/common/http/http2/codec_impl.cc

@@ -812,8 +812,10 @@ int ClientConnectionImpl::onHeader(const nghttp2_frame* frame, HeaderString&& na
 
 ServerConnectionImpl::ServerConnectionImpl(Network::Connection& connection,
                                            Http::ServerConnectionCallbacks& callbacks,
-                                           Stats::Scope& scope, const Http2Settings& http2_settings)
+                                           Stats::Scope& scope, const Http2Settings& http2_settings,
+                                           std::chrono::milliseconds delayed_close_timeout)
     : ConnectionImpl(connection, scope, http2_settings), callbacks_(callbacks) {
+  connection_.setDelayedCloseTimeout(delayed_close_timeout);
   Http2Options http2_options(http2_settings);
   nghttp2_session_server_new2(&session_, http2_callbacks_.callbacks(), base(),
                               http2_options.options());

source/common/http/http2/codec_impl.h

@@ -323,8 +323,10 @@ class ClientConnectionImpl : public ClientConnection, public ConnectionImpl {
  */
 class ServerConnectionImpl : public ServerConnection, public ConnectionImpl {
 public:
-  ServerConnectionImpl(Network::Connection& connection, ServerConnectionCallbacks& callbacks,
-                       Stats::Scope& scope, const Http2Settings& http2_settings);
+  ServerConnectionImpl(
+      Network::Connection& connection, ServerConnectionCallbacks& callbacks, Stats::Scope& scope,
+      const Http2Settings& http2_settings,
+      std::chrono::milliseconds delayed_close_timeout = std::chrono::milliseconds(0));
 
 private:
   // ConnectionImpl

source/common/http/http2/conn_pool.cc

@@ -241,7 +241,7 @@ ConnPoolImpl::ActiveClient::ActiveClient(ConnPoolImpl& parent)
                                parent_.host_->cluster().stats().upstream_cx_rx_bytes_buffered_,
                                parent_.host_->cluster().stats().upstream_cx_tx_bytes_total_,
                                parent_.host_->cluster().stats().upstream_cx_tx_bytes_buffered_,
-                               &parent_.host_->cluster().stats().bind_errors_});
+                               &parent_.host_->cluster().stats().bind_errors_, nullptr});
 }
 
 ConnPoolImpl::ActiveClient::~ActiveClient() {

source/common/network/connection_impl.cc

@@ -70,6 +70,12 @@ ConnectionImpl::ConnectionImpl(Event::Dispatcher& dispatcher, ConnectionSocketPt
 ConnectionImpl::~ConnectionImpl() {
   ASSERT(fd() == -1, "ConnectionImpl was unexpectedly torn down without being closed.");
 
+  if (delayed_close_timer_) {
+    // It's ok to disable even if the timer has already fired.
+    delayed_close_timer_->disableTimer();
+    delayed_close_timer_.reset();
+  }
+
   // In general we assume that owning code has called close() previously to the destructor being
   // run. This generally must be done so that callbacks run in the correct context (vs. deferred
   // deletion). Hence the assert above. However, call close() here just to be completely sure that
@@ -106,10 +112,29 @@ void ConnectionImpl::close(ConnectionCloseType type) {
 
     closeSocket(ConnectionEvent::LocalClose);
   } else {
-    // TODO(mattklein123): We need a flush timer here. We might never get open socket window.
-    ASSERT(type == ConnectionCloseType::FlushWrite);
-    close_with_flush_ = true;
+    RELEASE_ASSERT(type == ConnectionCloseType::FlushWrite ||
+                       type == ConnectionCloseType::FlushWriteAndDelay,
+                   "");
+    delayed_close_ = true;
+    bool delayed_close_timeout_set = delayedCloseTimeout().count() > 0;
+
+    // All close types that follow do not actually close() the socket immediately so that buffered
+    // data can be written. However, we do want to stop reading to apply TCP backpressure.
     read_enabled_ = false;
+
+    // Force a closeSocket() after the write buffer is flushed if the close_type calls for it or if
+    // no delayed close timeout is set.
+    close_after_flush_ = !delayed_close_timeout_set || type == ConnectionCloseType::FlushWrite;
+
+    // Create and activate a timer which will immediately close the connection if triggered.
+    // A config value of 0 disables the timeout.
+    if (delayed_close_timeout_set) {
+      delayed_close_timer_ = dispatcher_.createTimer([this]() -> void { onDelayedCloseTimeout(); });
+      ENVOY_CONN_LOG(debug, "setting delayed close timer with timeout {} ms", *this,
+                     delayedCloseTimeout().count());
+      delayed_close_timer_->enableTimer(delayedCloseTimeout());
+    }
+
     file_event_->setEnabled(Event::FileReadyType::Write |
                             (enable_half_close_ ? 0 : Event::FileReadyType::Closed));
   }
@@ -118,7 +143,7 @@ void ConnectionImpl::close(ConnectionCloseType type) {
 Connection::State ConnectionImpl::state() const {
   if (fd() == -1) {
     return State::Closed;
-  } else if (close_with_flush_) {
+  } else if (delayed_close_) {
     return State::Closing;
   } else {
     return State::Open;
@@ -488,7 +513,7 @@ void ConnectionImpl::onWriteReady() {
     // write callback. This can happen if we manage to complete the SSL handshake in the write
     // callback, raise a connected event, and close the connection.
     closeSocket(ConnectionEvent::RemoteClose);
-  } else if ((close_with_flush_ && new_buffer_size == 0) || bothSidesHalfClosed()) {
+  } else if ((close_after_flush_ && new_buffer_size == 0) || bothSidesHalfClosed()) {
     ENVOY_CONN_LOG(debug, "write flush complete", *this);
     closeSocket(ConnectionEvent::LocalClose);
   } else if (result.action_ == PostIoAction::KeepOpen && result.bytes_processed_ > 0) {
@@ -535,6 +560,13 @@ bool ConnectionImpl::bothSidesHalfClosed() {
   return read_end_stream_ && write_end_stream_ && write_buffer_->length() == 0;
 }
 
+void ConnectionImpl::onDelayedCloseTimeout() {
+  ENVOY_CONN_LOG(debug, "triggered delayed close", *this);
+  ASSERT(connection_stats_->delayed_close_timeouts_);
+  connection_stats_->delayed_close_timeouts_->inc();
+  closeSocket(ConnectionEvent::LocalClose);
+}
+
 ClientConnectionImpl::ClientConnectionImpl(
     Event::Dispatcher& dispatcher, const Address::InstanceConstSharedPtr& remote_address,
     const Network::Address::InstanceConstSharedPtr& source_address,

source/common/network/connection_impl.h

@@ -114,6 +114,11 @@ class ConnectionImpl : public virtual Connection,
   // Obtain global next connection ID. This should only be used in tests.
   static uint64_t nextGlobalIdForTest() { return next_global_id_; }
 
+  void setDelayedCloseTimeout(std::chrono::milliseconds timeout) override {
+    delayed_close_timeout_ = timeout;
+  }
+  std::chrono::milliseconds delayedCloseTimeout() const override { return delayed_close_timeout_; }
+
 protected:
   void closeSocket(ConnectionEvent close_type);
 
@@ -128,6 +133,7 @@ class ConnectionImpl : public virtual Connection,
   // a generic pointer.
   Buffer::InstancePtr write_buffer_;
   uint32_t read_buffer_limit_ = 0;
+  std::chrono::milliseconds delayed_close_timeout_{0};
 
 protected:
   bool connecting_{false};
@@ -146,14 +152,19 @@ class ConnectionImpl : public virtual Connection,
   // Returns true iff end of stream has been both written and read.
   bool bothSidesHalfClosed();
 
+  // Callback issued when a delayed close timeout triggers.
+  void onDelayedCloseTimeout();
+
   static std::atomic<uint64_t> next_global_id_;
 
   Event::Dispatcher& dispatcher_;
   const uint64_t id_;
+  Event::TimerPtr delayed_close_timer_;
   std::list<ConnectionCallbacks*> callbacks_;
   std::list<BytesSentCb> bytes_sent_callbacks_;
   bool read_enabled_{true};
-  bool close_with_flush_{false};
+  bool close_after_flush_{false};
+  bool delayed_close_{false};
   bool above_high_watermark_{false};
   bool detect_early_close_{true};
   bool enable_half_close_{false};

source/common/tcp/conn_pool.cc

@@ -337,7 +337,7 @@ ConnPoolImpl::ActiveConn::ActiveConn(ConnPoolImpl& parent)
                              parent_.host_->cluster().stats().upstream_cx_rx_bytes_buffered_,
                              parent_.host_->cluster().stats().upstream_cx_tx_bytes_total_,
                              parent_.host_->cluster().stats().upstream_cx_tx_bytes_buffered_,
-                             &parent_.host_->cluster().stats().bind_errors_});
+                             &parent_.host_->cluster().stats().bind_errors_, nullptr});
 
   // We just universally set no delay on connections. Theoretically we might at some point want
   // to make this configurable.

source/common/tcp_proxy/tcp_proxy.cc

@@ -167,7 +167,7 @@ void Filter::initialize(Network::ReadFilterCallbacks& callbacks, bool set_connec
         {config_->stats().downstream_cx_rx_bytes_total_,
          config_->stats().downstream_cx_rx_bytes_buffered_,
          config_->stats().downstream_cx_tx_bytes_total_,
-         config_->stats().downstream_cx_tx_bytes_buffered_, nullptr});
+         config_->stats().downstream_cx_tx_bytes_buffered_, nullptr, nullptr});
   }
 }