diff --git a/ydb/core/driver_lib/run/run.cpp b/ydb/core/driver_lib/run/run.cpp index dafe627e7278..ecbff23a49d2 100644 --- a/ydb/core/driver_lib/run/run.cpp +++ b/ydb/core/driver_lib/run/run.cpp @@ -973,13 +973,13 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) { sslData.DoRequestClientCertificate = appConfig.GetClientCertificateAuthorization().GetRequestClientCertificate(); sslOpts.SetSslData(sslData); - GRpcServers.push_back({ "grpcs", new NYdbGrpc::TGRpcServer(sslOpts) }); + GRpcServers.push_back({ "grpcs", new NYdbGrpc::TGRpcServer(sslOpts, Counters) }); fillFn(grpcConfig, *GRpcServers.back().second, sslOpts); } if (grpcConfig.GetPort()) { - GRpcServers.push_back({ "grpc", new NYdbGrpc::TGRpcServer(opts) }); + GRpcServers.push_back({ "grpc", new NYdbGrpc::TGRpcServer(opts, Counters) }); fillFn(grpcConfig, *GRpcServers.back().second, opts); } @@ -996,7 +996,7 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) { xopts.SetEndpointId(ex.GetEndpointId()); } - GRpcServers.push_back({ "grpc", new NYdbGrpc::TGRpcServer(xopts) }); + GRpcServers.push_back({ "grpc", new NYdbGrpc::TGRpcServer(xopts, Counters) }); fillFn(ex, *GRpcServers.back().second, xopts); } @@ -1035,7 +1035,7 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) { Y_ABORT_UNLESS(xopts.SslData->Cert, "Cert not set"); Y_ABORT_UNLESS(xopts.SslData->Key, "Key not set"); - GRpcServers.push_back({ "grpcs", new NYdbGrpc::TGRpcServer(xopts) }); + GRpcServers.push_back({ "grpcs", new NYdbGrpc::TGRpcServer(xopts, Counters) }); fillFn(ex, *GRpcServers.back().second, xopts); } } diff --git a/ydb/core/grpc_streaming/grpc_streaming_ut.cpp b/ydb/core/grpc_streaming/grpc_streaming_ut.cpp index d6c723a9b27b..f0f63a51505b 100644 --- a/ydb/core/grpc_streaming/grpc_streaming_ut.cpp +++ b/ydb/core/grpc_streaming/grpc_streaming_ut.cpp @@ -95,12 +95,13 @@ class TGRpcTestServer { Server->GetRuntime()->SetLogPriority(NKikimrServices::GRPC_SERVER, NActors::NLog::PRI_DEBUG); + TIntrusivePtr<::NMonitoring::TDynamicCounters> counters(MakeIntrusive<::NMonitoring::TDynamicCounters>()); + NYdbGrpc::TServerOptions options; options.SetPort(grpc); - GRpcServer.Reset(new NYdbGrpc::TGRpcServer(options)); + GRpcServer.Reset(new NYdbGrpc::TGRpcServer(options, counters)); auto* as = Server->GetRuntime()->GetAnyNodeActorSystem(); - TIntrusivePtr<::NMonitoring::TDynamicCounters> counters(MakeIntrusive<::NMonitoring::TDynamicCounters>()); GRpcServer->AddService(new TStreamingService(as, counters)); GRpcServer->Start(); diff --git a/ydb/core/testlib/test_client.cpp b/ydb/core/testlib/test_client.cpp index f8552fecb17e..b039fe64fb47 100644 --- a/ydb/core/testlib/test_client.cpp +++ b/ydb/core/testlib/test_client.cpp @@ -346,7 +346,10 @@ namespace Tests { } void TServer::EnableGRpc(const NYdbGrpc::TServerOptions& options, ui32 grpcServiceNodeId) { - GRpcServer.reset(new NYdbGrpc::TGRpcServer(options)); + GRpcServerRootCounters = MakeIntrusive<::NMonitoring::TDynamicCounters>(); + auto& counters = GRpcServerRootCounters; + + GRpcServer.reset(new NYdbGrpc::TGRpcServer(options), counters); auto grpcService = new NGRpcProxy::TGRpcService(); auto system(Runtime->GetActorSystem(grpcServiceNodeId)); @@ -378,9 +381,6 @@ namespace Tests { auto grpcMon = system->Register(NGRpcService::CreateGrpcMonService(), TMailboxType::ReadAsFilled, appData.UserPoolId); system->RegisterLocalService(NGRpcService::GrpcMonServiceId(), grpcMon); - GRpcServerRootCounters = MakeIntrusive<::NMonitoring::TDynamicCounters>(); - auto& counters = GRpcServerRootCounters; - // Setup discovery for typically used services on the node { TIntrusivePtr desc = new NGRpcService::TGrpcEndpointDescription(); diff --git a/ydb/library/grpc/server/grpc_server.cpp b/ydb/library/grpc/server/grpc_server.cpp index cb8432563f60..2a593ec54c18 100644 --- a/ydb/library/grpc/server/grpc_server.cpp +++ b/ydb/library/grpc/server/grpc_server.cpp @@ -1,5 +1,8 @@ #include "grpc_server.h" +#include +#include + #include #include #include @@ -18,19 +21,31 @@ namespace NYdbGrpc { -using NThreading::TFuture; - -static void PullEvents(grpc::ServerCompletionQueue* cq) { +static void PullEvents(grpc::ServerCompletionQueue* cq, TIntrusivePtr<::NMonitoring::TDynamicCounters> counters) { TThread::SetCurrentThreadName("grpc_server"); + auto okCounter = counters->GetCounter("RequestExecuted", true); + auto errorCounter = counters->GetCounter("RequestDestroyed", true); + auto cpuTime = counters->GetCounter("ThreadCPU", true); + + NMonotonic::TMonotonic lastCpuTimeTs = {}; while (true) { void* tag; // uniquely identifies a request. bool ok; + auto now = NMonotonic::TMonotonic::Now(); + if (now - lastCpuTimeTs >= TDuration::Seconds(1)) { + lastCpuTimeTs = now; + *cpuTime = ThreadCPUTime(); + } + if (cq->Next(&tag, &ok)) { IQueueEvent* const ev(static_cast(tag)); - if (!ev->Execute(ok)) { + if (ev->Execute(ok)) { + okCounter->Inc(); + } else { ev->DestroyRequest(); + errorCounter->Inc(); } } else { break; @@ -103,8 +118,9 @@ void TGrpcServiceProtectiable::DecRequest() { } } -TGRpcServer::TGRpcServer(const TServerOptions& opts) +TGRpcServer::TGRpcServer(const TServerOptions& opts, TIntrusivePtr<::NMonitoring::TDynamicCounters> counters) : Options_(opts) + , Counters_(std::move(counters)) , Limiter_(Options_.MaxGlobalRequestInFlight) {} @@ -237,10 +253,12 @@ void TGRpcServer::Start() { } Ts.reserve(Options_.WorkerThreads); + auto grpcCounters = Counters_->GetSubgroup("counters", "grpc"); for (size_t i = 0; i < Options_.WorkerThreads; ++i) { auto* cq = &CQS_[i % CQS_.size()]; - Ts.push_back(SystemThreadFactory()->Run([cq] { - PullEvents(cq->get()); + auto workerCounters = grpcCounters->GetSubgroup("worker", ToString(i)); + Ts.push_back(SystemThreadFactory()->Run([cq, workerCounters] { + PullEvents(cq->get(), std::move(workerCounters)); })); } diff --git a/ydb/library/grpc/server/grpc_server.h b/ydb/library/grpc/server/grpc_server.h index 5ab48d2f24fe..664ecefb8889 100644 --- a/ydb/library/grpc/server/grpc_server.h +++ b/ydb/library/grpc/server/grpc_server.h @@ -18,6 +18,10 @@ #include +namespace NMonitoring { + struct TDynamicCounters; +} // NMonitoring + namespace NYdbGrpc { struct TSslData { @@ -349,7 +353,7 @@ class TGrpcServiceBase: public TGrpcServiceProtectiable { class TGRpcServer { public: using IGRpcServicePtr = TIntrusivePtr; - TGRpcServer(const TServerOptions& opts); + TGRpcServer(const TServerOptions& opts, TIntrusivePtr<::NMonitoring::TDynamicCounters> counters); ~TGRpcServer(); void AddService(IGRpcServicePtr service); void Start(); @@ -365,6 +369,7 @@ class TGRpcServer { using IThreadRef = TAutoPtr; const TServerOptions Options_; + TIntrusivePtr<::NMonitoring::TDynamicCounters> Counters_; std::unique_ptr Server_; std::vector> CQS_; TVector Ts;