http: unified the filter chain creation in the FilterManager

source/common/http/conn_manager_impl.cc

@@ -1388,7 +1388,7 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(RequestHeaderMapSharedPt
 
   filter_manager_.streamInfo().setRequestHeaders(*request_headers_);
 
-  const bool upgrade_rejected = filter_manager_.createFilterChain() == false;
+  const bool upgrade_rejected = filter_manager_.createDownstreamFilterChain();
 
   if (connection_manager_.config_->flushAccessLogOnNewRequest()) {
     log(AccessLog::AccessLogType::DownstreamStart);

source/common/http/filter_manager.cc

@@ -1036,7 +1036,7 @@ void DownstreamFilterManager::prepareLocalReplyViaFilterChain(
   // For early error handling, do a best-effort attempt to create a filter chain
   // to ensure access logging. If the filter chain already exists this will be
   // a no-op.
-  createFilterChain();
+  createDownstreamFilterChain();
 
   if (prepared_local_reply_) {
     return;
@@ -1077,6 +1077,11 @@ void DownstreamFilterManager::executeLocalReplyIfPrepared() {
   Utility::encodeLocalReply(state_.destroyed_, std::move(prepared_local_reply_));
 }
 
+bool DownstreamFilterManager::createDownstreamFilterChain() {
+  return createFilterChain(filter_chain_factory_, /*allow_upgrade_filter_chain=*/true, false)
+      .upgrade_rejected;
+}
+
 void DownstreamFilterManager::sendLocalReplyViaFilterChain(
     bool is_grpc_request, Code code, absl::string_view body,
     const std::function<void(ResponseHeaderMap& headers)>& modify_headers, bool is_head_request,
@@ -1086,7 +1091,7 @@ void DownstreamFilterManager::sendLocalReplyViaFilterChain(
   // For early error handling, do a best-effort attempt to create a filter chain
   // to ensure access logging. If the filter chain already exists this will be
   // a no-op.
-  createFilterChain();
+  createDownstreamFilterChain();
 
   Utility::sendLocalReply(
       state_.destroyed_,
@@ -1623,11 +1628,9 @@ void FilterManager::contextOnContinue(ScopeTrackedObjectStack& tracked_object_st
   tracked_object_stack.add(filter_manager_callbacks_.scope());
 }
 
-bool FilterManager::createFilterChain() {
-  if (state_.created_filter_chain_) {
-    return false;
-  }
-  bool upgrade_rejected = false;
+FilterManager::CreateFilterChainResult
+FilterManager::createUpgradeFilterChain(const FilterChainFactory& filter_chain_factory,
+                                        const FilterChainOptionsImpl& options) {
   const HeaderEntry* upgrade = nullptr;
   if (filter_manager_callbacks_.requestHeaders()) {
     upgrade = filter_manager_callbacks_.requestHeaders()->Upgrade();
@@ -1638,28 +1641,55 @@ bool FilterManager::createFilterChain() {
     }
   }
 
+  if (upgrade == nullptr) {
+    // No upgrade header, no upgrade filter chain.
+    return {false, false};
+  }
+
+  const Router::RouteEntry::UpgradeMap* upgrade_map = filter_manager_callbacks_.upgradeMap();
+  if (filter_chain_factory.createUpgradeFilterChain(upgrade->value().getStringView(), upgrade_map,
+                                                    *this, options)) {
+    filter_manager_callbacks_.upgradeFilterChainCreated();
+    // The upgrade filter chain is created successfully.
+    return {true, false};
+  }
+
+  // The upgrade filter chain is rejected. Fall through to the default filter chain.
+  // The default filter chain will be used to handle the upgrade failure local reply.
+  return {false, true};
+}
+
+FilterManager::CreateFilterChainResult
+FilterManager::createFilterChain(const FilterChainFactory& filter_chain_factory,
+                                 bool allow_upgrade_filter_chain, bool only_create_if_configured) {
+  if (state_.created_filter_chain_) {
+    return {true, false};
+  }
+
   // This filter chain options is only used for the downstream HTTP filter chains for now. So, try
   // to set valid initial route only when the downstream callbacks is available.
   FilterChainOptionsImpl options(
       filter_manager_callbacks_.downstreamCallbacks().has_value() ? streamInfo().route() : nullptr);
 
-  state_.created_filter_chain_ = true;
-  if (upgrade != nullptr) {
-    const Router::RouteEntry::UpgradeMap* upgrade_map = filter_manager_callbacks_.upgradeMap();
+  bool upgrade_rejected = false;
 
-    if (filter_chain_factory_.createUpgradeFilterChain(upgrade->value().getStringView(),
-                                                       upgrade_map, *this, options)) {
-      filter_manager_callbacks_.upgradeFilterChainCreated();
-      return true;
-    } else {
-      upgrade_rejected = true;
-      // Fall through to the default filter chain. The function calling this
-      // will send a local reply indicating that the upgrade failed.
+  if (allow_upgrade_filter_chain) {
+    const auto upgrade_result = createUpgradeFilterChain(filter_chain_factory, options);
+    if (upgrade_result.created) {
+      ASSERT(!upgrade_result.upgrade_rejected);
+      state_.created_filter_chain_ = true;
+      return upgrade_result;
     }
+    // The upgrade filter chain may be unnecessary or be rejected. Fall through to the default
+    // filter chain anyway.
+    upgrade_rejected = upgrade_result.upgrade_rejected;
   }
 
-  filter_chain_factory_.createFilterChain(*this, false, options);
-  return !upgrade_rejected;
+  const bool created =
+      filter_chain_factory.createFilterChain(*this, only_create_if_configured, options);
+  state_.created_filter_chain_ = created;
+
+  return {created, upgrade_rejected};
 }
 
 void ActiveStreamDecoderFilter::requestDataDrained() {
@@ -1714,11 +1744,11 @@ bool ActiveStreamDecoderFilter::recreateStream(const ResponseHeaderMap* headers)
 
   if (headers != nullptr) {
     // The call to setResponseHeaders is needed to ensure that the headers are properly logged in
-    // access logs before the stream is destroyed. Since the function expects a ResponseHeaderPtr&&,
-    // ownership of the headers must be passed. This cannot happen earlier in the flow (such as in
-    // the call to setupRedirect) because at that point it is still possible for the headers to be
-    // used in a different logical branch. We work around this by creating a copy and passing
-    // ownership of the copy instead.
+    // access logs before the stream is destroyed. Since the function expects a
+    // ResponseHeaderPtr&&, ownership of the headers must be passed. This cannot happen earlier in
+    // the flow (such as in the call to setupRedirect) because at that point it is still possible
+    // for the headers to be used in a different logical branch. We work around this by creating a
+    // copy and passing ownership of the copy instead.
     ResponseHeaderMapPtr headers_copy = createHeaderMap<ResponseHeaderMapImpl>(*headers);
     parent_.filter_manager_callbacks_.setResponseHeaders(std::move(headers_copy));
     parent_.filter_manager_callbacks_.chargeStats(*headers);

source/common/http/filter_manager.h

@@ -639,11 +639,10 @@ class FilterManager : public ScopeTrackedObject,
   FilterManager(FilterManagerCallbacks& filter_manager_callbacks, Event::Dispatcher& dispatcher,
                 OptRef<const Network::Connection> connection, uint64_t stream_id,
                 Buffer::BufferMemoryAccountSharedPtr account, bool proxy_100_continue,
-                uint32_t buffer_limit, const FilterChainFactory& filter_chain_factory)
+                uint32_t buffer_limit)
       : filter_manager_callbacks_(filter_manager_callbacks), dispatcher_(dispatcher),
         connection_(connection), stream_id_(stream_id), account_(std::move(account)),
-        proxy_100_continue_(proxy_100_continue), buffer_limit_(buffer_limit),
-        filter_chain_factory_(filter_chain_factory) {}
+        proxy_100_continue_(proxy_100_continue), buffer_limit_(buffer_limit) {}
 
   ~FilterManager() override {
     ASSERT(state_.destroyed_);
@@ -837,8 +836,23 @@ class FilterManager : public ScopeTrackedObject,
   virtual StreamInfo::StreamInfo& streamInfo() PURE;
   virtual const StreamInfo::StreamInfo& streamInfo() const PURE;
 
-  // Set up the Encoder/Decoder filter chain.
-  bool createFilterChain();
+  struct CreateFilterChainResult {
+    bool created{};
+    bool upgrade_rejected{};
+  };
+
+  /**
+   * Set up the Encoder/Decoder filter chain.
+   * @param filter_chain_factory the factory to create the filter chain.
+   * @param allow_upgrade_filter_chain whether to allow the creation of an upgrade filter chain.
+   *        This only should be set to true for downstream HTTP filter chain.
+   * @param only_create_if_configured whether to only create the filter chain if it is configured
+   *        explicitly. This only makes sense for upstream HTTP filter chain.
+   *
+   */
+  CreateFilterChainResult createFilterChain(const FilterChainFactory& filter_chain_factory,
+                                            bool allow_upgrade_filter_chain,
+                                            bool only_create_if_configured);
 
   OptRef<const Network::Connection> connection() const { return connection_; }
 
@@ -966,6 +980,9 @@ class FilterManager : public ScopeTrackedObject,
   // Indicates which filter to start the iteration with.
   enum class FilterIterationStartState { AlwaysStartFromNext, CanStartFromCurrent };
 
+  CreateFilterChainResult createUpgradeFilterChain(const FilterChainFactory& filter_chain_factory,
+                                                   const FilterChainOptionsImpl& options);
+
   // Returns the encoder filter to start iteration with.
   std::list<ActiveStreamEncoderFilterPtr>::iterator
   commonEncodePrefix(ActiveStreamEncoderFilter* filter, bool end_stream,
@@ -1053,7 +1070,6 @@ class FilterManager : public ScopeTrackedObject,
       std::make_shared<Network::Socket::Options>();
   absl::optional<Upstream::LoadBalancerContext::OverrideHost> upstream_override_host_;
 
-  const FilterChainFactory& filter_chain_factory_;
   // TODO(snowp): Once FM has been moved to its own file we'll make these private classes of FM,
   // at which point they no longer need to be friends.
   friend ActiveStreamFilterBase;
@@ -1108,11 +1124,11 @@ class DownstreamFilterManager : public FilterManager {
                           StreamInfo::FilterStateSharedPtr parent_filter_state,
                           Server::OverloadManager& overload_manager)
       : FilterManager(filter_manager_callbacks, dispatcher, connection, stream_id, account,
-                      proxy_100_continue, buffer_limit, filter_chain_factory),
+                      proxy_100_continue, buffer_limit),
         stream_info_(protocol, time_source, connection.connectionInfoProviderSharedPtr(),
                      StreamInfo::FilterState::LifeSpan::FilterChain,
                      std::move(parent_filter_state)),
-        local_reply_(local_reply),
+        local_reply_(local_reply), filter_chain_factory_(filter_chain_factory),
         downstream_filter_load_shed_point_(overload_manager.getLoadShedPoint(
             Server::LoadShedPointName::get().HttpDownstreamFilterCheck)),
         use_filter_manager_state_for_downstream_end_stream_(Runtime::runtimeFeatureEnabled(
@@ -1136,6 +1152,8 @@ class DownstreamFilterManager : public FilterManager {
     stream_info_.setDownstreamRemoteAddress(downstream_remote_address);
   }
 
+  bool createDownstreamFilterChain();
+
   /**
    * Called before local reply is made by the filter manager.
    * @param data the data associated with the local reply.
@@ -1212,6 +1230,7 @@ class DownstreamFilterManager : public FilterManager {
 private:
   OverridableRemoteConnectionInfoSetterStreamInfo stream_info_;
   const LocalReply::LocalReply& local_reply_;
+  const FilterChainFactory& filter_chain_factory_;
   Utility::PreparedLocalReplyPtr prepared_local_reply_{nullptr};
   Server::LoadShedPoint* downstream_filter_load_shed_point_{nullptr};
   // Set by the envoy.reloadable_features.use_filter_manager_state_for_downstream_end_stream runtime

source/common/router/router.h

@@ -199,7 +199,7 @@ class FilterUtility {
 /**
  * Configuration for the router filter.
  */
-class FilterConfig : Http::FilterChainFactory {
+class FilterConfig : public Http::FilterChainFactory {
 public:
   FilterConfig(Server::Configuration::CommonFactoryContext& factory_context,
                Stats::StatName stat_prefix, const LocalInfo::LocalInfo& local_info,

source/common/router/upstream_request.cc

@@ -47,11 +47,9 @@ class UpstreamFilterManager : public Http::FilterManager {
   UpstreamFilterManager(Http::FilterManagerCallbacks& filter_manager_callbacks,
                         Event::Dispatcher& dispatcher, OptRef<const Network::Connection> connection,
                         uint64_t stream_id, Buffer::BufferMemoryAccountSharedPtr account,
-                        bool proxy_100_continue, uint32_t buffer_limit,
-                        const Http::FilterChainFactory& filter_chain_factory,
-                        UpstreamRequest& request)
+                        bool proxy_100_continue, uint32_t buffer_limit, UpstreamRequest& request)
       : FilterManager(filter_manager_callbacks, dispatcher, connection, stream_id, account,
-                      proxy_100_continue, buffer_limit, filter_chain_factory),
+                      proxy_100_continue, buffer_limit),
         upstream_request_(request) {}
 
   StreamInfo::StreamInfo& streamInfo() override {
@@ -142,18 +140,20 @@ UpstreamRequest::UpstreamRequest(RouterFilterInterface& parent,
   filter_manager_ = std::make_unique<UpstreamFilterManager>(
       *filter_manager_callbacks_, parent_.callbacks()->dispatcher(), UpstreamRequest::connection(),
       parent_.callbacks()->streamId(), parent_.callbacks()->account(), true,
-      parent_.callbacks()->decoderBufferLimit(), *parent_.cluster(), *this);
+      parent_.callbacks()->decoderBufferLimit(), *this);
   // Attempt to create custom cluster-specified filter chain
-  bool created = parent_.cluster()->createFilterChain(*filter_manager_,
-                                                      /*only_create_if_configured=*/true);
+  bool created = filter_manager_
+                     ->createFilterChain(*parent_.cluster(), false,
+                                         /*only_create_if_configured=*/true)
+                     .created;
   if (!created) {
     // Attempt to create custom router-specified filter chain.
-    created = parent_.config().createFilterChain(*filter_manager_);
+    created = filter_manager_->createFilterChain(parent_.config(), false, false).created;
   }
   if (!created) {
     // Neither cluster nor router have a custom filter chain; add the default
     // cluster filter chain, which only consists of the codec filter.
-    created = parent_.cluster()->createFilterChain(*filter_manager_, false);
+    created = filter_manager_->createFilterChain(*parent_.cluster(), false, false).created;
   }
   // There will always be a codec filter present, which sets the upstream
   // interface. Fast-fail any tests that don't set up mocks correctly.