Add API field in Router for Upstream Filters. (envoyproxy#23367)

Upstream filters specified in the router will apply if there are no filters specified in the cluster.

Risk Level: Low.
Testing: Unit and Integration testing.

Signed-off-by: Paul Gallagher <pgal@google.com>

api/envoy/extensions/filters/http/router/v3/BUILD

@@ -7,6 +7,7 @@ licenses(["notice"])  # Apache 2
 api_proto_package(
     deps = [
         "//envoy/config/accesslog/v3:pkg",
+        "//envoy/extensions/filters/network/http_connection_manager/v3:pkg",
         "@com_github_cncf_udpa//udpa/annotations:pkg",
     ],
 )

api/envoy/extensions/filters/http/router/v3/router.proto

@@ -3,6 +3,7 @@ syntax = "proto3";
 package envoy.extensions.filters.http.router.v3;
 
 import "envoy/config/accesslog/v3/accesslog.proto";
+import "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.proto";
 
 import "google/protobuf/wrappers.proto";
 
@@ -20,7 +21,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // Router :ref:`configuration overview <config_http_filters_router>`.
 // [#extension: envoy.filters.http.router]
 
-// [#next-free-field: 8]
+// [#next-free-field: 9]
 message Router {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.filter.http.router.v2.Router";
@@ -89,4 +90,20 @@ message Router {
   // :ref:`gRPC stats filter<config_http_filters_grpc_stats>` documentation
   // for more details.
   bool suppress_grpc_request_failure_code_stats = 7;
+
+  // .. note::
+  //   Upstream HTTP filters are currently in alpha.
+  //
+  // Optional HTTP filters for the upstream filter chain.
+  //
+  // These filters will be applied for all requests that pass through the router.
+  // They will also be applied to shadowed requests.
+  // Upstream filters cannot change route or cluster.
+  // Upstream filters specified on the cluster will override these filters.
+  //
+  // If using upstream filters, please be aware that local errors sent by
+  // upstream filters will not trigger retries, and local errors sent by
+  // upstream filters will count as a final response if hedging is configured.
+  // [#extension-category: envoy.filters.http.upstream]
+  repeated network.http_connection_manager.v3.HttpFilter upstream_http_filters = 8;
 }

envoy/http/filter_factory.h

@@ -67,8 +67,12 @@ class FilterChainFactory {
    * Called when a new HTTP stream is created on the connection.
    * @param manager supplies the "sink" that is used for actually creating the filter chain. @see
    *                FilterChainManager.
+   * @param only_create_if_configured if true, only creates filter chain if there is a non-default
+   *                                  configured filter chain. Default false.
+   * @return whather a filter chain has been created.
    */
-  virtual void createFilterChain(FilterChainManager& manager) const PURE;
+  virtual bool createFilterChain(FilterChainManager& manager,
+                                 bool only_create_if_configured = false) const PURE;
 
   /**
    * Called when a new upgrade stream is created on the connection.

source/common/filter/BUILD

@@ -20,7 +20,6 @@ envoy_cc_library(
         "//envoy/thread_local:thread_local_interface",
         "//source/common/common:containers_lib",
         "//source/common/config:subscription_base_interface",
-        "//source/common/config:subscription_factory_lib",
         "//source/common/config:utility_lib",
         "//source/common/grpc:common_lib",
         "//source/common/init:manager_lib",

source/common/router/BUILD

@@ -292,11 +292,13 @@ envoy_cc_library(
         "//envoy/http:codec_interface",
         "//envoy/http:codes_interface",
         "//envoy/http:conn_pool_interface",
+        "//envoy/http:filter_factory_interface",
         "//envoy/http:filter_interface",
         "//envoy/http:stateful_session_interface",
         "//envoy/local_info:local_info_interface",
         "//envoy/router:shadow_writer_interface",
         "//envoy/runtime:runtime_interface",
+        "//envoy/server:factory_context_interface",
         "//envoy/server:filter_config_interface",
         "//envoy/stats:stats_interface",
         "//envoy/stats:stats_macros",
@@ -317,6 +319,7 @@ envoy_cc_library(
         "//source/common/config:utility_lib",
         "//source/common/grpc:common_lib",
         "//source/common/http:codes_lib",
+        "//source/common/http:filter_chain_helper_lib",
         "//source/common/http:filter_manager_lib",
         "//source/common/http:header_map_lib",
         "//source/common/http:headers_lib",
@@ -330,6 +333,7 @@ envoy_cc_library(
         "//source/common/stream_info:uint32_accessor_lib",
         "//source/common/tracing:http_tracer_lib",
         "//source/common/upstream:load_balancer_lib",
+        "//source/common/upstream:upstream_http_factory_context_lib",
         "//source/extensions/common/proxy_protocol:proxy_protocol_header_lib",
         "//source/extensions/filters/http/common:factory_base_lib",
         "@envoy_api//envoy/extensions/filters/http/router/v3:pkg_cc_proto",

source/common/router/router.h

@@ -12,10 +12,12 @@
 #include "envoy/http/codec.h"
 #include "envoy/http/codes.h"
 #include "envoy/http/filter.h"
+#include "envoy/http/filter_factory.h"
 #include "envoy/http/stateful_session.h"
 #include "envoy/local_info/local_info.h"
 #include "envoy/router/shadow_writer.h"
 #include "envoy/runtime/runtime.h"
+#include "envoy/server/factory_context.h"
 #include "envoy/server/filter_config.h"
 #include "envoy/stats/scope.h"
 #include "envoy/stats/stats_macros.h"
@@ -30,13 +32,15 @@
 #include "source/common/common/logger.h"
 #include "source/common/config/utility.h"
 #include "source/common/config/well_known_names.h"
+#include "source/common/http/filter_chain_helper.h"
 #include "source/common/http/utility.h"
 #include "source/common/router/config_impl.h"
 #include "source/common/router/context_impl.h"
 #include "source/common/router/upstream_request.h"
 #include "source/common/stats/symbol_table.h"
 #include "source/common/stream_info/stream_info_impl.h"
 #include "source/common/upstream/load_balancer_impl.h"
+#include "source/common/upstream/upstream_http_factory_context_impl.h"
 
 namespace Envoy {
 namespace Router {
@@ -195,7 +199,7 @@ class FilterUtility {
 /**
  * Configuration for the router filter.
  */
-class FilterConfig {
+class FilterConfig : Http::FilterChainFactory {
 public:
   FilterConfig(Stats::StatName stat_prefix, const LocalInfo::LocalInfo& local_info,
                Stats::Scope& scope, Upstream::ClusterManager& cm, Runtime::Loader& runtime,
@@ -235,7 +239,43 @@ class FilterConfig {
     for (const auto& upstream_log : config.upstream_log()) {
       upstream_logs_.push_back(AccessLog::AccessLogFactory::fromProto(upstream_log, context));
     }
+    if (config.upstream_http_filters_size() > 0) {
+      auto& server_factory_ctx = context.getServerFactoryContext();
+      const Http::FilterChainUtility::FiltersList& upstream_http_filters =
+          config.upstream_http_filters();
+      std::shared_ptr<Http::UpstreamFilterConfigProviderManager> filter_config_provider_manager =
+          Http::FilterChainUtility::createSingletonUpstreamFilterConfigProviderManager(
+              server_factory_ctx);
+      std::string prefix = context.scope().symbolTable().toString(context.scope().prefix());
+      upstream_ctx_ = std::make_unique<Upstream::UpstreamHttpFactoryContextImpl>(
+          server_factory_ctx, context.initManager(), context.scope());
+      Http::FilterChainHelper<Server::Configuration::UpstreamHttpFactoryContext,
+                              Server::Configuration::UpstreamHttpFilterConfigFactory>
+          helper(*filter_config_provider_manager, server_factory_ctx, *upstream_ctx_, prefix);
+      helper.processFilters(upstream_http_filters, "router upstream http", "router upstream http",
+                            upstream_http_filter_factories_);
+    }
+  }
+
+  bool createFilterChain(Http::FilterChainManager& manager,
+                         bool only_create_if_configured = false) const override {
+    // Currently there is no default filter chain, so only_create_if_configured true doesn't make
+    // sense.
+    ASSERT(!only_create_if_configured);
+    if (upstream_http_filter_factories_.empty()) {
+      return false;
+    }
+    Http::FilterChainUtility::createFilterChainForFactories(manager,
+                                                            upstream_http_filter_factories_);
+    return true;
+  }
+
+  bool createUpgradeFilterChain(absl::string_view, const UpgradeMap*,
+                                Http::FilterChainManager&) const override {
+    // Upgrade filter chains not yet supported for upstream filters.
+    return false;
   }
+
   using HeaderVector = std::vector<Http::LowerCaseString>;
   using HeaderVectorPtr = std::unique_ptr<HeaderVector>;
 
@@ -261,6 +301,8 @@ class FilterConfig {
   Http::Context& http_context_;
   Stats::StatName zone_name_;
   Stats::StatName empty_stat_name_;
+  std::unique_ptr<Server::Configuration::UpstreamHttpFactoryContext> upstream_ctx_;
+  Http::FilterChainUtility::FilterFactoriesList upstream_http_filter_factories_;
 
 private:
   ShadowWriterPtr shadow_writer_;

source/common/router/upstream_request.cc

@@ -125,10 +125,21 @@ UpstreamRequest::UpstreamRequest(RouterFilterInterface& parent,
       parent_.callbacks()->connection(), parent_.callbacks()->streamId(),
       parent_.callbacks()->account(), true, parent_.callbacks()->decoderBufferLimit(),
       *parent_.cluster(), *this);
-  parent_.cluster()->createFilterChain(*filter_manager_);
-  // The cluster will always create a codec filter, which sets the upstream
+  // Attempt to create custom cluster-specified filter chain
+  bool created = parent_.cluster()->createFilterChain(*filter_manager_,
+                                                      /*only_create_if_configured=*/true);
+  if (!created) {
+    // Attempt to create custom router-specified filter chain.
+    created = parent_.config().createFilterChain(*filter_manager_);
+  }
+  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);
+  }
+  // There will always be a codec filter present, which sets the upstream
   // interface. Fast-fail any tests that don't set up mocks correctly.
-  ASSERT(upstream_interface_.has_value());
+  ASSERT(created && upstream_interface_.has_value());
 }
 
 UpstreamRequest::~UpstreamRequest() { cleanUp(); }

source/common/upstream/BUILD

@@ -583,3 +583,13 @@ envoy_cc_library(
         "@envoy_api//envoy/config/cluster/v3:pkg_cc_proto",
     ],
 )
+
+envoy_cc_library(
+    name = "upstream_http_factory_context_lib",
+    hdrs = ["upstream_http_factory_context_impl.h"],
+    deps = [
+        "//envoy/init:manager_interface",
+        "//envoy/server:factory_context_interface",
+        "//envoy/stats:stats_interface",
+    ],
+)

source/common/upstream/upstream_http_factory_context_impl.h

@@ -0,0 +1,34 @@
+#pragma once
+
+#include "envoy/init/manager.h"
+#include "envoy/server/factory_context.h"
+#include "envoy/stats/scope.h"
+
+namespace Envoy {
+namespace Upstream {
+
+/*
+ * Upstream Factory Context used by both Clusters and Routers to configure
+ * upstream filters.
+ */
+class UpstreamHttpFactoryContextImpl : public Server::Configuration::UpstreamHttpFactoryContext {
+public:
+  UpstreamHttpFactoryContextImpl(Server::Configuration::ServerFactoryContext& context,
+                                 Init::Manager& init_manager, Stats::Scope& scope)
+      : server_context_(context), init_manager_(init_manager), scope_(scope) {}
+
+  Server::Configuration::ServerFactoryContext& getServerFactoryContext() const override {
+    return server_context_;
+  }
+
+  Init::Manager& initManager() override { return init_manager_; }
+  Stats::Scope& scope() override { return scope_; }
+
+private:
+  Server::Configuration::ServerFactoryContext& server_context_;
+  Init::Manager& init_manager_;
+  Stats::Scope& scope_;
+};
+
+} // namespace Upstream
+} // namespace Envoy

source/common/upstream/upstream_impl.cc

@@ -1132,6 +1132,7 @@ ClusterInfoImpl::ClusterInfoImpl(
 
   if (http_protocol_options_) {
     Http::FilterChainUtility::FiltersList http_filters = http_protocol_options_->http_filters_;
+    has_configured_http_filters_ = !http_filters.empty();
     if (http_filters.empty()) {
       auto* codec_filter = http_filters.Add();
       codec_filter->set_name("envoy.filters.http.upstream_codec");

source/common/upstream/upstream_impl.h

@@ -55,6 +55,7 @@
 #include "source/common/upstream/outlier_detection_impl.h"
 #include "source/common/upstream/resource_manager_impl.h"
 #include "source/common/upstream/transport_socket_match_impl.h"
+#include "source/common/upstream/upstream_http_factory_context_impl.h"
 #include "source/extensions/upstreams/http/config.h"
 #include "source/server/transport_socket_config_impl.h"
 
@@ -676,25 +677,6 @@ class MainPrioritySetImpl : public PrioritySetImpl, public Logger::Loggable<Logg
   mutable HostMapSharedPtr mutable_cross_priority_host_map_;
 };
 
-class UpstreamHttpFactoryContextImpl : public Server::Configuration::UpstreamHttpFactoryContext {
-public:
-  UpstreamHttpFactoryContextImpl(Server::Configuration::ServerFactoryContext& context,
-                                 Init::Manager& init_manager, Stats::Scope& scope)
-      : server_context_(context), init_manager_(init_manager), scope_(scope) {}
-
-  Server::Configuration::ServerFactoryContext& getServerFactoryContext() const override {
-    return server_context_;
-  }
-
-  Init::Manager& initManager() override { return init_manager_; }
-  Stats::Scope& scope() override { return scope_; }
-
-private:
-  Server::Configuration::ServerFactoryContext& server_context_;
-  Init::Manager& init_manager_;
-  Stats::Scope& scope_;
-};
-
 /**
  * Implementation of ClusterInfo that reads from JSON.
  */
@@ -856,8 +838,13 @@ class ClusterInfoImpl : public ClusterInfo,
   upstreamHttpProtocol(absl::optional<Http::Protocol> downstream_protocol) const override;
 
   // Http::FilterChainFactory
-  void createFilterChain(Http::FilterChainManager& manager) const override {
+  bool createFilterChain(Http::FilterChainManager& manager,
+                         bool only_create_if_configured) const override {
+    if (!has_configured_http_filters_ && only_create_if_configured) {
+      return false;
+    }
     Http::FilterChainUtility::createFilterChainForFactories(manager, http_filter_factories_);
+    return true;
   }
   bool createUpgradeFilterChain(absl::string_view, const UpgradeMap*,
                                 Http::FilterChainManager&) const override {
@@ -952,6 +939,8 @@ class ClusterInfoImpl : public ClusterInfo,
   const std::unique_ptr<Server::Configuration::CommonFactoryContext> factory_context_;
   std::vector<Network::FilterFactoryCb> filter_factories_;
   Http::FilterChainUtility::FilterFactoriesList http_filter_factories_;
+  // true iff the cluster proto specified upstream http filters.
+  bool has_configured_http_filters_{false};
   mutable Http::Http1::CodecStats::AtomicPtr http1_codec_stats_;
   mutable Http::Http2::CodecStats::AtomicPtr http2_codec_stats_;
   mutable Http::Http3::CodecStats::AtomicPtr http3_codec_stats_;

source/extensions/filters/network/http_connection_manager/config.cc

@@ -687,8 +687,9 @@ HttpConnectionManagerConfig::createCodec(Network::Connection& connection,
   PANIC_DUE_TO_CORRUPT_ENUM;
 }
 
-void HttpConnectionManagerConfig::createFilterChain(Http::FilterChainManager& manager) const {
+bool HttpConnectionManagerConfig::createFilterChain(Http::FilterChainManager& manager, bool) const {
   Http::FilterChainUtility::createFilterChainForFactories(manager, filter_factories_);
+  return true;
 }
 
 bool HttpConnectionManagerConfig::createUpgradeFilterChain(

source/extensions/filters/network/http_connection_manager/config.h

@@ -134,7 +134,7 @@ class HttpConnectionManagerConfig : Logger::Loggable<Logger::Id::config>,
       FilterConfigProviderManager& filter_config_provider_manager);
 
   // Http::FilterChainFactory
-  void createFilterChain(Http::FilterChainManager& manager) const override;
+  bool createFilterChain(Http::FilterChainManager& manager, bool = false) const override;
   using FilterFactoriesList =
       std::list<Filter::FilterConfigProviderPtr<Filter::NamedHttpFilterFactoryCb>>;
   struct FilterConfig {

source/server/admin/admin.cc

@@ -261,11 +261,12 @@ bool AdminImpl::createNetworkFilterChain(Network::Connection& connection,
   return true;
 }
 
-void AdminImpl::createFilterChain(Http::FilterChainManager& manager) const {
+bool AdminImpl::createFilterChain(Http::FilterChainManager& manager, bool) const {
   Http::FilterFactoryCb factory = [this](Http::FilterChainFactoryCallbacks& callbacks) {
     callbacks.addStreamFilter(std::make_shared<AdminFilter>(createRequestFunction()));
   };
   manager.applyFilterFactoryCb({}, factory);
+  return true;
 }
 
 namespace {

source/server/admin/admin.h

@@ -112,7 +112,7 @@ class AdminImpl : public Admin,
                                     Network::UdpReadFilterCallbacks&) override {}
 
   // Http::FilterChainFactory
-  void createFilterChain(Http::FilterChainManager& manager) const override;
+  bool createFilterChain(Http::FilterChainManager& manager, bool) const override;
   bool createUpgradeFilterChain(absl::string_view, const Http::FilterChainFactory::UpgradeMap*,
                                 Http::FilterChainManager&) const override {
     return false;

test/common/grpc/grpc_client_integration_test_harness.h

@@ -246,16 +246,7 @@ class GrpcClientIntegrationTest : public GrpcClientIntegrationParamTest {
       : method_descriptor_(helloworld::Greeter::descriptor()->FindMethodByName("SayHello")),
         api_(Api::createApiForTest(*stats_store_, test_time_.timeSystem())),
         dispatcher_(api_->allocateDispatcher("test_thread")),
-        http_context_(stats_store_->symbolTable()), router_context_(stats_store_->symbolTable()) {
-    ON_CALL(*cm_.thread_local_cluster_.cluster_.info_, createFilterChain(_))
-        .WillByDefault(Invoke([&](Http::FilterChainManager& manager) -> void {
-          Http::FilterFactoryCb factory_cb =
-              [](Http::FilterChainFactoryCallbacks& callbacks) -> void {
-            callbacks.addStreamDecoderFilter(std::make_shared<Router::UpstreamCodecFilter>());
-          };
-          manager.applyFilterFactoryCb({}, factory_cb);
-        }));
-  }
+        http_context_(stats_store_->symbolTable()), router_context_(stats_store_->symbolTable()) {}
 
   virtual void initialize() {
     if (fake_upstream_ == nullptr) {