Limit inflight coordination service session requests (#9168)

Author: snaury

ydb/core/driver_lib/run/run.cpp

@@ -785,7 +785,7 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) {
 
         if (hasKesus) {
             server.AddService(new NKesus::TKesusGRpcService(ActorSystem.Get(), Counters,
-                grpcRequestProxies[0], hasKesus.IsRlAllowed()));
+                AppData->InFlightLimiterRegistry, grpcRequestProxies[0], hasKesus.IsRlAllowed()));
         }
 
         if (hasPQ) {

ydb/core/testlib/test_client.cpp

@@ -421,7 +421,7 @@ namespace Tests {
         GRpcServer->AddService(new NGRpcService::V1::TGRpcPersQueueService(system, counters, NMsgBusProxy::CreatePersQueueMetaCacheV2Id(), grpcRequestProxies[0], true));
         GRpcServer->AddService(new NGRpcService::V1::TGRpcTopicService(system, counters, NMsgBusProxy::CreatePersQueueMetaCacheV2Id(), grpcRequestProxies[0], true));
         GRpcServer->AddService(new NGRpcService::TGRpcPQClusterDiscoveryService(system, counters, grpcRequestProxies[0]));
-        GRpcServer->AddService(new NKesus::TKesusGRpcService(system, counters, grpcRequestProxies[0], true));
+        GRpcServer->AddService(new NKesus::TKesusGRpcService(system, counters, appData.InFlightLimiterRegistry, grpcRequestProxies[0], true));
         GRpcServer->AddService(new NGRpcService::TGRpcCmsService(system, counters, grpcRequestProxies[0], true));
         GRpcServer->AddService(new NGRpcService::TGRpcDiscoveryService(system, counters, grpcRequestProxies[0], true));
         GRpcServer->AddService(new NGRpcService::TGRpcYdbClickhouseInternalService(system, counters, appData.InFlightLimiterRegistry, grpcRequestProxies[0], true));

ydb/services/kesus/grpc_service.cpp

@@ -22,6 +22,10 @@
 namespace NKikimr {
 namespace NKesus {
 
+// Note: this is an extremely high default to avoid breaking clients
+// TODO: make it configurable
+static constexpr i64 DEFAULT_MAX_SESSIONS_INFLIGHT = 100000;
+
 ////////////////////////////////////////////////////////////////////////////////
 
 class TGRpcSessionActor
@@ -613,29 +617,42 @@ class TGRpcSessionActor
 
 ////////////////////////////////////////////////////////////////////////////////
 
+TKesusGRpcService::TKesusGRpcService(
+        NActors::TActorSystem* system,
+        TIntrusivePtr<::NMonitoring::TDynamicCounters> counters,
+        TIntrusivePtr<NGRpcService::TInFlightLimiterRegistry> limiterRegistry,
+        const NActors::TActorId& proxyId,
+        bool rlAllowed)
+    : TBase(system, counters, proxyId, rlAllowed)
+    , LimiterRegistry_(limiterRegistry)
+{}
+
 void TKesusGRpcService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) {
-    auto getCounterBlock = NGRpcService::CreateCounterCb(Counters_, ActorSystem_);
     using NGRpcService::TRateLimiterMode;
+    auto getCounterBlock = NGRpcService::CreateCounterCb(Counters_, ActorSystem_);
+    auto getLimiter = CreateLimiterCb(LimiterRegistry_);
 
 #ifdef ADD_REQUEST
 #error ADD_REQUEST macro is already defined
 #endif
 
 #define ADD_REQUEST(NAME, IN, OUT, CB) \
-    MakeIntrusive<NGRpcService::TGRpcRequest<Ydb::Coordination::IN, Ydb::Coordination::OUT, TKesusGRpcService>>( \
-        this, \
-        &Service_, \
-        CQ_, \
-        [this](NYdbGrpc::IRequestContextBase* reqCtx) { \
-            NGRpcService::ReportGrpcReqToMon(*ActorSystem_, reqCtx->GetPeer()); \
-            ActorSystem_->Send(GRpcRequestProxyId_, \
-                new NGRpcService::TGrpcRequestOperationCall<Ydb::Coordination::IN, Ydb::Coordination::OUT> \
-                    (reqCtx, &CB, NGRpcService::TRequestAuxSettings{RLSWITCH(TRateLimiterMode::Rps), nullptr})); \
-        }, \
-        &Ydb::Coordination::V1::CoordinationService::AsyncService::Request ## NAME, \
-        "Coordination/" #NAME,             \
-        logger, \
-        getCounterBlock("coordination", #NAME))->Run();
+    for (auto* cq : CQS) { \
+        MakeIntrusive<NGRpcService::TGRpcRequest<Ydb::Coordination::IN, Ydb::Coordination::OUT, TKesusGRpcService>>( \
+            this, \
+            &Service_, \
+            cq, \
+            [this](NYdbGrpc::IRequestContextBase* reqCtx) { \
+                NGRpcService::ReportGrpcReqToMon(*ActorSystem_, reqCtx->GetPeer()); \
+                ActorSystem_->Send(GRpcRequestProxyId_, \
+                    new NGRpcService::TGrpcRequestOperationCall<Ydb::Coordination::IN, Ydb::Coordination::OUT> \
+                        (reqCtx, &CB, NGRpcService::TRequestAuxSettings{RLSWITCH(TRateLimiterMode::Rps), nullptr})); \
+            }, \
+            &Ydb::Coordination::V1::CoordinationService::AsyncService::Request ## NAME, \
+            "Coordination/" #NAME,             \
+            logger, \
+            getCounterBlock("coordination", #NAME))->Run(); \
+    }
 
     ADD_REQUEST(CreateNode, CreateNodeRequest, CreateNodeResponse, NGRpcService::DoCreateCoordinationNode);
     ADD_REQUEST(AlterNode, AlterNodeRequest, AlterNodeResponse, NGRpcService::DoAlterCoordinationNode);
@@ -644,19 +661,21 @@ void TKesusGRpcService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) {
 
 #undef ADD_REQUEST
 
-    TGRpcSessionActor::TGRpcRequest::Start(
-        this,
-        this->GetService(),
-        CQ_,
-        &Ydb::Coordination::V1::CoordinationService::AsyncService::RequestSession,
-        [this](TIntrusivePtr<TGRpcSessionActor::IContext> context) {
-            NGRpcService::ReportGrpcReqToMon(*ActorSystem_, context->GetPeerName());
-            ActorSystem_->Send(GRpcRequestProxyId_, new NGRpcService::TEvCoordinationSessionRequest(context));
-        },
-        *ActorSystem_,
-        "Coordination/Session",
-        getCounterBlock("coordination", "Session", true),
-        /* TODO: limiter */ nullptr);
+    for (auto* cq : CQS) {
+        TGRpcSessionActor::TGRpcRequest::Start(
+            this,
+            this->GetService(),
+            cq,
+            &Ydb::Coordination::V1::CoordinationService::AsyncService::RequestSession,
+            [this](TIntrusivePtr<TGRpcSessionActor::IContext> context) {
+                NGRpcService::ReportGrpcReqToMon(*ActorSystem_, context->GetPeerName());
+                ActorSystem_->Send(GRpcRequestProxyId_, new NGRpcService::TEvCoordinationSessionRequest(context));
+            },
+            *ActorSystem_,
+            "Coordination/Session",
+            getCounterBlock("coordination", "Session", true),
+            getLimiter("CoordinationService", "Session", DEFAULT_MAX_SESSIONS_INFLIGHT));
+    }
 }
 
 } // namespace NKesus

ydb/services/kesus/grpc_service.h

@@ -9,6 +9,7 @@
 #include <util/generic/hash_set.h>
 
 #include <ydb/core/grpc_services/base/base_service.h>
+#include <ydb/core/grpc_services/grpc_helper.h>
 
 
 namespace NKikimr {
@@ -17,14 +18,24 @@ namespace NKesus {
 class TKesusGRpcService
     : public ::NKikimr::NGRpcService::TGrpcServiceBase<Ydb::Coordination::V1::CoordinationService>
 {
+    using TBase = ::NKikimr::NGRpcService::TGrpcServiceBase<Ydb::Coordination::V1::CoordinationService>;
+
     class TContextBase;
     class TSessionContext;
 
 public:
-    using ::NKikimr::NGRpcService::TGrpcServiceBase<Ydb::Coordination::V1::CoordinationService>::TGrpcServiceBase;
+    TKesusGRpcService(
+        NActors::TActorSystem* system,
+        TIntrusivePtr<::NMonitoring::TDynamicCounters> counters,
+        TIntrusivePtr<NGRpcService::TInFlightLimiterRegistry> limiterRegistry,
+        const NActors::TActorId& proxyId,
+        bool rlAllowed);
 
 private:
     void SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger);
+
+private:
+    TIntrusivePtr<NGRpcService::TInFlightLimiterRegistry> LimiterRegistry_;
 };
 
 }