router filter: implement hedge_on_per_try_timeout.

api/envoy/api/v2/route/route.proto

@@ -143,7 +143,6 @@ message VirtualHost {
   // Indicates the hedge policy for all routes in this virtual host. Note that setting a
   // route level entry will take precedence over this config and it'll be treated
   // independently (e.g.: values are not inherited).
-  // [#not-implemented-hide:]
   HedgePolicy hedge_policy = 17;
 }
 
@@ -792,7 +791,6 @@ message RouteAction {
   // Indicates that the route has a hedge policy. Note that if this is set,
   // it'll take precedence over the virtual host level hedge policy entirely
   // (e.g.: policies are not merged, most internal one becomes the enforced policy).
-  // [#not-implemented-hide:]
   HedgePolicy hedge_policy = 27;
 }
 
@@ -859,22 +857,28 @@ message RetryPolicy {
   repeated uint32 retriable_status_codes = 7;
 }
 
-// HTTP request hedging TODO(mpuncel) docs
-// [#not-implemented-hide:]
+// HTTP request hedging :ref:`architecture overview <arch_overview_http_routing_hedging>`.
 message HedgePolicy {
   // Specifies the number of initial requests that should be sent upstream.
   // Must be at least 1.
   // Defaults to 1.
+  // [#not-implemented-hide:]
   google.protobuf.UInt32Value initial_requests = 1 [(validate.rules).uint32.gte = 1];
 
   // Specifies a probability that an additional upstream request should be sent
   // on top of what is specified by initial_requests.
   // Defaults to 0.
+  // [#not-implemented-hide:]
   envoy.type.FractionalPercent additional_request_chance = 2;
 
   // Indicates that a hedged request should be sent when the per-try timeout
   // is hit. This will only occur if the retry policy also indicates that a
-  // timed out request should be retried. Defaults to false.
+  // timed out request should be retried.
+  // Once a timed out request is retried due to per try timeout, the router
+  // filter will ensure that it is not retried again even if the returned
+  // response headers would otherwise be retried according the specified
+  // :ref:`RetryPolicy <envoy_api_msg_route.RetryPolicy>`.
+  // Defaults to false.
   bool hedge_on_per_try_timeout = 3;
 }
 

docs/root/configuration/cluster_manager/cluster_stats.rst

@@ -60,6 +60,7 @@ Every cluster has a statistics tree rooted at *cluster.<name>.* with the followi
   upstream_cx_max_requests, Counter, Total connections closed due to maximum requests
   upstream_cx_none_healthy, Counter, Total times connection not established due to no healthy hosts
   upstream_rq_total, Counter, Total requests
+  upstream_rq_hedge_abandoned, Counter, Number of hedged requests that were abandoned due to accepting another response.
   upstream_rq_active, Gauge, Total active requests
   upstream_rq_pending_total, Counter, Total requests pending a connection pool connection
   upstream_rq_pending_overflow, Counter, Total requests that overflowed connection pool circuit breaking and were failed

docs/root/intro/arch_overview/http_routing.rst

@@ -35,6 +35,7 @@ request. The router filter supports the following features:
 * Request timeout specified either via :ref:`HTTP
   header <config_http_filters_router_headers_consumed>` or via :ref:`route configuration
   <envoy_api_field_route.RouteAction.timeout>`.
+* :ref:`Request hedging <arch_overview_http_routing_hedging>` for retries in response to a request (per try) timeout.
 * Traffic shifting from one upstream cluster to another via :ref:`runtime values
   <envoy_api_field_route.RouteMatch.runtime_fraction>` (see :ref:`traffic shifting/splitting
   <config_http_conn_man_route_table_traffic_splitting>`).
@@ -87,6 +88,21 @@ headers <config_http_filters_router_headers_consumed>`. The following configurat
 Note that retries may be disabled depending on the contents of the :ref:`x-envoy-overloaded
 <config_http_filters_router_x-envoy-overloaded_consumed>`.
 
+.. _arch_overview_http_routing_hedging:
+
+Request Hedging
+---------------
+
+Envoy supports request hedging via specifying a :ref:`hedge policy
+<envoy_api_msg_route.HedgePolicy>`. This means that Envoy will race multiple
+simultaneous upstream requests and return the first valid response to the
+downstream according to retry policy.
+
+Currently hedging can only be applied to retries performed in response to a
+request timeout. The implementation ensures that the same upstream request is
+not retried twice, for instance if it times out and then later receives a 5xx
+response and the retry policy calls for retrying on 5xx.
+
 .. _arch_overview_http_routing_priority:
 
 Priority routing

docs/root/intro/version_history.rst

@@ -49,6 +49,7 @@ Version history
 * router: added reset reason to response body when upstream reset happens. After this change, the response body will be of the form `upstream connect error or disconnect/reset before headers. reset reason:`
 * router: added :ref:`rq_reset_after_downstream_response_started <config_http_filters_router_stats>` counter stat to router stats.
 * router: added per-route configuration of :ref:`internal redirects <envoy_api_field_route.RouteAction.internal_redirect_action>`.
+* router: added ability to issue a hedged retry in response to a per try timeout via a :ref:`hedge policy <envoy_api_msg_route.HedgePolicy>`.
 * stats: added support for histograms in prometheus
 * stats: added usedonly flag to prometheus stats to only output metrics which have been
   updated at least once.

docs/root/operations/admin.rst

@@ -84,6 +84,7 @@ modify different aspects of the server:
       cx_total, Counter, Total connections
       cx_active, Gauge, Total active connections
       cx_connect_fail, Counter, Total connection failures
+      rq_hedge_abandoned, Counter, Total hedged requests that were canceled and abandoned due to accepting another response.
       rq_total, Counter, Total requests
       rq_timeout, Counter, Total timed out requests
       rq_success, Counter, Total requests with non-5xx responses

include/envoy/router/router.h

@@ -240,6 +240,16 @@ class RetryState {
   virtual RetryStatus shouldRetryHeaders(const Http::HeaderMap& response_headers,
                                          DoRetryCallback callback) PURE;
 
+  /**
+   * Determines whether given response headers would be retried by the retry policy, assuming
+   * sufficient retry budget and circuit breaker headroom. This is useful in cases where
+   * the information about whether a response is "good" or not is useful, but a retry should
+   * not be attempted for other reasons.
+   * @param response_headers supplies the response headers.
+   * @return bool true if a retry would be warranted based on the retry policy.
+   */
+  virtual bool wouldRetryFromHeaders(const Http::HeaderMap& response_headers) PURE;
+
   /**
    * Determine whether a request should be retried after a reset based on the reason for the reset.
    * @param reset_reason supplies the reset reason.
@@ -253,6 +263,19 @@ class RetryState {
   virtual RetryStatus shouldRetryReset(const Http::StreamResetReason reset_reason,
                                        DoRetryCallback callback) PURE;
 
+  /**
+   * Determine whether a "hedged" retry should be sent after the per try
+   * timeout expires. This means the original request is not canceled, but a
+   * new one is sent to hedge against the original request taking even longer.
+   * @param callback supplies the callback that will be invoked when the retry should take place.
+   *                 This is used to add timed backoff, etc. The callback will never be called
+   *                 inline.
+   * @return RetryStatus if a retry should take place. @param callback will be called at some point
+   *         in the future. Otherwise a retry should not take place and the callback will never be
+   *         called. Calling code should proceed with error handling.
+   */
+  virtual RetryStatus shouldHedgeRetryPerTryTimeout(DoRetryCallback callback) PURE;
+
   /**
    * Called when a host was attempted but the request failed and is eligible for another retry.
    * Should be used to update whatever internal state depends on previously attempted hosts.

include/envoy/upstream/host_description.h

@@ -24,6 +24,7 @@ namespace Upstream {
   COUNTER(cx_total)                                                                                \
   GAUGE  (cx_active)                                                                               \
   COUNTER(cx_connect_fail)                                                                         \
+  COUNTER(rq_hedge_abandoned)                                                                      \
   COUNTER(rq_total)                                                                                \
   COUNTER(rq_timeout)                                                                              \
   COUNTER(rq_success)                                                                              \

include/envoy/upstream/upstream.h

@@ -474,6 +474,7 @@ class PrioritySet {
   COUNTER  (upstream_cx_max_requests)                                                              \
   COUNTER  (upstream_cx_none_healthy)                                                              \
   COUNTER  (upstream_rq_total)                                                                     \
+  COUNTER  (upstream_rq_hedge_abandoned)                                                           \
   GAUGE    (upstream_rq_active)                                                                    \
   COUNTER  (upstream_rq_completed)                                                                 \
   COUNTER  (upstream_rq_pending_total)                                                             \

source/common/http/conn_manager_impl.cc

@@ -1477,16 +1477,16 @@ bool ConnectionManagerImpl::ActiveStream::verbose() const {
 
 void ConnectionManagerImpl::ActiveStream::callHighWatermarkCallbacks() {
   ++high_watermark_count_;
-  if (watermark_callbacks_) {
-    watermark_callbacks_->onAboveWriteBufferHighWatermark();
+  for (auto watermark_callbacks : watermark_callbacks_) {
+    watermark_callbacks->onAboveWriteBufferHighWatermark();
   }
 }
 
 void ConnectionManagerImpl::ActiveStream::callLowWatermarkCallbacks() {
   ASSERT(high_watermark_count_ > 0);
   --high_watermark_count_;
-  if (watermark_callbacks_) {
-    watermark_callbacks_->onBelowWriteBufferLowWatermark();
+  for (auto watermark_callbacks : watermark_callbacks_) {
+    watermark_callbacks->onBelowWriteBufferLowWatermark();
   }
 }
 
@@ -1801,19 +1801,20 @@ void ConnectionManagerImpl::ActiveStreamDecoderFilter::
 
 void ConnectionManagerImpl::ActiveStreamDecoderFilter::addDownstreamWatermarkCallbacks(
     DownstreamWatermarkCallbacks& watermark_callbacks) {
-  // This is called exactly once per stream, by the router filter.
-  // If there's ever a need for another filter to subscribe to watermark callbacks this can be
-  // turned into a vector.
-  ASSERT(parent_.watermark_callbacks_ == nullptr);
-  parent_.watermark_callbacks_ = &watermark_callbacks;
+  // This is called exactly once per stream, by the router filter. Therefore we
+  // expect the same callbacks to not be registered twice.
+  ASSERT(std::find(parent_.watermark_callbacks_.begin(), parent_.watermark_callbacks_.end(),
+                   &watermark_callbacks) == parent_.watermark_callbacks_.end());
+  parent_.watermark_callbacks_.emplace(parent_.watermark_callbacks_.end(), &watermark_callbacks);
   for (uint32_t i = 0; i < parent_.high_watermark_count_; ++i) {
     watermark_callbacks.onAboveWriteBufferHighWatermark();
   }
 }
 void ConnectionManagerImpl::ActiveStreamDecoderFilter::removeDownstreamWatermarkCallbacks(
     DownstreamWatermarkCallbacks& watermark_callbacks) {
-  ASSERT(parent_.watermark_callbacks_ == &watermark_callbacks);
-  parent_.watermark_callbacks_ = nullptr;
+  ASSERT(std::find(parent_.watermark_callbacks_.begin(), parent_.watermark_callbacks_.end(),
+                   &watermark_callbacks) != parent_.watermark_callbacks_.end());
+  parent_.watermark_callbacks_.remove(&watermark_callbacks);
 }
 
 bool ConnectionManagerImpl::ActiveStreamDecoderFilter::recreateStream() {

source/common/http/conn_manager_impl.h

@@ -442,7 +442,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     StreamInfo::StreamInfoImpl stream_info_;
     absl::optional<Router::RouteConstSharedPtr> cached_route_;
     absl::optional<Upstream::ClusterInfoConstSharedPtr> cached_cluster_info_;
-    DownstreamWatermarkCallbacks* watermark_callbacks_{nullptr};
+    std::list<DownstreamWatermarkCallbacks*> watermark_callbacks_{};
     uint32_t buffer_limit_{0};
     uint32_t high_watermark_count_{0};
     const std::string* decorated_operation_{nullptr};

source/common/router/BUILD

@@ -145,6 +145,7 @@ envoy_cc_library(
         "//source/common/common:enum_to_int",
         "//source/common/common:hash_lib",
         "//source/common/common:hex_lib",
+        "//source/common/common:linked_object",
         "//source/common/common:minimal_logger_lib",
         "//source/common/common:utility_lib",
         "//source/common/grpc:common_lib",

source/common/router/retry_state_impl.cc

@@ -193,6 +193,13 @@ RetryStatus RetryStateImpl::shouldRetryReset(Http::StreamResetReason reset_reaso
   return shouldRetry(wouldRetryFromReset(reset_reason), callback);
 }
 
+RetryStatus RetryStateImpl::shouldHedgeRetryPerTryTimeout(DoRetryCallback callback) {
+  // A hedged retry on per try timeout is always retried if there are retries
+  // left. NOTE: this is different than non-hedged per try timeouts which are only retried
+  // if RETRY_ON_5XX or RETRY_ON_GATEWAY_ERROR
+  return shouldRetry([]() -> bool { return true; }, callback);
+}
+
 bool RetryStateImpl::wouldRetryFromHeaders(const Http::HeaderMap& response_headers) {
   if (response_headers.EnvoyOverloaded() != nullptr) {
     return false;

source/common/router/retry_state_impl.h

@@ -38,8 +38,12 @@ class RetryStateImpl : public RetryState {
   bool enabled() override { return retry_on_ != 0; }
   RetryStatus shouldRetryHeaders(const Http::HeaderMap& response_headers,
                                  DoRetryCallback callback) override;
+  // Returns true if the retry policy would retry the passed headers. Does not
+  // take into account circuit breaking or remaining tries.
+  bool wouldRetryFromHeaders(const Http::HeaderMap& response_headers) override;
   RetryStatus shouldRetryReset(const Http::StreamResetReason reset_reason,
                                DoRetryCallback callback) override;
+  RetryStatus shouldHedgeRetryPerTryTimeout(DoRetryCallback callback) override;
 
   void onHostAttempted(Upstream::HostDescriptionConstSharedPtr host) override {
     std::for_each(retry_host_predicates_.begin(), retry_host_predicates_.end(),
@@ -75,7 +79,6 @@ class RetryStateImpl : public RetryState {
   void enableBackoffTimer();
   void resetRetry();
   bool wouldRetryFromReset(const Http::StreamResetReason reset_reason);
-  bool wouldRetryFromHeaders(const Http::HeaderMap& response_headers);
   RetryStatus shouldRetry(bool would_retry, DoRetryCallback callback);
 
   const Upstream::ClusterInfo& cluster_;