overload_manager: make proactive resource monitors trigger actions (envoyproxy#42018)

Also, publish a resource pressure gauge like normal resource monitors.

Signed-off-by: Greg Greenway <ggreenway@apple.com>

Author: ggreenway

changelogs/current.yaml

@@ -32,6 +32,12 @@ minor_behavior_changes:
     <envoy_v3_api_field_config.route.v3.RouteAction.path_rewrite>` to
     :ref:`RouteAction <envoy_v3_api_msg_config.route.v3.RouteAction>` to support substitution
     formatting for path header rewriting.
+- area: overload_manager
+  change: |
+    Fixed :ref:`downstream connections monitor
+    <envoy_v3_api_msg_extensions.resource_monitors.downstream_connections.v3.DownstreamConnectionsConfig>`
+    to correctly trigger configured actions, and create a ``pressure`` metric like other resource monitors.
+    Previously, this extension would never trigger an action.
 - area: tracing
   change: |
     The :ref:`request header custom tag <envoy_v3_api_field_type.tracing.v3.CustomTag.request_header>`

envoy/server/overload/thread_local_overload_state.h

@@ -28,6 +28,14 @@ class OverloadProactiveResourceNameValues {
       proactive_action_name_to_resource_ = {
           {GlobalDownstreamMaxConnections,
            OverloadProactiveResourceName::GlobalDownstreamMaxConnections}};
+
+  const std::string& resourceToName(OverloadProactiveResourceName resource) const {
+    switch (resource) {
+    case OverloadProactiveResourceName::GlobalDownstreamMaxConnections:
+      return GlobalDownstreamMaxConnections;
+    }
+    PANIC_DUE_TO_CORRUPT_ENUM;
+  }
 };
 
 using OverloadProactiveResources = ConstSingleton<OverloadProactiveResourceNameValues>;

envoy/server/proactive_resource_monitor.h

@@ -50,7 +50,8 @@ class ProactiveResource {
   ProactiveResource(const std::string& name, ProactiveResourceMonitorPtr monitor,
                     Stats::Scope& stats_scope)
       : name_(name), monitor_(std::move(monitor)),
-        failed_updates_counter_(makeCounter(stats_scope, name, "failed_updates")) {}
+        failed_updates_counter_(makeCounter(stats_scope, name, "failed_updates")),
+        pressure_gauge_(makeGauge(stats_scope, name, "pressure")) {}
 
   bool tryAllocateResource(int64_t increment) {
     if (monitor_->tryAllocateResource(increment)) {
@@ -70,6 +71,13 @@ class ProactiveResource {
     }
   }
 
+  double updateResourcePressure() {
+    const double pressure = static_cast<double>(monitor_->currentResourceUsage()) /
+                            static_cast<double>(monitor_->maxResourceUsage());
+    pressure_gauge_.set(pressure * 100);
+    return pressure;
+  }
+
   ProactiveResourceMonitorOptRef getProactiveResourceMonitorForTest() {
     return makeOptRefFromPtr<ProactiveResourceMonitor>(monitor_.get());
   };
@@ -78,12 +86,22 @@ class ProactiveResource {
   const std::string name_;
   ProactiveResourceMonitorPtr monitor_;
   Stats::Counter& failed_updates_counter_;
+  Stats::Gauge& pressure_gauge_;
+
+  Stats::StatNameManagedStorage makeName(Stats::Scope& scope, absl::string_view a,
+                                         absl::string_view b) {
+    return {absl::StrCat("overload.", a, ".", b), scope.symbolTable()};
+  }
 
   Stats::Counter& makeCounter(Stats::Scope& scope, absl::string_view a, absl::string_view b) {
-    Stats::StatNameManagedStorage stat_name(absl::StrCat("overload.", a, ".", b),
-                                            scope.symbolTable());
+    auto stat_name = makeName(scope, a, b);
     return scope.counterFromStatName(stat_name.statName());
   }
+
+  Stats::Gauge& makeGauge(Stats::Scope& scope, absl::string_view a, absl::string_view b) {
+    auto stat_name = makeName(scope, a, b);
+    return scope.gaugeFromStatName(stat_name.statName(), Stats::Gauge::ImportMode::NeverImport);
+  }
 };
 
 } // namespace Server

source/server/overload_manager_impl.cc

@@ -581,12 +581,18 @@ void OverloadManagerImpl::start() {
     // Start a new flush epoch. If all resource updates complete before this callback runs, the last
     // resource update will call flushResourceUpdates to flush the whole batch early.
     ++flush_epoch_;
-    flush_awaiting_updates_ = resources_.size();
+    flush_awaiting_updates_ = resources_.size() + proactive_resources_->size();
 
     for (auto& resource : resources_) {
       resource.second.update(flush_epoch_);
     }
 
+    for (auto& resource : *proactive_resources_) {
+      const double pressure = resource.second.updateResourcePressure();
+      updateResourcePressure(OverloadProactiveResources::get().resourceToName(resource.first),
+                             pressure, flush_epoch_);
+    }
+
     // Record delay.
     auto now = time_source_.monotonicTime();
     std::chrono::milliseconds delay =

test/server/overload_manager_impl_test.cc

@@ -299,9 +299,9 @@ constexpr char proactiveResourceConfig[] = R"YAML(
   actions:
     - name: envoy.overload_actions.shrink_heap
       triggers:
-        - name: envoy.resource_monitors.fake_resource1
+        - name: envoy.resource_monitors.global_downstream_max_connections
           threshold:
-            value: 0.9
+            value: 0.5
 )YAML";
 
 TEST_F(OverloadManagerImplTest, CallbackOnlyFiresWhenStateChanges) {
@@ -930,12 +930,21 @@ TEST_F(OverloadManagerImplTest, ProactiveResourceAllocateAndDeallocateResourceTe
   Stats::Counter& failed_updates =
       stats_.counter("overload.envoy.resource_monitors.global_downstream_max_connections."
                      "failed_updates");
+  Stats::Gauge& pressure =
+      stats_.gauge("overload.envoy.resource_monitors.global_downstream_max_connections.pressure",
+                   Stats::Gauge::ImportMode::NeverImport);
+
   manager->start();
   EXPECT_TRUE(manager->getThreadLocalOverloadState().isResourceMonitorEnabled(
       OverloadProactiveResourceName::GlobalDownstreamMaxConnections));
   bool resource_allocated = manager->getThreadLocalOverloadState().tryAllocateResource(
       Server::OverloadProactiveResourceName::GlobalDownstreamMaxConnections, 1);
   EXPECT_TRUE(resource_allocated);
+
+  EXPECT_EQ(pressure.value(), 0);
+  timer_cb_();
+  EXPECT_EQ(pressure.value(), 33);
+
   auto monitor = manager->getThreadLocalOverloadState().getProactiveResourceMonitorForTest(
       Server::OverloadProactiveResourceName::GlobalDownstreamMaxConnections);
   EXPECT_NE(absl::nullopt, monitor);
@@ -954,6 +963,34 @@ TEST_F(OverloadManagerImplTest, ProactiveResourceAllocateAndDeallocateResourceTe
   manager->stop();
 }
 
+// Test that proactive monitors trigger the configured actions when they reach the threshold.
+TEST_F(OverloadManagerImplTest, ProactiveResourceTriggers) {
+  setDispatcherExpectation();
+
+  auto manager(createOverloadManager(proactiveResourceConfig));
+  bool is_active = false;
+  manager->registerForAction("envoy.overload_actions.shrink_heap", dispatcher_,
+                             [&](OverloadActionState state) { is_active = state.isSaturated(); });
+
+  manager->start();
+
+  // Trigger threshold is 50%, max is 3.
+
+  ASSERT_TRUE(factory5_.monitor_->tryAllocateResource(1));
+  timer_cb_();
+  EXPECT_FALSE(is_active);
+
+  ASSERT_TRUE(factory5_.monitor_->tryAllocateResource(1));
+  timer_cb_();
+  EXPECT_TRUE(is_active);
+
+  ASSERT_TRUE(factory5_.monitor_->tryDeallocateResource(1));
+  timer_cb_();
+  EXPECT_FALSE(is_active);
+
+  manager->stop();
+}
+
 class OverloadManagerSimulatedTimeTest : public OverloadManagerImplTest,
                                          public Envoy::Event::TestUsingSimulatedTime {};