diff --git a/CODEOWNERS b/CODEOWNERS
index 631ab8e8b5a..c88f01af905 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -10,3 +10,5 @@
 /*/extensions/filters/http/header_to_metadata @rgs1 @zuercher
 # alts transport socket extension
 /*/extensions/transport_sockets/alts @lizan @yangminzhu
+# sni_cluster extension
+/*/extensions/filters/network/sni_cluster @rshriram @lizan
diff --git a/docs/root/configuration/network_filters/network_filters.rst b/docs/root/configuration/network_filters/network_filters.rst
index ed805b7de2d..5ef34c7dd47 100644
--- a/docs/root/configuration/network_filters/network_filters.rst
+++ b/docs/root/configuration/network_filters/network_filters.rst
@@ -19,3 +19,4 @@ filters.
   redis_proxy_filter
   tcp_proxy_filter
   thrift_proxy_filter
+  sni_cluster_filter
diff --git a/docs/root/configuration/network_filters/sni_cluster_filter.rst b/docs/root/configuration/network_filters/sni_cluster_filter.rst
new file mode 100644
index 00000000000..9a18fd129cd
--- /dev/null
+++ b/docs/root/configuration/network_filters/sni_cluster_filter.rst
@@ -0,0 +1,13 @@
+.. _config_network_filters_sni_cluster:
+
+Upstream Cluster from SNI
+=========================
+
+The `sni_cluster` is a network filter that uses the SNI value in a TLS
+connection as the upstream cluster name. The filter will not modify the
+upstream cluster for non-TLS connections.
+
+This filter has no configuration. It must be installed before the
+:ref:`tcp_proxy <config_network_filters_tcp_proxy>` filter.
+
+* :ref:`v2 API reference <envoy_api_field_listener.Filter.name>`
diff --git a/docs/root/configuration/network_filters/tcp_proxy_filter.rst b/docs/root/configuration/network_filters/tcp_proxy_filter.rst
index d20a44408d8..bddad546054 100644
--- a/docs/root/configuration/network_filters/tcp_proxy_filter.rst
+++ b/docs/root/configuration/network_filters/tcp_proxy_filter.rst
@@ -7,6 +7,16 @@ TCP proxy
 * :ref:`v1 API reference <config_network_filters_tcp_proxy_v1>`
 * :ref:`v2 API reference <envoy_api_msg_config.filter.network.tcp_proxy.v2.TcpProxy>`
 
+.. _config_network_filters_tcp_proxy_dynamic_cluster:
+
+Dynamic cluster selection
+-------------------------
+
+The upstream cluster used by the TCP proxy filter can be dynamically set by
+other network filters on a per-connection basis by setting a per-connection
+state object under the key `envoy.tcp_proxy.cluster`. See the
+implementation for the details.
+
 .. _config_network_filters_tcp_proxy_stats:
 
 Statistics
diff --git a/docs/root/intro/version_history.rst b/docs/root/intro/version_history.rst
index 8f4cdb1932c..c3251ef57bc 100644
--- a/docs/root/intro/version_history.rst
+++ b/docs/root/intro/version_history.rst
@@ -94,6 +94,8 @@ Version history
 * config: Fixed stat inconsistency between xDS and ADS implementation. :ref:`update_failure <config_cluster_manager_cds>`  
   stat is incremented in case of network failure and :ref:`update_rejected <config_cluster_manager_cds>` stat is incremented 
   in case of schema/validation error.
+* :ref:`sni_cluster <config_network_filters_sni_cluster>`: introduced a new network filter that forwards connections to the
+  upstream cluster specified by the SNI value presented by the client during a TLS handshake.
 * config: Added a stat :ref:`connected_state <management_server_stats>` that indicates current connected state of Envoy with 
   management server. 
 
diff --git a/source/common/tcp_proxy/BUILD b/source/common/tcp_proxy/BUILD
index 7fcc846da64..7a912338037 100644
--- a/source/common/tcp_proxy/BUILD
+++ b/source/common/tcp_proxy/BUILD
@@ -19,6 +19,7 @@ envoy_cc_library(
         "//include/envoy/event:dispatcher_interface",
         "//include/envoy/network:connection_interface",
         "//include/envoy/network:filter_interface",
+        "//include/envoy/request_info:filter_state_interface",
         "//include/envoy/router:router_interface",
         "//include/envoy/server:filter_config_interface",
         "//include/envoy/stats:stats_interface",
diff --git a/source/common/tcp_proxy/tcp_proxy.cc b/source/common/tcp_proxy/tcp_proxy.cc
index 5efb1f16e9c..c44e35c85c9 100644
--- a/source/common/tcp_proxy/tcp_proxy.cc
+++ b/source/common/tcp_proxy/tcp_proxy.cc
@@ -22,6 +22,8 @@
 namespace Envoy {
 namespace TcpProxy {
 
+const std::string PerConnectionCluster::Key = "envoy.tcp_proxy.cluster";
+
 Config::Route::Route(
     const envoy::config::filter::network::tcp_proxy::v2::TcpProxy::DeprecatedV1::TCPRoute& config) {
   cluster_name_ = config.cluster();
@@ -108,6 +110,13 @@ Config::Config(const envoy::config::filter::network::tcp_proxy::v2::TcpProxy& co
 }
 
 const std::string& Config::getRegularRouteFromEntries(Network::Connection& connection) {
+  // First check if the per-connection state to see if we need to route to a pre-selected cluster
+  if (connection.perConnectionState().hasData<PerConnectionCluster>(PerConnectionCluster::Key)) {
+    const PerConnectionCluster& per_connection_cluster =
+        connection.perConnectionState().getData<PerConnectionCluster>(PerConnectionCluster::Key);
+    return per_connection_cluster.value();
+  }
+
   for (const Config::Route& route : routes_) {
     if (!route.source_port_ranges_.empty() &&
         !Network::Utility::portInRangeList(*connection.remoteAddress(),
diff --git a/source/common/tcp_proxy/tcp_proxy.h b/source/common/tcp_proxy/tcp_proxy.h
index 7f40ef6a453..e749400fe1b 100644
--- a/source/common/tcp_proxy/tcp_proxy.h
+++ b/source/common/tcp_proxy/tcp_proxy.h
@@ -11,6 +11,7 @@
 #include "envoy/event/timer.h"
 #include "envoy/network/connection.h"
 #include "envoy/network/filter.h"
+#include "envoy/request_info/filter_state.h"
 #include "envoy/runtime/runtime.h"
 #include "envoy/server/filter_config.h"
 #include "envoy/stats/scope.h"
@@ -154,6 +155,19 @@ class Config {
 
 typedef std::shared_ptr<Config> ConfigSharedPtr;
 
+/**
+ * Per-connection TCP Proxy Cluster configuration.
+ */
+class PerConnectionCluster : public RequestInfo::FilterState::Object {
+public:
+  PerConnectionCluster(absl::string_view cluster) : cluster_(cluster) {}
+  const std::string& value() const { return cluster_; }
+  static const std::string Key;
+
+private:
+  const std::string cluster_;
+};
+
 /**
  * An implementation of a TCP (L3/L4) proxy. This filter will instantiate a new outgoing TCP
  * connection using the defined load balancing proxy for the configured cluster. All data will
diff --git a/source/extensions/extensions_build_config.bzl b/source/extensions/extensions_build_config.bzl
index eaf1045aaba..f3d7f9ef60d 100644
--- a/source/extensions/extensions_build_config.bzl
+++ b/source/extensions/extensions_build_config.bzl
@@ -70,6 +70,7 @@ EXTENSIONS = {
     "envoy.filters.network.redis_proxy":                "//source/extensions/filters/network/redis_proxy:config",
     "envoy.filters.network.tcp_proxy":                  "//source/extensions/filters/network/tcp_proxy:config",
     "envoy.filters.network.thrift_proxy":               "//source/extensions/filters/network/thrift_proxy:config",
+    "envoy.filters.network.sni_cluster":                "//source/extensions/filters/network/sni_cluster:config",
 
     #
     # Resource monitors
@@ -181,6 +182,7 @@ WINDOWS_EXTENSIONS = {
     #"envoy.filters.network.ratelimit":                  "//source/extensions/filters/network/ratelimit:config",
     "envoy.filters.network.tcp_proxy":                  "//source/extensions/filters/network/tcp_proxy:config",
     #"envoy.filters.network.thrift_proxy":               "//source/extensions/filters/network/thrift_proxy:config",
+    #"envoy.filters.network.sni_cluster":                "//source/extensions/filters/network/sni_cluster:config",
 
     #
     # Stat sinks
diff --git a/source/extensions/filters/network/sni_cluster/BUILD b/source/extensions/filters/network/sni_cluster/BUILD
new file mode 100644
index 00000000000..60eec7e5c92
--- /dev/null
+++ b/source/extensions/filters/network/sni_cluster/BUILD
@@ -0,0 +1,34 @@
+licenses(["notice"])  # Apache 2
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+    name = "sni_cluster",
+    srcs = ["sni_cluster.cc"],
+    hdrs = ["sni_cluster.h"],
+    deps = [
+        "//include/envoy/network:connection_interface",
+        "//include/envoy/network:filter_interface",
+        "//source/common/common:assert_lib",
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/tcp_proxy",
+    ],
+)
+
+envoy_cc_library(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        ":sni_cluster",
+        "//include/envoy/registry",
+        "//include/envoy/server:filter_config_interface",
+        "//source/extensions/filters/network:well_known_names",
+    ],
+)
diff --git a/source/extensions/filters/network/sni_cluster/config.cc b/source/extensions/filters/network/sni_cluster/config.cc
new file mode 100644
index 00000000000..8d7b6eba067
--- /dev/null
+++ b/source/extensions/filters/network/sni_cluster/config.cc
@@ -0,0 +1,41 @@
+#include "extensions/filters/network/sni_cluster/config.h"
+
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+
+#include "extensions/filters/network/sni_cluster/sni_cluster.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace SniCluster {
+
+Network::FilterFactoryCb
+SniClusterNetworkFilterConfigFactory::createFilterFactory(const Json::Object&,
+                                                          Server::Configuration::FactoryContext&) {
+  // Only used in v1 filters.
+  NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+}
+
+Network::FilterFactoryCb SniClusterNetworkFilterConfigFactory::createFilterFactoryFromProto(
+    const Protobuf::Message&, Server::Configuration::FactoryContext&) {
+  return [](Network::FilterManager& filter_manager) -> void {
+    filter_manager.addReadFilter(std::make_shared<SniClusterFilter>());
+  };
+}
+
+ProtobufTypes::MessagePtr SniClusterNetworkFilterConfigFactory::createEmptyConfigProto() {
+  return std::make_unique<ProtobufWkt::Empty>();
+}
+
+/**
+ * Static registration for the sni_cluster filter. @see RegisterFactory.
+ */
+static Registry::RegisterFactory<SniClusterNetworkFilterConfigFactory,
+                                 Server::Configuration::NamedNetworkFilterConfigFactory>
+    registered_;
+
+} // namespace SniCluster
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/sni_cluster/config.h b/source/extensions/filters/network/sni_cluster/config.h
new file mode 100644
index 00000000000..b4edcd51360
--- /dev/null
+++ b/source/extensions/filters/network/sni_cluster/config.h
@@ -0,0 +1,31 @@
+#pragma once
+
+#include "envoy/server/filter_config.h"
+
+#include "extensions/filters/network/well_known_names.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace SniCluster {
+
+/**
+ * Config registration for the sni_cluster filter. @see NamedNetworkFilterConfigFactory.
+ */
+class SniClusterNetworkFilterConfigFactory
+    : public Server::Configuration::NamedNetworkFilterConfigFactory {
+public:
+  // NamedNetworkFilterConfigFactory
+  Network::FilterFactoryCb createFilterFactory(const Json::Object&,
+                                               Server::Configuration::FactoryContext&) override;
+  Network::FilterFactoryCb
+  createFilterFactoryFromProto(const Protobuf::Message&,
+                               Server::Configuration::FactoryContext&) override;
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override;
+  std::string name() override { return NetworkFilterNames::get().SniCluster; }
+};
+
+} // namespace SniCluster
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/sni_cluster/sni_cluster.cc b/source/extensions/filters/network/sni_cluster/sni_cluster.cc
new file mode 100644
index 00000000000..f5acb2b917b
--- /dev/null
+++ b/source/extensions/filters/network/sni_cluster/sni_cluster.cc
@@ -0,0 +1,30 @@
+#include "extensions/filters/network/sni_cluster/sni_cluster.h"
+
+#include "envoy/network/connection.h"
+
+#include "common/common/assert.h"
+#include "common/tcp_proxy/tcp_proxy.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace SniCluster {
+
+Network::FilterStatus SniClusterFilter::onNewConnection() {
+  absl::string_view sni = read_callbacks_->connection().requestedServerName();
+  ENVOY_CONN_LOG(trace, "sni_cluster: new connection with server name {}",
+                 read_callbacks_->connection(), sni);
+
+  if (!sni.empty()) {
+    // Set the tcp_proxy cluster to the same value as SNI
+    read_callbacks_->connection().perConnectionState().setData(
+        TcpProxy::PerConnectionCluster::Key, std::make_unique<TcpProxy::PerConnectionCluster>(sni));
+  }
+
+  return Network::FilterStatus::Continue;
+}
+
+} // namespace SniCluster
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/sni_cluster/sni_cluster.h b/source/extensions/filters/network/sni_cluster/sni_cluster.h
new file mode 100644
index 00000000000..d062368ac66
--- /dev/null
+++ b/source/extensions/filters/network/sni_cluster/sni_cluster.h
@@ -0,0 +1,34 @@
+#pragma once
+
+#include "envoy/network/filter.h"
+
+#include "common/common/logger.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace SniCluster {
+
+/**
+ * Implementation of the sni_cluster filter that sets the upstream cluster name from
+ * the SNI field in the TLS connection.
+ */
+class SniClusterFilter : public Network::ReadFilter, Logger::Loggable<Logger::Id::filter> {
+public:
+  // Network::ReadFilter
+  Network::FilterStatus onData(Buffer::Instance&, bool) override {
+    return Network::FilterStatus::Continue;
+  }
+  Network::FilterStatus onNewConnection() override;
+  void initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) override {
+    read_callbacks_ = &callbacks;
+  }
+
+private:
+  Network::ReadFilterCallbacks* read_callbacks_{};
+};
+
+} // namespace SniCluster
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy
diff --git a/source/extensions/filters/network/well_known_names.h b/source/extensions/filters/network/well_known_names.h
index 06123fd6df5..de09145f83a 100644
--- a/source/extensions/filters/network/well_known_names.h
+++ b/source/extensions/filters/network/well_known_names.h
@@ -32,6 +32,8 @@ class NetworkFilterNameValues {
   const std::string ThriftProxy = "envoy.filters.network.thrift_proxy";
   // Role based access control filter
   const std::string Rbac = "envoy.filters.network.rbac";
+  // SNI Cluster filter
+  const std::string SniCluster = "envoy.filters.network.sni_cluster";
 
   // Converts names from v1 to v2
   const Config::V1Converter v1_converter_;
diff --git a/test/common/tcp_proxy/tcp_proxy_test.cc b/test/common/tcp_proxy/tcp_proxy_test.cc
index d3c71741a41..26edf3a1da3 100644
--- a/test/common/tcp_proxy/tcp_proxy_test.cc
+++ b/test/common/tcp_proxy/tcp_proxy_test.cc
@@ -1142,5 +1142,24 @@ TEST_F(TcpProxyRoutingTest, RoutableConnection) {
   EXPECT_EQ(non_routable_cx, config_->stats().downstream_cx_no_route_.value());
 }
 
+// Test that the tcp proxy uses the cluster from FilterState if set
+TEST_F(TcpProxyRoutingTest, UseClusterFromPerConnectionCluster) {
+  setup();
+
+  RequestInfo::FilterStateImpl per_connection_state;
+  per_connection_state.setData("envoy.tcp_proxy.cluster",
+                               std::make_unique<PerConnectionCluster>("filter_state_cluster"));
+  ON_CALL(connection_, perConnectionState()).WillByDefault(ReturnRef(per_connection_state));
+  EXPECT_CALL(Const(connection_), perConnectionState())
+      .WillRepeatedly(ReturnRef(per_connection_state));
+
+  // Expect filter to try to open a connection to specified cluster.
+  EXPECT_CALL(factory_context_.cluster_manager_,
+              tcpConnPoolForCluster("filter_state_cluster", _, _))
+      .WillOnce(Return(nullptr));
+
+  filter_->onNewConnection();
+}
+
 } // namespace TcpProxy
 } // namespace Envoy
diff --git a/test/extensions/filters/network/sni_cluster/BUILD b/test/extensions/filters/network/sni_cluster/BUILD
new file mode 100644
index 00000000000..a7d00cbff51
--- /dev/null
+++ b/test/extensions/filters/network/sni_cluster/BUILD
@@ -0,0 +1,25 @@
+licenses(["notice"])  # Apache 2
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_test",
+    "envoy_package",
+)
+load(
+    "//test/extensions:extensions_build_system.bzl",
+    "envoy_extension_cc_test",
+)
+
+envoy_package()
+
+envoy_extension_cc_test(
+    name = "sni_cluster_test",
+    srcs = ["sni_cluster_test.cc"],
+    extension_name = "envoy.filters.network.sni_cluster",
+    deps = [
+        "//source/extensions/filters/network/sni_cluster",
+        "//source/extensions/filters/network/sni_cluster:config",
+        "//test/mocks/network:network_mocks",
+        "//test/mocks/server:server_mocks",
+    ],
+)
diff --git a/test/extensions/filters/network/sni_cluster/sni_cluster_test.cc b/test/extensions/filters/network/sni_cluster/sni_cluster_test.cc
new file mode 100644
index 00000000000..4d833d378ef
--- /dev/null
+++ b/test/extensions/filters/network/sni_cluster/sni_cluster_test.cc
@@ -0,0 +1,76 @@
+#include "common/tcp_proxy/tcp_proxy.h"
+
+#include "extensions/filters/network/sni_cluster/config.h"
+#include "extensions/filters/network/sni_cluster/sni_cluster.h"
+
+#include "test/mocks/network/mocks.h"
+#include "test/mocks/server/mocks.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+using testing::_;
+using testing::Matcher;
+using testing::NiceMock;
+using testing::Return;
+using testing::ReturnRef;
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace SniCluster {
+
+// Test that a SniCluster filter config works.
+TEST(SniCluster, ConfigTest) {
+  NiceMock<Server::Configuration::MockFactoryContext> context;
+  SniClusterNetworkFilterConfigFactory factory;
+
+  Network::FilterFactoryCb cb =
+      factory.createFilterFactoryFromProto(*factory.createEmptyConfigProto(), context);
+  Network::MockConnection connection;
+  EXPECT_CALL(connection, addReadFilter(_));
+  cb(connection);
+}
+
+// Test that per connection filter config is set if SNI is available
+TEST(SniCluster, SetTcpProxyClusterOnlyIfSniIsPresent) {
+  NiceMock<Network::MockReadFilterCallbacks> filter_callbacks;
+
+  RequestInfo::FilterStateImpl per_connection_state;
+  ON_CALL(filter_callbacks.connection_, perConnectionState())
+      .WillByDefault(ReturnRef(per_connection_state));
+  ON_CALL(Const(filter_callbacks.connection_), perConnectionState())
+      .WillByDefault(ReturnRef(per_connection_state));
+
+  SniClusterFilter filter;
+  filter.initializeReadFilterCallbacks(filter_callbacks);
+
+  // no sni
+  {
+    ON_CALL(filter_callbacks.connection_, requestedServerName())
+        .WillByDefault(Return(EMPTY_STRING));
+    filter.onNewConnection();
+
+    EXPECT_FALSE(per_connection_state.hasData<TcpProxy::PerConnectionCluster>(
+        TcpProxy::PerConnectionCluster::Key));
+  }
+
+  // with sni
+  {
+    ON_CALL(filter_callbacks.connection_, requestedServerName())
+        .WillByDefault(Return("filter_state_cluster"));
+    filter.onNewConnection();
+
+    EXPECT_TRUE(per_connection_state.hasData<TcpProxy::PerConnectionCluster>(
+        TcpProxy::PerConnectionCluster::Key));
+
+    auto per_connection_cluster = per_connection_state.getData<TcpProxy::PerConnectionCluster>(
+        TcpProxy::PerConnectionCluster::Key);
+    EXPECT_EQ(per_connection_cluster.value(), "filter_state_cluster");
+  }
+}
+
+} // namespace SniCluster
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy