adaptive concurrency: Fix min_concurrency behavior when gradient shrinks

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);
   }
 }