diff --git a/api/envoy/api/v2/BUILD b/api/envoy/api/v2/BUILD
index 3e557a10239e..6b703d81e8c4 100644
--- a/api/envoy/api/v2/BUILD
+++ b/api/envoy/api/v2/BUILD
@@ -72,6 +72,7 @@ api_proto_library(
         "//envoy/api/v2/core:health_check",
         "//envoy/api/v2/core:protocol",
         "//envoy/api/v2/endpoint",
+        "//envoy/api/v2/listener", # TODO(alanconway): for Filter definition
         "//envoy/type:percent",
     ],
 )
diff --git a/api/envoy/api/v2/cds.proto b/api/envoy/api/v2/cds.proto
index 8359cd51964b..0d6414219539 100644
--- a/api/envoy/api/v2/cds.proto
+++ b/api/envoy/api/v2/cds.proto
@@ -11,6 +11,7 @@ import "envoy/api/v2/core/config_source.proto";
 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/listener/listener.proto"; // TODO(alanconway): for Filter definition
 import "envoy/api/v2/cluster/circuit_breaker.proto";
 import "envoy/api/v2/cluster/outlier_detection.proto";
 import "envoy/api/v2/eds.proto";
@@ -468,6 +469,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 listener.Filter filters = 34 [(gogoproto.nullable) = false];
 }
 
 // An extensible structure containing the address Envoy should bind to when
diff --git a/include/envoy/upstream/upstream.h b/include/envoy/upstream/upstream.h
index d549a5ea8f7d..c78d027a1d63 100644
--- a/include/envoy/upstream/upstream.h
+++ b/include/envoy/upstream/upstream.h
@@ -534,6 +534,11 @@ class ClusterInfo {
    *         after a host is removed from service discovery.
    */
   virtual bool drainConnectionsOnHostRemoval() const PURE;
+
+  /**
+   * Create network filters on a new upstream connection.
+   */
+  virtual void createNetworkFilters(Network::Connection& connection) const PURE;
 };
 
 typedef std::shared_ptr<const ClusterInfo> ClusterInfoConstSharedPtr;
diff --git a/source/common/upstream/BUILD b/source/common/upstream/BUILD
index d7491ec73f6b..8a08b0a7dc03 100644
--- a/source/common/upstream/BUILD
+++ b/source/common/upstream/BUILD
@@ -361,6 +361,7 @@ envoy_cc_library(
         "//include/envoy/network:dns_interface",
         "//include/envoy/runtime:runtime_interface",
         "//include/envoy/server:transport_socket_config_interface",
+        "//include/envoy/server:filter_config_interface",
         "//include/envoy/ssl:context_manager_interface",
         "//include/envoy/thread_local:thread_local_interface",
         "//include/envoy/upstream:cluster_manager_interface",
@@ -373,6 +374,7 @@ envoy_cc_library(
         "//source/common/config:metadata_lib",
         "//source/common/stats:stats_lib",
         "//source/common/upstream:locality_lib",
+        "//source/server:configuration_lib", # TODO(alanconway): bad dependency
         "@envoy_api//envoy/api/v2/core:base_cc",
         "@envoy_api//envoy/api/v2/endpoint:endpoint_cc",
     ],
diff --git a/source/common/upstream/upstream_impl.cc b/source/common/upstream/upstream_impl.cc
index 05427bdad6fd..719f81956dbc 100644
--- a/source/common/upstream/upstream_impl.cc
+++ b/source/common/upstream/upstream_impl.cc
@@ -12,6 +12,7 @@
 #include "envoy/event/timer.h"
 #include "envoy/network/dns.h"
 #include "envoy/server/transport_socket_config.h"
+#include "envoy/server/filter_config.h"
 #include "envoy/ssl/context_manager.h"
 #include "envoy/upstream/health_checker.h"
 
@@ -33,6 +34,7 @@
 #include "common/upstream/original_dst_cluster.h"
 
 #include "extensions/transport_sockets/well_known_names.h"
+#include "server/configuration_impl.h" // TODO(alanconway): bad dependency
 
 namespace Envoy {
 namespace Upstream {
@@ -135,6 +137,7 @@ HostImpl::createConnection(Event::Dispatcher& dispatcher, const ClusterInfo& clu
       address, cluster.sourceAddress(), cluster.transportSocketFactory().createTransportSocket(),
       connection_options);
   connection->setBufferLimits(cluster.perConnectionBufferLimitBytes());
+  cluster.createNetworkFilters(*connection);
   return connection;
 }
 
@@ -253,6 +256,30 @@ ClusterLoadReportStats ClusterInfoImpl::generateLoadReportStats(Stats::Scope& sc
   return {ALL_CLUSTER_LOAD_REPORT_STATS(POOL_COUNTER(scope))};
 }
 
+// TODO(alanconway): dummy factory context, how do we implement this?  Some of
+// the methods are listener specific, should they throw, return null values, return
+// values derived from the Cluster or be moved/removed?
+class ClusterInfoImpl::FactoryContextImpl : public Server::Configuration::FactoryContext {
+ public:
+  AccessLog::AccessLogManager& accessLogManager() override { NOT_IMPLEMENTED; }
+  Upstream::ClusterManager& clusterManager() override { NOT_IMPLEMENTED; }
+  Event::Dispatcher& dispatcher() override { NOT_IMPLEMENTED; }
+  bool healthCheckFailed() override { NOT_IMPLEMENTED; }
+  Tracing::HttpTracer& httpTracer() override { NOT_IMPLEMENTED; }
+  const LocalInfo::LocalInfo& localInfo() const override { NOT_IMPLEMENTED; }
+  Envoy::Runtime::RandomGenerator& random() override { NOT_IMPLEMENTED; }
+  RateLimit::ClientPtr rateLimitClient(const absl::optional<std::chrono::milliseconds>&) override { NOT_IMPLEMENTED; }
+  Envoy::Runtime::Loader& runtime() override { NOT_IMPLEMENTED; }
+  Singleton::Manager& singletonManager() override { NOT_IMPLEMENTED; }
+  ThreadLocal::Instance& threadLocal() override { NOT_IMPLEMENTED; }
+  Server::Admin& admin() override { NOT_IMPLEMENTED; }
+  Init::Manager& initManager() override { NOT_IMPLEMENTED; }
+  Stats::Scope& listenerScope() override { NOT_IMPLEMENTED; }
+  Stats::Scope& scope() override { NOT_IMPLEMENTED; }
+  const envoy::api::v2::core::Metadata& listenerMetadata() const override { NOT_IMPLEMENTED; }
+  Network::DrainDecision& drainDecision() override { NOT_IMPLEMENTED; }
+};
+
 ClusterInfoImpl::ClusterInfoImpl(const envoy::api::v2::Cluster& config,
                                  const envoy::api::v2::core::BindConfig& bind_config,
                                  Runtime::Loader& runtime, Stats::Store& stats,
@@ -340,6 +367,30 @@ ClusterInfoImpl::ClusterInfoImpl(const envoy::api::v2::Cluster& config,
     idle_timeout_ = std::chrono::milliseconds(
         DurationUtil::durationToMilliseconds(config.common_http_protocol_options().idle_timeout()));
   }
+
+  auto filters = config.filters();
+  for (ssize_t i = 0; i < filters.size(); i++) {
+    const auto& proto_config = filters[i];
+    const ProtobufTypes::String name = proto_config.name();
+    const Json::ObjectSharedPtr filter_config =
+        MessageUtil::getJsonObjectFromMessage(proto_config.config());
+    ENVOY_LOG(debug, "filter #{} name: {} config: {}", i, name, filter_config->asJsonString());
+    // Now see if there is a factory that will accept the config.
+    auto& factory =
+      Config::Utility::getAndCheckFactory<Server::Configuration::NamedNetworkFilterConfigFactory>(name);
+    Network::FilterFactoryCb callback;
+    if (filter_config->getBoolean("deprecated_v1", false)) {
+      callback = factory.createFilterFactory(*filter_config->getObject("value", true), *factory_context_);
+    } else {
+      auto message = Config::Utility::translateToFactoryConfig(proto_config, factory);
+      callback = factory.createFilterFactoryFromProto(*message, *factory_context_);
+    }
+    filter_factories_.push_back(callback);
+  }
+}
+
+void ClusterInfoImpl::createNetworkFilters(Network::Connection& connection) const {
+  Server::Configuration::FilterChainUtility::buildFilterChain(connection, filter_factories_);
 }
 
 ClusterSharedPtr ClusterImplBase::create(const envoy::api::v2::Cluster& cluster, ClusterManager& cm,
diff --git a/source/common/upstream/upstream_impl.h b/source/common/upstream/upstream_impl.h
index 12984dd0634a..0ef5dfef5acc 100644
--- a/source/common/upstream/upstream_impl.h
+++ b/source/common/upstream/upstream_impl.h
@@ -16,6 +16,8 @@
 #include "envoy/event/timer.h"
 #include "envoy/local_info/local_info.h"
 #include "envoy/network/dns.h"
+
+#include "envoy/network/filter.h"
 #include "envoy/runtime/runtime.h"
 #include "envoy/server/transport_socket_config.h"
 #include "envoy/ssl/context_manager.h"
@@ -307,7 +309,8 @@ class PrioritySetImpl : public PrioritySet {
  * Implementation of ClusterInfo that reads from JSON.
  */
 class ClusterInfoImpl : public ClusterInfo,
-                        public Server::Configuration::TransportSocketFactoryContext {
+                        public Server::Configuration::TransportSocketFactoryContext,
+                        protected Logger::Loggable<Logger::Id::upstream> {
 public:
   ClusterInfoImpl(const envoy::api::v2::Cluster& config,
                   const envoy::api::v2::core::BindConfig& bind_config, Runtime::Loader& runtime,
@@ -362,6 +365,8 @@ class ClusterInfoImpl : public ClusterInfo,
 
   bool drainConnectionsOnHostRemoval() const override { return drain_connections_on_host_removal_; }
 
+  void createNetworkFilters(Network::Connection&) const;
+
 private:
   struct ResourceManagers {
     ResourceManagers(const envoy::api::v2::Cluster& config, Runtime::Loader& runtime,
@@ -401,6 +406,10 @@ class ClusterInfoImpl : public ClusterInfo,
   const envoy::api::v2::Cluster::CommonLbConfig common_lb_config_;
   const Network::ConnectionSocket::OptionsSharedPtr cluster_socket_options_;
   const bool drain_connections_on_host_removal_;
+
+  class FactoryContextImpl;
+  std::unique_ptr<FactoryContextImpl> factory_context_;
+  std::vector<Network::FilterFactoryCb> filter_factories_;
 };
 
 /**