router: allow retry of streaming/incomplete requests

docs/root/configuration/http/http_filters/router_filter.rst

@@ -388,7 +388,6 @@ owning HTTP connection manager.
   rq_direct_response, Counter, Total requests that resulted in a direct response
   rq_total, Counter, Total routed requests
   rq_reset_after_downstream_response_started, Counter, Total requests that were reset after downstream response had started
-  rq_retry_skipped_request_not_complete, Counter, Total retries that were skipped as the request is not yet complete
 
 .. _config_http_filters_router_vcluster_stats:
 

docs/root/version_history/current.rst

@@ -10,6 +10,7 @@ Changes
   are applied to using :ref:`HTTP headers <config_http_filters_fault_injection_http_header>` to the HTTP fault filter.
 * http: fixed a bug where the upgrade header was not cleared on responses to non-upgrade requests.
   Can be reverted temporarily by setting runtime feature `envoy.reloadable_features.fix_upgrade_response` to false.
+* router: allow retries of streaming or incomplete requests. This removes stat `rq_retry_skipped_request_not_complete`.
 * tracing: tracing configuration has been made fully dynamic and every HTTP connection manager
   can now have a separate :ref:`tracing provider <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.Tracing.provider>`.
 

include/envoy/stream_info/stream_info.h

@@ -91,6 +91,10 @@ struct ResponseCodeDetailValues {
   // Envoy is doing non-streaming proxying, and the request payload exceeded
   // configured limits.
   const std::string RequestPayloadTooLarge = "request_payload_too_large";
+  // Envoy is doing streaming proxying, but too much data arrived while waiting
+  // to attempt a retry.
+  const std::string RequestPayloadExceededRetryBufferLimit =
+      "request_payload_exceeded_retry_buffer_limit";
   // Envoy is doing non-streaming proxying, and the response payload exceeded
   // configured limits.
   const std::string ResponsePayloadTooLArge = "response_payload_too_large";

source/common/router/router.cc

@@ -667,11 +667,12 @@ void Filter::sendNoHealthyUpstreamResponse() {
 }
 
 Http::FilterDataStatus Filter::decodeData(Buffer::Instance& data, bool end_stream) {
-  // upstream_requests_.size() cannot be 0 because we add to it unconditionally
-  // in decodeHeaders(). It cannot be > 1 because that only happens when a per
+  // upstream_requests_.size() cannot be > 1 because that only happens when a per
   // try timeout occurs with hedge_on_per_try_timeout enabled but the per
-  // try timeout timer is not started until onUpstreamComplete().
-  ASSERT(upstream_requests_.size() == 1);
+  // try timeout timer is not started until onRequestComplete(). It could be zero
+  // if the first request attempt has already failed and a retry is waiting for
+  // a backoff timer.
+  ASSERT(upstream_requests_.size() <= 1);
 
   bool buffering = (retry_state_ && retry_state_->enabled()) || !active_shadow_policies_.empty();
   if (buffering &&
@@ -681,13 +682,31 @@ Http::FilterDataStatus Filter::decodeData(Buffer::Instance& data, bool end_strea
     retry_state_.reset();
     buffering = false;
     active_shadow_policies_.clear();
+
+    // If we had to abandon buffering and there's no request in progress, abort the request and
+    // clean up. This happens if the initial upstream request failed, and we are currently waiting
+    // for a backoff timer before starting the next upstream attempt.
+    if (upstream_requests_.empty()) {
+      cleanup();
+      callbacks_->sendLocalReply(
+          Http::Code::InsufficientStorage, "exceeded request buffer limit while retrying upstream",
+          modify_headers_, absl::nullopt,
+          StreamInfo::ResponseCodeDetails::get().RequestPayloadExceededRetryBufferLimit);
+      return Http::FilterDataStatus::StopIterationNoBuffer;
+    }
   }
 
+  // If we aren't buffering and there is no active request, an abort should have occurred
+  // already.
+  ASSERT(buffering || !upstream_requests_.empty());
+
   if (buffering) {
     // If we are going to buffer for retries or shadowing, we need to make a copy before encoding
     // since it's all moves from here on.
-    Buffer::OwnedImpl copy(data);
-    upstream_requests_.front()->encodeData(copy, end_stream);
+    if (!upstream_requests_.empty()) {
+      Buffer::OwnedImpl copy(data);
+      upstream_requests_.front()->encodeData(copy, end_stream);
+    }
 
     // If we are potentially going to retry or shadow this request we need to buffer.
     // This will not cause the connection manager to 413 because before we hit the
@@ -709,11 +728,12 @@ Http::FilterDataStatus Filter::decodeData(Buffer::Instance& data, bool end_strea
 Http::FilterTrailersStatus Filter::decodeTrailers(Http::RequestTrailerMap& trailers) {
   ENVOY_STREAM_LOG(debug, "router decoding trailers:\n{}", *callbacks_, trailers);
 
-  // upstream_requests_.size() cannot be 0 because we add to it unconditionally
-  // in decodeHeaders(). It cannot be > 1 because that only happens when a per
+  // upstream_requests_.size() cannot be > 1 because that only happens when a per
   // try timeout occurs with hedge_on_per_try_timeout enabled but the per
-  // try timeout timer is not started until onUpstreamComplete().
-  ASSERT(upstream_requests_.size() == 1);
+  // try timeout timer is not started until onRequestComplete(). It could be zero
+  // if the first request attempt has already failed and a retry is waiting for
+  // a backoff timer.
+  ASSERT(upstream_requests_.size() <= 1);
   downstream_trailers_ = &trailers;
   for (auto& upstream_request : upstream_requests_) {
     upstream_request->encodeTrailers(trailers);
@@ -724,8 +744,10 @@ Http::FilterTrailersStatus Filter::decodeTrailers(Http::RequestTrailerMap& trail
 
 Http::FilterMetadataStatus Filter::decodeMetadata(Http::MetadataMap& metadata_map) {
   Http::MetadataMapPtr metadata_map_ptr = std::make_unique<Http::MetadataMap>(metadata_map);
-  ASSERT(upstream_requests_.size() == 1);
-  upstream_requests_.front()->encodeMetadata(std::move(metadata_map_ptr));
+  if (!upstream_requests_.empty()) {
+    // TODO(soya3129): Save metadata for retry, redirect and shadowing case.
+    upstream_requests_.front()->encodeMetadata(std::move(metadata_map_ptr));
+  }
   return Http::FilterMetadataStatus::Continue;
 }
 
@@ -869,8 +891,9 @@ void Filter::onSoftPerTryTimeout(UpstreamRequest& upstream_request) {
     RetryStatus retry_status =
         retry_state_->shouldHedgeRetryPerTryTimeout([this]() -> void { doRetry(); });
 
-    if (retry_status == RetryStatus::Yes && setupRetry()) {
-      setupRetry();
+    if (retry_status == RetryStatus::Yes) {
+      pending_retries_++;
+
       // Don't increment upstream_host->stats().rq_error_ here, we'll do that
       // later if 1) we hit global timeout or 2) we get bad response headers
       // back.
@@ -996,7 +1019,9 @@ bool Filter::maybeRetryReset(Http::StreamResetReason reset_reason,
 
   const RetryStatus retry_status =
       retry_state_->shouldRetryReset(reset_reason, [this]() -> void { doRetry(); });
-  if (retry_status == RetryStatus::Yes && setupRetry()) {
+  if (retry_status == RetryStatus::Yes) {
+    pending_retries_++;
+
     if (upstream_request.upstreamHost()) {
       upstream_request.upstreamHost()->stats().rq_error_.inc();
     }
@@ -1184,19 +1209,18 @@ void Filter::onUpstreamHeaders(uint64_t response_code, Http::ResponseHeaderMapPt
     } else {
       const RetryStatus retry_status =
           retry_state_->shouldRetryHeaders(*headers, [this]() -> void { doRetry(); });
-      // Capture upstream_host since setupRetry() in the following line will clear
-      // upstream_request.
-      const auto upstream_host = upstream_request.upstreamHost();
-      if (retry_status == RetryStatus::Yes && setupRetry()) {
+      if (retry_status == RetryStatus::Yes) {
+        pending_retries_++;
+        upstream_request.upstreamHost()->stats().rq_error_.inc();
+        Http::CodeStats& code_stats = httpContext().codeStats();
+        code_stats.chargeBasicResponseStat(cluster_->statsScope(), config_.retry_,
+                                           static_cast<Http::Code>(response_code));
+
         if (!end_stream) {
           upstream_request.resetStream();
         }
         upstream_request.removeFromList(upstream_requests_);
 
-        Http::CodeStats& code_stats = httpContext().codeStats();
-        code_stats.chargeBasicResponseStat(cluster_->statsScope(), config_.retry_,
-                                           static_cast<Http::Code>(response_code));
-        upstream_host->stats().rq_error_.inc();
         return;
       } else if (retry_status == RetryStatus::NoOverflow) {
         callbacks_->streamInfo().setResponseFlag(StreamInfo::ResponseFlag::UpstreamOverflow);
@@ -1375,23 +1399,6 @@ void Filter::onUpstreamComplete(UpstreamRequest& upstream_request) {
   cleanup();
 }
 
-bool Filter::setupRetry() {
-  // If we responded before the request was complete we don't bother doing a retry. This may not
-  // catch certain cases where we are in full streaming mode and we have a connect timeout or an
-  // overflow of some kind. However, in many cases deployments will use the buffer filter before
-  // this filter which will make this a non-issue. The implementation of supporting retry in cases
-  // where the request is not complete is more complicated so we will start with this for now.
-  if (!downstream_end_stream_) {
-    config_.stats_.rq_retry_skipped_request_not_complete_.inc();
-    return false;
-  }
-  pending_retries_++;
-
-  ENVOY_STREAM_LOG(debug, "performing retry", *callbacks_);
-
-  return true;
-}
-
 bool Filter::setupRedirect(const Http::ResponseHeaderMap& headers,
                            UpstreamRequest& upstream_request) {
   ENVOY_STREAM_LOG(debug, "attempting internal redirect", *callbacks_);
@@ -1412,7 +1419,7 @@ bool Filter::setupRedirect(const Http::ResponseHeaderMap& headers,
 
   const StreamInfo::FilterStateSharedPtr& filter_state = callbacks_->streamInfo().filterState();
 
-  // As with setupRetry, redirects are not supported for streaming requests yet.
+  // Redirects are not supported for streaming requests yet.
   if (downstream_end_stream_ &&
       !callbacks_->decodingBuffer() && // Redirects with body not yet supported.
       location != nullptr &&
@@ -1432,6 +1439,8 @@ bool Filter::setupRedirect(const Http::ResponseHeaderMap& headers,
 }
 
 void Filter::doRetry() {
+  ENVOY_STREAM_LOG(debug, "performing retry", *callbacks_);
+
   is_retry_ = true;
   attempt_count_++;
   ASSERT(pending_retries_ > 0);
@@ -1454,18 +1463,18 @@ void Filter::doRetry() {
     downstream_headers_->setEnvoyAttemptCount(attempt_count_);
   }
 
-  ASSERT(response_timeout_ || timeout_.global_timeout_.count() == 0);
   UpstreamRequest* upstream_request_tmp = upstream_request.get();
   upstream_request->moveIntoList(std::move(upstream_request), upstream_requests_);
-  upstream_requests_.front()->encodeHeaders(!callbacks_->decodingBuffer() && !downstream_trailers_);
+  upstream_requests_.front()->encodeHeaders(!callbacks_->decodingBuffer() &&
+                                            !downstream_trailers_ && downstream_end_stream_);
   // It's possible we got immediately reset which means the upstream request we just
   // added to the front of the list might have been removed, so we need to check to make
   // sure we don't encodeData on the wrong request.
   if (!upstream_requests_.empty() && (upstream_requests_.front().get() == upstream_request_tmp)) {
     if (callbacks_->decodingBuffer()) {
       // If we are doing a retry we need to make a copy.
       Buffer::OwnedImpl copy(*callbacks_->decodingBuffer());
-      upstream_requests_.front()->encodeData(copy, !downstream_trailers_);
+      upstream_requests_.front()->encodeData(copy, !downstream_trailers_ && downstream_end_stream_);
     }
 
     if (downstream_trailers_) {

source/common/router/router.h

@@ -46,8 +46,7 @@ namespace Router {
   COUNTER(rq_redirect)                                                                             \
   COUNTER(rq_direct_response)                                                                      \
   COUNTER(rq_total)                                                                                \
-  COUNTER(rq_reset_after_downstream_response_started)                                              \
-  COUNTER(rq_retry_skipped_request_not_complete)
+  COUNTER(rq_reset_after_downstream_response_started)
 // clang-format on
 
 /**
@@ -492,8 +491,6 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   // for the remaining upstream requests to return.
   void resetOtherUpstreams(UpstreamRequest& upstream_request);
   void sendNoHealthyUpstreamResponse();
-  // TODO(soya3129): Save metadata for retry, redirect and shadowing case.
-  bool setupRetry();
   bool setupRedirect(const Http::ResponseHeaderMap& headers, UpstreamRequest& upstream_request);
   void updateOutlierDetection(Upstream::Outlier::Result result, UpstreamRequest& upstream_request,
                               absl::optional<uint64_t> code);