http: allow trigger GOAWAY from l7 filters (#36306)

Risk Level: low
Testing: integration test

---------

Signed-off-by: Boteng Yao <[email protected]>

envoy/http/filter.h

@@ -644,6 +644,11 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
                               const absl::optional<Grpc::Status::GrpcStatus> grpc_status,
                               absl::string_view details) PURE;
 
+  /**
+   * Attempt to send GOAWAY and close the connection, and no filter chain will move forward.
+   */
+  virtual void sendGoAwayAndClose() PURE;
+
   /**
    * Adds decoded metadata. This function can only be called in
    * StreamDecoderFilter::decodeHeaders/Data/Trailers(). Do not call in

source/common/http/async_client_impl.h

@@ -235,6 +235,7 @@ class AsyncStreamImpl : public virtual AsyncClient::Stream,
       watermark_callbacks_->get().removeDownstreamWatermarkCallbacks(callbacks);
     }
   }
+  void sendGoAwayAndClose() override {}
 
   void setDecoderBufferLimit(uint32_t) override {
     IS_ENVOY_BUG("decoder buffer limits should not be overridden on async streams.");

source/common/http/conn_manager_impl.cc

@@ -757,6 +757,18 @@ void ConnectionManagerImpl::onDrainTimeout() {
   checkForDeferredClose(false);
 }
 
+void ConnectionManagerImpl::sendGoAwayAndClose() {
+  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");
+}
+
 void ConnectionManagerImpl::chargeTracingStats(const Tracing::Reason& tracing_reason,
                                                ConnectionManagerTracingStats& tracing_stats) {
   switch (tracing_reason) {

source/common/http/conn_manager_impl.h

@@ -189,6 +189,9 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
                         absl::string_view details) override {
       return filter_manager_.sendLocalReply(code, body, modify_headers, grpc_status, details);
     }
+
+    void sendGoAwayAndClose() override { return connection_manager_.sendGoAwayAndClose(); }
+
     AccessLog::InstanceSharedPtrVector accessLogHandlers() override {
       const AccessLog::InstanceSharedPtrVector& config_log_handlers =
           connection_manager_.config_->accessLogs();
@@ -597,6 +600,8 @@ 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();
+
   // 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
   // the stream was alive for short period of time. This period is specified
@@ -646,6 +651,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
   const Server::OverloadActionState& overload_stop_accepting_requests_ref_;
   const Server::OverloadActionState& overload_disable_keepalive_ref_;
   TimeSource& time_source_;
+  bool go_away_sent_{false};
   bool remote_close_{};
   // Hop by hop headers should always be cleared for Envoy-as-a-proxy but will
   // not be for Envoy-mobile.

source/common/http/filter_manager.cc

@@ -453,6 +453,8 @@ void ActiveStreamDecoderFilter::sendLocalReply(
   ActiveStreamFilterBase::sendLocalReply(code, body, modify_headers, grpc_status, details);
 }
 
+void ActiveStreamDecoderFilter::sendGoAwayAndClose() { parent_.sendGoAwayAndClose(); }
+
 void ActiveStreamDecoderFilter::encode1xxHeaders(ResponseHeaderMapPtr&& headers) {
   // If Envoy is not configured to proxy 100-Continue responses, swallow the 100 Continue
   // here. This avoids the potential situation where Envoy strips Expect: 100-Continue and sends a
@@ -868,7 +870,7 @@ void FilterManager::decodeMetadata(ActiveStreamDecoderFilter* filter, MetadataMa
                      metadata_map);
     if (state_.decoder_filter_chain_aborted_) {
       // If the decoder filter chain has been aborted, then either:
-      // 1. This filter has sent a local reply from decode metadata.
+      // 1. This filter has sent a local reply or GoAway from decode metadata.
       // 2. This filter is the terminal http filter, and an upstream HTTP filter has sent a local
       // reply.
       ASSERT((status == FilterMetadataStatus::StopIterationForLocalReply) ||

source/common/http/filter_manager.h

@@ -262,6 +262,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;
 
   // 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()
@@ -449,6 +450,11 @@ class FilterManagerCallbacks {
    */
   virtual void endStream() PURE;
 
+  /**
+   * Attempt to send GOAWAY and close the connection.
+   */
+  virtual void sendGoAwayAndClose() PURE;
+
   /**
    * Called when the stream write buffer is no longer above the low watermark.
    */
@@ -857,6 +863,16 @@ class FilterManager : public ScopeTrackedObject,
 
   virtual bool shouldLoadShed() { return false; };
 
+  void sendGoAwayAndClose() {
+    // 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();
+  }
+
 protected:
   struct State {
     State() = default;

source/common/router/upstream_request.h

@@ -338,6 +338,7 @@ class UpstreamRequestFilterManagerCallbacks : public Http::FilterManagerCallback
   void disarmRequestTimeout() override {}
   void resetIdleTimer() override {}
   void onLocalReply(Http::Code) override {}
+  void sendGoAwayAndClose() override {}
   // Upgrade filter chains not supported.
   const Router::RouteEntry::UpgradeMap* upgradeMap() override { return nullptr; }
 

source/common/tcp_proxy/tcp_proxy.h

@@ -520,6 +520,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 encode1xxHeaders(Http::ResponseHeaderMapPtr&&) override {}
     Http::ResponseHeaderMapOptRef informationalHeaders() override { return {}; }
     void encodeHeaders(Http::ResponseHeaderMapPtr&&, bool, absl::string_view) override {}

test/integration/BUILD

@@ -557,6 +557,7 @@ envoy_cc_test(
         "//test/integration/filters:on_local_reply_filter_config_lib",
         "//test/integration/filters:request_metadata_filter_config_lib",
         "//test/integration/filters:response_metadata_filter_config_lib",
+        "//test/integration/filters:send_goaway_filter_lib",
         "//test/integration/filters:set_response_code_filter_config_proto_cc_proto",
         "//test/integration/filters:set_response_code_filter_lib",
         "//test/integration/filters:stop_in_headers_continue_in_data_filter_lib",

test/integration/filters/BUILD

@@ -121,6 +121,21 @@ envoy_cc_test_library(
     ],
 )
 
+envoy_cc_test_library(
+    name = "send_goaway_filter_lib",
+    srcs = [
+        "send_goaway_filter.cc",
+    ],
+    deps = [
+        ":common_lib",
+        "//envoy/http:filter_interface",
+        "//envoy/registry",
+        "//envoy/server:filter_config_interface",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+        "//test/extensions/filters/http/common:empty_http_filter_config_lib",
+    ],
+)
+
 envoy_cc_test_library(
     name = "tee_filter_lib",
     srcs = [

test/integration/filters/send_goaway_filter.cc

@@ -0,0 +1,79 @@
+#include <string>
+
+#include "envoy/http/filter.h"
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+
+#include "source/common/common/enum_to_int.h"
+#include "source/common/http/utility.h"
+#include "source/extensions/filters/http/common/pass_through_filter.h"
+
+#include "test/extensions/filters/http/common/empty_http_filter_config.h"
+#include "test/integration/filters/common.h"
+
+namespace Envoy {
+
+class GoAwayDuringDecoding : public Http::PassThroughFilter {
+public:
+  constexpr static char name[] = "send-goaway-during-decode-filter";
+
+  Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap& request_headers, bool) override {
+    auto local_reply = request_headers.get(Http::LowerCaseString("send-local-reply"));
+    if (!local_reply.empty()) {
+      decoder_callbacks_->sendLocalReply(Http::Code::Gone, "", nullptr, std::nullopt,
+                                         "local_reply_test");
+      return Http::FilterHeadersStatus::StopIteration;
+    }
+
+    auto result = request_headers.get(Http::LowerCaseString("skip-goaway"));
+    if (!result.empty() && result[0]->value() == "true") {
+      go_away_skiped_ = true;
+      return Http::FilterHeadersStatus::Continue;
+    }
+    decoder_callbacks_->sendGoAwayAndClose();
+    result = request_headers.get(Http::LowerCaseString("continue-filter-chain"));
+    if (!result.empty() && result[0]->value() == "true") {
+      return Http::FilterHeadersStatus::Continue;
+    }
+    return Http::FilterHeadersStatus::StopIteration;
+  }
+
+  Http::FilterDataStatus decodeData(Buffer::Instance&, bool) override {
+    return Http::FilterDataStatus::Continue;
+  }
+
+  // Due to the above local reply, this method should never be invoked in tests.
+  Http::FilterMetadataStatus decodeMetadata(Http::MetadataMap&) override {
+    ASSERT(go_away_skiped_);
+    return Http::FilterMetadataStatus::Continue;
+  }
+
+  Http::FilterHeadersStatus encodeHeaders(Http::ResponseHeaderMap& headers, bool) override {
+    if (auto status = Http::Utility::getResponseStatus(headers);
+        status == enumToInt(Http::Code::Gone)) {
+      decoder_callbacks_->sendGoAwayAndClose();
+      return Http::FilterHeadersStatus::Continue;
+    }
+
+    auto result = headers.get(Http::LowerCaseString("send-encoder-goaway"));
+    if (result.empty()) {
+      return Http::FilterHeadersStatus::Continue;
+    }
+    decoder_callbacks_->sendGoAwayAndClose();
+    result = headers.get(Http::LowerCaseString("continue-encoder-filter-chain"));
+    if (!result.empty() && result[0]->value() == "true") {
+      return Http::FilterHeadersStatus::Continue;
+    }
+    return Http::FilterHeadersStatus::StopIteration;
+  }
+
+private:
+  bool go_away_skiped_ = false;
+};
+
+constexpr char GoAwayDuringDecoding::name[];
+static Registry::RegisterFactory<SimpleFilterConfig<GoAwayDuringDecoding>,
+                                 Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+
+} // namespace Envoy