feat core: add congestion-control-sensor testpoints

commit_hash:fc35a41d16b98e629ca3a4740eae5c355cd2fde5

Author: kpavlov00

core/src/congestion_control/watchdog.cpp

@@ -44,18 +44,36 @@ void Watchdog::Check() {
     auto cis = cis_.Lock();
     for (const auto& ci : *cis) {
         auto data = ci.sensor.FetchCurrent();
+
+        TESTPOINT_CALLBACK_NONCORO(
+            "congestion-control-sensor",
+            formats::json::Value{},
+            tp_,
+            [&data](const formats::json::Value& doc) {
+                data.current_load = doc["current-load"].As<std::uint64_t>(data.current_load);
+                data.overload_events_count = doc["overload-events-count"].As<std::uint64_t>(data.overload_events_count);
+                data.no_overload_events_count =
+                    doc["no-overload-events-count"].As<std::uint64_t>(data.no_overload_events_count);
+                LOG_INFO() << "Forcing CC Sensor data from testpoint: " << doc;
+            }
+        );
+
         ci.controller.Feed(data);
+
         auto limit = ci.controller.GetLimit();
 
         TESTPOINT_CALLBACK_NONCORO(
             "congestion-control",
             formats::json::Value{},
             tp_,
             [&limit](const formats::json::Value& doc) {
-                limit.load_limit = doc["force-rps-limit"].As<std::optional<size_t>>();
-                LOG_ERROR() << "Forcing RPS limit from testpoint: " << limit.load_limit;
+                if (doc.HasMember("force-rps-limit")) {
+                    limit.load_limit = doc["force-rps-limit"].As<std::optional<std::size_t>>();
+                    LOG_ERROR() << "Forcing RPS limit from testpoint: " << limit.load_limit;
+                }
             }
         );
+
         ci.limiter.SetLimit(limit);
 
         TESTPOINT_NONCORO("congestion-control-apply", formats::json::Value{}, tp_);

grpc/functional_tests/basic_chaos/tests-nonchaos/conftest.py

@@ -1,10 +1,38 @@
+from typing import Any
+
 import pytest
 
 import samples.greeter_pb2_grpc as greeter_pb2_grpc
 
 pytest_plugins = ['pytest_userver.plugins.grpc']
 
 
+@pytest.fixture(scope='session')
+def service_env():
+    return {'CPU_LIMIT': '1c'}
+
+
+@pytest.fixture(scope='session')
+def congestion_control_fake_mode() -> bool:
+    return False
+
+
+@pytest.fixture(scope='session')
+def dynamic_config_fallback_patch() -> dict[str, Any]:
+    return {
+        'USERVER_RPS_CCONTROL_ENABLED': True,
+        'USERVER_RPS_CCONTROL': {
+            'min-limit': 1,
+            'up-rate-percent': 100,
+            'down-rate-percent': 100,
+            'overload-on-seconds': 1,
+            'overload-off-seconds': 1,
+            'no-limit-seconds': 1,
+            'start-limit-factor': 0.01,
+        },
+    }
+
+
 @pytest.fixture
 def grpc_client(grpc_channel):
     return greeter_pb2_grpc.GreeterServiceStub(grpc_channel)

grpc/functional_tests/basic_chaos/tests-nonchaos/test_basic.py

@@ -4,19 +4,22 @@
 import samples.greeter_pb2 as greeter_pb2
 
 
-async def test_grpc_cc_enabled(grpc_client, service_client, testpoint):
+async def test_grpc_congestion_control_limit(grpc_client, service_client, testpoint):
+    @testpoint('congestion-control-sensor')
+    def tp_congestion_control_sensor(data):
+        return {}
+
     @testpoint('congestion-control')
-    def tp_cc_enable(data):
+    def tp_congestion_control(data):
         return {'force-rps-limit': 0}
 
     @testpoint('congestion-control-apply')
-    def tp_cc_apply(data):
+    def tp_congestion_control_apply(data):
         return {}
 
     # wait until server obtains the new limit, up to 1 second
     await service_client.enable_testpoints()
-    await tp_cc_enable.wait_call()
-    await tp_cc_apply.wait_call()
+    await tp_congestion_control.wait_call()
 
     # Random non-ping handler is throttled
     with pytest.raises(grpc.RpcError) as error:
@@ -27,11 +30,65 @@ def tp_cc_apply(data):
 
     # A hack to disable CC for other tests
     @testpoint('congestion-control')
-    def tp_cc_disable(data):
+    def tp_congestion_control_disable(data):
+        return {'force-rps-limit': None}
+
+    await tp_congestion_control_disable.wait_call()
+
+
+async def test_grpc_congestion_control_activate(grpc_client, service_client, testpoint):
+    @testpoint('congestion-control-sensor')
+    def tp_congestion_control_sensor(data):
+        return {
+            'current-load': 100,
+            'overload-events-count': 4,
+            'no-overload-events-count': 96,
+        }
+
+    @testpoint('congestion-control')
+    def tp_congestion_control(data):
+        return {}
+
+    @testpoint('congestion-control-apply')
+    def tp_congestion_control_apply(data):
         return {}
 
-    await tp_cc_disable.wait_call()
-    await tp_cc_apply.wait_call()
+    # is_overloaded -> true
+    await service_client.enable_testpoints()
+    await tp_congestion_control_sensor.wait_call()
+    await tp_congestion_control.wait_call()
+
+    # current_limit -> 1
+    await tp_congestion_control_sensor.wait_call()
+    await tp_congestion_control.wait_call()
+
+    with pytest.raises(grpc.RpcError) as error:
+        for i in range(0, 2):
+            request = greeter_pb2.GreetingRequest(name=f'Name0{i}/1')
+            await grpc_client.SayHello(request, wait_for_ready=True)
+    assert error.value.details() == 'Congestion control: rate limit exceeded'
+    assert error.value.code() == grpc.StatusCode.RESOURCE_EXHAUSTED
+
+    @testpoint('congestion-control-sensor')
+    def tp_congestion_control_sensor_disable(data):
+        return {
+            'current-load': 100,
+            'overload-events-count': 0,
+            'no-overload-events-count': 100,
+        }
+
+    # current_limit -> 1
+    await tp_congestion_control_sensor_disable.wait_call()
+    await tp_congestion_control.wait_call()
+
+    # current_limit -> null
+    await tp_congestion_control_sensor_disable.wait_call()
+    await tp_congestion_control.wait_call()
+
+    for i in range(1, 10):
+        request = greeter_pb2.GreetingRequest(name=f'Name{i}/2')
+        response = await grpc_client.SayHello(request, wait_for_ready=True)
+        assert response.greeting == f'Hello, Name{i}/2!'
 
 
 async def test_grpc_cancellation(grpc_client, service_client, testpoint):