[Overload] Active downstream connections resource monitor

CODEOWNERS

@@ -128,6 +128,7 @@ extensions/filters/common/original_src @snowp @klarose
 /*/extensions/resource_monitors/injected_resource @eziskind @htuch
 /*/extensions/resource_monitors/common @eziskind @htuch
 /*/extensions/resource_monitors/fixed_heap @eziskind @htuch
+/*/extensions/resource_monitors/downstream_connections @antoniovicente @mattklein123 @nezdolik
 /*/extensions/retry/priority @snowp @alyssawilk
 /*/extensions/retry/priority/previous_priorities @snowp @alyssawilk
 /*/extensions/retry/host @snowp @alyssawilk

api/BUILD

@@ -223,6 +223,7 @@ proto_library(
         "//envoy/extensions/rate_limit_descriptors/expr/v3:pkg",
         "//envoy/extensions/rbac/matchers/upstream_ip_port/v3:pkg",
         "//envoy/extensions/request_id/uuid/v3:pkg",
+        "//envoy/extensions/resource_monitors/downstream_connections/v3:pkg",
         "//envoy/extensions/resource_monitors/fixed_heap/v3:pkg",
         "//envoy/extensions/resource_monitors/injected_resource/v3:pkg",
         "//envoy/extensions/retry/host/omit_canary_hosts/v3:pkg",

api/envoy/extensions/resource_monitors/downstream_connections/v3/BUILD

@@ -0,0 +1,9 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = ["@com_github_cncf_udpa//udpa/annotations:pkg"],
+)

api/envoy/extensions/resource_monitors/downstream_connections/v3/downstream_connections.proto

@@ -0,0 +1,23 @@
+syntax = "proto3";
+
+package envoy.extensions.resource_monitors.downstream_connections.v3;
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.resource_monitors.downstream_connections.v3";
+option java_outer_classname = "DownstreamConnectionsProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/resource_monitors/downstream_connections/v3;downstream_connectionsv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Downstream connections]
+// [#extension: envoy.resource_monitors.downstream_connections]
+
+// The downstream connections resource monitor tracks the global number of open downstream connections.
+// [#not-implemented-hide:]
+message DownstreamConnectionsConfig {
+  // Maximum threshold for global open active downstream connections, defaults to 0.
+  // If monitor is configured via Overload manager api and has no value set, Envoy will reject all incoming connections.
+  int64 max_active_downstream_connections = 1 [(validate.rules).int64 = {gt: 0}];
+}

api/versioning/BUILD

@@ -160,6 +160,7 @@ proto_library(
         "//envoy/extensions/rate_limit_descriptors/expr/v3:pkg",
         "//envoy/extensions/rbac/matchers/upstream_ip_port/v3:pkg",
         "//envoy/extensions/request_id/uuid/v3:pkg",
+        "//envoy/extensions/resource_monitors/downstream_connections/v3:pkg",
         "//envoy/extensions/resource_monitors/fixed_heap/v3:pkg",
         "//envoy/extensions/resource_monitors/injected_resource/v3:pkg",
         "//envoy/extensions/retry/host/omit_canary_hosts/v3:pkg",

source/extensions/extensions_build_config.bzl

@@ -157,6 +157,7 @@ EXTENSIONS = {
 
     "envoy.resource_monitors.fixed_heap":               "//source/extensions/resource_monitors/fixed_heap:config",
     "envoy.resource_monitors.injected_resource":        "//source/extensions/resource_monitors/injected_resource:config",
+    "envoy.resource_monitors.downstream_connections":   "//source/extensions/resource_monitors/downstream_connections:config",
 
     #
     # Stat sinks

source/extensions/extensions_metadata.yaml

@@ -519,6 +519,11 @@ envoy.request_id.uuid:
   - envoy.request_id
   security_posture: robust_to_untrusted_downstream_and_upstream
   status: stable
+envoy.resource_monitors.downstream_connections:
+  categories:
+  - envoy.resource_monitors
+  security_posture: data_plane_agnostic
+  status: alpha
 envoy.resource_monitors.fixed_heap:
   categories:
   - envoy.resource_monitors

source/extensions/resource_monitors/common/factory_base.h

@@ -37,6 +37,35 @@ class FactoryBase : public Server::Configuration::ResourceMonitorFactory {
   const std::string name_;
 };
 
+template <class ConfigProto>
+class ProactiveFactoryBase : public Server::Configuration::ProactiveResourceMonitorFactory {
+public:
+  Server::ProactiveResourceMonitorPtr createProactiveResourceMonitor(
+      const Protobuf::Message& config,
+      Server::Configuration::ResourceMonitorFactoryContext& context) override {
+    return createProactiveResourceMonitorFromProtoTyped(
+        MessageUtil::downcastAndValidate<const ConfigProto&>(config,
+                                                             context.messageValidationVisitor()),
+        context);
+  }
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return std::make_unique<ConfigProto>();
+  }
+
+  std::string name() const override { return name_; }
+
+protected:
+  ProactiveFactoryBase(const std::string& name) : name_(name) {}
+
+private:
+  virtual Server::ProactiveResourceMonitorPtr createProactiveResourceMonitorFromProtoTyped(
+      const ConfigProto& config,
+      Server::Configuration::ResourceMonitorFactoryContext& context) PURE;
+
+  const std::string name_;
+};
+
 } // namespace Common
 } // namespace ResourceMonitors
 } // namespace Extensions

source/extensions/resource_monitors/downstream_connections/BUILD

@@ -0,0 +1,35 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_extension",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_library(
+    name = "downstream_connections_monitor",
+    srcs = ["downstream_connections_monitor.cc"],
+    hdrs = ["downstream_connections_monitor.h"],
+    deps = [
+        "//envoy/server:proactive_resource_monitor_interface",
+        "//envoy/server:resource_monitor_config_interface",
+        "//source/common/common:assert_lib",
+        "@envoy_api//envoy/extensions/resource_monitors/downstream_connections/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        ":downstream_connections_monitor",
+        "//envoy/registry",
+        "//source/common/common:assert_lib",
+        "//source/extensions/resource_monitors/common:factory_base_lib",
+        "@envoy_api//envoy/extensions/resource_monitors/downstream_connections/v3:pkg_cc_proto",
+    ],
+)

source/extensions/resource_monitors/downstream_connections/config.cc

@@ -0,0 +1,33 @@
+#include "source/extensions/resource_monitors/downstream_connections/config.h"
+
+#include "envoy/extensions/resource_monitors/downstream_connections/v3/downstream_connections.pb.h"
+#include "envoy/extensions/resource_monitors/downstream_connections/v3/downstream_connections.pb.validate.h"
+#include "envoy/registry/registry.h"
+
+#include "source/common/protobuf/utility.h"
+#include "source/extensions/resource_monitors/downstream_connections/downstream_connections_monitor.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace ResourceMonitors {
+namespace DownstreamConnections {
+
+Server::ProactiveResourceMonitorPtr
+ActiveDownstreamConnectionsMonitorFactory::createProactiveResourceMonitorFromProtoTyped(
+    const envoy::extensions::resource_monitors::downstream_connections::v3::
+        DownstreamConnectionsConfig& config,
+    Server::Configuration::ResourceMonitorFactoryContext& /*unused_context*/) {
+  return std::make_unique<ActiveDownstreamConnectionsResourceMonitor>(config);
+}
+
+/**
+ * Static registration for the downstream connections resource monitor factory. @see
+ * RegistryFactory.
+ */
+REGISTER_FACTORY(ActiveDownstreamConnectionsMonitorFactory,
+                 Server::Configuration::ProactiveResourceMonitorFactory);
+
+} // namespace DownstreamConnections
+} // namespace ResourceMonitors
+} // namespace Extensions
+} // namespace Envoy

source/extensions/resource_monitors/downstream_connections/config.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include "envoy/extensions/resource_monitors/downstream_connections/v3/downstream_connections.pb.h"
+#include "envoy/extensions/resource_monitors/downstream_connections/v3/downstream_connections.pb.validate.h"
+#include "envoy/server/resource_monitor_config.h"
+
+#include "source/extensions/resource_monitors/common/factory_base.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace ResourceMonitors {
+namespace DownstreamConnections {
+
+class ActiveDownstreamConnectionsMonitorFactory
+    : public Common::ProactiveFactoryBase<
+          envoy::extensions::resource_monitors::downstream_connections::v3::
+              DownstreamConnectionsConfig> {
+public:
+  ActiveDownstreamConnectionsMonitorFactory()
+      : ProactiveFactoryBase("envoy.resource_monitors.downstream_connections") {}
+
+private:
+  Server::ProactiveResourceMonitorPtr createProactiveResourceMonitorFromProtoTyped(
+      const envoy::extensions::resource_monitors::downstream_connections::v3::
+          DownstreamConnectionsConfig& config,
+      Server::Configuration::ResourceMonitorFactoryContext& context) override;
+};
+
+} // namespace DownstreamConnections
+} // namespace ResourceMonitors
+} // namespace Extensions
+} // namespace Envoy

source/extensions/resource_monitors/downstream_connections/downstream_connections_monitor.cc

@@ -0,0 +1,45 @@
+#include "source/extensions/resource_monitors/downstream_connections/downstream_connections_monitor.h"
+
+#include "envoy/extensions/resource_monitors/downstream_connections/v3/downstream_connections.pb.h"
+
+#include "source/common/common/assert.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace ResourceMonitors {
+namespace DownstreamConnections {
+
+ActiveDownstreamConnectionsResourceMonitor::ActiveDownstreamConnectionsResourceMonitor(
+    const envoy::extensions::resource_monitors::downstream_connections::v3::
+        DownstreamConnectionsConfig& config)
+    : max_(config.max_active_downstream_connections()), current_(0){};
+
+bool ActiveDownstreamConnectionsResourceMonitor::tryAllocateResource(int64_t increment) {
+  int64_t new_val = (current_ += increment);
+  if (new_val > max_ || new_val < 0) {
+    current_ -= increment;
+    return false;
+  }
+  return true;
+}
+
+bool ActiveDownstreamConnectionsResourceMonitor::tryDeallocateResource(int64_t decrement) {
+  RELEASE_ASSERT(decrement <= current_,
+                 "Cannot deallocate resource, current resource usage is lower than decrement");
+  int64_t new_val = (current_ -= decrement);
+  if (new_val < 0) {
+    current_ += decrement;
+    return false;
+  }
+  return true;
+}
+
+int64_t ActiveDownstreamConnectionsResourceMonitor::currentResourceUsage() const {
+  return current_.load();
+}
+int64_t ActiveDownstreamConnectionsResourceMonitor::maxResourceUsage() const { return max_; };
+
+} // namespace DownstreamConnections
+} // namespace ResourceMonitors
+} // namespace Extensions
+} // namespace Envoy

source/extensions/resource_monitors/downstream_connections/downstream_connections_monitor.h

@@ -0,0 +1,34 @@
+#pragma once
+
+#include "envoy/extensions/resource_monitors/downstream_connections/v3/downstream_connections.pb.h"
+#include "envoy/server/proactive_resource_monitor.h"
+
+#include "source/common/common/assert.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace ResourceMonitors {
+namespace DownstreamConnections {
+
+class ActiveDownstreamConnectionsResourceMonitor : public Server::ProactiveResourceMonitor {
+public:
+  ActiveDownstreamConnectionsResourceMonitor(
+      const envoy::extensions::resource_monitors::downstream_connections::v3::
+          DownstreamConnectionsConfig& config);
+
+  bool tryAllocateResource(int64_t increment) override;
+
+  bool tryDeallocateResource(int64_t decrement) override;
+
+  int64_t currentResourceUsage() const override;
+  int64_t maxResourceUsage() const override;
+
+protected:
+  int64_t max_;
+  std::atomic<int64_t> current_;
+};
+
+} // namespace DownstreamConnections
+} // namespace ResourceMonitors
+} // namespace Extensions
+} // namespace Envoy

test/extensions/resource_monitors/downstream_connections/BUILD

@@ -0,0 +1,38 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_package",
+)
+load(
+    "//test/extensions:extensions_build_system.bzl",
+    "envoy_extension_cc_test",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_package()
+
+envoy_extension_cc_test(
+    name = "downstream_connections_monitor_test",
+    srcs = ["downstream_connections_monitor_test.cc"],
+    extension_names = ["envoy.resource_monitors.downstream_connections"],
+    external_deps = ["abseil_optional"],
+    deps = [
+        "//source/extensions/resource_monitors/downstream_connections:downstream_connections_monitor",
+        "@envoy_api//envoy/extensions/resource_monitors/downstream_connections/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_extension_cc_test(
+    name = "config_test",
+    srcs = ["config_test.cc"],
+    extension_names = ["envoy.resource_monitors.downstream_connections"],
+    deps = [
+        "//envoy/registry",
+        "//source/extensions/resource_monitors/downstream_connections:config",
+        "//source/server:resource_monitor_config_lib",
+        "//test/mocks/event:event_mocks",
+        "//test/mocks/server:options_mocks",
+        "//test/test_common:utility_lib",
+        "@envoy_api//envoy/extensions/resource_monitors/downstream_connections/v3:pkg_cc_proto",
+    ],
+)