upstream: update host's socket factory when metadata is updated.

source/common/upstream/upstream_impl.h

@@ -86,6 +86,7 @@ class HostDescriptionImpl : virtual public HostDescription,
       uint32_t priority, TimeSource& time_source);
 
   Network::TransportSocketFactory& transportSocketFactory() const override {
+    absl::ReaderMutexLock lock(&metadata_mutex_);
     return socket_factory_;
   }
 
@@ -103,8 +104,13 @@ class HostDescriptionImpl : virtual public HostDescription,
     return metadata_;
   }
   void metadata(MetadataConstSharedPtr new_metadata) override {
-    absl::WriterMutexLock lock(&metadata_mutex_);
-    metadata_ = new_metadata;
+    auto& new_socket_factory = resolveTransportSocketFactory(address_, new_metadata.get());
+    {
+      absl::WriterMutexLock lock(&metadata_mutex_);
+      metadata_ = new_metadata;
+      // Update data members dependent on metadata.
+      socket_factory_ = new_socket_factory;
+    }
   }
 
   const ClusterInfo& cluster() const override { return *cluster_; }
@@ -178,7 +184,8 @@ class HostDescriptionImpl : virtual public HostDescription,
   Outlier::DetectorHostMonitorPtr outlier_detector_;
   HealthCheckHostMonitorPtr health_checker_;
   std::atomic<uint32_t> priority_;
-  Network::TransportSocketFactory& socket_factory_;
+  std::reference_wrapper<Network::TransportSocketFactory>
+      socket_factory_ ABSL_GUARDED_BY(metadata_mutex_);
   const MonotonicTime creation_time_;
 };
 

test/common/upstream/BUILD

@@ -95,6 +95,7 @@ envoy_cc_test(
         "//source/common/config:utility_lib",
         "//source/common/upstream:eds_lib",
         "//source/extensions/transport_sockets/raw_buffer:config",
+        "//source/extensions/transport_sockets/tls:config",
         "//source/server:transport_socket_config_lib",
         "//test/common/stats:stat_test_utility_lib",
         "//test/mocks/local_info:local_info_mocks",

test/common/upstream/eds_test.cc

@@ -56,6 +56,36 @@ class EdsTest : public testing::Test {
                  Cluster::InitializePhase::Secondary);
   }
 
+  //  Define a cluster with secure and unsecure (default) transport
+  //  sockets.
+  void resetClusterWithTransportSockets() {
+    resetCluster(R"EOF(
+      name: name
+      connect_timeout: 0.25s
+      type: EDS
+      lb_policy: ROUND_ROBIN
+      eds_cluster_config:
+        service_name: fare
+        eds_config:
+          api_config_source:
+            api_type: REST
+            cluster_names:
+            - eds
+            refresh_delay: 1s
+      transport_socket_matches:
+      - match:
+          secure: enabled
+        name: secure-mode
+        transport_socket:
+          name: envoy.transport_sockets.tls
+      - match: {}
+        name: default-mode
+        transport_socket:
+          name: envoy.transport_sockets.raw_buffer
+ )EOF",
+                 Cluster::InitializePhase::Secondary);
+  }
+
   void resetClusterDrainOnHostRemoval() {
     resetCluster(R"EOF(
         name: name
@@ -117,7 +147,7 @@ class EdsTest : public testing::Test {
 
   bool initialized_{};
   Stats::TestUtil::TestStore stats_;
-  Ssl::MockContextManager ssl_context_manager_;
+  NiceMock<Ssl::MockContextManager> ssl_context_manager_;
   envoy::config::cluster::v3::Cluster eds_cluster_;
   NiceMock<MockClusterManager> cm_;
   NiceMock<Event::MockDispatcher> dispatcher_;
@@ -452,6 +482,51 @@ TEST_F(EdsTest, EndpointMetadata) {
             "v2");
 }
 
+// Test verifies that updating metadata updates
+// data members dependent on metadata values.
+// Specifically, it transport socket matcher has changed,
+// the transport socket factory should also be updated.
+TEST_F(EdsTest, EndpointMetadataWithTransportSocket) {
+  envoy::config::endpoint::v3::ClusterLoadAssignment cluster_load_assignment;
+  cluster_load_assignment.set_cluster_name("fare");
+  resetClusterWithTransportSockets();
+
+  auto health_checker = std::make_shared<MockHealthChecker>();
+  EXPECT_CALL(*health_checker, start());
+  EXPECT_CALL(*health_checker, addHostCheckCompleteCb(_)).Times(2);
+  cluster_->setHealthChecker(health_checker);
+
+  // Add single endpoint to the cluster.
+  auto* endpoints = cluster_load_assignment.add_endpoints();
+  auto* endpoint = endpoints->add_lb_endpoints();
+
+  auto* socket_address = endpoint->mutable_endpoint()->mutable_address()->mutable_socket_address();
+  socket_address->set_address("1.2.3.4");
+  socket_address->set_port_value(80);
+
+  doOnConfigUpdateVerifyNoThrow(cluster_load_assignment);
+
+  auto& hosts = cluster_->prioritySet().hostSetsPerPriority()[0]->hosts();
+  ASSERT_EQ(hosts.size(), 1);
+  auto* upstream_host = hosts[0].get();
+
+  // Verify that default transport socket is raw (does not implement secure transport).
+  EXPECT_FALSE(upstream_host->transportSocketFactory().implementsSecureTransport());
+
+  // Create metadata with transport socket match pointing to secure mode.
+  auto metadata = new envoy::config::core::v3::Metadata();
+  MetadataConstSharedPtr metadata_sharedptr(metadata);
+  Config::Metadata::mutableMetadataValue(
+      *metadata, Config::MetadataFilters::get().ENVOY_TRANSPORT_SOCKET_MATCH, "secure")
+      .set_string_value("enabled");
+
+  // Update metadata.
+  upstream_host->metadata(metadata_sharedptr);
+
+  // Transport socket factory should point to tls, which implements secure transport.
+  EXPECT_TRUE(upstream_host->transportSocketFactory().implementsSecureTransport());
+}
+
 // Validate that onConfigUpdate() updates endpoint health status.
 TEST_F(EdsTest, EndpointHealthStatus) {
   envoy::config::endpoint::v3::ClusterLoadAssignment cluster_load_assignment;