http: unified the filter chain creation in the FilterManager

source/common/http/conn_manager_impl.cc

@@ -1399,8 +1399,13 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(RequestHeaderMapSharedPt
   }
 
   filter_manager_.streamInfo().setRequestHeaders(*request_headers_);
-
-  const bool upgrade_rejected = filter_manager_.createFilterChain() == false;
+  const FilterManager::CreateChainResult create_chain_result =
+      filter_manager_.createDownstreamFilterChain();
+  if (create_chain_result.upgradeAccepted()) {
+    connection_manager_.stats_.named_.downstream_cx_upgrades_total_.inc();
+    connection_manager_.stats_.named_.downstream_cx_upgrades_active_.inc();
+    state_.successful_upgrade_ = true;
+  }
 
   if (connection_manager_.config_->flushAccessLogOnNewRequest()) {
     log(AccessLog::AccessLogType::DownstreamStart);
@@ -1410,7 +1415,7 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(RequestHeaderMapSharedPt
   // should return 404. The current returns no response if there is no router filter.
   if (hasCachedRoute()) {
     // Do not allow upgrades if the route does not support it.
-    if (upgrade_rejected) {
+    if (create_chain_result.upgradeRejected()) {
       // While downstream servers should not send upgrade payload without the upgrade being
       // accepted, err on the side of caution and refuse to process any further requests on this
       // connection, to avoid a class of HTTP/1.1 smuggling bugs where Upgrade or CONNECT payload

source/common/http/conn_manager_impl.h

@@ -286,11 +286,6 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     }
     void onDecoderFilterBelowWriteBufferLowWatermark() override;
     void onDecoderFilterAboveWriteBufferHighWatermark() override;
-    void upgradeFilterChainCreated() override {
-      connection_manager_.stats_.named_.downstream_cx_upgrades_total_.inc();
-      connection_manager_.stats_.named_.downstream_cx_upgrades_active_.inc();
-      state_.successful_upgrade_ = true;
-    }
     void disarmRequestTimeout() override;
     void resetIdleTimer() override;
     void recreateStream(StreamInfo::FilterStateSharedPtr filter_state) override;

source/common/http/filter_manager.cc

@@ -1042,7 +1042,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;
@@ -1083,6 +1083,10 @@ void DownstreamFilterManager::executeLocalReplyIfPrepared() {
   Utility::encodeLocalReply(state_.destroyed_, std::move(prepared_local_reply_));
 }
 
+FilterManager::CreateChainResult DownstreamFilterManager::createDownstreamFilterChain() {
+  return createFilterChain(filter_chain_factory_, false);
+}
+
 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,
@@ -1092,7 +1096,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_,
@@ -1629,11 +1633,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::UpgradeResult
+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();
@@ -1644,28 +1646,49 @@ bool FilterManager::createFilterChain() {
     }
   }
 
+  if (upgrade == nullptr) {
+    // No upgrade header, no upgrade filter chain.
+    return UpgradeResult::UpgradeUnneeded;
+  }
+
+  const Router::RouteEntry::UpgradeMap* upgrade_map = filter_manager_callbacks_.upgradeMap();
+  return filter_chain_factory.createUpgradeFilterChain(upgrade->value().getStringView(),
+                                                       upgrade_map, *this, options)
+             ? UpgradeResult::UpgradeAccepted
+             : UpgradeResult::UpgradeRejected;
+}
+
+FilterManager::CreateChainResult
+FilterManager::createFilterChain(const FilterChainFactory& filter_chain_factory,
+                                 bool only_create_if_configured) {
+  if (state_.create_chain_result_.created()) {
+    return state_.create_chain_result_;
+  }
+
+  OptRef<DownstreamStreamFilterCallbacks> downstream_callbacks =
+      filter_manager_callbacks_.downstreamCallbacks();
+
   // 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);
+  FilterChainOptionsImpl options(downstream_callbacks.has_value() ? streamInfo().route() : nullptr);
 
-  state_.created_filter_chain_ = true;
-  if (upgrade != nullptr) {
-    const Router::RouteEntry::UpgradeMap* upgrade_map = filter_manager_callbacks_.upgradeMap();
+  UpgradeResult upgrade = UpgradeResult::UpgradeUnneeded;
 
-    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.
+  // Only try the upgrade filter chain for downstream filter chains.
+  if (downstream_callbacks.has_value()) {
+    upgrade = createUpgradeFilterChain(filter_chain_factory, options);
+    if (upgrade == UpgradeResult::UpgradeAccepted) {
+      // Upgrade filter chain is created. Return the result directly.
+      state_.create_chain_result_ = CreateChainResult(true, upgrade);
+      return state_.create_chain_result_;
     }
+    // If the upgrade is unnecessary or the upgrade filter chain is rejected, fall through to
+    // create the default filter chain.
   }
 
-  filter_chain_factory_.createFilterChain(*this, false, options);
-  return !upgrade_rejected;
+  state_.create_chain_result_ = CreateChainResult(
+      filter_chain_factory.createFilterChain(*this, only_create_if_configured, options), upgrade);
+  return state_.create_chain_result_;
 }
 
 void ActiveStreamDecoderFilter::requestDataDrained() {
@@ -1720,11 +1743,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

@@ -465,11 +465,6 @@ class FilterManagerCallbacks {
    */
   virtual void onDecoderFilterAboveWriteBufferHighWatermark() PURE;
 
-  /**
-   * Called when the FilterManager creates an Upgrade filter chain.
-   */
-  virtual void upgradeFilterChainCreated() PURE;
-
   /**
    * Called when request activity indicates that the request timeout should be disarmed.
    */
@@ -648,11 +643,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_);
@@ -839,15 +833,56 @@ class FilterManager : public ScopeTrackedObject,
    * a local reply without the overhead of creating and traversing the filters.
    */
   void skipFilterChainCreation() {
-    ASSERT(!state_.created_filter_chain_);
-    state_.created_filter_chain_ = true;
+    ASSERT(!state_.create_chain_result_.created());
+    state_.create_chain_result_ = CreateChainResult(true);
   }
 
   virtual StreamInfo::StreamInfo& streamInfo() PURE;
   virtual const StreamInfo::StreamInfo& streamInfo() const PURE;
 
-  // Set up the Encoder/Decoder filter chain.
-  bool createFilterChain();
+  enum class UpgradeResult : uint8_t { UpgradeUnneeded, UpgradeAccepted, UpgradeRejected };
+
+  /**
+   * Filter chain creation result.
+   */
+  class CreateChainResult {
+  public:
+    CreateChainResult() = default;
+
+    /**
+     * @param created whether the filter chain was created.
+     * @param upgrade the upgrade result.
+     */
+    CreateChainResult(bool created, UpgradeResult upgrade = UpgradeResult::UpgradeUnneeded)
+        : created_(created), upgrade_(upgrade) {}
+
+    /**
+     * @return whether the filter chain was created.
+     */
+    bool created() const { return created_; }
+    /**
+     * @return whether the upgrade was accepted.
+     */
+    bool upgradeAccepted() const { return upgrade_ == UpgradeResult::UpgradeAccepted; }
+    /**
+     * @return whether the upgrade was rejected.
+     */
+    bool upgradeRejected() const { return upgrade_ == UpgradeResult::UpgradeRejected; }
+
+  private:
+    bool created_ = false;
+    UpgradeResult upgrade_ = UpgradeResult::UpgradeUnneeded;
+  };
+
+  /**
+   * Set up the Encoder/Decoder filter chain.
+   * @param filter_chain_factory the factory to create the 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.
+   *
+   */
+  CreateChainResult createFilterChain(const FilterChainFactory& filter_chain_factory,
+                                      bool only_create_if_configured);
 
   OptRef<const Network::Connection> connection() const { return connection_; }
 
@@ -900,7 +935,6 @@ class FilterManager : public ScopeTrackedObject,
     // By default, we will assume there are no 1xx. If encode1xxHeaders
     // is ever called, this is set to true so commonContinue resumes processing the 1xx.
     bool has_1xx_headers_{};
-    bool created_filter_chain_{};
     // These two are latched on initial header read, to determine if the original headers
     // constituted a HEAD or gRPC request, respectively.
     bool is_head_request_{};
@@ -923,6 +957,9 @@ class FilterManager : public ScopeTrackedObject,
     bool decoder_filters_streaming_{true};
     bool destroyed_{false};
 
+    // Result of filter chain creation.
+    CreateChainResult create_chain_result_{};
+
     // Used to track which filter is the latest filter that has received data.
     ActiveStreamEncoderFilter* latest_data_encoding_filter_{};
     ActiveStreamDecoderFilter* latest_data_decoding_filter_{};
@@ -985,6 +1022,9 @@ class FilterManager : public ScopeTrackedObject,
   // Indicates which filter to start the iteration with.
   enum class FilterIterationStartState { AlwaysStartFromNext, CanStartFromCurrent };
 
+  UpgradeResult 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,
@@ -1072,7 +1112,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;
@@ -1127,11 +1166,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(
@@ -1155,6 +1194,8 @@ class DownstreamFilterManager : public FilterManager {
     stream_info_.setDownstreamRemoteAddress(downstream_remote_address);
   }
 
+  CreateChainResult createDownstreamFilterChain();
+
   /**
    * Called before local reply is made by the filter manager.
    * @param data the data associated with the local reply.
@@ -1231,6 +1272,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,21 @@ 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(),
+                                         /*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).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).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.

source/common/router/upstream_request.h

@@ -346,9 +346,6 @@ class UpstreamRequestFilterManagerCallbacks : public Http::FilterManagerCallback
   void recreateStream(StreamInfo::FilterStateSharedPtr) override {
     IS_ENVOY_BUG("recreateStream called from upstream HTTP filter");
   }
-  void upgradeFilterChainCreated() override {
-    IS_ENVOY_BUG("upgradeFilterChainCreated called from upstream HTTP filter");
-  }
   OptRef<UpstreamStreamFilterCallbacks> upstreamCallbacks() override { return {*this}; }
 
   // Http::UpstreamStreamFilterCallbacks