router: add support for retry options predicate extensions

api/envoy/config/route/v3/route_components.proto

@@ -1164,7 +1164,7 @@ message RouteAction {
 }
 
 // HTTP retry :ref:`architecture overview <arch_overview_http_routing_retry>`.
-// [#next-free-field: 12]
+// [#next-free-field: 13]
 message RetryPolicy {
   option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.route.RetryPolicy";
 
@@ -1329,6 +1329,11 @@ message RetryPolicy {
   // details.
   repeated RetryHostPredicate retry_host_predicate = 5;
 
+  // Retry options predicates that will be applied prior to retrying a request. These predicates
+  // allow customizing request behavior between retries.
+  // [#comment: add [#extension-category: envoy.retry_options_predicates] when there are built-in extensions]
+  repeated core.v3.TypedExtensionConfig retry_options_predicates = 12;
+
   // The maximum number of times host selection will be reattempted before giving up, at which
   // point the host that was last selected will be routed to. If unspecified, this will default to
   // retrying once.

docs/root/version_history/current.rst

@@ -113,6 +113,10 @@ New Features
 * overload: add a new overload action that resets streams using a lot of memory. To enable the tracking of allocated bytes in buffers that a stream is using  we need to configure the minimum threshold for tracking via:ref:`buffer_factory_config <envoy_v3_api_field_config.overload.v3.OverloadManager.buffer_factory_config>`. We have an overload action ``Envoy::Server::OverloadActionNameValues::ResetStreams`` that takes advantage of the tracking to  reset the most expensive stream first.
 * rbac: added :ref:`destination_port_range <envoy_v3_api_field_config.rbac.v3.Permission.destination_port_range>` for matching range of destination ports.
 * route config: added :ref:`dynamic_metadata <envoy_v3_api_field_config.route.v3.RouteMatch.dynamic_metadata>` for routing based on dynamic metadata.
+* router: added retry options predicate extensions configured via
+  :ref:` <envoy_v3_api_field_config.route.v3.RetryPolicy.retry_options_predicates>`. These
+  extensions allow modification of requests between retries at the router level. There are not
+  currently any built-in extensions that implement this extension point.
 * sxg_filter: added filter to transform response to SXG package to :ref:`contrib images <install_contrib>`. This can be enabled by setting :ref:`SXG <envoy_v3_api_msg_extensions.filters.http.sxg.v3alpha.SXG>` configuration.
 * thrift_proxy: added support for :ref:`mirroring requests <envoy_v3_api_field_extensions.filters.network.thrift_proxy.v3.RouteAction.request_mirror_policies>`.
 

envoy/router/router.h

@@ -231,6 +231,13 @@ class RetryPolicy {
    */
   virtual Upstream::RetryPrioritySharedPtr retryPriority() const PURE;
 
+  /**
+   * @return the retry options predicates for this policy. Each policy will be applied prior
+   * to retrying a request, allowing for request behavior to be customized.
+   */
+  virtual const std::vector<Upstream::RetryOptionsPredicateConstSharedPtr>&
+  retryOptionsPredicates() const PURE;
+
   /**
    * Number of times host selection should be reattempted when selecting a host
    * for a retry attempt.

envoy/upstream/cluster_manager.h

@@ -406,6 +406,11 @@ class ClusterManagerFactory {
    * Returns the secret manager.
    */
   virtual Secret::SecretManager& secretManager() PURE;
+
+  /**
+   * Returns the singleton manager.
+   */
+  virtual Singleton::Manager& singletonManager() PURE;
 };
 
 /**

envoy/upstream/retry.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "envoy/config/typed_config.h"
+#include "envoy/singleton/manager.h"
 #include "envoy/upstream/types.h"
 #include "envoy/upstream/upstream.h"
 
@@ -92,13 +93,57 @@ class RetryHostPredicate {
 
 using RetryHostPredicateSharedPtr = std::shared_ptr<RetryHostPredicate>;
 
+/**
+ * A predicate that is applied prior to retrying a request. Each predicate can customize request
+ * behavior prior to the request being retried.
+ */
+class RetryOptionsPredicate {
+public:
+  struct UpdateOptionsParameters {
+    // Stream info for the previous request attempt that is about to be retried.
+    const StreamInfo::StreamInfo& retriable_request_stream_info_;
+    // The current upstream socket options that were used for connection pool selection on the
+    // previous attempt.
+    Network::Socket::OptionsSharedPtr current_upstream_socket_options_;
+  };
+
+  struct UpdateOptionsReturn {
+    // New upstream socket options to apply to the next request attempt. If changed, will effect
+    // connection pool selection similar to that which was done for the initial request.
+    absl::optional<Network::Socket::OptionsSharedPtr> new_upstream_socket_options_;
+  };
+
+  virtual ~RetryOptionsPredicate() = default;
+
+  /**
+   * Update request options.
+   * @param parameters supplies the update parameters.
+   * @return the new options to apply. Each option is wrapped in an optional and is only applied
+   *         if valid.
+   */
+  virtual UpdateOptionsReturn updateOptions(const UpdateOptionsParameters& parameters) const PURE;
+};
+
+using RetryOptionsPredicateConstSharedPtr = std::shared_ptr<const RetryOptionsPredicate>;
+
+/**
+ * Context for all retry extensions.
+ */
+class RetryExtensionFactoryContext {
+public:
+  virtual ~RetryExtensionFactoryContext() = default;
+
+  /**
+   * @return Singleton::Manager& the server-wide singleton manager.
+   */
+  virtual Singleton::Manager& singletonManager() PURE;
+};
+
 /**
  * Factory for RetryPriority.
  */
 class RetryPriorityFactory : public Config::TypedFactory {
 public:
-  ~RetryPriorityFactory() override = default;
-
   virtual RetryPrioritySharedPtr
   createRetryPriority(const Protobuf::Message& config,
                       ProtobufMessage::ValidationVisitor& validation_visitor,
@@ -112,13 +157,23 @@ class RetryPriorityFactory : public Config::TypedFactory {
  */
 class RetryHostPredicateFactory : public Config::TypedFactory {
 public:
-  ~RetryHostPredicateFactory() override = default;
-
   virtual RetryHostPredicateSharedPtr createHostPredicate(const Protobuf::Message& config,
                                                           uint32_t retry_count) PURE;
 
   std::string category() const override { return "envoy.retry_host_predicates"; }
 };
 
+/**
+ * Factory for RetryOptionsPredicate.
+ */
+class RetryOptionsPredicateFactory : public Config::TypedFactory {
+public:
+  virtual RetryOptionsPredicateConstSharedPtr
+  createOptionsPredicate(const Protobuf::Message& config,
+                         RetryExtensionFactoryContext& context) PURE;
+
+  std::string category() const override { return "envoy.retry_options_predicates"; }
+};
+
 } // namespace Upstream
 } // namespace Envoy

source/common/http/async_client_impl.cc

@@ -42,7 +42,7 @@ AsyncClientImpl::AsyncClientImpl(Upstream::ClusterInfoConstSharedPtr cluster,
       config_(http_context.asyncClientStatPrefix(), local_info, stats_store, cm, runtime, random,
               std::move(shadow_writer), true, false, false, false, false, {},
               dispatcher.timeSource(), http_context, router_context),
-      dispatcher_(dispatcher) {}
+      dispatcher_(dispatcher), singleton_manager_(cm.clusterManagerFactory().singletonManager()) {}
 
 AsyncClientImpl::~AsyncClientImpl() {
   while (!active_streams_.empty()) {
@@ -81,8 +81,8 @@ AsyncStreamImpl::AsyncStreamImpl(AsyncClientImpl& parent, AsyncClient::StreamCal
       router_(parent.config_),
       stream_info_(Protocol::Http11, parent.dispatcher().timeSource(), nullptr),
       tracing_config_(Tracing::EgressConfig::get()),
-      route_(std::make_shared<RouteImpl>(parent_.cluster_->name(), options.timeout,
-                                         options.hash_policy, options.retry_policy)),
+      route_(std::make_shared<RouteImpl>(parent_, options.timeout, options.hash_policy,
+                                         options.retry_policy)),
       send_xff_(options.send_xff) {
 
   stream_info_.dynamicMetadata().MergeFrom(options.metadata);

source/common/http/async_client_impl.h

@@ -40,6 +40,7 @@
 #include "source/common/router/router.h"
 #include "source/common/stream_info/stream_info_impl.h"
 #include "source/common/tracing/http_tracer_impl.h"
+#include "source/common/upstream/retry_factory.h"
 
 namespace Envoy {
 namespace Http {
@@ -67,6 +68,7 @@ class AsyncClientImpl final : public AsyncClient {
   Router::FilterConfig config_;
   Event::Dispatcher& dispatcher_;
   std::list<std::unique_ptr<AsyncStreamImpl>> active_streams_;
+  Singleton::Manager& singleton_manager_;
 
   friend class AsyncStreamImpl;
   friend class AsyncRequestImpl;
@@ -124,45 +126,6 @@ class AsyncStreamImpl : public AsyncClient::Stream,
         rate_limit_policy_entry_;
   };
 
-  struct NullRetryPolicy : public Router::RetryPolicy {
-    // Router::RetryPolicy
-    std::chrono::milliseconds perTryTimeout() const override {
-      return std::chrono::milliseconds(0);
-    }
-    std::vector<Upstream::RetryHostPredicateSharedPtr> retryHostPredicates() const override {
-      return {};
-    }
-    Upstream::RetryPrioritySharedPtr retryPriority() const override { return {}; }
-
-    uint32_t hostSelectionMaxAttempts() const override { return 1; }
-    uint32_t numRetries() const override { return 1; }
-    uint32_t retryOn() const override { return 0; }
-    const std::vector<uint32_t>& retriableStatusCodes() const override {
-      return retriable_status_codes_;
-    }
-    const std::vector<Http::HeaderMatcherSharedPtr>& retriableHeaders() const override {
-      return retriable_headers_;
-    }
-    const std::vector<Http::HeaderMatcherSharedPtr>& retriableRequestHeaders() const override {
-      return retriable_request_headers_;
-    }
-    absl::optional<std::chrono::milliseconds> baseInterval() const override {
-      return absl::nullopt;
-    }
-    absl::optional<std::chrono::milliseconds> maxInterval() const override { return absl::nullopt; }
-    const std::vector<Router::ResetHeaderParserSharedPtr>& resetHeaders() const override {
-      return reset_headers_;
-    }
-    std::chrono::milliseconds resetMaxInterval() const override {
-      return std::chrono::milliseconds(300000);
-    }
-
-    const std::vector<uint32_t> retriable_status_codes_{};
-    const std::vector<Http::HeaderMatcherSharedPtr> retriable_headers_{};
-    const std::vector<Http::HeaderMatcherSharedPtr> retriable_request_headers_{};
-    const std::vector<Router::ResetHeaderParserSharedPtr> reset_headers_{};
-  };
-
   struct NullConfig : public Router::Config {
     Router::RouteConstSharedPtr route(const Http::RequestHeaderMap&, const StreamInfo::StreamInfo&,
                                       uint64_t) const override {
@@ -208,20 +171,21 @@ class AsyncStreamImpl : public AsyncClient::Stream,
 
   struct RouteEntryImpl : public Router::RouteEntry {
     RouteEntryImpl(
-        const std::string& cluster_name, const absl::optional<std::chrono::milliseconds>& timeout,
+        AsyncClientImpl& parent, const absl::optional<std::chrono::milliseconds>& timeout,
         const Protobuf::RepeatedPtrField<envoy::config::route::v3::RouteAction::HashPolicy>&
             hash_policy,
         const absl::optional<envoy::config::route::v3::RetryPolicy>& retry_policy)
-        : cluster_name_(cluster_name), timeout_(timeout) {
+        : cluster_name_(parent.cluster_->name()), timeout_(timeout) {
       if (!hash_policy.empty()) {
         hash_policy_ = std::make_unique<HashPolicyImpl>(hash_policy);
       }
       if (retry_policy.has_value()) {
         // ProtobufMessage::getStrictValidationVisitor() ?  how often do we do this?
+        Upstream::RetryExtensionFactoryContextImpl factory_context(parent.singleton_manager_);
         retry_policy_ = std::make_unique<Router::RetryPolicyImpl>(
-            retry_policy.value(), ProtobufMessage::getNullValidationVisitor());
+            retry_policy.value(), ProtobufMessage::getNullValidationVisitor(), factory_context);
       } else {
-        retry_policy_ = std::make_unique<NullRetryPolicy>();
+        retry_policy_ = std::make_unique<Router::RetryPolicyImpl>();
       }
     }
 
@@ -327,12 +291,11 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   };
 
   struct RouteImpl : public Router::Route {
-    RouteImpl(const std::string& cluster_name,
-              const absl::optional<std::chrono::milliseconds>& timeout,
+    RouteImpl(AsyncClientImpl& parent, const absl::optional<std::chrono::milliseconds>& timeout,
               const Protobuf::RepeatedPtrField<envoy::config::route::v3::RouteAction::HashPolicy>&
                   hash_policy,
               const absl::optional<envoy::config::route::v3::RetryPolicy>& retry_policy)
-        : route_entry_(cluster_name, timeout, hash_policy, retry_policy), typed_metadata_({}) {}
+        : route_entry_(parent, timeout, hash_policy, retry_policy), typed_metadata_({}) {}
 
     // Router::Route
     const Router::DirectResponseEntry* directResponseEntry() const override { return nullptr; }

source/common/router/BUILD

@@ -63,6 +63,7 @@ envoy_cc_library(
         "//source/common/http:utility_lib",
         "//source/common/protobuf:utility_lib",
         "//source/common/tracing:http_tracer_lib",
+        "//source/common/upstream:retry_factory_lib",
         "//source/extensions/filters/http/common:utility_lib",
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
         "@envoy_api//envoy/config/route/v3:pkg_cc_proto",

source/common/router/config_impl.cc

@@ -37,6 +37,7 @@
 #include "source/common/router/retry_state_impl.h"
 #include "source/common/runtime/runtime_features.h"
 #include "source/common/tracing/http_tracer_impl.h"
+#include "source/common/upstream/retry_factory.h"
 #include "source/extensions/filters/http/common/utility.h"
 
 #include "absl/strings/match.h"
@@ -87,7 +88,8 @@ HedgePolicyImpl::HedgePolicyImpl(const envoy::config::route::v3::HedgePolicy& he
 HedgePolicyImpl::HedgePolicyImpl() : initial_requests_(1), hedge_on_per_try_timeout_(false) {}
 
 RetryPolicyImpl::RetryPolicyImpl(const envoy::config::route::v3::RetryPolicy& retry_policy,
-                                 ProtobufMessage::ValidationVisitor& validation_visitor)
+                                 ProtobufMessage::ValidationVisitor& validation_visitor,
+                                 Upstream::RetryExtensionFactoryContext& factory_context)
     : retriable_headers_(
           Http::HeaderUtility::buildHeaderMatcherVector(retry_policy.retriable_headers())),
       retriable_request_headers_(
@@ -116,6 +118,16 @@ RetryPolicyImpl::RetryPolicyImpl(const envoy::config::route::v3::RetryPolicy& re
                                      retry_priority, validation_visitor, factory));
   }
 
+  for (const auto& options_predicate : retry_policy.retry_options_predicates()) {
+    auto& factory =
+        Envoy::Config::Utility::getAndCheckFactory<Upstream::RetryOptionsPredicateFactory>(
+            options_predicate);
+    retry_options_predicates_.emplace_back(
+        factory.createOptionsPredicate(*Envoy::Config::Utility::translateToFactoryConfig(
+                                           options_predicate, validation_visitor, factory),
+                                       factory_context));
+  }
+
   auto host_selection_attempts = retry_policy.host_selection_retry_max_attempts();
   if (host_selection_attempts) {
     host_selection_attempts_ = host_selection_attempts;
@@ -348,7 +360,8 @@ RouteEntryImplBase::RouteEntryImplBase(const VirtualHostImpl& vhost,
       prefix_rewrite_redirect_(route.redirect().prefix_rewrite()),
       strip_query_(route.redirect().strip_query()),
       hedge_policy_(buildHedgePolicy(vhost.hedgePolicy(), route.route())),
-      retry_policy_(buildRetryPolicy(vhost.retryPolicy(), route.route(), validator)),
+      retry_policy_(
+          buildRetryPolicy(vhost.retryPolicy(), route.route(), validator, factory_context)),
       internal_redirect_policy_(
           buildInternalRedirectPolicy(route.route(), validator, route.name())),
       rate_limit_policy_(route.route().rate_limits(), validator),
@@ -891,15 +904,18 @@ HedgePolicyImpl RouteEntryImplBase::buildHedgePolicy(
 RetryPolicyImpl RouteEntryImplBase::buildRetryPolicy(
     const absl::optional<envoy::config::route::v3::RetryPolicy>& vhost_retry_policy,
     const envoy::config::route::v3::RouteAction& route_config,
-    ProtobufMessage::ValidationVisitor& validation_visitor) const {
+    ProtobufMessage::ValidationVisitor& validation_visitor,
+    Server::Configuration::ServerFactoryContext& factory_context) const {
+  Upstream::RetryExtensionFactoryContextImpl retry_factory_context(
+      factory_context.singletonManager());
   // Route specific policy wins, if available.
   if (route_config.has_retry_policy()) {
-    return RetryPolicyImpl(route_config.retry_policy(), validation_visitor);
+    return RetryPolicyImpl(route_config.retry_policy(), validation_visitor, retry_factory_context);
   }
 
   // If not, we fallback to the virtual host policy if there is one.
   if (vhost_retry_policy) {
-    return RetryPolicyImpl(vhost_retry_policy.value(), validation_visitor);
+    return RetryPolicyImpl(vhost_retry_policy.value(), validation_visitor, retry_factory_context);
   }
 
   // Otherwise, an empty policy will do.