http2: send graceful goaways on close (#41284)

### Description

With this change envoy will send one GOAWAY frame with a last_stream_id
set to 2^31-1, as described in [RFC
9113](https://www.rfc-editor.org/rfc/rfc9113.html#name-goaway). After a
reasonable interval (defaults to 1s), envoy will send a followup GOAWAY
frame with last_stream_id set to the highest received stream ID at that
point in time.
This interval is configurable using `graceful_goaway_timeout` field.

---
Commit Message: http2: send graceful GoAway on close
Additional Description: According to RFC 9113, envoy should send a
goaway with last_stream_id set to 2^31-1 before sending a final goaway
with the highest received stream ID.
Risk Level: Low
Testing: Tests added. Manually verified.
Docs Changes: NA
Release Notes: Changelog added
Platform Specific Features:
Runtime guard: `envoy.reloadable_features.http2_graceful_goaway`
Fixes #39876
[Optional [API
Considerations](https://github.com/envoyproxy/envoy/blob/main/api/review_checklist.md):]

---------

Signed-off-by: Anurag Aggarwal <[email protected]>

Author: kanurag94

changelogs/current.yaml

@@ -124,6 +124,9 @@ new_features:
     Added :ref:`vhost_header <envoy_v3_api_field_config.route.v3.RouteConfiguration.vhost_header>` to
     :ref:`RouteConfiguration <envoy_v3_api_msg_config.route.v3.RouteConfiguration>`, allowing use of a different
     header for vhost matching.
+- area: http2
+  change: |
+    Added new parameter to the ``sendGoAwayAndClose`` to support gracefully closing of HTTP/2 connection.
 - area: logging
   change: |
     Added support for the not-equal operator in access log filter rules, in

envoy/http/filter.h

@@ -668,7 +668,7 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
   /**
    * Attempt to send GOAWAY and close the connection, and no filter chain will move forward.
    */
-  virtual void sendGoAwayAndClose() PURE;
+  virtual void sendGoAwayAndClose(bool graceful = false) PURE;
 
   /**
    * Adds decoded metadata. This function can only be called in

source/common/http/async_client_impl.h

@@ -228,7 +228,7 @@ class AsyncStreamImpl : public virtual AsyncClient::Stream,
   }
   void addDownstreamWatermarkCallbacks(DownstreamWatermarkCallbacks&) override {}
   void removeDownstreamWatermarkCallbacks(DownstreamWatermarkCallbacks&) override {}
-  void sendGoAwayAndClose() override {}
+  void sendGoAwayAndClose(bool graceful [[maybe_unused]] = false) override {}
 
   void setDecoderBufferLimit(uint64_t) override {
     IS_ENVOY_BUG("decoder buffer limits should not be overridden on async streams.");

source/common/http/conn_manager_impl.cc

@@ -780,16 +780,33 @@ void ConnectionManagerImpl::onDrainTimeout() {
   checkForDeferredClose(false);
 }
 
-void ConnectionManagerImpl::sendGoAwayAndClose() {
+void ConnectionManagerImpl::sendGoAwayAndClose(bool graceful) {
   ENVOY_CONN_LOG(trace, "connection manager sendGoAwayAndClose was triggerred from filters.",
                  read_callbacks_->connection());
   if (go_away_sent_) {
     return;
   }
-  codec_->goAway();
-  go_away_sent_ = true;
-  doConnectionClose(Network::ConnectionCloseType::FlushWriteAndDelay, absl::nullopt,
-                    "forced_goaway");
+
+  // Use graceful drain sequence if graceful shutdown is requested.
+  // startDrainSequence() works for both HTTP/1 and HTTP/2:
+  // - HTTP/1: shutdownNotice() + goAway() provides graceful close
+  // - HTTP/2: shutdownNotice() sends GOAWAY with high stream ID, then goAway() sends final GOAWAY
+  if (graceful) {
+    if (drain_state_ == DrainState::NotDraining) {
+      startDrainSequence();
+    }
+    // Consider the "go away" process started once draining begins.
+    // The actual GOAWAY frame will be sent in onDrainTimeout(), but we want to
+    // prevent multiple calls to sendGoAwayAndClose() from starting multiple drain sequences.
+    go_away_sent_ = true;
+  } else {
+    // Immediate close - send GOAWAY and close immediately
+    codec_->shutdownNotice();
+    codec_->goAway();
+    go_away_sent_ = true;
+    doConnectionClose(Network::ConnectionCloseType::FlushWriteAndDelay, absl::nullopt,
+                      "forced_goaway");
+  }
 }
 
 void ConnectionManagerImpl::chargeTracingStats(const Tracing::Reason& tracing_reason,

source/common/http/conn_manager_impl.h

@@ -191,7 +191,9 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
       return filter_manager_.sendLocalReply(code, body, modify_headers, grpc_status, details);
     }
 
-    void sendGoAwayAndClose() override { return connection_manager_.sendGoAwayAndClose(); }
+    void sendGoAwayAndClose(bool graceful = false) override {
+      return connection_manager_.sendGoAwayAndClose(graceful);
+    }
 
     AccessLog::InstanceSharedPtrVector accessLogHandlers() override {
       const AccessLog::InstanceSharedPtrVector& config_log_handlers =
@@ -601,7 +603,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
   void doConnectionClose(absl::optional<Network::ConnectionCloseType> close_type,
                          absl::optional<StreamInfo::CoreResponseFlag> response_flag,
                          absl::string_view details);
-  void sendGoAwayAndClose();
+  void sendGoAwayAndClose(bool graceful = false);
 
   // Returns true if a RST_STREAM for the given stream is premature. Premature
   // means the RST_STREAM arrived before response headers were sent and than

source/common/http/filter_manager.cc

@@ -487,7 +487,9 @@ void ActiveStreamDecoderFilter::sendLocalReply(
   ActiveStreamFilterBase::sendLocalReply(code, body, modify_headers, grpc_status, details);
 }
 
-void ActiveStreamDecoderFilter::sendGoAwayAndClose() { parent_.sendGoAwayAndClose(); }
+void ActiveStreamDecoderFilter::sendGoAwayAndClose(bool graceful) {
+  parent_.sendGoAwayAndClose(graceful);
+}
 
 void ActiveStreamDecoderFilter::encode1xxHeaders(ResponseHeaderMapPtr&& headers) {
   // If Envoy is not configured to proxy 100-Continue responses, swallow the 100 Continue

source/common/http/filter_manager.h

@@ -300,7 +300,7 @@ struct ActiveStreamDecoderFilter : public ActiveStreamFilterBase,
   void setUpstreamOverrideHost(Upstream::LoadBalancerContext::OverrideHost) override;
   absl::optional<Upstream::LoadBalancerContext::OverrideHost> upstreamOverrideHost() const override;
   bool shouldLoadShed() const override;
-  void sendGoAwayAndClose() override;
+  void sendGoAwayAndClose(bool graceful = false) override;
 
   // Each decoder filter instance checks if the request passed to the filter is gRPC
   // so that we can issue gRPC local responses to gRPC requests. Filter's decodeHeaders()
@@ -489,7 +489,7 @@ class FilterManagerCallbacks {
   /**
    * Attempt to send GOAWAY and close the connection.
    */
-  virtual void sendGoAwayAndClose() PURE;
+  virtual void sendGoAwayAndClose(bool graceful = false) PURE;
 
   /**
    * Called when the stream write buffer is no longer above the low watermark.
@@ -906,14 +906,14 @@ class FilterManager : public ScopeTrackedObject,
 
   virtual bool shouldLoadShed() { return false; };
 
-  void sendGoAwayAndClose() {
+  void sendGoAwayAndClose(bool graceful = false) {
     // Stop filter chain iteration by checking encoder or decoder chain.
     if (state_.filter_call_state_ & FilterCallState::IsDecodingMask) {
       state_.decoder_filter_chain_aborted_ = true;
     } else if (state_.filter_call_state_ & FilterCallState::IsEncodingMask) {
       state_.encoder_filter_chain_aborted_ = true;
     }
-    filter_manager_callbacks_.sendGoAwayAndClose();
+    filter_manager_callbacks_.sendGoAwayAndClose(graceful);
   }
 
 protected:

source/common/router/upstream_request.h

@@ -344,7 +344,7 @@ class UpstreamRequestFilterManagerCallbacks : public Http::FilterManagerCallback
   void disarmRequestTimeout() override {}
   void resetIdleTimer() override {}
   void onLocalReply(Http::Code) override {}
-  void sendGoAwayAndClose() override {}
+  void sendGoAwayAndClose(bool graceful [[maybe_unused]] = false) override {}
   // Upgrade filter chains not supported.
   const Router::RouteEntry::UpgradeMap* upgradeMap() override { return nullptr; }
 

source/common/tcp_proxy/tcp_proxy.h

@@ -563,7 +563,7 @@ class Filter : public Network::ReadFilter,
                         std::function<void(Http::ResponseHeaderMap& headers)>,
                         const absl::optional<Grpc::Status::GrpcStatus>,
                         absl::string_view) override {}
-    void sendGoAwayAndClose() override {}
+    void sendGoAwayAndClose(bool graceful [[maybe_unused]] = false) override {}
     void encode1xxHeaders(Http::ResponseHeaderMapPtr&&) override {}
     Http::ResponseHeaderMapOptRef informationalHeaders() override { return {}; }
     void encodeHeaders(Http::ResponseHeaderMapPtr&&, bool, absl::string_view) override {}

test/common/http/conn_manager_impl_test.cc

@@ -4933,5 +4933,43 @@ TEST_F(HttpConnectionManagerImplTest, ShouldDrainConnectionUponCompletionHttp11)
   response_encoder_.stream_.codec_callbacks_->onCodecEncodeComplete();
 }
 
+// Test graceful shutdown behavior with sendGoAwayAndClose(true)
+TEST_F(HttpConnectionManagerImplTest, SendGoAwayAndCloseGraceful) {
+  setup();
+  // Mock codec to return HTTP/2 protocol for graceful shutdown support.
+  EXPECT_CALL(*codec_, protocol()).WillRepeatedly(Return(Protocol::Http2));
+  MockStreamDecoderFilter* filter = new NiceMock<MockStreamDecoderFilter>();
+  EXPECT_CALL(filter_factory_, createFilterChain(_))
+      .WillOnce(Invoke([&](FilterChainManager& manager) -> bool {
+        auto factory = createDecoderFilterFactoryCb(StreamDecoderFilterSharedPtr{filter});
+        manager.applyFilterFactoryCb({}, factory);
+        return true;
+      }));
+  EXPECT_CALL(*filter, decodeHeaders(_, true))
+      .WillOnce(Invoke([&](RequestHeaderMap&, bool) -> FilterHeadersStatus {
+        // Trigger graceful shutdown during request processing
+        filter->callbacks_->sendGoAwayAndClose(true);
+        return FilterHeadersStatus::StopIteration;
+      }));
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> Http::Status {
+    decoder_ = &conn_manager_->newStream(response_encoder_);
+    RequestHeaderMapPtr headers{
+        new TestRequestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
+    decoder_->decodeHeaders(std::move(headers), true);
+    return Http::okStatus();
+  }));
+  // Expect graceful shutdown to start drain timer and send shutdown notice
+  Event::MockTimer* drain_timer = setUpTimer();
+  EXPECT_CALL(*drain_timer, enableTimer(_, _));
+  EXPECT_CALL(*codec_, shutdownNotice());
+  Buffer::OwnedImpl fake_input;
+  conn_manager_->onData(fake_input, false);
+
+  // Complete the existing stream
+  ResponseHeaderMapPtr response_headers{new TestResponseHeaderMapImpl{{":status", "200"}}};
+  filter->callbacks_->streamInfo().setResponseCodeDetails("");
+  filter->callbacks_->encodeHeaders(std::move(response_headers), true, "details");
+}
+
 } // namespace Http
 } // namespace Envoy