udp_proxy: support outlier detection in http tunneling

changelogs/current.yaml

@@ -17,5 +17,9 @@ new_features:
   change: |
     Added :ref:`virtualClusterName() <config_http_filters_lua_stream_info_virtual_cluster_name>` API to the Stream Info
     Object to get the name of the virtual cluster matched.
+- area: udp_proxy
+  change: |
+    Added support for outlier detection in UDP proxy. This change can be temporarily reverted by setting runtime guard
+    ``envoy.reloadable_features.enable_udp_proxy_outlier_detection`` to ``false``.
 
 deprecated:

source/common/runtime/runtime_features.cc

@@ -39,6 +39,7 @@ RUNTIME_GUARD(envoy_reloadable_features_dns_details);
 RUNTIME_GUARD(envoy_reloadable_features_dns_nodata_noname_is_success);
 RUNTIME_GUARD(envoy_reloadable_features_enable_compression_bomb_protection);
 RUNTIME_GUARD(envoy_reloadable_features_enable_include_histograms);
+RUNTIME_GUARD(envoy_reloadable_features_enable_udp_proxy_outlier_detection);
 RUNTIME_GUARD(envoy_reloadable_features_explicit_internal_address_config);
 RUNTIME_GUARD(envoy_reloadable_features_ext_proc_timeout_error);
 RUNTIME_GUARD(envoy_reloadable_features_extend_h3_accept_untrusted);

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

@@ -929,7 +929,7 @@ void TunnelingConnectionPoolImpl::onPoolFailure(Http::ConnectionPool::PoolFailur
   // removed by onStreamFailure, which will cause downstream_info_ to be freed.
   downstream_info_.upstreamInfo()->setUpstreamHost(host);
   downstream_info_.upstreamInfo()->setUpstreamTransportFailureReason(failure_reason);
-  callbacks_->onStreamFailure(reason, failure_reason, host);
+  callbacks_->onStreamFailure(reason, failure_reason, *host);
 }
 
 void TunnelingConnectionPoolImpl::onPoolReady(Http::RequestEncoder& request_encoder,
@@ -1032,7 +1032,7 @@ bool UdpProxyFilter::TunnelingActiveSession::createConnectionPool() {
 
 void UdpProxyFilter::TunnelingActiveSession::onStreamFailure(
     ConnectionPool::PoolFailureReason reason, absl::string_view failure_reason,
-    Upstream::HostDescriptionConstSharedPtr) {
+    const Upstream::HostDescription& host) {
   ENVOY_LOG(debug, "Failed to create upstream stream: {}", failure_reason);
 
   conn_pool_.reset();
@@ -1045,20 +1045,28 @@ void UdpProxyFilter::TunnelingActiveSession::onStreamFailure(
     break;
   case ConnectionPool::PoolFailureReason::Timeout:
     udp_session_info_.setResponseFlag(StreamInfo::CoreResponseFlag::UpstreamConnectionFailure);
+    if (Runtime::runtimeFeatureEnabled(
+            "envoy.reloadable_features.enable_udp_proxy_outlier_detection")) {
+      host.outlierDetector().putResult(Upstream::Outlier::Result::LocalOriginTimeout);
+    }
     onUpstreamEvent(Network::ConnectionEvent::RemoteClose);
     break;
   case ConnectionPool::PoolFailureReason::RemoteConnectionFailure:
     if (connecting_) {
       udp_session_info_.setResponseFlag(StreamInfo::CoreResponseFlag::UpstreamConnectionFailure);
     }
+    if (Runtime::runtimeFeatureEnabled(
+            "envoy.reloadable_features.enable_udp_proxy_outlier_detection")) {
+      host.outlierDetector().putResult(Upstream::Outlier::Result::LocalOriginConnectFailed);
+    }
     onUpstreamEvent(Network::ConnectionEvent::RemoteClose);
     break;
   }
 }
 
 void UdpProxyFilter::TunnelingActiveSession::onStreamReady(StreamInfo::StreamInfo* upstream_info,
                                                            std::unique_ptr<HttpUpstream>&& upstream,
-                                                           Upstream::HostDescriptionConstSharedPtr&,
+                                                           const Upstream::HostDescription& host,
                                                            const Network::ConnectionInfoProvider&,
                                                            Ssl::ConnectionInfoConstSharedPtr) {
   // TODO(ohadvano): save the host description to host_ field. This requires refactoring because
@@ -1072,6 +1080,10 @@ void UdpProxyFilter::TunnelingActiveSession::onStreamReady(StreamInfo::StreamInf
   connecting_ = false;
   can_send_upstream_ = true;
   cluster_->cluster_stats_.sess_tunnel_success_.inc();
+  if (Runtime::runtimeFeatureEnabled(
+          "envoy.reloadable_features.enable_udp_proxy_outlier_detection")) {
+    host.outlierDetector().putResult(Upstream::Outlier::Result::LocalOriginConnectSuccessFinal);
+  }
 
   if (filter_.config_->flushAccessLogOnTunnelConnected()) {
     fillSessionStreamInfo();

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

@@ -203,7 +203,7 @@ class HttpStreamCallbacks {
    * @param ssl_info supplies the ssl information of the upstream connection.
    */
   virtual void onStreamReady(StreamInfo::StreamInfo* info, std::unique_ptr<HttpUpstream>&& upstream,
-                             Upstream::HostDescriptionConstSharedPtr& host,
+                             const Upstream::HostDescription& host,
                              const Network::ConnectionInfoProvider& address_provider,
                              Ssl::ConnectionInfoConstSharedPtr ssl_info) PURE;
 
@@ -216,7 +216,7 @@ class HttpStreamCallbacks {
    */
   virtual void onStreamFailure(ConnectionPool::PoolFailureReason reason,
                                absl::string_view failure_reason,
-                               Upstream::HostDescriptionConstSharedPtr host) PURE;
+                               const Upstream::HostDescription& host) PURE;
 
   /**
    * Called to reset the idle timer.
@@ -418,13 +418,13 @@ class TunnelingConnectionPoolImpl : public TunnelingConnectionPool,
 
   // TunnelCreationCallbacks
   void onStreamSuccess(Http::RequestEncoder& request_encoder) override {
-    callbacks_->onStreamReady(upstream_info_, std::move(upstream_), upstream_host_,
+    callbacks_->onStreamReady(upstream_info_, std::move(upstream_), *upstream_host_,
                               request_encoder.getStream().connectionInfoProvider(), ssl_info_);
   }
 
   void onStreamFailure() override {
     callbacks_->onStreamFailure(ConnectionPool::PoolFailureReason::RemoteConnectionFailure, "",
-                                upstream_host_);
+                                *upstream_host_);
   }
 
   // Http::ConnectionPool::Callbacks
@@ -754,12 +754,11 @@ class UdpProxyFilter : public Network::UdpListenerReadFilter,
 
     // HttpStreamCallbacks
     void onStreamReady(StreamInfo::StreamInfo*, std::unique_ptr<HttpUpstream>&&,
-                       Upstream::HostDescriptionConstSharedPtr&,
-                       const Network::ConnectionInfoProvider&,
+                       const Upstream::HostDescription&, const Network::ConnectionInfoProvider&,
                        Ssl::ConnectionInfoConstSharedPtr) override;
 
     void onStreamFailure(ConnectionPool::PoolFailureReason, absl::string_view,
-                         Upstream::HostDescriptionConstSharedPtr) override;
+                         const Upstream::HostDescription&) override;
 
     void resetIdleTimer() override { ActiveSession::resetIdleTimer(); }
 

test/extensions/filters/udp/udp_proxy/mocks.h

@@ -107,12 +107,12 @@ class MockHttpStreamCallbacks : public HttpStreamCallbacks {
 
   MOCK_METHOD(void, onStreamReady,
               (StreamInfo::StreamInfo * info, std::unique_ptr<HttpUpstream>&& upstream,
-               Upstream::HostDescriptionConstSharedPtr& host,
+               const Upstream::HostDescription& host,
                const Network::ConnectionInfoProvider& address_provider,
                Ssl::ConnectionInfoConstSharedPtr ssl_info));
   MOCK_METHOD(void, onStreamFailure,
               (ConnectionPool::PoolFailureReason reason, absl::string_view failure_reason,
-               Upstream::HostDescriptionConstSharedPtr host));
+               const Upstream::HostDescription& host));
   MOCK_METHOD(void, resetIdleTimer, ());
 };
 

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

@@ -453,6 +453,7 @@ use_original_src_ip: true
                                                                     ENVOY_SOCKET_IPV6_TRANSPARENT};
   inline static const std::string upstream_ip_address_ = "20.0.0.1:443";
   inline static const std::string peer_ip_address_ = "10.0.0.1:1000";
+  NiceMock<Upstream::MockHostDescription> upstream_host_;
 };
 
 class UdpProxyFilterIpv6Test : public UdpProxyFilterTest {
@@ -1878,7 +1879,6 @@ stat_prefix: foo
 
   NiceMock<StreamInfo::MockStreamInfo> stream_info;
   stream_info.downstream_connection_info_provider_->setConnectionID(0);
-  Upstream::HostDescriptionConstSharedPtr upstream_host;
   Network::ConnectionInfoSetterImpl address_provider(nullptr, nullptr);
 
   auto session = filter_->createTunnelingSession();
@@ -1899,7 +1899,7 @@ stat_prefix: foo
   }));
 
   session->onNewSession();
-  session->onStreamReady(&stream_info, std::unique_ptr<HttpUpstream>{upstream}, upstream_host,
+  session->onStreamReady(&stream_info, std::unique_ptr<HttpUpstream>{upstream}, upstream_host_,
                          address_provider, nullptr);
 }
 
@@ -1928,7 +1928,6 @@ stat_prefix: foo
 
   NiceMock<StreamInfo::MockStreamInfo> stream_info;
   stream_info.downstream_connection_info_provider_->setConnectionID(0);
-  Upstream::HostDescriptionConstSharedPtr upstream_host;
   Network::ConnectionInfoSetterImpl address_provider(nullptr, nullptr);
 
   auto session = filter_->createTunnelingSession();
@@ -1963,10 +1962,101 @@ stat_prefix: foo
       }));
 
   session->onNewSession();
-  session->onStreamReady(&stream_info, std::unique_ptr<HttpUpstream>{upstream}, upstream_host,
+  session->onStreamReady(&stream_info, std::unique_ptr<HttpUpstream>{upstream}, upstream_host_,
                          address_provider, nullptr);
 }
 
+TEST_F(UdpProxyFilterTest, TunnelingSessionOutlierDetectionConnectSuccessFinal) {
+  Event::MockTimer* idle_timer = new Event::MockTimer(&callbacks_.udp_listener_.dispatcher_);
+  EXPECT_CALL(*idle_timer, enableTimer(_, _)).Times(0);
+
+  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
+tunneling_config:
+  proxy_host: host.com
+  target_host: host.com
+  default_target_port: 30
+  )EOF"),
+        true);
+
+  NiceMock<StreamInfo::MockStreamInfo> stream_info;
+  stream_info.downstream_connection_info_provider_->setConnectionID(0);
+  auto* upstream = new NiceMock<SessionFilters::MockHttpUpstream>();
+  Network::ConnectionInfoSetterImpl address_provider(nullptr, nullptr);
+
+  auto session = filter_->createTunnelingSession();
+  session->onNewSession();
+
+  EXPECT_CALL(upstream_host_.outlier_detector_,
+              putResult(Upstream::Outlier::Result::LocalOriginConnectSuccessFinal, _));
+  session->onStreamReady(&stream_info, std::unique_ptr<HttpUpstream>{upstream}, upstream_host_,
+                         address_provider, nullptr);
+}
+
+TEST_F(UdpProxyFilterTest, TunnelingSessionOutlierDetectionConnectFailed) {
+  Event::MockTimer* idle_timer = new Event::MockTimer(&callbacks_.udp_listener_.dispatcher_);
+  EXPECT_CALL(*idle_timer, enableTimer(_, _)).Times(0);
+
+  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
+tunneling_config:
+  proxy_host: host.com
+  target_host: host.com
+  default_target_port: 30
+  )EOF"),
+        true);
+
+  auto session = filter_->createTunnelingSession();
+  session->onNewSession();
+
+  EXPECT_CALL(upstream_host_.outlier_detector_,
+              putResult(Upstream::Outlier::Result::LocalOriginConnectFailed, _));
+  session->onStreamFailure(ConnectionPool::PoolFailureReason::RemoteConnectionFailure, "",
+                           upstream_host_);
+}
+
+TEST_F(UdpProxyFilterTest, TunnelingSessionOutlierDetectionTimeout) {
+  Event::MockTimer* idle_timer = new Event::MockTimer(&callbacks_.udp_listener_.dispatcher_);
+  EXPECT_CALL(*idle_timer, enableTimer(_, _)).Times(0);
+
+  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
+tunneling_config:
+  proxy_host: host.com
+  target_host: host.com
+  default_target_port: 30
+  )EOF"),
+        true);
+
+  auto session = filter_->createTunnelingSession();
+  session->onNewSession();
+
+  EXPECT_CALL(upstream_host_.outlier_detector_,
+              putResult(Upstream::Outlier::Result::LocalOriginTimeout, _));
+  session->onStreamFailure(ConnectionPool::PoolFailureReason::Timeout, "", upstream_host_);
+}
+
 using MockUdpTunnelingConfig = SessionFilters::MockUdpTunnelingConfig;
 using MockUpstreamTunnelCallbacks = SessionFilters::MockUpstreamTunnelCallbacks;
 using MockTunnelCreationCallbacks = SessionFilters::MockTunnelCreationCallbacks;
@@ -2367,14 +2457,19 @@ TEST_F(TunnelingConnectionPoolImplTest, PoolReady) {
 TEST_F(TunnelingConnectionPoolImplTest, OnStreamFailure) {
   setup();
   createNewStream();
+  pool_->onPoolReady(request_encoder_, upstream_host_, stream_info_, absl::nullopt);
+
   EXPECT_CALL(stream_callbacks_,
               onStreamFailure(ConnectionPool::PoolFailureReason::RemoteConnectionFailure, "", _));
   pool_->onStreamFailure();
+  pool_->onDownstreamEvent(Network::ConnectionEvent::LocalClose);
 }
 
 TEST_F(TunnelingConnectionPoolImplTest, OnStreamSuccess) {
   setup();
   createNewStream();
+  pool_->onPoolReady(request_encoder_, upstream_host_, stream_info_, absl::nullopt);
+
   EXPECT_CALL(stream_callbacks_, onStreamReady(_, _, _, _, _));
   pool_->onStreamSuccess(request_encoder_);
 }