Http Filter: Add connection_pool_cardinality http filter

CODEOWNERS

@@ -391,6 +391,8 @@ extensions/upstreams/tcp @ggreenway @mattklein123
 # Dynamic Modules
 /*/extensions/dynamic_modules @mattklein123 @mathetake @marc-barry
 /*/extensions/filters/http/dynamic_modules @mattklein123 @mathetake @marc-barry
+# Connection Pool Cardinality
+/*/extensions/filters/http/connection_pool_cardinality @kbaichoo @UNOWNED
 
 # HTTP credential injector
 /*/extensions/filters/http/credential_injector @zhaohuabing @kyessenov

api/BUILD

@@ -182,6 +182,7 @@ proto_library(
         "//envoy/extensions/filters/http/composite/v3:pkg",
         "//envoy/extensions/filters/http/compressor/v3:pkg",
         "//envoy/extensions/filters/http/connect_grpc_bridge/v3:pkg",
+        "//envoy/extensions/filters/http/connection_pool_cardinality/v3:pkg",
         "//envoy/extensions/filters/http/cors/v3:pkg",
         "//envoy/extensions/filters/http/credential_injector/v3:pkg",
         "//envoy/extensions/filters/http/csrf/v3:pkg",

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

@@ -0,0 +1,9 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = ["@com_github_cncf_xds//udpa/annotations:pkg"],
+)

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

@@ -0,0 +1,32 @@
+syntax = "proto3";
+
+package envoy.extensions.filters.http.connection_pool_cardinality.v3;
+
+import "udpa/annotations/status.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.filters.http.connection_pool_cardinality.v3";
+option java_outer_classname = "ConnectionPoolCardinalityProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/connection_pool_cardinality/v3;connection_pool_cardinalityv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: ConnectionPoolCardinality]
+
+// [#extension: envoy.filters.http.connection_pool_cardinality]
+
+// This filter bumps up the number of connection pools to create for the upstreams that the
+// downstream requests will be balanced across. This is particularly useful when running envoy
+// as a sidecar where you may want additional connections to be established for the hop between
+// the proxy and the application for ingress requests as it will reduce queueing during load spikes.
+//
+// To validate if this is useful for your scenario, configure the filter and see if the cluster
+// stat ``upstream_rq_pending_total`` decreases when compared to when the filter isn't used.
+message ConnectionPoolCardinalityConfig {
+  // Number of connection pools to distribute the downstream requests across.
+  // If unset or <= 1, this will be a NOP and no additional connection pools will be created.
+  //
+  // With steady state traffic, discounting idle connections / other connection closes, this
+  // will establish a minimum of ||num_envoy_workers * connection_pool_count * num_endpoints ||
+  // clusters where traffic is routed.
+  uint32 connection_pool_count = 1;
+}

api/versioning/BUILD

@@ -121,6 +121,7 @@ proto_library(
         "//envoy/extensions/filters/http/composite/v3:pkg",
         "//envoy/extensions/filters/http/compressor/v3:pkg",
         "//envoy/extensions/filters/http/connect_grpc_bridge/v3:pkg",
+        "//envoy/extensions/filters/http/connection_pool_cardinality/v3:pkg",
         "//envoy/extensions/filters/http/cors/v3:pkg",
         "//envoy/extensions/filters/http/credential_injector/v3:pkg",
         "//envoy/extensions/filters/http/csrf/v3:pkg",

changelogs/current.yaml

@@ -192,6 +192,11 @@ new_features:
     Added :ref:`transform http filter <config_http_filters_transform>` which adds the ability to modify request
     and response bodies in any position of HTTP filter chain.
     This also make it possible to refresh routes based on the attributes in the request body.
+- area: http filter
+  change: |
+    Added :ref:`connection pool cardinality http filter <config_http_filters_connection_pool_cardinality>` which
+    allows one to create more connection pools to given upstreams for a downstream request. This is useful when
+    using envoy as an ingress sidecar to the local application.
 - area: matcher
   change: |
     Removed work-in-progress annotations from RBAC filter ``matcher`` and ``shadow_matcher`` fields in both HTTP

docs/root/configuration/http/http_filters/connection_pool_cardinality.rst

@@ -0,0 +1,30 @@
+.. _config_http_filters_connection_pool_cardinality:
+
+Connection Pool Cardinality
+===
+
+This filter allows envoy to create additional connection pools all the upstreams that downstream
+requests that flow through this filter goes through. For example using it in the ingress http
+connection manager that connects to the local application when running Envoy as a sidecar.
+
+To tune this monitor the connections active with the clusters with which you want to bump the
+number of connection pools up to and select a number that magnifies the number of connections
+without adding significant memory burden. For example in the ingress sidecar scenario, by default
+we'd have a minimum of `|| num_workers ||` connections to the local application, but we can
+bump it to a minimum of `|| num_workers * connection_pool_cardinality ||` with this filter. So if
+we have 4 envoy workers and a connection pool cardinality of 10, we'd go from a minimum of
+4 connection -> 40 connections between the envoy and the local application for ingress.
+
+Configuration
+-------------
+
+Example configuration:
+
+.. code-block:: yaml
+
+  http_filters:
+  - name: envoy.filters.http.connection_pool_cardinality
+    typed_config:
+      "@type": type.googleapis.com/envoy.extensions.filters.http.connection_pool_cardinality.v3.ConnectionPoolCardinalityConfig
+      connection_pool_count: 10
+

docs/root/configuration/http/http_filters/http_filters.rst

@@ -21,6 +21,7 @@ HTTP filters
   compressor_filter
   composite_filter
   connect_grpc_bridge_filter
+  connection_pool_cardinality
   cors_filter
   credential_injector_filter
   csrf_filter

source/extensions/extensions_build_config.bzl

@@ -162,6 +162,7 @@ EXTENSIONS = {
     "envoy.filters.http.cache_v2":                      "//source/extensions/filters/http/cache_v2:config",
     "envoy.filters.http.cdn_loop":                      "//source/extensions/filters/http/cdn_loop:config",
     "envoy.filters.http.compressor":                    "//source/extensions/filters/http/compressor:config",
+    "envoy.filters.http.connection_pool_cardinality":   "//source/extensions/filters/http/connection_pool_cardinality:config",
     "envoy.filters.http.cors":                          "//source/extensions/filters/http/cors:config",
     "envoy.filters.http.composite":                     "//source/extensions/filters/http/composite:config",
     "envoy.filters.http.connect_grpc_bridge":           "//source/extensions/filters/http/connect_grpc_bridge:config",

source/extensions/extensions_metadata.yaml

@@ -2106,3 +2106,10 @@ envoy.filters.http.dynamic_modules:
   status_upstream: alpha
   type_urls:
   - envoy.extensions.filters.http.dynamic_modules.v3.DynamicModuleFilter
+envoy.filters.http.connection_pool_cardinality:
+  categories:
+  - envoy.filters.http
+  security_posture: robust_to_untrusted_downstream_and_upstream
+  status: alpha
+  type_urls:
+  - envoy.extensions.filters.http.connection_pool_cardinality.v3.ConnectionPoolCardinalityConfig

source/extensions/filters/http/connection_pool_cardinality/BUILD

@@ -0,0 +1,41 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_extension",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    repository = "@envoy",
+    deps = [
+        ":filter_lib",
+        "//envoy/registry",
+        "//envoy/singleton:manager_interface",
+        "//source/common/common:logger_lib",
+        "//source/extensions/filters/http/common:factory_base_lib",
+        "@envoy_api//envoy/extensions/filters/http/connection_pool_cardinality/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_library(
+    name = "filter_lib",
+    srcs = ["filter.cc"],
+    hdrs = ["filter.h"],
+    repository = "@envoy",
+    deps = [
+        "//envoy/common:hashable_interface",
+        "//envoy/common:random_generator_interface",
+        "//envoy/http:filter_interface",
+        "//envoy/server:filter_config_interface",
+        "//envoy/stats:stats_interface",
+        "//source/common/protobuf:utility_lib",
+        "//source/extensions/filters/http/common:pass_through_filter_lib",
+    ],
+)

source/extensions/filters/http/connection_pool_cardinality/config.cc

@@ -0,0 +1,29 @@
+#include "source/extensions/filters/http/connection_pool_cardinality/config.h"
+
+#include "source/extensions/filters/http/connection_pool_cardinality/filter.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace ConnectionPoolCardinality {
+
+Envoy::Http::FilterFactoryCb
+ConnectionPoolCardinalityFilterFactory::createFilterFactoryFromProtoTyped(
+    const envoy::extensions::filters::http::connection_pool_cardinality::v3::
+        ConnectionPoolCardinalityConfig& config,
+    const std::string& /*stats_prefix*/, Server::Configuration::FactoryContext& context) {
+
+  uint32_t connection_pool_count = std::max(1U, config.connection_pool_count());
+  return [connection_pool_count, &context](Http::FilterChainFactoryCallbacks& callbacks) -> void {
+    callbacks.addStreamDecoderFilter(std::make_shared<Filter>(
+        connection_pool_count, context.serverFactoryContext().api().randomGenerator()));
+  };
+}
+
+REGISTER_FACTORY(ConnectionPoolCardinalityFilterFactory,
+                 Server::Configuration::NamedHttpFilterConfigFactory);
+
+} // namespace ConnectionPoolCardinality
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/http/connection_pool_cardinality/config.h

@@ -0,0 +1,34 @@
+#pragma once
+
+#include "envoy/extensions/filters/http/connection_pool_cardinality/v3/connection_pool_cardinality.pb.h"
+#include "envoy/extensions/filters/http/connection_pool_cardinality/v3/connection_pool_cardinality.pb.validate.h"
+
+#include "source/extensions/filters/http/common/factory_base.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace ConnectionPoolCardinality {
+
+class ConnectionPoolCardinalityFilterFactory
+    : public Common::FactoryBase<envoy::extensions::filters::http::connection_pool_cardinality::v3::
+                                     ConnectionPoolCardinalityConfig> {
+public:
+  ConnectionPoolCardinalityFilterFactory()
+      : FactoryBase("envoy.filters.http.connection_pool_cardinality") {}
+  std::string name() const override { return "envoy.filters.http.connection_pool_cardinality"; }
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return std::make_unique<envoy::extensions::filters::http::connection_pool_cardinality::v3::
+                                ConnectionPoolCardinalityConfig>();
+  }
+
+  Http::FilterFactoryCb createFilterFactoryFromProtoTyped(
+      const envoy::extensions::filters::http::connection_pool_cardinality::v3::
+          ConnectionPoolCardinalityConfig& config,
+      const std::string&, Server::Configuration::FactoryContext& context) override;
+};
+
+} // namespace ConnectionPoolCardinality
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/http/connection_pool_cardinality/filter.cc

@@ -0,0 +1,33 @@
+#include "source/extensions/filters/http/connection_pool_cardinality/filter.h"
+
+#include "envoy/stream_info/stream_info.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace ConnectionPoolCardinality {
+
+Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap&, bool) {
+  if (connection_pool_count_ <= 1) {
+    // No dynamic state needed if only one connection pool
+    return Http::FilterHeadersStatus::Continue;
+  }
+
+  const uint32_t pool_index = random_.random() % connection_pool_count_;
+
+  // Set dynamic metadata for connection pool selection
+  auto filter_state = decoder_callbacks_->streamInfo().filterState();
+  filter_state->setData(
+      "connection_pool_cardinality", std::make_shared<ConnPoolCardinality>(pool_index),
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::Request,
+      StreamInfo::StreamSharingMayImpactPooling::SharedWithUpstreamConnection);
+
+  ENVOY_STREAM_LOG(trace, "Selected connection pool index: {}", *decoder_callbacks_, pool_index);
+
+  return Http::FilterHeadersStatus::Continue;
+}
+
+} // namespace ConnectionPoolCardinality
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/http/connection_pool_cardinality/filter.h

@@ -0,0 +1,41 @@
+#pragma once
+
+#include "envoy/common/hashable.h"
+#include "envoy/common/random_generator.h"
+#include "envoy/http/filter.h"
+
+#include "source/common/common/logger.h"
+#include "source/extensions/filters/http/common/pass_through_filter.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace ConnectionPoolCardinality {
+
+struct ConnPoolCardinality : public StreamInfo::FilterState::Object, public Envoy::Hashable {
+  explicit ConnPoolCardinality(uint32_t i) : idx_(i) {}
+  absl::optional<uint64_t> hash() const override { return idx_; }
+
+  const uint32_t idx_;
+};
+
+/**
+ * A HTTP filter that adds random dynamic state for downstream requests.
+ * This enables load balancing requests across multiple connection pools.
+ */
+class Filter : public Http::PassThroughDecoderFilter, public Logger::Loggable<Logger::Id::misc> {
+public:
+  Filter(uint32_t connection_pool_count, Random::RandomGenerator& random)
+      : connection_pool_count_(connection_pool_count), random_(random) {}
+  Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap& headers,
+                                          bool end_stream) override;
+
+private:
+  const uint32_t connection_pool_count_;
+  Random::RandomGenerator& random_;
+};
+
+} // namespace ConnectionPoolCardinality
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

test/extensions/filters/http/connection_pool_cardinality/BUILD

@@ -0,0 +1,54 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_package",
+)
+load(
+    "//test/extensions:extensions_build_system.bzl",
+    "envoy_extension_cc_test",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_package()
+
+envoy_extension_cc_test(
+    name = "filter_test",
+    srcs = ["filter_test.cc"],
+    extension_names = ["envoy.filters.http.connection_pool_cardinality"],
+    rbe_pool = "6gig",
+    deps = [
+        "//source/extensions/filters/http/connection_pool_cardinality:config",
+        "//source/extensions/filters/http/connection_pool_cardinality:filter_lib",
+        "//test/mocks:common_lib",
+        "//test/mocks/http:http_mocks",
+        "//test/mocks/server:factory_context_mocks",
+        "//test/mocks/stats:stats_mocks",
+        "//test/mocks/stream_info:stream_info_mocks",
+        "//test/test_common:utility_lib",
+    ],
+)
+
+envoy_extension_cc_test(
+    name = "config_test",
+    srcs = ["config_test.cc"],
+    extension_names = ["envoy.filters.http.connection_pool_cardinality"],
+    rbe_pool = "6gig",
+    deps = [
+        "//source/extensions/filters/http/connection_pool_cardinality:config",
+        "//test/mocks/server:factory_context_mocks",
+        "//test/test_common:utility_lib",
+        "@envoy_api//envoy/extensions/filters/http/connection_pool_cardinality/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_extension_cc_test(
+    name = "integration_test",
+    srcs = ["integration_test.cc"],
+    extension_names = ["envoy.filters.http.connection_pool_cardinality"],
+    rbe_pool = "6gig",
+    deps = [
+        "//source/extensions/filters/http/connection_pool_cardinality:config",
+        "//test/integration:http_protocol_integration_lib",
+        "@envoy_api//envoy/extensions/filters/http/connection_pool_cardinality/v3:pkg_cc_proto",
+    ],
+)