bug grpc: enable Congestion Control for grpc services (rebase)

kpavlov00 · kpavlov00 · commit 47105c3820f7 · 2026-03-13T09:59:38.000+03:00
origin: **bug grpc: enable Congestion Control for grpc services** <https://nda.ya.ru/t/Ajxwo_qT7WdSzs> commit_hash:e5b223f405e65d1843713fcf0195135eccad56aa
diff --git a/core/include/userver/congestion_control/limiter.hpp b/core/include/userver/congestion_control/limiter.hpp
@@ -8,10 +8,12 @@ USERVER_NAMESPACE_BEGIN
 namespace congestion_control {
 
 struct Limit {
-    std::optional<size_t> load_limit;
-    size_t current_load{0};
+    std::optional<std::size_t> load_limit;
+    std::size_t current_load{0};
 
-    std::string ToLogString() { return "limit=" + (load_limit ? std::to_string(*load_limit) : std::string("(none)")); }
+    std::string ToLogString() const {
+        return "limit=" + (load_limit ? std::to_string(*load_limit) : std::string("(none)"));
+    }
 };
 
 class Limiter {
diff --git a/core/include/userver/server/congestion_control/limiter.hpp b/core/include/userver/server/congestion_control/limiter.hpp
@@ -14,12 +14,16 @@ namespace server::congestion_control {
 class Limitee {
 public:
     virtual void SetLimit(std::optional<size_t> new_limit) = 0;
+
+    virtual std::size_t GetLimitableHandlersCount() const = 0;
 };
 
 class Limiter final : public USERVER_NAMESPACE::congestion_control::Limiter {
 public:
     void SetLimit(const USERVER_NAMESPACE::congestion_control::Limit& new_limit) override;
 
+    std::size_t GetLimitableHandlersCount() const;
+
     void RegisterLimitee(Limitee& limitee);
 
 private:
diff --git a/core/include/userver/server/server.hpp b/core/include/userver/server/server.hpp
@@ -50,8 +50,6 @@ class Server final : public congestion_control::Limitee, public congestion_contr
 
     void AddHandler(const handlers::HttpHandlerBase& handler, engine::TaskProcessor& task_processor);
 
-    size_t GetThrottlableHandlersCount() const;
-
     const http::HttpRequestHandler& GetHttpRequestHandler(bool is_monitor = false) const;
 
     void StartMonitorPort();
@@ -63,6 +61,8 @@ class Server final : public congestion_control::Limitee, public congestion_contr
 
     void SetLimit(std::optional<size_t> new_limit) override;
 
+    size_t GetLimitableHandlersCount() const override;
+
     void SetRpsRatelimit(std::optional<size_t> rps);
 
     void SetRpsRatelimitStatusCode(http::HttpStatus status_code);
diff --git a/core/src/congestion_control/component.cpp b/core/src/congestion_control/component.cpp
@@ -149,12 +149,12 @@ void Component::OnConfigUpdate(const dynamic_config::Snapshot& cfg) {
 }
 
 void Component::OnAllComponentsLoaded() {
-    LOG_DEBUG()
-        << "Found " << pimpl_->server.GetThrottlableHandlersCount()
-        << " registered HTTP handlers with enabled throttling";
-    if (pimpl_->server.GetThrottlableHandlersCount() == 0) {
+    const auto limitable_handlers_count = pimpl_->server_limiter.GetLimitableHandlersCount();
+    LOG_DEBUG() << "Found " << limitable_handlers_count << " registered handlers with enabled throttling";
+
+    if (limitable_handlers_count == 0) {
         pimpl_->force_disabled = true;
-        LOG_WARNING() << "No throttlable HTTP handlers registered, disabling";
+        LOG_WARNING() << "No throttlable handlers registered, disabling";
 
         // apply fake_mode
         OnConfigUpdate(pimpl_->dynamic_config.GetSnapshot());
diff --git a/core/src/server/congestion_control/limiter.cpp b/core/src/server/congestion_control/limiter.cpp
@@ -1,5 +1,8 @@
 #include <userver/server/congestion_control/limiter.hpp>
 
+#include <boost/range/adaptor/transformed.hpp>
+#include <boost/range/numeric.hpp>
+
 USERVER_NAMESPACE_BEGIN
 
 namespace server::congestion_control {
@@ -13,6 +16,14 @@ void Limiter::SetLimit(const USERVER_NAMESPACE::congestion_control::Limit& new_l
     }
 }
 
+std::size_t Limiter::GetLimitableHandlersCount() const {
+    auto lock = limitees_.Lock();
+    return boost::accumulate(
+        *lock | boost::adaptors::transformed([](auto ptr) { return ptr->GetLimitableHandlersCount(); }),
+        std::size_t{0}
+    );
+}
+
 void Limiter::RegisterLimitee(Limitee& limitee) {
     auto lock = limitees_.Lock();
     lock->emplace_back(&limitee);
diff --git a/core/src/server/server.cpp b/core/src/server/server.cpp
@@ -292,8 +292,6 @@ RequestsView& ServerImpl::GetRequestsView() {
     return requests_view_;
 }
 
-void Server::SetLimit(std::optional<size_t> new_limit) { SetRpsRatelimit(new_limit); }
-
 void ServerImpl::WriteTotalHandlerStatistics(utils::statistics::Writer& writer) const {
     handlers::HttpHandlerStatisticsSnapshot total;
 
@@ -381,8 +379,6 @@ void Server::AddHandler(const handlers::HttpHandlerBase& handler, engine::TaskPr
     pimpl_->AddHandler(handler, task_processor);
 }
 
-size_t Server::GetThrottlableHandlersCount() const { return pimpl_->GetThrottlableHandlersCount(); }
-
 const http::HttpRequestHandler& Server::GetHttpRequestHandler(bool is_monitor) const {
     return pimpl_->GetHttpRequestHandler(is_monitor);
 }
@@ -404,6 +400,10 @@ void Server::Stop() { pimpl_->Stop(); }
 
 RequestsView& Server::GetRequestsView() { return pimpl_->GetRequestsView(); }
 
+void Server::SetLimit(std::optional<size_t> new_limit) { SetRpsRatelimit(new_limit); }
+
+size_t Server::GetLimitableHandlersCount() const { return pimpl_->GetThrottlableHandlersCount(); }
+
 void Server::SetRpsRatelimit(std::optional<size_t> rps) { pimpl_->SetRpsRatelimit(rps); }
 
 void Server::SetRpsRatelimitStatusCode(http::HttpStatus status_code) { pimpl_->SetRpsRatelimitStatusCode(status_code); }
diff --git a/grpc/functional_tests/middleware_server/src/service.cpp b/grpc/functional_tests/middleware_server/src/service.cpp
@@ -2,18 +2,38 @@
 
 #include <fmt/format.h>
 
+#include <userver/components/component_context.hpp>
+#include <userver/congestion_control/component.hpp>
 #include <userver/engine/async.hpp>
 #include <userver/engine/sleep.hpp>
+#include <userver/utils/assert.hpp>
 #include <userver/yaml_config/merge_schemas.hpp>
 
 #include <userver/ugrpc/server/service_component_base.hpp>
 
 namespace functional_tests {
 
+namespace {
+
+void EnsureCongestionControlEnbled(const congestion_control::Controller& congestion_control_controller) {
+    UINVARIANT(congestion_control_controller.IsEnabled(), "CongestionControl Controller should be enabled");
+}
+
+}  // namespace
+
+GreeterServiceComponent::GreeterServiceComponent(
+    const components::ComponentConfig& config,
+    const components::ComponentContext& context
+)
+    : samples::api::GreeterServiceBase::Component(config, context),
+      congestion_control_controller_{context.FindComponent<congestion_control::Component>().GetServerController()}
+{}
+
 GreeterServiceComponent::SayHelloResult GreeterServiceComponent::SayHello(
     CallContext& /*context*/,
     samples::api::GreetingRequest&& request
 ) {
+    EnsureCongestionControlEnbled(congestion_control_controller_);
     samples::api::GreetingResponse response;
     response.set_greeting(fmt::format("Hello, {}!", request.name()));
     return response;
@@ -24,6 +44,7 @@ GreeterServiceComponent::SayHelloResponseStreamResult GreeterServiceComponent::S
     samples::api::GreetingRequest&& request,
     SayHelloResponseStreamWriter& writer
 ) {
+    EnsureCongestionControlEnbled(congestion_control_controller_);
     std::string message = fmt::format("{}, {}", "Hello", request.name());
     constexpr auto kCountSend = 5;
     constexpr std::chrono::milliseconds kTimeInterval{200};
@@ -41,6 +62,7 @@ GreeterServiceComponent::SayHelloRequestStreamResult GreeterServiceComponent::Sa
     CallContext& /*context*/,
     SayHelloRequestStreamReader& reader
 ) {
+    EnsureCongestionControlEnbled(congestion_control_controller_);
     std::string income_message;
     samples::api::GreetingRequest request;
     while (reader.Read(request)) {
@@ -55,6 +77,7 @@ GreeterServiceComponent::SayHelloStreamsResult GreeterServiceComponent::SayHello
     CallContext& /*context*/,
     SayHelloStreamsReaderWriter& stream
 ) {
+    EnsureCongestionControlEnbled(congestion_control_controller_);
     constexpr std::chrono::milliseconds kTimeInterval{200};
     std::string income_message;
     samples::api::GreetingRequest request;
diff --git a/grpc/functional_tests/middleware_server/src/service.hpp b/grpc/functional_tests/middleware_server/src/service.hpp
@@ -1,6 +1,8 @@
 #pragma once
 #include <userver/utest/using_namespace_userver.hpp>
 
+#include <userver/congestion_control/controller.hpp>
+
 #include <samples/greeter_client.usrv.pb.hpp>
 #include <samples/greeter_service.usrv.pb.hpp>
 
@@ -10,8 +12,7 @@ class GreeterServiceComponent final : public samples::api::GreeterServiceBase::C
 public:
     static constexpr std::string_view kName = "greeter-service";
 
-    GreeterServiceComponent(const components::ComponentConfig& config, const components::ComponentContext& context)
-        : samples::api::GreeterServiceBase::Component(config, context) {}
+    GreeterServiceComponent(const components::ComponentConfig& config, const components::ComponentContext& context);
 
     SayHelloResult SayHello(CallContext& context, samples::api::GreetingRequest&& request) final;
 
@@ -26,6 +27,9 @@ class GreeterServiceComponent final : public samples::api::GreeterServiceBase::C
     SayHelloStreamsResult SayHelloStreams(CallContext& context, SayHelloStreamsReaderWriter& stream) final;
 
     static yaml_config::Schema GetStaticConfigSchema();
+
+private:
+    const congestion_control::Controller& congestion_control_controller_;
 };
 
 }  // namespace functional_tests
diff --git a/grpc/functional_tests/middleware_server/tests/conftest.py b/grpc/functional_tests/middleware_server/tests/conftest.py
@@ -1,10 +1,27 @@
+from typing import Any
+
 import pytest
 
 import samples.greeter_pb2_grpc as greeter_services
 
 pytest_plugins = ['pytest_userver.plugins.grpc']
 
 
+@pytest.fixture(scope='session')
+def dynamic_config_fallback_patch() -> dict[str, Any]:
+    return {'USERVER_RPS_CCONTROL_ENABLED': True}
+
+
+@pytest.fixture(scope='session')
+def congestion_control_fake_mode() -> bool:
+    return False
+
+
+@pytest.fixture(scope='session')
+def service_env():
+    return {'CPU_LIMIT': '1c'}
+
+
 @pytest.fixture
 def grpc_client(grpc_channel):
     return greeter_services.GreeterServiceStub(grpc_channel)
diff --git a/grpc/include/userver/ugrpc/server/middlewares/congestion_control/component.hpp b/grpc/include/userver/ugrpc/server/middlewares/congestion_control/component.hpp
@@ -3,6 +3,9 @@
 /// @file userver/ugrpc/server/middlewares/congestion_control/component.hpp
 /// @brief @copybrief ugrpc::server::middlewares::congestion_control::Component
 
+#include <atomic>
+#include <cstddef>
+
 #include <userver/server/congestion_control/limiter.hpp>
 #include <userver/ugrpc/server/middlewares/base.hpp>
 #include <userver/utils/token_bucket.hpp>
@@ -51,10 +54,13 @@ class Component final
 
     void SetLimit(std::optional<size_t> new_limit) override;
 
+    std::size_t GetLimitableHandlersCount() const override;
+
 private:
     std::shared_ptr<utils::TokenBucket> rate_limit_{
         std::make_shared<utils::TokenBucket>(utils::TokenBucket::MakeUnbounded())
     };
+    mutable std::atomic<std::size_t> limitable_handlers_count_{0};
 };
 
 }  // namespace ugrpc::server::middlewares::congestion_control
diff --git a/grpc/src/ugrpc/server/middlewares/congestion_control/component.cpp b/grpc/src/ugrpc/server/middlewares/congestion_control/component.cpp
@@ -58,6 +58,7 @@ std::shared_ptr<const MiddlewareBase> Component::CreateMiddleware(
     const yaml_config::YamlConfig& middleware_config
 ) const {
     auto middleware = std::make_shared<Middleware>(middleware_config.As<Settings>(), rate_limit_);
+    ++limitable_handlers_count_;
     return middleware;
 }
 
@@ -76,6 +77,8 @@ void Component::SetLimit(std::optional<size_t> new_limit) {
     }
 }
 
+std::size_t Component::GetLimitableHandlersCount() const { return limitable_handlers_count_; }
+
 yaml_config::Schema Component::GetMiddlewareConfigSchema() const { return GetStaticConfigSchema(); }
 
 yaml_config::Schema Component::GetStaticConfigSchema() {
diff --git a/scripts/docs/en/userver/congestion_control.md b/scripts/docs/en/userver/congestion_control.md
@@ -11,7 +11,7 @@ by dynamic config options @ref USERVER_RPS_CCONTROL and @ref USERVER_RPS_CCONTRO
 CC can run in `fake-mode` with no RPS limit (but FSM works). CC goes into `fake-mode` in the following cases:
 
 * there are no reliable guarantees on CPU, in this case RPS-limit would be triggered too often,
-* service has no HTTP-handles others than @ref server::handlers::Ping.
+* service has no handlers others than @ref server::handlers::Ping, which has throttling disabled.
 
 `fake-mode` can be useful for more flexible traffic restriction settings, according to it's more complex logic, which
 can be implemented in a middleware.
@@ -21,14 +21,14 @@ can be implemented in a middleware.
 @ref congestion_control::Component can be useful if your service stops handling requests when overloaded, significantly
 increasing response time, responding with HTTP 500 codes to requests, eating memory.
 
-Including CC in your service will help you handle some reasonable request flow returning HTTP error codes to the rest. 
+Including CC in your service will help you handle some reasonable request flow returning HTTP error codes to the rest.
 
 ### Usage restrictions
 
 @ref congestion_control::Component cannot be useful if:
 
 1. Your service has an exact RPS limit requirement. The heuristic algorithm depends on how much and in what mode the
-   CPU is used in the current container. 
+   CPU is used in the current container.
 
 2. CPU is not a limiting resource.
 
@@ -68,7 +68,7 @@ In some situations default settings are ineffective. For example:
 
 * if request processing is done without context switch more than 10 ms,
 
-in those situations congestion_control::Component settings need adjusting. 
+in those situations congestion_control::Component settings need adjusting.
 
 Basic dynamic configuration options:
 
@@ -77,13 +77,13 @@ Basic dynamic configuration options:
 * @ref USERVER_RPS_CCONTROL - basic CC settings
 
 * @ref USERVER_TASK_PROCESSOR_QOS - note
-  `default-service.default-task-processor.wait_queue_overload.sensor_time_limit_us`. 
-  This setting defines wait in queue time after which the overload events for RPS congestion control are generated. 
+  `default-service.default-task-processor.wait_queue_overload.sensor_time_limit_us`.
+  This setting defines wait in queue time after which the overload events for RPS congestion control are generated.
   It is recommended to set this setting >= 2000 (2 ms) because system scheduler (CFS) time unit by default equals 2 ms.
 
 ## Diagnostics
 
-In case RPS mechanism is triggered it is recommended to ensure that there is no mistake. If RPS triggering coincided 
+In case RPS mechanism is triggered it is recommended to ensure that there is no mistake. If RPS triggering coincided
 with peak CPU consumption than there is no mistake and the lack of resources situation needs to be resolved:
 
 1. You need to locate slow code and optimise it for the workload.
diff --git a/testsuite/pytest_plugins/pytest_userver/plugins/base.py b/testsuite/pytest_plugins/pytest_userver/plugins/base.py
@@ -136,6 +136,16 @@ def monitor_port(pytestconfig, _original_service_config, choose_free_port) -> in
     )
 
 
+@pytest.fixture(scope='session')
+def congestion_control_fake_mode() -> bool:
+    """
+    Returns congestion control fake-mode value.
+
+    @ingroup userver_testsuite_fixtures
+    """
+    return True
+
+
 def _get_port(
     original_service_config,
     choose_free_port,
@@ -215,13 +225,13 @@ def _is_port_free(port_num: int, family: int, address: str) -> bool:
 
 
 @pytest.fixture(scope='session')
-def userver_base_prepare_service_config():
+def userver_base_prepare_service_config(congestion_control_fake_mode):
     def patch_config(config, config_vars):
         components = config['components_manager']['components']
         if 'congestion-control' in components:
             if components['congestion-control'] is None:
                 components['congestion-control'] = {}
 
-            components['congestion-control']['fake-mode'] = True
+            components['congestion-control']['fake-mode'] = congestion_control_fake_mode
 
     return patch_config
