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/intro/version_history.rst

@@ -3,6 +3,7 @@ Version history
 
 1.15.0 (Pending)
 ================
+* router: allow retries of streaming or incomplete requests
 
 1.14.1 (April 8, 2020)
 ======================

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
+    // cleanup.
+    if (upstream_requests_.empty()) {
+      cleanup();
+      callbacks_->sendLocalReply(
+          Http::Code::ServiceUnavailable, "exceeded request buffer limit while retrying upstream",
+          modify_headers_, absl::nullopt,
+          StreamInfo::ResponseCodeDetails::get()
+              .RequestPayloadTooLarge /* TODO: better code than RequestPayloadTooLarge */);
+      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,11 @@ 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: buffer this if there's no request? It doesn't look like the upstream request has
+    // any handling for metadata in retries already, so I don't think this is a regression.
+    upstream_requests_.front()->encodeMetadata(std::move(metadata_map_ptr));
+  }
   return Http::FilterMetadataStatus::Continue;
 }
 
@@ -869,7 +892,7 @@ void Filter::onSoftPerTryTimeout(UpstreamRequest& upstream_request) {
     RetryStatus retry_status =
         retry_state_->shouldHedgeRetryPerTryTimeout([this]() -> void { doRetry(); });
 
-    if (retry_status == RetryStatus::Yes && setupRetry()) {
+    if (retry_status == RetryStatus::Yes) {
       setupRetry();
       // 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
@@ -996,7 +1019,8 @@ 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) {
+    setupRetry();
     if (upstream_request.upstreamHost()) {
       upstream_request.upstreamHost()->stats().rq_error_.inc();
     }
@@ -1187,7 +1211,8 @@ void Filter::onUpstreamHeaders(uint64_t response_code, Http::ResponseHeaderMapPt
       // 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) {
+        setupRetry();
         if (!end_stream) {
           upstream_request.resetStream();
         }
@@ -1375,21 +1400,10 @@ 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;
-  }
+void Filter::setupRetry() {
   pending_retries_++;
 
   ENVOY_STREAM_LOG(debug, "performing retry", *callbacks_);
-
-  return true;
 }
 
 bool Filter::setupRedirect(const Http::ResponseHeaderMap& headers,
@@ -1412,7 +1426,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 &&
@@ -1454,18 +1468,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
 
 /**
@@ -493,7 +492,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   void resetOtherUpstreams(UpstreamRequest& upstream_request);
   void sendNoHealthyUpstreamResponse();
   // TODO(soya3129): Save metadata for retry, redirect and shadowing case.
-  bool setupRetry();
+  void setupRetry();
   bool setupRedirect(const Http::ResponseHeaderMap& headers, UpstreamRequest& upstream_request);
   void updateOutlierDetection(Upstream::Outlier::Result result, UpstreamRequest& upstream_request,
                               absl::optional<uint64_t> code);

test/common/router/router_test.cc

@@ -2752,7 +2752,7 @@ TEST_F(RouterTest, BadHeadersDroppedIfPreviousRetryScheduled) {
   response_decoder2->decodeHeaders(std::move(response_headers2), true);
 }
 
-TEST_F(RouterTest, RetryRequestNotComplete) {
+/*TEST_F(RouterTest, RetryRequestNotComplete) {
   NiceMock<Http::MockRequestEncoder> encoder1;
   Http::ResponseDecoder* response_decoder = nullptr;
   EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
@@ -2780,7 +2780,7 @@ TEST_F(RouterTest, RetryRequestNotComplete) {
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
   EXPECT_EQ(1UL, stats_store_.counter("test.rq_retry_skipped_request_not_complete").value());
-}
+  }*/
 
 // Two requests are sent (slow request + hedged retry) and then global timeout
 // is hit. Verify everything gets cleaned up.

test/integration/protocol_integration_test.cc

@@ -299,6 +299,117 @@ TEST_P(ProtocolIntegrationTest, Retry) {
   EXPECT_EQ(512U, response->body().size());
 }
 
+TEST_P(ProtocolIntegrationTest, RetryStreaming) {
+  initialize();
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto encoder_decoder =
+      codec_client_->startRequest(Http::TestRequestHeaderMapImpl{{":method", "POST"},
+                                                                 {":path", "/test/long/url"},
+                                                                 {":scheme", "http"},
+                                                                 {":authority", "host"},
+                                                                 {"x-forwarded-for", "10.0.0.1"},
+                                                                 {"x-envoy-retry-on", "5xx"}});
+  auto& encoder = encoder_decoder.first;
+  auto& response = encoder_decoder.second;
+
+  // Send some data, but not the entire body.
+  std::string data(1024, 'a');
+  Buffer::OwnedImpl send1(data);
+  encoder.encodeData(send1, false);
+
+  ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_));
+  ASSERT_TRUE(fake_upstream_connection_->waitForNewStream(*dispatcher_, upstream_request_));
+
+  // Send back an upstream failure.
+  upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "503"}}, false);
+
+  if (fake_upstreams_[0]->httpType() == FakeHttpConnection::Type::HTTP1) {
+    ASSERT_TRUE(fake_upstream_connection_->waitForDisconnect());
+    ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_));
+  } else {
+    ASSERT_TRUE(upstream_request_->waitForReset());
+  }
+
+  // Wait for a retry. Ensure all data, both before and after the retry, is received.
+  ASSERT_TRUE(fake_upstream_connection_->waitForNewStream(*dispatcher_, upstream_request_));
+
+  // Finish the request.
+  std::string data2(512, 'b');
+  Buffer::OwnedImpl send2(data2);
+  encoder.encodeData(send2, true);
+  std::string combined_request_data = data + data2;
+  ASSERT_TRUE(upstream_request_->waitForData(*dispatcher_, combined_request_data));
+
+  upstream_request_->encodeHeaders(default_response_headers_, false);
+  upstream_request_->encodeData(512, true);
+
+  response->waitForEndStream();
+  EXPECT_TRUE(upstream_request_->complete());
+  EXPECT_EQ(combined_request_data.size(), upstream_request_->bodyLength());
+
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+  EXPECT_EQ(512U, response->body().size());
+}
+
+TEST_P(ProtocolIntegrationTest, RetryStreamingCancelDueToBufferOverflow) {
+  config_helper_.addConfigModifier(
+      [](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
+             hcm) {
+        auto* route = hcm.mutable_route_config()->mutable_virtual_hosts(0)->mutable_routes(0);
+
+        route->mutable_per_request_buffer_limit_bytes()->set_value(1024);
+        route->mutable_route()
+            ->mutable_retry_policy()
+            ->mutable_retry_back_off()
+            ->mutable_base_interval()
+            ->MergeFrom(
+                ProtobufUtil::TimeUtil::MillisecondsToDuration(100000000)); // Effectively infinity.
+      });
+  initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto encoder_decoder =
+      codec_client_->startRequest(Http::TestRequestHeaderMapImpl{{":method", "POST"},
+                                                                 {":path", "/test/long/url"},
+                                                                 {":scheme", "http"},
+                                                                 {":authority", "host"},
+                                                                 {"x-forwarded-for", "10.0.0.1"},
+                                                                 {"x-envoy-retry-on", "5xx"}});
+  auto& encoder = encoder_decoder.first;
+  auto& response = encoder_decoder.second;
+
+  // Send some data, but less than the buffer limit, and not end-stream
+  std::string data(64, 'a');
+  Buffer::OwnedImpl send1(data);
+  encoder.encodeData(send1, false);
+
+  ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_));
+  ASSERT_TRUE(fake_upstream_connection_->waitForNewStream(*dispatcher_, upstream_request_));
+
+  // Send back an upstream failure.
+  upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "503"}}, false);
+
+  if (fake_upstreams_[0]->httpType() == FakeHttpConnection::Type::HTTP1) {
+    ASSERT_TRUE(fake_upstream_connection_->waitForDisconnect());
+  } else {
+    ASSERT_TRUE(upstream_request_->waitForReset());
+  }
+
+  // Overflow the request buffer limit. Because the retry base interval is infinity, no
+  // request will be in progress. This will cause the request to be aborted and an error
+  // to be returned to the client.
+  std::string data2(2048, 'b');
+  Buffer::OwnedImpl send2(data2);
+  encoder.encodeData(send2, false);
+
+  response->waitForEndStream();
+
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("503", response->headers().Status()->value().getStringView());
+  test_server_->waitForCounterEq("cluster.cluster_0.retry_or_shadow_abandoned", 1);
+}
+
 // Tests that the x-envoy-attempt-count header is properly set on the upstream request and the
 // downstream response, and updated after the request is retried.
 TEST_P(DownstreamProtocolIntegrationTest, RetryAttemptCountHeader) {