adaptive concurrency: Fix min_concurrency behavior when gradient shri…

…nks (envoyproxy#9947)

The min_concurrency setting dicated the concurrency limit when in a
min_rtt calculation window. However, if the gradient caused the
concurrency limit to shrink, it was possible for the limit to drop below
this configured minimum. This patch fixes this behavior so that the
calculated concurrency limit is always >= the configured minimum.

In addition, a minor change to the stat indicating whether the min_rtt
calculation window is active. When active, the stat was being set to the
concurrency limit; however, according to the documentation this should
be either 0 or 1. This patch also fixes this behavior.

Signed-off-by: Tony Allen <tony@allen.gg>

docs/root/intro/version_history.rst

@@ -5,6 +5,8 @@ Version history
 ================
 * access loggers: access logger extensions use the "envoy.access_loggers" name space. A mapping
   of extension names is available in the :ref:`deprecated <deprecated>` documentation.
+* adaptive concurrency: fixed bug that allowed concurrency limits to drop below the configured
+  minimum.
 * config: use type URL to select an extension whenever the config type URL (or its previous versions) uniquely identify a typed extension, see :ref:`extension configuration <config_overview_extension_configuration>`.
 * http: fixing a bug in HTTP/1.0 responses where Connection: keep-alive was not appended for connections which were kept alive.
 * retry: added a retry predicate that :ref:`rejects hosts based on metadata. <envoy_api_field_route.RetryPolicy.retry_host_predicate>`

source/extensions/filters/http/adaptive_concurrency/concurrency_controller/gradient_controller.cc

@@ -48,8 +48,8 @@ GradientController::GradientController(GradientControllerConfig config,
                                        const std::string& stats_prefix, Stats::Scope& scope,
                                        Runtime::RandomGenerator& random)
     : config_(std::move(config)), dispatcher_(dispatcher), scope_(scope),
-      stats_(generateStats(scope_, stats_prefix)), random_(random), deferred_limit_value_(1),
-      num_rq_outstanding_(0), concurrency_limit_(config_.minConcurrency()),
+      stats_(generateStats(scope_, stats_prefix)), random_(random), num_rq_outstanding_(0),
+      concurrency_limit_(config_.minConcurrency()),
       latency_sample_hist_(hist_fast_alloc(), hist_free) {
   min_rtt_calc_timer_ = dispatcher_.createTimer([this]() -> void { enterMinRTTSamplingWindow(); });
 
@@ -83,7 +83,7 @@ GradientControllerStats GradientController::generateStats(Stats::Scope& scope,
 void GradientController::enterMinRTTSamplingWindow() {
   absl::MutexLock ml(&sample_mutation_mtx_);
 
-  stats_.min_rtt_calculation_active_.set(config_.minConcurrency());
+  stats_.min_rtt_calculation_active_.set(1);
 
   // Set the minRTT flag to indicate we're gathering samples to update the value. This will
   // prevent the sample window from resetting until enough requests are gathered to complete the
@@ -162,9 +162,10 @@ uint32_t GradientController::calculateNewLimit() {
   stats_.burst_queue_size_.set(burst_headroom);
 
   // The final concurrency value factors in the burst headroom and must be clamped to keep the value
-  // in the range [1, configured_max].
+  // in the range [configured_min, configured_max].
   const uint32_t new_limit = limit + burst_headroom;
-  return std::max<uint32_t>(1, std::min<uint32_t>(config_.maxConcurrencyLimit(), new_limit));
+  return std::max<uint32_t>(config_.minConcurrency(),
+                            std::min<uint32_t>(config_.maxConcurrencyLimit(), new_limit));
 }
 
 RequestForwardingAction GradientController::forwardingDecision() {

test/extensions/filters/http/adaptive_concurrency/concurrency_controller/gradient_controller_test.cc

@@ -219,29 +219,32 @@ TEST_F(GradientControllerTest, MinRTTLogicTest) {
     value: 0.0
   interval: 30s
   request_count: 50
-  min_concurrency: 1
+  min_concurrency: 7
 )EOF";
 
   auto controller = makeController(yaml);
   const auto min_rtt = std::chrono::milliseconds(13);
 
-  // The controller should be measuring minRTT upon creation, so the concurrency window is 1 (the
+  // The controller should be measuring minRTT upon creation, so the concurrency window is 7 (the
   // min concurrency).
   EXPECT_EQ(
       1,
       stats_.gauge("test_prefix.min_rtt_calculation_active", Stats::Gauge::ImportMode::Accumulate)
           .value());
-  EXPECT_EQ(controller->concurrencyLimit(), 1);
-  tryForward(controller, true);
+  EXPECT_EQ(controller->concurrencyLimit(), 7);
+  for (int i = 0; i < 7; ++i) {
+    tryForward(controller, true);
+  }
   tryForward(controller, false);
   tryForward(controller, false);
-  controller->recordLatencySample(min_rtt);
+  for (int i = 0; i < 7; ++i) {
+    controller->recordLatencySample(min_rtt);
+  }
 
-  // 49 more requests should cause the minRTT to be done calculating.
-  for (int i = 0; i < 49; ++i) {
-    EXPECT_EQ(controller->concurrencyLimit(), 1);
+  // 43 more requests should cause the minRTT to be done calculating.
+  for (int i = 0; i < 43; ++i) {
+    EXPECT_EQ(controller->concurrencyLimit(), 7);
     tryForward(controller, true);
-    tryForward(controller, false);
     controller->recordLatencySample(min_rtt);
   }
 
@@ -356,10 +359,11 @@ TEST_F(GradientControllerTest, ConcurrencyLimitBehaviorTestBasic) {
   request_count: 5
   buffer:
     value: 10
+  min_concurrency: 7
 )EOF";
 
   auto controller = makeController(yaml);
-  EXPECT_EQ(controller->concurrencyLimit(), 3);
+  EXPECT_EQ(controller->concurrencyLimit(), 7);
 
   // Force a minRTT of 5ms.
   advancePastMinRTTStage(controller, yaml, std::chrono::milliseconds(5));
@@ -370,8 +374,8 @@ TEST_F(GradientControllerTest, ConcurrencyLimitBehaviorTestBasic) {
   // the max gradient.
   time_system_.sleep(std::chrono::milliseconds(101));
   dispatcher_->run(Event::Dispatcher::RunType::Block);
-  EXPECT_GE(controller->concurrencyLimit(), 3);
-  EXPECT_LE(controller->concurrencyLimit() / 3.0, 2.0);
+  EXPECT_GE(controller->concurrencyLimit(), 7);
+  EXPECT_LE(controller->concurrencyLimit() / 7.0, 2.0);
 
   // Make it seem as if the recorded latencies are consistently lower than the measured minRTT.
   // Ensure that it grows.
@@ -398,6 +402,7 @@ TEST_F(GradientControllerTest, ConcurrencyLimitBehaviorTestBasic) {
     time_system_.sleep(std::chrono::milliseconds(101));
     dispatcher_->run(Event::Dispatcher::RunType::Block);
     EXPECT_LT(controller->concurrencyLimit(), last_concurrency);
+    EXPECT_GE(controller->concurrencyLimit(), 7);
   }
 }