udp-proxy: Fix crash related to dynamic updates to UDP proxy via CDS/…

…EDS (#33824)

Signed-off-by: s.adams@f5.com <s.adams@f5.com>
Signed-off-by: Shaun Adams <shaun.adams@volunteers.acasi.info>
Co-authored-by: phlax <phlax@users.noreply.github.com>

changelogs/current.yaml

@@ -46,6 +46,9 @@ bug_fixes:
     Upgraded c-ares library to 1.20.1 and added fix to c-ares DNS implementation to additionally check for ``ARES_EREFUSED``,
     ``ARES_ESERVFAIL``and ``ARES_ENOTIMP`` status. Without this fix, ``DestroyChannelOnRefused`` and
     ``CustomResolverValidAfterChannelDestruction`` unit test will break.
+- area: udp
+  change: |
+    Fixed a bug that would cause Envoy to crash when updates to a pre-existing cluster were made (e.g. ``HostSet`` changes).
 - area: ext_authz
   change: |
     Handle ``append_action`` from :ref:`external authorization service <envoy_v3_api_msg_service.auth.v3.CheckResponse>`

source/extensions/filters/udp/udp_proxy/udp_proxy_filter.cc

@@ -49,11 +49,11 @@ void UdpProxyFilter::onClusterAddOrUpdate(absl::string_view cluster_name,
          &cluster_infos_[cluster_name]->cluster_ != &cluster);
 
   if (config_->usingPerPacketLoadBalancing()) {
-    cluster_infos_.emplace(cluster_name,
-                           std::make_unique<PerPacketLoadBalancingClusterInfo>(*this, cluster));
+    cluster_infos_.insert_or_assign(
+        cluster_name, std::make_unique<PerPacketLoadBalancingClusterInfo>(*this, cluster));
   } else {
-    cluster_infos_.emplace(cluster_name,
-                           std::make_unique<StickySessionClusterInfo>(*this, cluster));
+    cluster_infos_.insert_or_assign(cluster_name,
+                                    std::make_unique<StickySessionClusterInfo>(*this, cluster));
   }
 }
 

test/extensions/filters/udp/udp_proxy/udp_proxy_filter_test.cc

@@ -826,6 +826,51 @@ stat_prefix: foo
   EXPECT_EQ(0, config_->stats().downstream_sess_active_.value());
 }
 
+// Test updates to existing cluster (e.g. priority set changes, etc).
+TEST_F(UdpProxyFilterTest, ClusterDynamicInfoMapUpdate) {
+  InSequence s;
+
+  setup(readConfig(R"EOF(
+stat_prefix: foo
+matcher:
+  on_no_match:
+    action:
+      name: route
+      typed_config:
+        '@type': type.googleapis.com/envoy.extensions.filters.udp.udp_proxy.v3.Route
+        cluster: fake_cluster
+  )EOF"),
+        false);
+
+  // Initial ThreadLocalCluster, scoped lifetime
+  // mimics replacement of old ThreadLocalCluster via postThreadLocalClusterUpdate.
+  {
+    NiceMock<Upstream::MockThreadLocalCluster> other_thread_local_cluster;
+    other_thread_local_cluster.cluster_.info_->name_ = "fake_cluster";
+    Upstream::ThreadLocalClusterCommand command =
+        [&other_thread_local_cluster]() -> Upstream::ThreadLocalCluster& {
+      return other_thread_local_cluster;
+    };
+    cluster_update_callbacks_->onClusterAddOrUpdate(other_thread_local_cluster.info()->name(),
+                                                    command);
+  }
+
+  // Push new cluster (getter), we expect this to result in new ClusterInfos object in map.
+  {
+    Upstream::ThreadLocalClusterCommand command = [this]() -> Upstream::ThreadLocalCluster& {
+      return factory_context_.server_factory_context_.cluster_manager_.thread_local_cluster_;
+    };
+    cluster_update_callbacks_->onClusterAddOrUpdate(
+        factory_context_.server_factory_context_.cluster_manager_.thread_local_cluster_.info()
+            ->name(),
+        command);
+  }
+
+  expectSessionCreate(upstream_address_);
+  test_sessions_[0].expectWriteToUpstream("hello", 0, nullptr, true);
+  recvDataFromDownstream("10.0.0.1:1000", "10.0.0.2:80", "hello");
+}
+
 // Hitting the maximum per-cluster connection/session circuit breaker.
 TEST_F(UdpProxyFilterTest, MaxSessionsCircuitBreaker) {
   InSequence s;