support backoff between retries in udp_proxy (#37912)

Commit Message: support backoff between retries in udp_proxy
Additional Description: This feature also fixes a bug in the retry
mechanism in udp proxy in the following situation:
1. We are tunneling UDP over HTTP.
2. A new stream is created on existing upstream connection
(multiplexing) and waiting for response headers.
3. The upstream connection is closed.
4. During the close process, all streams created on this connection will
be reset.
5. The udp_proxy receives a callback on the stream reset.
6. They retry to connect.
7. The closed connection is picked as it is still in the connection pool
(we are still in the process of the close).
8. The new stream that is created on the second attempt will be reset
immediately.
9. The same process will happen (step 5 - step 8) until we reach the
max_connect_attepts.
10. This means that we are doing only one real attempt in this
situation.

Risk Level: medium
Testing: unit tests, integration tests
Docs Changes: added
Release Notes: added

---------

Signed-off-by: Issa Abu Kalbein <[email protected]>
Co-authored-by: Issa Abu Kalbein <[email protected]>

api/envoy/extensions/filters/udp/udp_proxy/v3/udp_proxy.proto

@@ -3,6 +3,7 @@ syntax = "proto3";
 package envoy.extensions.filters.udp.udp_proxy.v3;
 
 import "envoy/config/accesslog/v3/accesslog.proto";
+import "envoy/config/core/v3/backoff.proto";
 import "envoy/config/core/v3/base.proto";
 import "envoy/config/core/v3/config_source.proto";
 import "envoy/config/core/v3/udp_socket_config.proto";
@@ -91,6 +92,9 @@ message UdpProxyConfig {
       // The maximum number of unsuccessful connection attempts that will be made before giving up.
       // If the parameter is not specified, 1 connection attempt will be made.
       google.protobuf.UInt32Value max_connect_attempts = 1;
+
+      // Sets the backoff strategy. If not set, the retries are performed without backoff.
+      config.core.v3.BackoffStrategy backoff_options = 2;
     }
 
     // The hostname to send in the synthesized CONNECT headers to the upstream proxy.

changelogs/current.yaml

@@ -411,6 +411,11 @@ new_features:
   change: |
     Add the option to reduce the rate limit budget based on request/response contexts on stream done.
     See :ref:`apply_on_stream_done <envoy_v3_api_field_config.route.v3.RateLimit.apply_on_stream_done>` for more details.
+- area: udp_proxy
+  change: |
+    Added support for :ref:`backoff_options
+    <envoy_v3_api_field_extensions.filters.udp.udp_proxy.v3.UdpProxyConfig.UdpTunnelingConfig.RetryOptions.backoff_options>`
+    to configure the backoff strategy for UDP proxy retries when tunneling over HTTP.
 
 deprecated:
 - area: rbac

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

@@ -82,6 +82,21 @@ TunnelingConfigImpl::TunnelingConfigImpl(const TunnelingConfig& config,
       THROW_OR_RETURN_VALUE(Formatter::SubstitutionFormatStringUtils::fromProtoConfig(
                                 target_substitution_format_config, context),
                             Formatter::FormatterBasePtr<Formatter::HttpFormatterContext>);
+
+  if (config.has_retry_options() && config.retry_options().has_backoff_options()) {
+    const uint64_t base_interval_ms =
+        PROTOBUF_GET_MS_REQUIRED(config.retry_options().backoff_options(), base_interval);
+    const uint64_t max_interval_ms = PROTOBUF_GET_MS_OR_DEFAULT(
+        config.retry_options().backoff_options(), max_interval, base_interval_ms * 10);
+
+    if (max_interval_ms < base_interval_ms) {
+      throw EnvoyException(
+          "max_backoff_interval must be greater or equal to base_backoff_interval");
+    }
+
+    backoff_strategy_ = std::make_unique<JitteredExponentialBackOffStrategy>(
+        base_interval_ms, max_interval_ms, context.serverFactoryContext().api().randomGenerator());
+  }
 }
 
 UdpProxyFilterConfigImpl::UdpProxyFilterConfigImpl(

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

@@ -61,6 +61,7 @@ class TunnelingConfigImpl : public UdpTunnelingConfig {
   bool usePost() const override { return use_post_; };
   const std::string& postPath() const override { return post_path_; }
   Http::HeaderEvaluator& headerEvaluator() const override { return *header_parser_; };
+  const BackOffStrategyPtr& backoffStrategy() const override { return backoff_strategy_; };
   uint32_t maxConnectAttempts() const override { return max_connect_attempts_; };
   bool bufferEnabled() const override { return buffer_enabled_; };
   uint32_t maxBufferedDatagrams() const override { return max_buffered_datagrams_; };
@@ -100,6 +101,7 @@ class TunnelingConfigImpl : public UdpTunnelingConfig {
   bool use_post_;
   std::string post_path_;
   const uint32_t max_connect_attempts_;
+  BackOffStrategyPtr backoff_strategy_;
   bool buffer_enabled_;
   uint32_t max_buffered_datagrams_;
   uint64_t max_buffered_bytes_;

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

@@ -1098,9 +1098,12 @@ void UdpProxyFilter::TunnelingActiveSession::onUpstreamEvent(Network::Connection
       event == Network::ConnectionEvent::LocalClose) {
     upstream_.reset();
 
-    if (!connecting || !establishUpstreamConnection()) {
+    if (!connecting) {
       filter_.removeSession(this);
+      return;
     }
+
+    resetRetryTimer();
   }
 }
 
@@ -1202,6 +1205,43 @@ void UdpProxyFilter::TunnelingActiveSession::onIdleTimer() {
   filter_.removeSession(this);
 }
 
+void UdpProxyFilter::TunnelingActiveSession::onRetryTimer() {
+  if (!establishUpstreamConnection()) {
+    filter_.removeSession(this);
+  }
+}
+
+void UdpProxyFilter::TunnelingActiveSession::resetRetryTimer() {
+  // Create the retry timer on the first retry.
+  if (!retry_timer_) {
+    retry_timer_ =
+        filter_.read_callbacks_->udpListener().dispatcher().createTimer([this] { onRetryTimer(); });
+  }
+
+  // If the backoff strategy is not configured, the next backoff time will be 0.
+  // This will allow the retry to happen on different event loop iteration, which
+  // will allow the connection pool to be cleaned up from the previous closed connection.
+  uint64_t next_backoff_ms = 0;
+
+  if (filter_.config_->tunnelingConfig()->backoffStrategy()) {
+    next_backoff_ms = filter_.config_->tunnelingConfig()->backoffStrategy()->nextBackOffMs();
+  }
+
+  retry_timer_->enableTimer(std::chrono::milliseconds(next_backoff_ms));
+}
+
+void UdpProxyFilter::TunnelingActiveSession::disableRetryTimer() {
+  if (retry_timer_ != nullptr) {
+    retry_timer_->disableTimer();
+    retry_timer_.reset();
+  }
+}
+
+void UdpProxyFilter::TunnelingActiveSession::onSessionComplete() {
+  disableRetryTimer();
+  ActiveSession::onSessionComplete();
+}
+
 } // namespace UdpProxy
 } // namespace UdpFilters
 } // namespace Extensions

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

@@ -98,6 +98,7 @@ class UdpTunnelingConfig {
   virtual const std::string& postPath() const PURE;
   virtual Http::HeaderEvaluator& headerEvaluator() const PURE;
   virtual uint32_t maxConnectAttempts() const PURE;
+  virtual const BackOffStrategyPtr& backoffStrategy() const PURE;
   virtual bool bufferEnabled() const PURE;
   virtual uint32_t maxBufferedDatagrams() const PURE;
   virtual uint64_t maxBufferedBytes() const PURE;
@@ -602,7 +603,7 @@ class UdpProxyFilter : public Network::UdpListenerReadFilter,
     bool onContinueFilterChain(ActiveReadFilter* filter);
     void onInjectReadDatagramToFilterChain(ActiveReadFilter* filter, Network::UdpRecvData& data);
     void onInjectWriteDatagramToFilterChain(ActiveWriteFilter* filter, Network::UdpRecvData& data);
-    void onSessionComplete();
+    virtual void onSessionComplete();
 
     // SessionFilters::FilterChainFactoryCallbacks
     void addReadFilter(ReadFilterSharedPtr filter) override {
@@ -743,6 +744,7 @@ class UdpProxyFilter : public Network::UdpListenerReadFilter,
     bool createUpstream() override;
     void writeUpstream(Network::UdpRecvData& data) override;
     void onIdleTimer() override;
+    void onSessionComplete() override;
 
     // UpstreamTunnelCallbacks
     void onUpstreamEvent(Network::ConnectionEvent event) override;
@@ -768,7 +770,11 @@ class UdpProxyFilter : public Network::UdpListenerReadFilter,
     bool createConnectionPool();
     void maybeBufferDatagram(Network::UdpRecvData& data);
     void flushBuffer();
+    void onRetryTimer();
+    void resetRetryTimer();
+    void disableRetryTimer();
 
+    Event::TimerPtr retry_timer_;
     TunnelingConnectionPoolFactoryPtr conn_pool_factory_;
     std::unique_ptr<UdpLoadBalancerContext> load_balancer_context_;
     TunnelingConnectionPoolPtr conn_pool_;

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

@@ -54,6 +54,7 @@ class MockUdpTunnelingConfig : public UdpTunnelingConfig {
   MOCK_METHOD(const std::string&, postPath, (), (const));
   MOCK_METHOD(Http::HeaderEvaluator&, headerEvaluator, (), (const));
   MOCK_METHOD(uint32_t, maxConnectAttempts, (), (const));
+  MOCK_METHOD(const BackOffStrategyPtr&, backoffStrategy, (), (const));
   MOCK_METHOD(bool, bufferEnabled, (), (const));
   MOCK_METHOD(uint32_t, maxBufferedDatagrams, (), (const));
   MOCK_METHOD(uint64_t, maxBufferedBytes, (), (const));

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

@@ -2560,6 +2560,22 @@ TEST(TunnelingConfigImplTest, BufferingState) {
   }
 }
 
+TEST(TunnelingConfigImplTest, InvalidBackoffConfig) {
+  NiceMock<Server::Configuration::MockFactoryContext> context;
+  TunnelingConfig proto_config;
+  proto_config.mutable_retry_options()
+      ->mutable_backoff_options()
+      ->mutable_base_interval()
+      ->set_seconds(5);
+  proto_config.mutable_retry_options()
+      ->mutable_backoff_options()
+      ->mutable_max_interval()
+      ->set_seconds(1);
+  EXPECT_THROW_WITH_MESSAGE(
+      TunnelingConfigImpl(proto_config, context), EnvoyException,
+      "max_backoff_interval must be greater or equal to base_backoff_interval");
+}
+
 } // namespace
 } // namespace UdpProxy
 } // namespace UdpFilters