diff --git a/source/common/upstream/BUILD b/source/common/upstream/BUILD
index 31dfdec940..03cff715ed 100644
--- a/source/common/upstream/BUILD
+++ b/source/common/upstream/BUILD
@@ -265,6 +265,7 @@ envoy_cc_library(
         "//include/envoy/upstream:cluster_factory_interface",
         "//source/common/common:empty_string",
         "//source/common/network:address_lib",
+        "//source/common/network:transport_socket_options_lib",
         "//source/common/network:utility_lib",
         "//source/extensions/clusters:well_known_names",
         "@envoy_api//envoy/config/cluster/v3:pkg_cc_proto",
diff --git a/source/common/upstream/original_dst_cluster.cc b/source/common/upstream/original_dst_cluster.cc
index 69b0e3b7f2..6bbeb18279 100644
--- a/source/common/upstream/original_dst_cluster.cc
+++ b/source/common/upstream/original_dst_cluster.cc
@@ -13,14 +13,25 @@
 
 #include "common/http/headers.h"
 #include "common/network/address_impl.h"
+#include "common/network/transport_socket_options_impl.h"
 #include "common/network/utility.h"
 #include "common/protobuf/protobuf.h"
 #include "common/protobuf/utility.h"
 
 namespace Envoy {
 namespace Upstream {
+namespace OriginalDst {
 
-HostConstSharedPtr OriginalDstCluster::LoadBalancer::chooseHost(LoadBalancerContext* context) {
+Upstream::Host::CreateConnectionData Cluster::Host::createConnection(
+    Event::Dispatcher& dispatcher, const Network::ConnectionSocket::OptionsSharedPtr& options,
+    Network::TransportSocketOptionsSharedPtr transport_socket_options) const {
+  return HostImpl::createConnection(dispatcher, options,
+                                    transport_socket_options != nullptr
+                                        ? transport_socket_options
+                                        : default_transport_socket_options_);
+}
+
+HostConstSharedPtr Cluster::LoadBalancer::chooseHost(LoadBalancerContext* context) {
   if (context) {
     // Check if override host header is present, if yes use it otherwise check local address.
     Network::Address::InstanceConstSharedPtr dst_host = nullptr;
@@ -51,22 +62,73 @@ HostConstSharedPtr OriginalDstCluster::LoadBalancer::chooseHost(LoadBalancerCont
       if (dst_ip) {
         Network::Address::InstanceConstSharedPtr host_ip_port(
             Network::Utility::copyInternetAddressAndPort(*dst_ip));
+        Network::TransportSocketOptionsSharedPtr transport_socket_options;
+
+        // Create an override transport socket options that automatically provides both SNI as well
+        // as SAN verification for the new host if the cluster has been configured with TLS.
+        if (parent_->implements_secure_transport_) {
+          absl::string_view sni;
+
+          // First check if have requested server name in the downstream connection
+          // This may be set by a TLS inspector listener socket
+          const Network::Connection* downstream_connection = context->downstreamConnection();
+          if (downstream_connection) {
+            sni = downstream_connection->requestedServerName();
+          }
+
+          if (sni.length() > 0) {
+            ENVOY_LOG(debug, "Using SNI from downstream connection: {}.", sni);
+          } else {
+            // If no SNI from downstream connection, check if have HTTP Host in downstream headers.
+            const Http::HeaderMap* downstream_headers = context->downstreamHeaders();
+            if (downstream_headers) {
+              auto host_header = downstream_headers->get(Http::Headers::get().Host);
+              if (host_header != nullptr) {
+                sni = host_header->value().getStringView();
+                // port in Host is optional, so cut it off, if any
+                const auto pos = sni.rfind(':');
+                if (pos != std::string::npos) {
+                  sni.remove_suffix(sni.length() - pos);
+                }
+                // Remove brackets if any
+                if (sni.front() == '[' && sni.back() == ']') {
+                  sni.remove_prefix(1);
+                  sni.remove_suffix(1);
+                }
+                // Try parse the host as an IP address
+                try {
+                  Network::Utility::parseInternetAddress(std::string(sni));
+                  sni = "";
+                } catch (const EnvoyException&) {
+                  // Parsing as an IP address failed, so it should be a domain name
+                  ENVOY_LOG(debug, "Using SNI from Host header: {}.", sni);
+                }
+              }
+            }
+          }
+
+          if (sni.length() > 0) {
+            transport_socket_options = std::make_shared<Network::TransportSocketOptionsImpl>(
+                sni, std::vector<std::string>{std::string(sni)});
+          }
+        }
+
         // Create a host we can use immediately.
         auto info = parent_->info();
-        HostSharedPtr host(std::make_shared<HostImpl>(
+        HostSharedPtr host(std::make_shared<Host>(
             info, info->name() + dst_addr.asString(), std::move(host_ip_port),
             envoy::config::core::v3::Metadata::default_instance(), 1,
             envoy::config::core::v3::Locality().default_instance(),
             envoy::config::endpoint::v3::Endpoint::HealthCheckConfig().default_instance(), 0,
-            envoy::config::core::v3::UNKNOWN));
+            envoy::config::core::v3::UNKNOWN, transport_socket_options));
         ENVOY_LOG(debug, "Created host {}.", host->address()->asString());
 
         // Tell the cluster about the new host
         // lambda cannot capture a member by value.
-        std::weak_ptr<OriginalDstCluster> post_parent = parent_;
+        std::weak_ptr<Cluster> post_parent = parent_;
         parent_->dispatcher_.post([post_parent, host]() mutable {
           // The main cluster may have disappeared while this post was queued.
-          if (std::shared_ptr<OriginalDstCluster> parent = post_parent.lock()) {
+          if (std::shared_ptr<Cluster> parent = post_parent.lock()) {
             parent->addHost(host);
           }
         });
@@ -82,7 +144,7 @@ HostConstSharedPtr OriginalDstCluster::LoadBalancer::chooseHost(LoadBalancerCont
 }
 
 Network::Address::InstanceConstSharedPtr
-OriginalDstCluster::LoadBalancer::requestOverrideHost(LoadBalancerContext* context) {
+Cluster::LoadBalancer::requestOverrideHost(LoadBalancerContext* context) {
   Network::Address::InstanceConstSharedPtr request_host;
   const Http::HeaderMap* downstream_headers = context->downstreamHeaders();
   if (downstream_headers &&
@@ -102,10 +164,9 @@ OriginalDstCluster::LoadBalancer::requestOverrideHost(LoadBalancerContext* conte
   return request_host;
 }
 
-OriginalDstCluster::OriginalDstCluster(
-    const envoy::config::cluster::v3::Cluster& config, Runtime::Loader& runtime,
-    Server::Configuration::TransportSocketFactoryContext& factory_context,
-    Stats::ScopePtr&& stats_scope, bool added_via_api)
+Cluster::Cluster(const envoy::config::cluster::v3::Cluster& config, Runtime::Loader& runtime,
+                 Server::Configuration::TransportSocketFactoryContext& factory_context,
+                 Stats::ScopePtr&& stats_scope, bool added_via_api)
     : ClusterImplBase(config, runtime, factory_context, std::move(stats_scope), added_via_api),
       dispatcher_(factory_context.dispatcher()),
       cleanup_interval_ms_(
@@ -114,15 +175,31 @@ OriginalDstCluster::OriginalDstCluster(
       use_http_header_(info_->lbOriginalDstConfig()
                            ? info_->lbOriginalDstConfig().value().use_http_header()
                            : false),
+      implements_secure_transport_(info_->transportSocketMatcher()
+                                       .resolve(envoy::config::core::v3::Metadata())
+                                       .factory_.implementsSecureTransport()),
       host_map_(std::make_shared<HostMap>()) {
   // TODO(dio): Remove hosts check once the hosts field is removed.
   if (config.has_load_assignment() || !config.hidden_envoy_deprecated_hosts().empty()) {
     throw EnvoyException("ORIGINAL_DST clusters must have no load assignment or hosts configured");
   }
+  // Block certain TLS context parameters that don't make sense on a cluster-wide scale. We will
+  // support these parameters dynamically in the future. This is not an exhaustive list of
+  // parameters that don't make sense but should be the most obvious ones that a user might set
+  // in error.
+  if (!config.hidden_envoy_deprecated_tls_context().sni().empty() ||
+      !config.hidden_envoy_deprecated_tls_context().common_tls_context()
+                                                   .validation_context()
+                                                   .hidden_envoy_deprecated_verify_subject_alt_name()
+                                                   .empty()) {
+    throw EnvoyException(
+        "ORIGINAL_DST cluster cannot configure 'sni' or 'verify_subject_alt_name'");
+  }
+
   cleanup_timer_->enableTimer(cleanup_interval_ms_);
 }
 
-void OriginalDstCluster::addHost(HostSharedPtr& host) {
+void Cluster::addHost(HostSharedPtr& host) {
   HostMapSharedPtr new_host_map = std::make_shared<HostMap>(*getCurrentHostMap());
   auto pair = new_host_map->emplace(host->address()->asString(), host);
   bool added = pair.second;
@@ -140,7 +217,7 @@ void OriginalDstCluster::addHost(HostSharedPtr& host) {
   }
 }
 
-void OriginalDstCluster::cleanup() {
+void Cluster::cleanup() {
   HostVectorSharedPtr keeping_hosts(new HostVector);
   HostVector to_be_removed;
   ENVOY_LOG(trace, "Stale original dst hosts cleanup triggered.");
@@ -163,7 +240,7 @@ void OriginalDstCluster::cleanup() {
 
   if (!to_be_removed.empty()) {
     HostMapSharedPtr new_host_map = std::make_shared<HostMap>(*host_map);
-    for (const HostSharedPtr& host : to_be_removed) {
+    for (const Upstream::HostSharedPtr& host : to_be_removed) {
       new_host_map->erase(host->address()->asString());
     }
     setHostMap(new_host_map);
@@ -175,8 +252,7 @@ void OriginalDstCluster::cleanup() {
   cleanup_timer_->enableTimer(cleanup_interval_ms_);
 }
 
-std::pair<ClusterImplBaseSharedPtr, ThreadAwareLoadBalancerPtr>
-OriginalDstClusterFactory::createClusterImpl(
+std::pair<ClusterImplBaseSharedPtr, ThreadAwareLoadBalancerPtr> ClusterFactory::createClusterImpl(
     const envoy::config::cluster::v3::Cluster& cluster, ClusterFactoryContext& context,
     Server::Configuration::TransportSocketFactoryContext& socket_factory_context,
     Stats::ScopePtr&& stats_scope) {
@@ -190,20 +266,19 @@ OriginalDstClusterFactory::createClusterImpl(
         envoy::config::cluster::v3::Cluster::DiscoveryType_Name(cluster.type())));
   }
 
-  // TODO(mattklein123): The original DST load balancer type should be deprecated and instead
-  //                     the cluster should directly supply the load balancer. This will remove
-  //                     a special case and allow this cluster to be compiled out as an extension.
-  auto new_cluster =
-      std::make_shared<OriginalDstCluster>(cluster, context.runtime(), socket_factory_context,
-                                           std::move(stats_scope), context.addedViaApi());
-  auto lb = std::make_unique<OriginalDstCluster::ThreadAwareLoadBalancer>(new_cluster);
+  // TODO(mattklein123): Compile this cluster as an extension when the now deprecated original DST
+  //                     load balancer type is removed.
+  auto new_cluster = std::make_shared<Cluster>(cluster, context.runtime(), socket_factory_context,
+                                               std::move(stats_scope), context.addedViaApi());
+  auto lb = std::make_unique<Cluster::ThreadAwareLoadBalancer>(new_cluster);
   return std::make_pair(new_cluster, std::move(lb));
 }
 
 /**
  * Static registration for the original dst cluster factory. @see RegisterFactory.
  */
-REGISTER_FACTORY(OriginalDstClusterFactory, ClusterFactory);
+REGISTER_FACTORY(ClusterFactory, Upstream::ClusterFactory);
 
+} // namespace OriginalDst
 } // namespace Upstream
 } // namespace Envoy
diff --git a/source/common/upstream/original_dst_cluster.h b/source/common/upstream/original_dst_cluster.h
index 148111cd0e..d0acdf4efa 100644
--- a/source/common/upstream/original_dst_cluster.h
+++ b/source/common/upstream/original_dst_cluster.h
@@ -20,31 +20,60 @@
 
 namespace Envoy {
 namespace Upstream {
-
-using HostMapSharedPtr = std::shared_ptr<HostMap>;
-using HostMapConstSharedPtr = std::shared_ptr<const HostMap>;
+namespace OriginalDst {
 
 /**
- * The OriginalDstCluster is a dynamic cluster that automatically adds hosts as needed based on the
+ * The OriginalDst Cluster is a dynamic cluster that automatically adds hosts as needed based on the
  * original destination address of the downstream connection. These hosts are also automatically
  * cleaned up after they have not seen traffic for a configurable cleanup interval time
  * ("cleanup_interval_ms").
  */
-class OriginalDstCluster : public ClusterImplBase {
+class Cluster : public ClusterImplBase {
 public:
-  OriginalDstCluster(const envoy::config::cluster::v3::Cluster& config, Runtime::Loader& runtime,
-                     Server::Configuration::TransportSocketFactoryContext& factory_context,
-                     Stats::ScopePtr&& stats_scope, bool added_via_api);
+  Cluster(const envoy::config::cluster::v3::Cluster& config, Runtime::Loader& runtime,
+          Server::Configuration::TransportSocketFactoryContext& factory_context,
+          Stats::ScopePtr&& stats_scope, bool added_via_api);
 
   // Upstream::Cluster
   InitializePhase initializePhase() const override { return InitializePhase::Primary; }
 
+  /**
+   * A host implementation that supports default transport socket options.
+   */
+  class Host : public HostImpl {
+  public:
+    Host(const ClusterInfoConstSharedPtr& cluster, const std::string& hostname,
+         const Network::Address::InstanceConstSharedPtr& address,
+         const envoy::config::core::v3::Metadata& metadata, uint32_t initial_weight,
+         const envoy::config::core::v3::Locality& locality,
+         const envoy::config::endpoint::v3::Endpoint::HealthCheckConfig& health_check_config,
+         uint32_t priority, const envoy::config::core::v3::HealthStatus health_status,
+         const Network::TransportSocketOptionsSharedPtr& default_transport_socket_options)
+        : HostImpl(cluster, hostname, address, metadata, initial_weight, locality,
+                   health_check_config, priority, health_status),
+          default_transport_socket_options_(default_transport_socket_options) {}
+
+    // Upstream::Host
+    CreateConnectionData createConnection(
+        Event::Dispatcher& dispatcher, const Network::ConnectionSocket::OptionsSharedPtr& options,
+        Network::TransportSocketOptionsSharedPtr transport_socket_options) const override;
+
+  private:
+    const Network::TransportSocketOptionsSharedPtr default_transport_socket_options_;
+
+    friend class ClusterTest;
+  };
+  using HostSharedPtr = std::shared_ptr<Host>;
+  using HostMap = std::unordered_map<std::string, HostSharedPtr>;
+  using HostMapSharedPtr = std::shared_ptr<HostMap>;
+  using HostMapConstSharedPtr = std::shared_ptr<const HostMap>;
+
   /**
    * Special Load Balancer for Original Dst Cluster.
    *
    * Load balancer gets called with the downstream context which can be used to make sure the
    * Original Dst cluster has a Host for the original destination. Normally load balancers can't
-   * modify clusters, but in this case we access a singleton OriginalDstCluster that we can ask to
+   * modify clusters, but in this case we access a singleton OriginalDst Cluster that we can ask to
    * add hosts on demand. Additions are synced with all other threads so that the host set in the
    * cluster remains (eventually) consistent. If multiple threads add a host to the same upstream
    * address then two distinct HostSharedPtr's (with the same upstream IP address) will be added,
@@ -52,7 +81,7 @@ public:
    */
   class LoadBalancer : public Upstream::LoadBalancer {
   public:
-    LoadBalancer(const std::shared_ptr<OriginalDstCluster>& parent)
+    LoadBalancer(const std::shared_ptr<Cluster>& parent)
         : parent_(parent), host_map_(parent->getCurrentHostMap()) {}
 
     // Upstream::LoadBalancer
@@ -61,23 +90,22 @@ public:
   private:
     Network::Address::InstanceConstSharedPtr requestOverrideHost(LoadBalancerContext* context);
 
-    const std::shared_ptr<OriginalDstCluster> parent_;
+    const std::shared_ptr<Cluster> parent_;
     HostMapConstSharedPtr host_map_;
   };
 
 private:
   struct LoadBalancerFactory : public Upstream::LoadBalancerFactory {
-    LoadBalancerFactory(const std::shared_ptr<OriginalDstCluster>& cluster) : cluster_(cluster) {}
+    LoadBalancerFactory(const std::shared_ptr<Cluster>& cluster) : cluster_(cluster) {}
 
     // Upstream::LoadBalancerFactory
     Upstream::LoadBalancerPtr create() override { return std::make_unique<LoadBalancer>(cluster_); }
 
-    const std::shared_ptr<OriginalDstCluster> cluster_;
+    const std::shared_ptr<Cluster> cluster_;
   };
 
   struct ThreadAwareLoadBalancer : public Upstream::ThreadAwareLoadBalancer {
-    ThreadAwareLoadBalancer(const std::shared_ptr<OriginalDstCluster>& cluster)
-        : cluster_(cluster) {}
+    ThreadAwareLoadBalancer(const std::shared_ptr<Cluster>& cluster) : cluster_(cluster) {}
 
     // Upstream::ThreadAwareLoadBalancer
     Upstream::LoadBalancerFactorySharedPtr factory() override {
@@ -85,7 +113,7 @@ private:
     }
     void initialize() override {}
 
-    const std::shared_ptr<OriginalDstCluster> cluster_;
+    const std::shared_ptr<Cluster> cluster_;
   };
 
   HostMapConstSharedPtr getCurrentHostMap() {
@@ -108,18 +136,20 @@ private:
   const std::chrono::milliseconds cleanup_interval_ms_;
   Event::TimerPtr cleanup_timer_;
   const bool use_http_header_;
+  const bool implements_secure_transport_;
 
   absl::Mutex host_map_lock_;
   HostMapConstSharedPtr host_map_ ABSL_GUARDED_BY(host_map_lock_);
 
-  friend class OriginalDstClusterFactory;
+  friend class ClusterFactory;
+  friend class ClusterTest;
 };
 
-using OriginalDstClusterSharedPtr = std::shared_ptr<OriginalDstCluster>;
+using ClusterSharedPtr = std::shared_ptr<Cluster>;
 
-class OriginalDstClusterFactory : public ClusterFactoryImplBase {
+class ClusterFactory : public ClusterFactoryImplBase {
 public:
-  OriginalDstClusterFactory()
+  ClusterFactory()
       : ClusterFactoryImplBase(Extensions::Clusters::ClusterTypes::get().OriginalDst) {}
 
 private:
@@ -130,5 +160,6 @@ private:
                     Stats::ScopePtr&& stats_scope) override;
 };
 
+} // namespace OriginalDst
 } // namespace Upstream
 } // namespace Envoy
diff --git a/test/common/upstream/BUILD b/test/common/upstream/BUILD
index c53b8368e9..b558158a05 100644
--- a/test/common/upstream/BUILD
+++ b/test/common/upstream/BUILD
@@ -235,6 +235,7 @@ envoy_cc_test(
 envoy_cc_test(
     name = "original_dst_cluster_test",
     srcs = ["original_dst_cluster_test.cc"],
+    data = ["//test/extensions/transport_sockets/tls/test_data:certs"],
     deps = [
         ":utility_lib",
         "//source/common/event:dispatcher_lib",
@@ -242,6 +243,7 @@ envoy_cc_test(
         "//source/common/upstream:original_dst_cluster_lib",
         "//source/common/upstream:upstream_lib",
         "//source/extensions/transport_sockets/raw_buffer:config",
+        "//source/extensions/transport_sockets/tls:config",
         "//test/mocks:common_lib",
         "//test/mocks/local_info:local_info_mocks",
         "//test/mocks/network:network_mocks",
@@ -250,6 +252,7 @@ envoy_cc_test(
         "//test/mocks/server:server_mocks",
         "//test/mocks/ssl:ssl_mocks",
         "//test/mocks/upstream:upstream_mocks",
+        "//test/test_common:environment_lib",
         "//test/test_common:utility_lib",
         "@envoy_api//envoy/config/cluster/v3:pkg_cc_proto",
     ],
diff --git a/test/common/upstream/original_dst_cluster_test.cc b/test/common/upstream/original_dst_cluster_test.cc
index 558f70bdee..9c2d95c5b9 100644
--- a/test/common/upstream/original_dst_cluster_test.cc
+++ b/test/common/upstream/original_dst_cluster_test.cc
@@ -24,6 +24,7 @@
 #include "test/mocks/server/mocks.h"
 #include "test/mocks/ssl/mocks.h"
 #include "test/mocks/upstream/mocks.h"
+#include "test/test_common/environment.h"
 #include "test/test_common/utility.h"
 
 #include "gmock/gmock.h"
@@ -36,7 +37,7 @@ using testing::SaveArg;
 
 namespace Envoy {
 namespace Upstream {
-namespace {
+namespace OriginalDst {
 
 class TestLoadBalancerContext : public LoadBalancerContextBase {
 public:
@@ -57,18 +58,18 @@ public:
   Http::HeaderMapPtr downstream_headers_;
 };
 
-class OriginalDstClusterTest : public testing::Test {
+class ClusterTest : public testing::Test {
 public:
   // cleanup timer must be created before the cluster (in setup()), so that we can set expectations
   // on it. Ownership is transferred to the cluster at the cluster constructor, so the cluster will
   // take care of destructing it!
-  OriginalDstClusterTest()
+  ClusterTest()
       : cleanup_timer_(new Event::MockTimer(&dispatcher_)),
         api_(Api::createApiForTest(stats_store_)) {}
 
-  void setupFromYaml(const std::string& yaml) { setup(parseClusterFromV2Yaml(yaml)); }
+  void setupFromYaml(const std::string& yaml, bool uses_tls = false) {
+    const envoy::config::cluster::v3::Cluster& cluster_config = parseClusterFromV2Yaml(yaml);
 
-  void setup(const envoy::config::cluster::v3::Cluster& cluster_config) {
     NiceMock<MockClusterManager> cm;
     Envoy::Stats::ScopePtr scope = stats_store_.createScope(fmt::format(
         "cluster.{}.", cluster_config.alt_stat_name().empty() ? cluster_config.name()
@@ -76,18 +77,37 @@ public:
     Envoy::Server::Configuration::TransportSocketFactoryContextImpl factory_context(
         admin_, ssl_context_manager_, *scope, cm, local_info_, dispatcher_, random_, stats_store_,
         singleton_manager_, tls_, validation_visitor_, *api_);
-    cluster_.reset(
-        new OriginalDstCluster(cluster_config, runtime_, factory_context, std::move(scope), false));
+    if (uses_tls) {
+      EXPECT_CALL(ssl_context_manager_, createSslClientContext(_, _));
+    }
+
+    cluster_.reset(new Cluster(cluster_config, runtime_, factory_context, std::move(scope), false));
+    thread_aware_lb_ = std::make_unique<Cluster::ThreadAwareLoadBalancer>(cluster_);
+    lb_factory_ = thread_aware_lb_->factory();
+    refreshLb();
+
     cluster_->prioritySet().addPriorityUpdateCb(
-        [&](uint32_t, const HostVector&, const HostVector&) -> void {
+        [&](uint32_t, const HostVector& hosts_added, const HostVector& hosts_removed) -> void {
+          // Mock Cluster Manager by refreshing the load balancer when membership has changed
+          if (hosts_added.size() || hosts_removed.size()) {
+            ENVOY_LOG_MISC(debug, "Refreshing LoadBalancer");
+            refreshLb();
+          }
           membership_updated_.ready();
         });
     cluster_->initialize([&]() -> void { initialized_.ready(); });
   }
 
+  void refreshLb() { lb_ = lb_factory_->create(); }
+
+  const Network::TransportSocketOptionsSharedPtr
+  getDefaultTransportSocketOptions(const Cluster::Host* host) {
+    return host->default_transport_socket_options_;
+  }
+
   Stats::IsolatedStoreImpl stats_store_;
   Ssl::MockContextManager ssl_context_manager_;
-  OriginalDstClusterSharedPtr cluster_;
+  ClusterSharedPtr cluster_;
   ReadyWatcher membership_updated_;
   ReadyWatcher initialized_;
   NiceMock<Runtime::MockLoader> runtime_;
@@ -100,27 +120,83 @@ public:
   NiceMock<ThreadLocal::MockInstance> tls_;
   NiceMock<ProtobufMessage::MockValidationVisitor> validation_visitor_;
   Api::ApiPtr api_;
+  Upstream::ThreadAwareLoadBalancerPtr thread_aware_lb_;
+  Upstream::LoadBalancerFactorySharedPtr lb_factory_;
+  Upstream::LoadBalancerPtr lb_;
+
+  const std::string default_yaml_ = R"EOF(
+    name: name
+    connect_timeout: 1.250s
+    type: original_dst
+    lb_policy: cluster_provided
+  )EOF";
+
+  const std::string default_tls_yaml_ = default_yaml_ + TestEnvironment::substitute(R"EOF(
+    tls_context:
+      common_tls_context:
+        validation_context:
+          trusted_ca:
+            filename: "{{ test_rundir }}/test/extensions/transport_sockets/tls/test_data/ca_cert.pem"
+  )EOF");
 };
 
-TEST(OriginalDstClusterConfigTest, GoodConfig) {
+TEST(ClusterConfigTest, GoodConfig) {
   const std::string yaml = R"EOF(
     name: name
     connect_timeout: 0.25s
     type: original_dst
     lb_policy: cluster_provided
     cleanup_interval: 1s
-  )EOF"; // Help Emacs balance quotation marks: "
+  )EOF";
 
   EXPECT_TRUE(parseClusterFromV2Yaml(yaml).has_cleanup_interval());
 }
 
-TEST_F(OriginalDstClusterTest, BadConfigWithLoadAssignment) {
+// Verify that using 'sni' causes a failure.
+TEST_F(ClusterTest, BadConfigWithInvalidSNI) {
+  const std::string yaml = TestEnvironment::substitute(R"EOF(
+    name: name
+    connect_timeout: 0.25s
+    type: original_dst
+    lb_policy: cluster_provided
+    tls_context:
+      sni: api.lyft.com
+      common_tls_context:
+        validation_context:
+          trusted_ca:
+            filename: "{{ test_rundir }}/test/extensions/transport_sockets/tls/test_data/ca_cert.pem"
+  )EOF");
+
+  EXPECT_THROW_WITH_MESSAGE(
+      setupFromYaml(yaml, true), EnvoyException,
+      "ORIGINAL_DST cluster cannot configure 'sni' or 'verify_subject_alt_name'");
+}
+
+TEST_F(ClusterTest, BadConfigWithInvalidVerifySubjectAltName) {
+  const std::string yaml = TestEnvironment::substitute(R"EOF(
+    name: name
+    connect_timeout: 0.25s
+    type: original_dst
+    lb_policy: CLUSTER_PROVIDED
+    tls_context:
+      common_tls_context:
+        validation_context:
+          trusted_ca:
+            filename: "{{ test_rundir }}/test/extensions/transport_sockets/tls/test_data/ca_cert.pem"
+          verify_subject_alt_name: [api.lyft.com]
+  )EOF");
+
+  EXPECT_THROW_WITH_MESSAGE(
+      setupFromYaml(yaml, true), EnvoyException,
+      "ORIGINAL_DST cluster cannot configure 'sni' or 'verify_subject_alt_name'");
+}
+
+TEST_F(ClusterTest, BadConfigWithLoadAssignment) {
   const std::string yaml = R"EOF(
     name: name
     connect_timeout: 0.25s
     type: ORIGINAL_DST
     lb_policy: ORIGINAL_DST_LB
-    cleanup_interval: 1s
     load_assignment:
       cluster_name: name
       endpoints:
@@ -137,13 +213,12 @@ TEST_F(OriginalDstClusterTest, BadConfigWithLoadAssignment) {
       "ORIGINAL_DST clusters must have no load assignment or hosts configured");
 }
 
-TEST_F(OriginalDstClusterTest, BadConfigWithDeprecatedHosts) {
+TEST_F(ClusterTest, BadConfigWithDeprecatedHosts) {
   const std::string yaml = R"EOF(
     name: name
     connect_timeout: 0.25s
     type: ORIGINAL_DST
     lb_policy: ORIGINAL_DST_LB
-    cleanup_interval: 1s
     hosts:
       - socket_address:
           address: 127.0.0.1
@@ -155,7 +230,7 @@ TEST_F(OriginalDstClusterTest, BadConfigWithDeprecatedHosts) {
       "ORIGINAL_DST clusters must have no load assignment or hosts configured");
 }
 
-TEST_F(OriginalDstClusterTest, CleanupInterval) {
+TEST_F(ClusterTest, CleanupInterval) {
   std::string yaml = R"EOF(
     name: name
     connect_timeout: 1.250s
@@ -173,7 +248,7 @@ TEST_F(OriginalDstClusterTest, CleanupInterval) {
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHosts().size());
 }
 
-TEST_F(OriginalDstClusterTest, NoContext) {
+TEST_F(ClusterTest, NoContext) {
   std::string yaml = R"EOF(
     name: name,
     connect_timeout: 0.125s
@@ -196,9 +271,8 @@ TEST_F(OriginalDstClusterTest, NoContext) {
   // No downstream connection => no host.
   {
     TestLoadBalancerContext lb_context(nullptr);
-    OriginalDstCluster::LoadBalancer lb(cluster_);
     EXPECT_CALL(dispatcher_, post(_)).Times(0);
-    HostConstSharedPtr host = lb.chooseHost(&lb_context);
+    HostConstSharedPtr host = lb_->chooseHost(&lb_context);
     EXPECT_EQ(host, nullptr);
   }
 
@@ -208,12 +282,8 @@ TEST_F(OriginalDstClusterTest, NoContext) {
     TestLoadBalancerContext lb_context(&connection);
 
     EXPECT_CALL(connection, localAddressRestored()).WillOnce(Return(false));
-    // First argument is normally the reference to the ThreadLocalCluster's HostSet, but in these
-    // tests we do not have the thread local clusters, so we pass a reference to the HostSet of the
-    // primary cluster. The implementation handles both cases the same.
-    OriginalDstCluster::LoadBalancer lb(cluster_);
     EXPECT_CALL(dispatcher_, post(_)).Times(0);
-    HostConstSharedPtr host = lb.chooseHost(&lb_context);
+    HostConstSharedPtr host = lb_->chooseHost(&lb_context);
     EXPECT_EQ(host, nullptr);
   }
 
@@ -224,24 +294,16 @@ TEST_F(OriginalDstClusterTest, NoContext) {
     connection.local_address_ = std::make_shared<Network::Address::PipeInstance>("unix://foo");
     EXPECT_CALL(connection, localAddressRestored()).WillRepeatedly(Return(true));
 
-    OriginalDstCluster::LoadBalancer lb(cluster_);
     EXPECT_CALL(dispatcher_, post(_)).Times(0);
-    HostConstSharedPtr host = lb.chooseHost(&lb_context);
+    HostConstSharedPtr host = lb_->chooseHost(&lb_context);
     EXPECT_EQ(host, nullptr);
   }
 }
 
-TEST_F(OriginalDstClusterTest, Membership) {
-  std::string yaml = R"EOF(
-    name: name
-    connect_timeout: 1.250s
-    type: ORIGINAL_DST
-    lb_policy: ORIGINAL_DST_LB
-  )EOF";
-
+TEST_F(ClusterTest, Membership) {
   EXPECT_CALL(initialized_, ready());
   EXPECT_CALL(*cleanup_timer_, enableTimer(_, _));
-  setupFromYaml(yaml);
+  setupFromYaml(default_yaml_, false);
 
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts().size());
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHosts().size());
@@ -261,8 +323,7 @@ TEST_F(OriginalDstClusterTest, Membership) {
 
   Event::PostCb post_cb;
   EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
-  // Mock the cluster manager by recreating the load balancer each time to get a fresh host map
-  HostConstSharedPtr host = OriginalDstCluster::LoadBalancer(cluster_).chooseHost(&lb_context);
+  HostConstSharedPtr host = lb_->chooseHost(&lb_context);
   post_cb();
   auto cluster_hosts = cluster_->prioritySet().hostSetsPerPriority()[0]->hosts();
 
@@ -281,8 +342,7 @@ TEST_F(OriginalDstClusterTest, Membership) {
             *cluster_->prioritySet().hostSetsPerPriority()[0]->hosts()[0]->address());
 
   // Same host is returned on the 2nd call
-  // Mock the cluster manager by recreating the load balancer with the new host map
-  HostConstSharedPtr host2 = OriginalDstCluster::LoadBalancer(cluster_).chooseHost(&lb_context);
+  HostConstSharedPtr host2 = lb_->chooseHost(&lb_context);
   EXPECT_EQ(host2, host);
 
   // Make host time out, no membership changes happen on the first timeout.
@@ -310,8 +370,7 @@ TEST_F(OriginalDstClusterTest, Membership) {
   // New host gets created
   EXPECT_CALL(membership_updated_, ready());
   EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
-  // Mock the cluster manager by recreating the load balancer with the new host map
-  HostConstSharedPtr host3 = OriginalDstCluster::LoadBalancer(cluster_).chooseHost(&lb_context);
+  HostConstSharedPtr host3 = lb_->chooseHost(&lb_context);
   post_cb();
   EXPECT_NE(host3, nullptr);
   EXPECT_NE(host3, host);
@@ -322,17 +381,10 @@ TEST_F(OriginalDstClusterTest, Membership) {
   EXPECT_EQ(host3, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts()[0]);
 }
 
-TEST_F(OriginalDstClusterTest, Membership2) {
-  std::string yaml = R"EOF(
-    name: name
-    connect_timeout: 1.250s
-    type: ORIGINAL_DST
-    lb_policy: ORIGINAL_DST_LB
-  )EOF";
-
+TEST_F(ClusterTest, Membership2) {
   EXPECT_CALL(initialized_, ready());
   EXPECT_CALL(*cleanup_timer_, enableTimer(_, _));
-  setupFromYaml(yaml);
+  setupFromYaml(default_yaml_);
 
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts().size());
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHosts().size());
@@ -353,18 +405,17 @@ TEST_F(OriginalDstClusterTest, Membership2) {
   connection2.local_address_ = std::make_shared<Network::Address::Ipv4Instance>("10.10.11.12");
   EXPECT_CALL(connection2, localAddressRestored()).WillRepeatedly(Return(true));
 
-  OriginalDstCluster::LoadBalancer lb(cluster_);
   EXPECT_CALL(membership_updated_, ready());
   Event::PostCb post_cb;
   EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
-  HostConstSharedPtr host1 = lb.chooseHost(&lb_context1);
+  HostConstSharedPtr host1 = lb_->chooseHost(&lb_context1);
   post_cb();
   ASSERT_NE(host1, nullptr);
   EXPECT_EQ(*connection1.local_address_, *host1->address());
 
   EXPECT_CALL(membership_updated_, ready());
   EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
-  HostConstSharedPtr host2 = lb.chooseHost(&lb_context2);
+  HostConstSharedPtr host2 = lb_->chooseHost(&lb_context2);
   post_cb();
   ASSERT_NE(host2, nullptr);
   EXPECT_EQ(*connection2.local_address_, *host2->address());
@@ -410,17 +461,10 @@ TEST_F(OriginalDstClusterTest, Membership2) {
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts().size());
 }
 
-TEST_F(OriginalDstClusterTest, Connection) {
-  std::string yaml = R"EOF(
-    name: name
-    connect_timeout: 1.250s
-    type: ORIGINAL_DST
-    lb_policy: ORIGINAL_DST_LB
-  )EOF";
-
+TEST_F(ClusterTest, Connection) {
   EXPECT_CALL(initialized_, ready());
   EXPECT_CALL(*cleanup_timer_, enableTimer(_, _));
-  setupFromYaml(yaml);
+  setupFromYaml(default_yaml_);
 
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts().size());
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHosts().size());
@@ -437,10 +481,9 @@ TEST_F(OriginalDstClusterTest, Connection) {
   connection.local_address_ = std::make_shared<Network::Address::Ipv6Instance>("FD00::1");
   EXPECT_CALL(connection, localAddressRestored()).WillRepeatedly(Return(true));
 
-  OriginalDstCluster::LoadBalancer lb(cluster_);
   Event::PostCb post_cb;
   EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
-  HostConstSharedPtr host = lb.chooseHost(&lb_context);
+  HostConstSharedPtr host = lb_->chooseHost(&lb_context);
   post_cb();
   ASSERT_NE(host, nullptr);
   EXPECT_EQ(*connection.local_address_, *host->address());
@@ -450,17 +493,10 @@ TEST_F(OriginalDstClusterTest, Connection) {
   host->createConnection(dispatcher_, nullptr, nullptr);
 }
 
-TEST_F(OriginalDstClusterTest, MultipleClusters) {
-  std::string yaml = R"EOF(
-    name: name
-    connect_timeout: 1.250s
-    type: ORIGINAL_DST
-    lb_policy: ORIGINAL_DST_LB
-  )EOF";
-
+TEST_F(ClusterTest, MultipleClusters) {
   EXPECT_CALL(initialized_, ready());
   EXPECT_CALL(*cleanup_timer_, enableTimer(_, _));
-  setupFromYaml(yaml);
+  setupFromYaml(default_yaml_);
 
   PrioritySetImpl second;
   cluster_->prioritySet().addPriorityUpdateCb(
@@ -486,10 +522,9 @@ TEST_F(OriginalDstClusterTest, MultipleClusters) {
   connection.local_address_ = std::make_shared<Network::Address::Ipv6Instance>("FD00::1");
   EXPECT_CALL(connection, localAddressRestored()).WillRepeatedly(Return(true));
 
-  OriginalDstCluster::LoadBalancer lb(cluster_);
   Event::PostCb post_cb;
   EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
-  HostConstSharedPtr host = lb.chooseHost(&lb_context);
+  HostConstSharedPtr host = lb_->chooseHost(&lb_context);
   post_cb();
   ASSERT_NE(host, nullptr);
   EXPECT_EQ(*connection.local_address_, *host->address());
@@ -502,12 +537,8 @@ TEST_F(OriginalDstClusterTest, MultipleClusters) {
   EXPECT_EQ(host, second.hostSetsPerPriority()[0]->hosts()[0]);
 }
 
-TEST_F(OriginalDstClusterTest, UseHttpHeaderEnabled) {
-  std::string yaml = R"EOF(
-    name: name
-    connect_timeout: 1.250s
-    type: ORIGINAL_DST
-    lb_policy: ORIGINAL_DST_LB
+TEST_F(ClusterTest, UseHttpHeaderEnabled) {
+  std::string yaml = default_yaml_ + R"EOF(
     original_dst_lb_config:
       use_http_header: true
   )EOF";
@@ -523,7 +554,6 @@ TEST_F(OriginalDstClusterTest, UseHttpHeaderEnabled) {
       0UL,
       cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHostsPerLocality().get().size());
 
-  OriginalDstCluster::LoadBalancer lb(cluster_);
   Event::PostCb post_cb;
 
   // HTTP header override.
@@ -532,7 +562,7 @@ TEST_F(OriginalDstClusterTest, UseHttpHeaderEnabled) {
 
   EXPECT_CALL(membership_updated_, ready());
   EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
-  HostConstSharedPtr host1 = lb.chooseHost(&lb_context1);
+  HostConstSharedPtr host1 = lb_->chooseHost(&lb_context1);
   post_cb();
   ASSERT_NE(host1, nullptr);
   EXPECT_EQ("127.0.0.1:5555", host1->address()->asString());
@@ -548,7 +578,7 @@ TEST_F(OriginalDstClusterTest, UseHttpHeaderEnabled) {
 
   EXPECT_CALL(membership_updated_, ready());
   EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
-  HostConstSharedPtr host2 = lb.chooseHost(&lb_context2);
+  HostConstSharedPtr host2 = lb_->chooseHost(&lb_context2);
   post_cb();
   ASSERT_NE(host2, nullptr);
   EXPECT_EQ("127.0.0.1:5556", host2->address()->asString());
@@ -558,7 +588,7 @@ TEST_F(OriginalDstClusterTest, UseHttpHeaderEnabled) {
 
   EXPECT_CALL(membership_updated_, ready()).Times(0);
   EXPECT_CALL(dispatcher_, post(_)).Times(0);
-  HostConstSharedPtr host3 = lb.chooseHost(&lb_context3);
+  HostConstSharedPtr host3 = lb_->chooseHost(&lb_context3);
   EXPECT_EQ(host3, nullptr);
   EXPECT_EQ(
       1, TestUtility::findCounter(stats_store_, "cluster.name.original_dst_host_invalid")->value());
@@ -569,23 +599,16 @@ TEST_F(OriginalDstClusterTest, UseHttpHeaderEnabled) {
 
   EXPECT_CALL(membership_updated_, ready()).Times(0);
   EXPECT_CALL(dispatcher_, post(_)).Times(0);
-  HostConstSharedPtr host4 = lb.chooseHost(&lb_context4);
+  HostConstSharedPtr host4 = lb_->chooseHost(&lb_context4);
   EXPECT_EQ(host4, nullptr);
   EXPECT_EQ(
       2, TestUtility::findCounter(stats_store_, "cluster.name.original_dst_host_invalid")->value());
 }
 
-TEST_F(OriginalDstClusterTest, UseHttpHeaderDisabled) {
-  std::string yaml = R"EOF(
-    name: name
-    connect_timeout: 1.250s
-    type: ORIGINAL_DST
-    lb_policy: ORIGINAL_DST_LB
-  )EOF";
-
+TEST_F(ClusterTest, UseHttpHeaderDisabled) {
   EXPECT_CALL(initialized_, ready());
   EXPECT_CALL(*cleanup_timer_, enableTimer(_, _));
-  setupFromYaml(yaml);
+  setupFromYaml(default_yaml_);
 
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts().size());
   EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHosts().size());
@@ -594,7 +617,6 @@ TEST_F(OriginalDstClusterTest, UseHttpHeaderDisabled) {
       0UL,
       cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHostsPerLocality().get().size());
 
-  OriginalDstCluster::LoadBalancer lb(cluster_);
   Event::PostCb post_cb;
 
   // Downstream connection with original_dst filter, HTTP header override ignored.
@@ -606,7 +628,7 @@ TEST_F(OriginalDstClusterTest, UseHttpHeaderDisabled) {
 
   EXPECT_CALL(membership_updated_, ready());
   EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
-  HostConstSharedPtr host1 = lb.chooseHost(&lb_context1);
+  HostConstSharedPtr host1 = lb_->chooseHost(&lb_context1);
   post_cb();
   ASSERT_NE(host1, nullptr);
   EXPECT_EQ(*connection1.local_address_, *host1->address());
@@ -620,7 +642,7 @@ TEST_F(OriginalDstClusterTest, UseHttpHeaderDisabled) {
 
   EXPECT_CALL(membership_updated_, ready()).Times(0);
   EXPECT_CALL(dispatcher_, post(_)).Times(0);
-  HostConstSharedPtr host2 = lb.chooseHost(&lb_context2);
+  HostConstSharedPtr host2 = lb_->chooseHost(&lb_context2);
   EXPECT_EQ(host2, nullptr);
 
   // Downstream connection over Unix Domain Socket, HTTP header override ignored.
@@ -631,10 +653,155 @@ TEST_F(OriginalDstClusterTest, UseHttpHeaderDisabled) {
 
   EXPECT_CALL(membership_updated_, ready()).Times(0);
   EXPECT_CALL(dispatcher_, post(_)).Times(0);
-  HostConstSharedPtr host3 = lb.chooseHost(&lb_context3);
+  HostConstSharedPtr host3 = lb_->chooseHost(&lb_context3);
   EXPECT_EQ(host3, nullptr);
 }
 
-} // namespace
+TEST_F(ClusterTest, SNIFromHostHeader) {
+  EXPECT_CALL(initialized_, ready());
+  EXPECT_CALL(*cleanup_timer_, enableTimer(_, _));
+  setupFromYaml(default_tls_yaml_, true);
+
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts().size());
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHosts().size());
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hostsPerLocality().get().size());
+  EXPECT_EQ(
+      0UL,
+      cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHostsPerLocality().get().size());
+
+  // SNI from Host header?
+  NiceMock<Network::MockConnection> connection;
+  connection.local_address_ = std::make_shared<Network::Address::Ipv4Instance>("10.10.11.11", 80);
+  EXPECT_CALL(connection, localAddressRestored()).WillRepeatedly(Return(true));
+  TestLoadBalancerContext lb_context(&connection, Http::Headers::get().Host.get(),
+                                     "www.google.com:80");
+  // Verify TestLoadBalancerContext works
+  auto downstream_headers = lb_context.downstreamHeaders();
+  ASSERT_NE(downstream_headers, nullptr);
+  auto host_header = downstream_headers->get(Http::Headers::get().Host);
+  ASSERT_NE(host_header, nullptr);
+  auto host_header_string = std::string(host_header->value().getStringView());
+  ASSERT_EQ("www.google.com:80", host_header_string);
+
+  Event::PostCb post_cb;
+  EXPECT_CALL(membership_updated_, ready());
+  EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
+  HostConstSharedPtr host = lb_->chooseHost(&lb_context);
+  post_cb();
+  ASSERT_NE(host, nullptr);
+  EXPECT_EQ("10.10.11.11:80", host->address()->asString());
+  const Cluster::Host* options_host = dynamic_cast<const Cluster::Host*>(host.get());
+  ASSERT_NE(options_host, nullptr);
+  const Network::TransportSocketOptionsSharedPtr options =
+      getDefaultTransportSocketOptions(options_host);
+  ASSERT_NE(options, nullptr);
+  auto sni_override = options->serverNameOverride();
+  ASSERT_EQ(true, !!sni_override);
+  ASSERT_EQ("www.google.com", sni_override.value());
+}
+
+TEST_F(ClusterTest, NoSNIFromHostHeaderIP) {
+  EXPECT_CALL(initialized_, ready());
+  EXPECT_CALL(*cleanup_timer_, enableTimer(_, _));
+  setupFromYaml(default_tls_yaml_, true);
+
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts().size());
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHosts().size());
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hostsPerLocality().get().size());
+  EXPECT_EQ(
+      0UL,
+      cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHostsPerLocality().get().size());
+
+  // SNI from Host header?
+  NiceMock<Network::MockConnection> connection;
+  connection.local_address_ = std::make_shared<Network::Address::Ipv4Instance>("10.10.11.11", 80);
+  EXPECT_CALL(connection, localAddressRestored()).WillRepeatedly(Return(true));
+  TestLoadBalancerContext lb_context(&connection, Http::Headers::get().Host.get(), "[ffd0::1]:80");
+
+  Event::PostCb post_cb;
+  EXPECT_CALL(membership_updated_, ready());
+  EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
+  HostConstSharedPtr host = lb_->chooseHost(&lb_context);
+  post_cb();
+  ASSERT_NE(host, nullptr);
+  EXPECT_EQ("10.10.11.11:80", host->address()->asString());
+  const Cluster::Host* options_host = dynamic_cast<const Cluster::Host*>(host.get());
+  ASSERT_NE(options_host, nullptr);
+  const Network::TransportSocketOptionsSharedPtr options =
+      getDefaultTransportSocketOptions(options_host);
+  ASSERT_EQ(options, nullptr);
+}
+
+TEST_F(ClusterTest, NoTLSNoSNIFromHostHeader) {
+  EXPECT_CALL(initialized_, ready());
+  EXPECT_CALL(*cleanup_timer_, enableTimer(_, _));
+  setupFromYaml(default_yaml_, false);
+
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts().size());
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHosts().size());
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hostsPerLocality().get().size());
+  EXPECT_EQ(
+      0UL,
+      cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHostsPerLocality().get().size());
+
+  // SNI from Host header?
+  NiceMock<Network::MockConnection> connection;
+  connection.local_address_ = std::make_shared<Network::Address::Ipv4Instance>("10.10.11.11", 80);
+  EXPECT_CALL(connection, localAddressRestored()).WillRepeatedly(Return(true));
+  TestLoadBalancerContext lb_context(&connection, Http::Headers::get().Host.get(),
+                                     "www.google.com:80");
+
+  Event::PostCb post_cb;
+  EXPECT_CALL(membership_updated_, ready());
+  EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
+  HostConstSharedPtr host = lb_->chooseHost(&lb_context);
+  post_cb();
+  ASSERT_NE(host, nullptr);
+  EXPECT_EQ("10.10.11.11:80", host->address()->asString());
+  const Cluster::Host* options_host = dynamic_cast<const Cluster::Host*>(host.get());
+  ASSERT_NE(options_host, nullptr);
+  const Network::TransportSocketOptionsSharedPtr options =
+      getDefaultTransportSocketOptions(options_host);
+  ASSERT_EQ(options, nullptr);
+}
+
+TEST_F(ClusterTest, SNIFromDownstreamConnection) {
+  EXPECT_CALL(initialized_, ready());
+  EXPECT_CALL(*cleanup_timer_, enableTimer(_, _));
+  setupFromYaml(default_tls_yaml_, true);
+
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hosts().size());
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHosts().size());
+  EXPECT_EQ(0UL, cluster_->prioritySet().hostSetsPerPriority()[0]->hostsPerLocality().get().size());
+  EXPECT_EQ(
+      0UL,
+      cluster_->prioritySet().hostSetsPerPriority()[0]->healthyHostsPerLocality().get().size());
+
+  // SNI from downstream connection, even if Host header is present
+  NiceMock<Network::MockConnection> connection;
+  connection.local_address_ = std::make_shared<Network::Address::Ipv4Instance>("10.10.11.11", 80);
+  EXPECT_CALL(connection, localAddressRestored()).WillRepeatedly(Return(true));
+  EXPECT_CALL(connection, requestedServerName()).WillRepeatedly(Return("www.isovalent.com"));
+  TestLoadBalancerContext lb_context(&connection, Http::Headers::get().Host.get(),
+                                     "www.google.com:80");
+
+  Event::PostCb post_cb;
+  EXPECT_CALL(membership_updated_, ready());
+  EXPECT_CALL(dispatcher_, post(_)).WillOnce(SaveArg<0>(&post_cb));
+  HostConstSharedPtr host = lb_->chooseHost(&lb_context);
+  post_cb();
+  ASSERT_NE(host, nullptr);
+  EXPECT_EQ("10.10.11.11:80", host->address()->asString());
+  const Cluster::Host* options_host = dynamic_cast<const Cluster::Host*>(host.get());
+  ASSERT_NE(options_host, nullptr);
+  const Network::TransportSocketOptionsSharedPtr options =
+      getDefaultTransportSocketOptions(options_host);
+  ASSERT_NE(options, nullptr);
+  auto sni_override = options->serverNameOverride();
+  ASSERT_EQ(true, !!sni_override);
+  ASSERT_EQ("www.isovalent.com", sni_override.value());
+}
+
+} // namespace OriginalDst
 } // namespace Upstream
 } // namespace Envoy
