filter: add network filters to the upstreams (#7503)

Signed-off-by: Kuat Yessenov <kuat@google.com>

api/envoy/api/v2/BUILD

@@ -67,6 +67,7 @@ api_proto_library_internal(
         ":eds",
         "//envoy/api/v2/auth:cert",
         "//envoy/api/v2/cluster:circuit_breaker",
+        "//envoy/api/v2/cluster:filter",
         "//envoy/api/v2/cluster:outlier_detection",
         "//envoy/api/v2/core:address",
         "//envoy/api/v2/core:base",
@@ -86,6 +87,7 @@ api_go_grpc_library(
         ":eds_go_grpc",
         "//envoy/api/v2/auth:cert_go_proto",
         "//envoy/api/v2/cluster:circuit_breaker_go_proto",
+        "//envoy/api/v2/cluster:filter_go_proto",
         "//envoy/api/v2/cluster:outlier_detection_go_proto",
         "//envoy/api/v2/core:address_go_proto",
         "//envoy/api/v2/core:base_go_proto",

api/envoy/api/v2/cds.proto

@@ -16,6 +16,7 @@ import "envoy/api/v2/discovery.proto";
 import "envoy/api/v2/core/health_check.proto";
 import "envoy/api/v2/core/protocol.proto";
 import "envoy/api/v2/cluster/circuit_breaker.proto";
+import "envoy/api/v2/cluster/filter.proto";
 import "envoy/api/v2/cluster/outlier_detection.proto";
 import "envoy/api/v2/eds.proto";
 import "envoy/type/percent.proto";
@@ -51,7 +52,7 @@ service ClusterDiscoveryService {
 // [#protodoc-title: Clusters]
 
 // Configuration for a single upstream cluster.
-// [#comment:next free field: 40]
+// [#comment:next free field: 41]
 message Cluster {
   // Supplies the name of the cluster which must be unique across all clusters.
   // The cluster name is used when emitting
@@ -632,6 +633,11 @@ message Cluster {
   // If this flag is not set to true, Envoy will wait until the hosts fail active health
   // checking before removing it from the cluster.
   bool drain_connections_on_host_removal = 32;
+
+  // An optional list of network filters that make up the filter chain for
+  // outgoing connections made by the cluster. Order matters as the filters are
+  // processed sequentially as connection events happen.
+  repeated cluster.Filter filters = 40;
 }
 
 // An extensible structure containing the address Envoy should bind to when

api/envoy/api/v2/cluster/BUILD

@@ -33,3 +33,16 @@ api_go_proto_library(
     name = "outlier_detection",
     proto = ":outlier_detection",
 )
+
+api_proto_library_internal(
+    name = "filter",
+    srcs = ["filter.proto"],
+    visibility = [
+        "//envoy/api/v2:__pkg__",
+    ],
+)
+
+api_go_proto_library(
+    name = "filter",
+    proto = ":filter",
+)

api/envoy/api/v2/cluster/filter.proto

@@ -0,0 +1,30 @@
+syntax = "proto3";
+
+package envoy.api.v2.cluster;
+
+option java_outer_classname = "FilterProto";
+option java_multiple_files = true;
+option java_package = "io.envoyproxy.envoy.api.v2.cluster";
+option csharp_namespace = "Envoy.Api.V2.ClusterNS";
+option ruby_package = "Envoy.Api.V2.ClusterNS";
+
+import "google/protobuf/any.proto";
+import "google/protobuf/struct.proto";
+
+import "validate/validate.proto";
+import "gogoproto/gogo.proto";
+
+option (gogoproto.equal_all) = true;
+
+// [#protodoc-title: Upstream filters]
+//
+// Upstream filters apply to the connections to the upstream cluster hosts.
+message Filter {
+  // The name of the filter to instantiate. The name must match a
+  // :ref:`supported filter <config_network_filters>`.
+  string name = 1 [(validate.rules).string.min_bytes = 1];
+
+  // Filter specific configuration which depends on the filter being
+  // instantiated. See the supported filters for further documentation.
+  google.protobuf.Any typed_config = 2;
+}

docs/build.sh

@@ -76,6 +76,7 @@ PROTO_RST="
   /envoy/api/v2/cds/envoy/api/v2/cds.proto.rst
   /envoy/api/v2/cluster/outlier_detection/envoy/api/v2/cluster/outlier_detection.proto.rst
   /envoy/api/v2/cluster/circuit_breaker/envoy/api/v2/cluster/circuit_breaker.proto.rst
+  /envoy/api/v2/cluster/filter/envoy/api/v2/cluster/filter.proto.rst
   /envoy/api/v2/rds/envoy/api/v2/rds.proto.rst
   /envoy/api/v2/route/route/envoy/api/v2/route/route.proto.rst
   /envoy/api/v2/srds/envoy/api/v2/srds.proto.rst

docs/root/api-v2/clusters/clusters.rst

@@ -8,6 +8,7 @@ Clusters
   ../api/v2/cds.proto
   ../api/v2/cluster/outlier_detection.proto
   ../api/v2/cluster/circuit_breaker.proto
+  ../api/v2/cluster/filter.proto
   ../api/v2/endpoint/endpoint.proto
   ../api/v2/eds.proto
   ../api/v2/core/health_check.proto

docs/root/intro/arch_overview/upstream/upstream.rst

@@ -11,3 +11,4 @@ Upstream clusters
   load_balancing/load_balancing
   outlier
   circuit_breaking
+  upstream_filters

docs/root/intro/arch_overview/upstream/upstream_filters.rst

@@ -0,0 +1,11 @@
+.. _arch_overview_upstream_filters:
+
+Upstream network filters
+========================
+
+Upstream clusters provide an ability to inject network level (L3/L4)
+:ref:`filters <arch_overview_network_filters>`. The filters apply to the
+connection to the upstream hosts, using the same API presented by listeners for
+the downstream connections. The write callbacks are invoked for any chunk of
+data sent to the upstream host, and the read callbacks are invoked for data
+received from the upstream host.

docs/root/intro/version_history.rst

@@ -14,6 +14,7 @@ Version history
 * router: added :ref:`rq_retry_skipped_request_not_complete <config_http_filters_router_stats>` counter stat to router stats.
 * tls: added verification of IP address SAN fields in certificates against configured SANs in the
   certificate validation context.
+* upstream: added network filter chains to upstream connections, see :ref:`filters<envoy_api_field_Cluster.filters>`.
 
 1.11.0 (July 11, 2019)
 ======================

include/envoy/server/filter_config.h

@@ -32,18 +32,11 @@ namespace Server {
 namespace Configuration {
 
 /**
- * Context passed to network and HTTP filters to access server resources.
- * TODO(mattklein123): When we lock down visibility of the rest of the code, filters should only
- * access the rest of the server via interfaces exposed here.
+ * Common interface for downstream and upstream network filters.
  */
-class FactoryContext {
+class CommonFactoryContext {
 public:
-  virtual ~FactoryContext() = default;
-
-  /**
-   * @return AccessLogManager for use by the entire server.
-   */
-  virtual AccessLog::AccessLogManager& accessLogManager() PURE;
+  virtual ~CommonFactoryContext() = default;
 
   /**
    * @return Upstream::ClusterManager& singleton for use by the entire server.
@@ -56,37 +49,6 @@ class FactoryContext {
    */
   virtual Event::Dispatcher& dispatcher() PURE;
 
-  /**
-   * @return const Network::DrainDecision& a drain decision that filters can use to determine if
-   *         they should be doing graceful closes on connections when possible.
-   */
-  virtual const Network::DrainDecision& drainDecision() PURE;
-
-  /**
-   * @return whether external healthchecks are currently failed or not.
-   */
-  virtual bool healthCheckFailed() PURE;
-
-  /**
-   * @return the server-wide http tracer.
-   */
-  virtual Tracing::HttpTracer& httpTracer() PURE;
-
-  /**
-   * @return the server's init manager. This can be used for extensions that need to initialize
-   *         after cluster manager init but before the server starts listening. All extensions
-   *         should register themselves during configuration load. initialize() will be called on
-   *         each registered target after cluster manager init but before the server starts
-   *         listening. Once all targets have initialized and invoked their callbacks, the server
-   *         will start listening.
-   */
-  virtual Init::Manager& initManager() PURE;
-
-  /**
-   * @return ServerLifecycleNotifier& the lifecycle notifier for the server.
-   */
-  virtual ServerLifecycleNotifier& lifecycleNotifier() PURE;
-
   /**
    * @return information about the local environment the server is running in.
    */
@@ -123,6 +85,68 @@ class FactoryContext {
    */
   virtual Server::Admin& admin() PURE;
 
+  /**
+   * @return TimeSource& a reference to the time source.
+   */
+  virtual TimeSource& timeSource() PURE;
+
+  /**
+   * @return ProtobufMessage::ValidationVisitor& validation visitor for filter configuration
+   *         messages.
+   */
+  virtual ProtobufMessage::ValidationVisitor& messageValidationVisitor() PURE;
+
+  /**
+   * @return Api::Api& a reference to the api object.
+   */
+  virtual Api::Api& api() PURE;
+};
+
+/**
+ * Context passed to network and HTTP filters to access server resources.
+ * TODO(mattklein123): When we lock down visibility of the rest of the code, filters should only
+ * access the rest of the server via interfaces exposed here.
+ */
+class FactoryContext : public virtual CommonFactoryContext {
+public:
+  ~FactoryContext() override = default;
+
+  /**
+   * @return AccessLogManager for use by the entire server.
+   */
+  virtual AccessLog::AccessLogManager& accessLogManager() PURE;
+
+  /**
+   * @return const Network::DrainDecision& a drain decision that filters can use to determine if
+   *         they should be doing graceful closes on connections when possible.
+   */
+  virtual const Network::DrainDecision& drainDecision() PURE;
+
+  /**
+   * @return whether external healthchecks are currently failed or not.
+   */
+  virtual bool healthCheckFailed() PURE;
+
+  /**
+   * @return the server-wide http tracer.
+   */
+  virtual Tracing::HttpTracer& httpTracer() PURE;
+
+  /**
+   * @return the server's init manager. This can be used for extensions that need to initialize
+   *         after cluster manager init but before the server starts listening. All extensions
+   *         should register themselves during configuration load. initialize() will be called on
+   *         each registered target after cluster manager init but before the server starts
+   *         listening. Once all targets have initialized and invoked their callbacks, the server
+   *         will start listening.
+   */
+  virtual Init::Manager& initManager() PURE;
+
+  /**
+   * @return ServerLifecycleNotifier& the lifecycle notifier for the server.
+   */
+  virtual ServerLifecycleNotifier& lifecycleNotifier() PURE;
+
   /**
    * @return Stats::Scope& the listener's stats scope.
    */
@@ -134,11 +158,6 @@ class FactoryContext {
    */
   virtual const envoy::api::v2::core::Metadata& listenerMetadata() const PURE;
 
-  /**
-   * @return TimeSource& a reference to the time source.
-   */
-  virtual TimeSource& timeSource() PURE;
-
   /**
    * @return OverloadManager& the overload manager for the server.
    */
@@ -158,17 +177,6 @@ class FactoryContext {
    * @return ProcessContext& a reference to the process context.
    */
   virtual ProcessContext& processContext() PURE;
-
-  /**
-   * @return ProtobufMessage::ValidationVisitor& validation visitor for filter configuration
-   *         messages.
-   */
-  virtual ProtobufMessage::ValidationVisitor& messageValidationVisitor() PURE;
-
-  /**
-   * @return Api::Api& a reference to the api object.
-   */
-  virtual Api::Api& api() PURE;
 };
 
 class ListenerFactoryContext : public virtual FactoryContext {
@@ -316,6 +324,34 @@ class NamedNetworkFilterConfigFactory : public ProtocolOptionsFactory {
   virtual std::string name() PURE;
 };
 
+/**
+ * Implemented by each upstream cluster network filter and registered via
+ * Registry::registerFactory() or the convenience class RegisterFactory.
+ */
+class NamedUpstreamNetworkFilterConfigFactory : public ProtocolOptionsFactory {
+public:
+  ~NamedUpstreamNetworkFilterConfigFactory() override = default;
+
+  /**
+   * Create a particular upstream network filter factory implementation. If the implementation is
+   * unable to produce a factory with the provided parameters, it should throw an EnvoyException in
+   * the case of general error. The returned callback should always be initialized.
+   */
+  virtual Network::FilterFactoryCb createFilterFactoryFromProto(const Protobuf::Message& config,
+                                                                CommonFactoryContext& context) PURE;
+
+  /**
+   * @return ProtobufTypes::MessagePtr create empty config proto message for v2.
+   */
+  virtual ProtobufTypes::MessagePtr createEmptyConfigProto() PURE;
+
+  /**
+   * @return std::string the identifying name for a particular implementation of a network filter
+   * produced by the factory.
+   */
+  virtual std::string name() PURE;
+};
+
 /**
  * Implemented by each HTTP filter and registered via Registry::registerFactory or the
  * convenience class RegisterFactory.

include/envoy/upstream/upstream.h

@@ -843,6 +843,11 @@ class ClusterInfo {
    */
   virtual absl::optional<std::string> eds_service_name() const PURE;
 
+  /**
+   * Create network filters on a new upstream connection.
+   */
+  virtual void createNetworkFilterChain(Network::Connection& connection) const PURE;
+
 protected:
   /**
    * Invoked by extensionProtocolOptionsTyped.

source/common/upstream/BUILD

@@ -425,6 +425,7 @@ envoy_cc_library(
         "//include/envoy/local_info:local_info_interface",
         "//include/envoy/network:dns_interface",
         "//include/envoy/runtime:runtime_interface",
+        "//include/envoy/server:filter_config_interface",
         "//include/envoy/server:transport_socket_config_interface",
         "//include/envoy/ssl:context_manager_interface",
         "//include/envoy/thread_local:thread_local_interface",

source/common/upstream/health_discovery_service.cc

@@ -236,9 +236,9 @@ ProdClusterInfoFactory::createClusterInfo(const CreateClusterInfoParams& params)
   Network::TransportSocketFactoryPtr socket_factory =
       Upstream::createTransportSocketFactory(params.cluster_, factory_context);
 
-  return std::make_unique<ClusterInfoImpl>(params.cluster_, params.bind_config_, params.runtime_,
-                                           std::move(socket_factory), std::move(scope),
-                                           params.added_via_api_, params.validation_visitor_);
+  return std::make_unique<ClusterInfoImpl>(
+      params.cluster_, params.bind_config_, params.runtime_, std::move(socket_factory),
+      std::move(scope), params.added_via_api_, params.validation_visitor_, factory_context);
 }
 
 void HdsCluster::startHealthchecks(AccessLog::AccessLogManager& access_log_manager,