envoyproxy · mattklein123 · May 4, 2020 · Mar 25, 2020 · Mar 26, 2020 · Mar 30, 2020
diff --git a/api/envoy/api/v2/core/protocol.proto b/api/envoy/api/v2/core/protocol.proto
@@ -85,8 +85,6 @@ message HttpProtocolOptions {
 
   // Total duration to keep alive an HTTP request/response stream. If the time limit is reached the stream will be
   // reset independent of any other timeouts. If not specified, this value is not set.
-  // The current implementation implements this timeout on downstream connections only.
-  // [#comment:TODO(shikugawa): add this functionality to upstream.]
   google.protobuf.Duration max_stream_duration = 4;
 
   // Action to take when a client request with a header name containing underscore characters is received.

diff --git a/api/envoy/config/core/v3/protocol.proto b/api/envoy/config/core/v3/protocol.proto
@@ -92,8 +92,6 @@ message HttpProtocolOptions {
 
   // Total duration to keep alive an HTTP request/response stream. If the time limit is reached the stream will be
   // reset independent of any other timeouts. If not specified, this value is not set.
-  // The current implementation implements this timeout on downstream connections only.
-  // [#comment:TODO(shikugawa): add this functionality to upstream.]
   google.protobuf.Duration max_stream_duration = 4;
 
   // Action to take when a client request with a header name containing underscore characters is received.

diff --git a/clang-tidy-fixes.yaml b/clang-tidy-fixes.yaml
diff --git a/docs/root/configuration/upstream/cluster_manager/cluster_stats.rst b/docs/root/configuration/upstream/cluster_manager/cluster_stats.rst
@@ -69,6 +69,7 @@ Every cluster has a statistics tree rooted at *cluster.<name>.* with the followi
   upstream_rq_cancelled, Counter, Total requests cancelled before obtaining a connection pool connection
   upstream_rq_maintenance_mode, Counter, Total requests that resulted in an immediate 503 due to :ref:`maintenance mode<config_http_filters_router_runtime_maintenance_mode>`
   upstream_rq_timeout, Counter, Total requests that timed out waiting for a response
+  upstream_rq_max_duration_reached, Counter, Total requests closed due to max duration reached
   upstream_rq_per_try_timeout, Counter, Total requests that hit the per try timeout
   upstream_rq_rx_reset, Counter, Total requests that were reset remotely
   upstream_rq_tx_reset, Counter, Total requests that were reset locally

diff --git a/generated_api_shadow/envoy/api/v2/core/protocol.proto b/generated_api_shadow/envoy/api/v2/core/protocol.proto
diff --git a/generated_api_shadow/envoy/config/core/v3/protocol.proto b/generated_api_shadow/envoy/config/core/v3/protocol.proto
diff --git a/include/envoy/upstream/upstream.h b/include/envoy/upstream/upstream.h
@@ -570,6 +570,7 @@ class PrioritySet {
   COUNTER(upstream_rq_cancelled)                                                                   \
   COUNTER(upstream_rq_completed)                                                                   \
   COUNTER(upstream_rq_maintenance_mode)                                                            \
+  COUNTER(upstream_rq_max_duration_reached)                                                        \
   COUNTER(upstream_rq_pending_failure_eject)                                                       \
   COUNTER(upstream_rq_pending_overflow)                                                            \
   COUNTER(upstream_rq_pending_total)                                                               \
@@ -727,6 +728,12 @@ class ClusterInfo {
    */
   virtual const envoy::config::core::v3::Http2ProtocolOptions& http2Options() const PURE;
 
+  /**
+   * @return const envoy::config::core::v3::HttpProtocolOptions for all of HTTP versions.
+   */
+  virtual const envoy::config::core::v3::HttpProtocolOptions&
+  commonHttpProtocolOptions() const PURE;
+
   /**
    * @param name std::string containing the well-known name of the extension for which protocol
    *        options are desired

diff --git a/source/common/router/upstream_request.cc b/source/common/router/upstream_request.cc
@@ -78,6 +78,9 @@ UpstreamRequest::~UpstreamRequest() {
     // Allows for testing.
     per_try_timeout_->disableTimer();
   }
+  if (max_stream_duration_timer_ != nullptr) {
+    max_stream_duration_timer_->disableTimer();
+  }
   clearRequestEncoder();
 
   // If desired, fire the per-try histogram when the UpstreamRequest
@@ -360,7 +363,18 @@ void UpstreamRequest::onPoolReady(
   // If end_stream is set in headers, and there are metadata to send, delays end_stream. The case
   // only happens when decoding headers filters return ContinueAndEndStream.
   const bool delay_headers_end_stream = end_stream && !downstream_metadata_map_vector_.empty();
+
+  if (upstream_host_->cluster().commonHttpProtocolOptions().has_max_stream_duration()) {
+    const auto max_stream_duration = std::chrono::milliseconds(DurationUtil::durationToMilliseconds(
+        upstream_host_->cluster().commonHttpProtocolOptions().max_stream_duration()));
+    if (max_stream_duration.count()) {
+      max_stream_duration_timer_ = parent_.callbacks()->dispatcher().createTimer(
+          [this]() -> void { onStreamMaxDurationReached(); });
+      max_stream_duration_timer_->enableTimer(max_stream_duration);
+    }
+  }
   upstream_->encodeHeaders(*parent_.downstreamHeaders(), end_stream && !delay_headers_end_stream);
+
   calling_encode_headers_ = false;
 
   // It is possible to get reset in the middle of an encodeHeaders() call. This happens for
@@ -398,6 +412,15 @@ void UpstreamRequest::onPoolReady(
   }
 }
 
+void UpstreamRequest::onStreamMaxDurationReached() {
+  ASSERT(upstream_host_);
+  upstream_host_->cluster().stats().upstream_rq_max_duration_reached_.inc();
+  max_stream_duration_timer_->disableTimer();
+  max_stream_duration_timer_.reset();
+  resetStream();
+  parent_.callbacks()->resetStream();
+}
+
 void UpstreamRequest::clearRequestEncoder() {
   // Before clearing the encoder, unsubscribe from callbacks.
   if (upstream_) {

diff --git a/source/common/router/upstream_request.h b/source/common/router/upstream_request.h
@@ -107,6 +107,7 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
   UpstreamRequest* upstreamRequest() override { return this; }
 
   void clearRequestEncoder();
+  void onStreamMaxDurationReached();
 
   struct DownstreamWatermarkManager : public Http::DownstreamWatermarkCallbacks {
     DownstreamWatermarkManager(UpstreamRequest& parent) : parent_(parent) {}
@@ -174,6 +175,8 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
   // Sentinel to indicate if timeout budget tracking is configured for the cluster,
   // and if so, if the per-try histogram should record a value.
   bool record_timeout_budget_ : 1;
+
+  Event::TimerPtr max_stream_duration_timer_;
 };
 
 class HttpConnPool : public GenericConnPool, public Http::ConnectionPool::Callbacks {

diff --git a/source/common/upstream/upstream_impl.cc b/source/common/upstream/upstream_impl.cc
@@ -679,6 +679,7 @@ ClusterInfoImpl::ClusterInfoImpl(
       features_(parseFeatures(config)),
       http1_settings_(Http::Utility::parseHttp1Settings(config.http_protocol_options())),
       http2_options_(Http2::Utility::initializeAndValidateOptions(config.http2_protocol_options())),
+      common_http_protocol_options_(config.common_http_protocol_options()),
       extension_protocol_options_(parseExtensionProtocolOptions(config, validation_visitor)),
       resource_managers_(config, runtime, name_, *stats_scope_),
       maintenance_mode_runtime_key_(absl::StrCat("upstream.maintenance_mode.", name_)),

diff --git a/source/common/upstream/upstream_impl.h b/source/common/upstream/upstream_impl.h
@@ -540,6 +540,9 @@ class ClusterInfoImpl : public ClusterInfo, protected Logger::Loggable<Logger::I
   const envoy::config::core::v3::Http2ProtocolOptions& http2Options() const override {
     return http2_options_;
   }
+  const envoy::config::core::v3::HttpProtocolOptions& commonHttpProtocolOptions() const override {
+    return common_http_protocol_options_;
+  }
   ProtocolOptionsConfigConstSharedPtr
   extensionProtocolOptions(const std::string& name) const override;
   LoadBalancerType lbType() const override { return lb_type_; }
@@ -627,6 +630,7 @@ class ClusterInfoImpl : public ClusterInfo, protected Logger::Loggable<Logger::I
   const uint64_t features_;
   const Http::Http1Settings http1_settings_;
   const envoy::config::core::v3::Http2ProtocolOptions http2_options_;
+  const envoy::config::core::v3::HttpProtocolOptions common_http_protocol_options_;
   const std::map<std::string, ProtocolOptionsConfigConstSharedPtr> extension_protocol_options_;
   mutable ResourceManagers resource_managers_;
   const std::string maintenance_mode_runtime_key_;

diff --git a/test/common/router/router_test.cc b/test/common/router/router_test.cc
@@ -125,6 +125,12 @@ class RouterTestBase : public testing::Test {
     EXPECT_CALL(*per_try_timeout_, disableTimer());
   }
 
+  void expectMaxStreamDurationTimer() {
+    max_stream_duration_timer_ = new Event::MockTimer(&callbacks_.dispatcher_);
+    EXPECT_CALL(*max_stream_duration_timer_, enableTimer(_, _));
+    EXPECT_CALL(*max_stream_duration_timer_, disableTimer());
+  }
+
   AssertionResult verifyHostUpstreamStats(uint64_t success, uint64_t error) {
     if (success != cm_.conn_pool_.host_->stats_.rq_success_.value()) {
       return AssertionFailure() << fmt::format("rq_success {} does not match expected {}",
@@ -348,6 +354,7 @@ class RouterTestBase : public testing::Test {
   RouterTestFilter router_;
   Event::MockTimer* response_timeout_{};
   Event::MockTimer* per_try_timeout_{};
+  Event::MockTimer* max_stream_duration_timer_{};
   Network::Address::InstanceConstSharedPtr host_address_{
       Network::Utility::resolveUrl("tcp://10.0.0.5:9211")};
   NiceMock<Http::MockRequestEncoder> original_encoder_;
@@ -3551,6 +3558,92 @@ TEST_F(RouterTest, RetryTimeoutDuringRetryDelay) {
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
 }
 
+TEST_F(RouterTest, MaxStreamDurationValidlyConfigured) {
+  NiceMock<Http::MockRequestEncoder> encoder1;
+  Http::ResponseDecoder* response_decoder = nullptr;
+  ON_CALL(cm_.conn_pool_.host_->cluster_, commonHttpProtocolOptions()).WillByDefault(Invoke([&]() {
+    auto common_http_protocol_options = envoy::config::core::v3::HttpProtocolOptions();
+    common_http_protocol_options.mutable_max_stream_duration()->MergeFrom(
+        ProtobufUtil::TimeUtil::MillisecondsToDuration(500));
+    return common_http_protocol_options;
+  }));
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke(
+          [&](Http::ResponseDecoder& decoder,
+              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+            response_decoder = &decoder;
+            callbacks.onPoolReady(encoder1, cm_.conn_pool_.host_, upstream_stream_info_);
+            return nullptr;
+          }));
+  expectMaxStreamDurationTimer();
+
+  Http::TestRequestHeaderMapImpl headers;
+  HttpTestUtility::addDefaultHeaders(headers);
+  router_.decodeHeaders(headers, false);
+  max_stream_duration_timer_->invokeCallback();
+
+  router_.onDestroy();
+  EXPECT_TRUE(verifyHostUpstreamStats(0, 0));
+}
+
+TEST_F(RouterTest, MaxStreamDurationDisabledIfSetToZero) {
+  NiceMock<Http::MockRequestEncoder> encoder1;
+  Http::ResponseDecoder* response_decoder = nullptr;
+
+  ON_CALL(cm_.conn_pool_.host_->cluster_, commonHttpProtocolOptions()).WillByDefault(Invoke([&]() {
+    auto common_http_protocol_options = envoy::config::core::v3::HttpProtocolOptions();
+    common_http_protocol_options.mutable_max_stream_duration()->MergeFrom(
+        ProtobufUtil::TimeUtil::MillisecondsToDuration(0));
+    return common_http_protocol_options;
+  }));
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke(
+          [&](Http::ResponseDecoder& decoder,
+              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+            response_decoder = &decoder;
+            callbacks.onPoolReady(encoder1, cm_.conn_pool_.host_, upstream_stream_info_);
+            return nullptr;
+          }));
+
+  // not to be called timer creation.
+  EXPECT_CALL(callbacks_.dispatcher_, createTimer_).Times(0);
+
+  Http::TestRequestHeaderMapImpl headers;
+  HttpTestUtility::addDefaultHeaders(headers);
+  router_.decodeHeaders(headers, false);
+
+  router_.onDestroy();
+  EXPECT_TRUE(verifyHostUpstreamStats(0, 0));
+}
+
+TEST_F(RouterTest, MaxStreamDurationCallbackNotCalled) {
+  NiceMock<Http::MockRequestEncoder> encoder1;
+  Http::ResponseDecoder* response_decoder = nullptr;
+  ON_CALL(cm_.conn_pool_.host_->cluster_, commonHttpProtocolOptions()).WillByDefault(Invoke([&]() {
+    auto common_http_protocol_options = envoy::config::core::v3::HttpProtocolOptions();
+    common_http_protocol_options.mutable_max_stream_duration()->MergeFrom(
+        ProtobufUtil::TimeUtil::MillisecondsToDuration(5000));
+    return common_http_protocol_options;
+  }));
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke(
+          [&](Http::ResponseDecoder& decoder,
+              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+            response_decoder = &decoder;
+            callbacks.onPoolReady(encoder1, cm_.conn_pool_.host_, upstream_stream_info_);
+            return nullptr;
+          }));
+
+  expectMaxStreamDurationTimer();
+
+  Http::TestRequestHeaderMapImpl headers;
+  HttpTestUtility::addDefaultHeaders(headers);
+  router_.decodeHeaders(headers, false);
+
+  router_.onDestroy();
+  EXPECT_TRUE(verifyHostUpstreamStats(0, 0));
+}
+
 TEST_F(RouterTest, RetryTimeoutDuringRetryDelayWithUpstreamRequestNoHost) {
   NiceMock<Http::MockRequestEncoder> encoder1;
   Http::ResponseDecoder* response_decoder = nullptr;

diff --git a/test/common/tcp_proxy/upstream_test.cc b/test/common/tcp_proxy/upstream_test.cc
@@ -26,6 +26,7 @@ class HttpUpstreamTest : public testing::Test {
   Http::MockRequestEncoder encoder_;
   NiceMock<Tcp::ConnectionPool::MockUpstreamCallbacks> callbacks_;
   std::unique_ptr<HttpUpstream> upstream_;
+  testing::NiceMock<Event::MockDispatcher> dispatcher_;
   std::string hostname_{"default.host.com"};
 };
 

diff --git a/test/integration/http_integration.cc b/test/integration/http_integration.cc
@@ -1209,6 +1209,36 @@ void HttpIntegrationTest::testAdminDrain(Http::CodecClient::Type admin_request_t
       nullptr, true));
 }
 
+void HttpIntegrationTest::testMaxStreamDuration() {
+  config_helper_.addConfigModifier([](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+    auto* static_resources = bootstrap.mutable_static_resources();
+    auto* cluster = static_resources->mutable_clusters(0);
+    auto* http_protocol_options = cluster->mutable_common_http_protocol_options();
+    http_protocol_options->mutable_max_stream_duration()->MergeFrom(
+        ProtobufUtil::TimeUtil::MillisecondsToDuration(200));
+  });
+
+  initialize();
+  fake_upstreams_[0]->set_allow_unexpected_disconnects(true);
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  auto encoder_decoder = codec_client_->startRequest(default_request_headers_);
+  request_encoder_ = &encoder_decoder.first;
+  auto response = std::move(encoder_decoder.second);
+
+  ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_));
+  ASSERT_TRUE(fake_upstream_connection_->waitForNewStream(*dispatcher_, upstream_request_));
+
+  test_server_->waitForCounterGe("cluster.cluster_0.upstream_rq_max_duration_reached", 1);
+
+  if (downstream_protocol_ == Http::CodecClient::Type::HTTP1) {
+    codec_client_->waitForDisconnect();
+  } else {
+    response->waitForReset();
+    codec_client_->close();
+  }
+}
+
 std::string HttpIntegrationTest::listenerStatPrefix(const std::string& stat_name) {
   if (version_ == Network::Address::IpVersion::v4) {
     return "listener.127.0.0.1_0." + stat_name;

diff --git a/test/integration/http_integration.h b/test/integration/http_integration.h
@@ -220,7 +220,8 @@ class HttpIntegrationTest : public BaseIntegrationTest {
                     bool response_trailers_present);
   // Test /drain_listener from admin portal.
   void testAdminDrain(Http::CodecClient::Type admin_request_type);
-
+  // test max stream duration
+  void testMaxStreamDuration();
   Http::CodecClient::Type downstreamProtocol() const { return downstream_protocol_; }
   // Prefix listener stat with IP:port, including IP version dependent loopback address.
   std::string listenerStatPrefix(const std::string& stat_name);

diff --git a/test/integration/protocol_integration_test.cc b/test/integration/protocol_integration_test.cc
@@ -635,6 +635,8 @@ TEST_P(ProtocolIntegrationTest, TwoRequests) { testTwoRequests(); }
 
 TEST_P(ProtocolIntegrationTest, TwoRequestsWithForcedBackup) { testTwoRequests(true); }
 
+TEST_P(ProtocolIntegrationTest, BasicMaxStreamDuration) { testMaxStreamDuration(); }
+
 // Verify that headers with underscores in their names are dropped from client requests
 // but remain in upstream responses.
 TEST_P(ProtocolIntegrationTest, HeadersWithUnderscoresDropped) {

diff --git a/test/integration/stats_integration_test.cc b/test/integration/stats_integration_test.cc
@@ -270,6 +270,7 @@ TEST_P(ClusterMemoryTestRunner, MemoryLargeClusterSizeWithFakeSymbolTable) {
   // 2020/03/16  9964     44085       44600   http2: support custom SETTINGS parameters.
   // 2020/03/24  10501    44261       44600   upstream: upstream_rq_retry_limit_exceeded.
   // 2020/04/02  10624    43356       44000   Use 100 clusters rather than 1000 to avoid timeouts
+  // 2020/04/07  10531    44437       44600   http: max stream duration upstream support.
 
   // Note: when adjusting this value: EXPECT_MEMORY_EQ is active only in CI
   // 'release' builds, where we control the platform and tool-chain. So you
@@ -283,8 +284,8 @@ TEST_P(ClusterMemoryTestRunner, MemoryLargeClusterSizeWithFakeSymbolTable) {
   // If you encounter a failure here, please see
   // https://github.com/envoyproxy/envoy/blob/master/source/docs/stats.md#stats-memory-tests
   // for details on how to fix.
-  EXPECT_MEMORY_EQ(m_per_cluster, 43356);
-  EXPECT_MEMORY_LE(m_per_cluster, 44000);
+  EXPECT_MEMORY_EQ(m_per_cluster, 44437);
+  EXPECT_MEMORY_LE(m_per_cluster, 44600);
 }
 
 TEST_P(ClusterMemoryTestRunner, MemoryLargeClusterSizeWithRealSymbolTable) {
@@ -327,6 +328,7 @@ TEST_P(ClusterMemoryTestRunner, MemoryLargeClusterSizeWithRealSymbolTable) {
   // 2020/03/16  9964     36220       36800   http2: support custom SETTINGS parameters.
   // 2020/03/24  10501    36300       36800   upstream: upstream_rq_retry_limit_exceeded.
   // 2020/04/02  10624    35564       36000   Use 100 clusters rather than 1000 to avoid timeouts
+  // 2020/04/07  10531    36380       36800   http: max stream duration upstream support.
 
   // Note: when adjusting this value: EXPECT_MEMORY_EQ is active only in CI
   // 'release' builds, where we control the platform and tool-chain. So you
@@ -340,8 +342,8 @@ TEST_P(ClusterMemoryTestRunner, MemoryLargeClusterSizeWithRealSymbolTable) {
   // If you encounter a failure here, please see
   // https://github.com/envoyproxy/envoy/blob/master/source/docs/stats.md#stats-memory-tests
   // for details on how to fix.
-  EXPECT_MEMORY_EQ(m_per_cluster, 35564);
-  EXPECT_MEMORY_LE(m_per_cluster, 36000);
+  EXPECT_MEMORY_EQ(m_per_cluster, 36380);
+  EXPECT_MEMORY_LE(m_per_cluster, 36800);
 }
 
 TEST_P(ClusterMemoryTestRunner, MemoryLargeHostSizeWithStats) {

diff --git a/test/mocks/upstream/cluster_info.cc b/test/mocks/upstream/cluster_info.cc
@@ -55,6 +55,8 @@ MockClusterInfo::MockClusterInfo()
   ON_CALL(*this, eds_service_name()).WillByDefault(ReturnPointee(&eds_service_name_));
   ON_CALL(*this, http1Settings()).WillByDefault(ReturnRef(http1_settings_));
   ON_CALL(*this, http2Options()).WillByDefault(ReturnRef(http2_options_));
+  ON_CALL(*this, commonHttpProtocolOptions())
+      .WillByDefault(ReturnRef(common_http_protocol_options_));
   ON_CALL(*this, extensionProtocolOptions(_)).WillByDefault(Return(extension_protocol_options_));
   ON_CALL(*this, maxResponseHeadersCount())
       .WillByDefault(ReturnPointee(&max_response_headers_count_));