Skip to content

Commit d7f99d9

Browse files
eivanov89eivanov89
authored
Ping service (#12100)
1 parent c8c54d4 commit d7f99d9

File tree

25 files changed

+974
-0
lines changed

25 files changed

+974
-0
lines changed

‎ydb/core/driver_lib/run/run.cpp‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,7 @@
124124
#include <ydb/services/backup/grpc_service.h>
125125
#include <ydb/services/ydb/ydb_logstore.h>
126126
#include <ydb/services/ydb/ydb_operation.h>
127+
#include <ydb/services/ydb/ydb_debug.h>
127128
#include <ydb/services/ydb/ydb_query.h>
128129
#include <ydb/services/ydb/ydb_scheme.h>
129130
#include <ydb/services/ydb/ydb_scripting.h>
@@ -883,6 +884,9 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) {
883884
grpcRequestProxies, hasDataStreams.IsRlAllowed(), grpcConfig.GetHandlersPerCompletionQueue()));
884885
}
885886

887+
server.AddService(new NGRpcService::TGRpcYdbDebugService(ActorSystem.Get(), Counters,
888+
grpcRequestProxies, hasDataStreams.IsRlAllowed(), grpcConfig.GetHandlersPerCompletionQueue()));
889+
886890
if (hasLogStore) {
887891
server.AddService(new NGRpcService::TGRpcYdbLogStoreService(ActorSystem.Get(), Counters,
888892
grpcRequestProxies[0], hasLogStore.IsRlAllowed()));

‎ydb/core/grpc_services/rpc_ping.cpp‎

Lines changed: 251 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,251 @@
1+
#include "service_debug.h"
2+
3+
#include <ydb/core/grpc_services/base/base.h>
4+
#include <ydb/core/kqp/compute_actor/kqp_compute_events.h>
5+
#include <ydb/core/tx/scheme_cache/scheme_cache.h>
6+
#include <ydb/core/tx/tx_proxy/proxy.h>
7+
8+
#include <ydb/public/api/protos/ydb_debug.pb.h>
9+
10+
#include <ydb/library/actors/core/actor_bootstrapped.h>
11+
12+
namespace NKikimr::NGRpcService {
13+
14+
using namespace Ydb;
15+
16+
namespace {
17+
18+
using namespace NActors;
19+
20+
////////////////////////////////////////////////////////////////////////////////
21+
22+
using TEvKqpProxyRequest = TGrpcRequestNoOperationCall<Debug::KqpProxyRequest, Debug::KqpProxyResponse>;
23+
24+
class TExecuteKqpPingRPC : public TActorBootstrapped<TExecuteKqpPingRPC> {
25+
public:
26+
static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
27+
return NKikimrServices::TActivity::OTHER;
28+
}
29+
30+
TExecuteKqpPingRPC(TEvKqpProxyRequest* request)
31+
: Request_(request)
32+
{}
33+
34+
void Bootstrap(const TActorContext &ctx) {
35+
this->Become(&TThis::StateWork);
36+
37+
Proceed(ctx);
38+
}
39+
40+
private:
41+
void StateWork(TAutoPtr<IEventHandle>& ev) {
42+
try {
43+
switch (ev->GetTypeRewrite()) {
44+
HFunc(NKqp::TEvKqp::TEvProxyPingResponse, Handle);
45+
default:
46+
UnexpectedEvent(__func__, ev);
47+
}
48+
} catch (const yexception& ex) {
49+
InternalError(ex.what());
50+
}
51+
}
52+
53+
void Proceed(const TActorContext &ctx) {
54+
LOG_TRACE_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << " sending ping to KQP proxy");
55+
if (!ctx.Send(NKqp::MakeKqpProxyID(ctx.SelfID.NodeId()), new NKqp::TEvKqp::TEvProxyPingRequest())) {
56+
LOG_ERROR_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << " failed to send ping");
57+
ReplyWithResult(StatusIds::INTERNAL_ERROR, ctx);
58+
}
59+
}
60+
61+
void Handle(NKqp::TEvKqp::TEvProxyPingResponse::TPtr&, const TActorContext& ctx) {
62+
LOG_TRACE_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << " got ping response");
63+
ReplyWithResult(StatusIds::SUCCESS, ctx);
64+
}
65+
66+
private:
67+
void ReplyWithResult(StatusIds::StatusCode status, const TActorContext &ctx) {
68+
Request_->ReplyWithYdbStatus(status);
69+
Die(ctx);
70+
}
71+
72+
void InternalError(const TString& message) {
73+
ALOG_ERROR(NKikimrServices::RPC_REQUEST, "Internal error, message: " << message);
74+
ReplyWithResult(StatusIds::INTERNAL_ERROR, TActivationContext::AsActorContext());
75+
}
76+
77+
void UnexpectedEvent(const TString& state, TAutoPtr<NActors::IEventHandle>& ev) {
78+
InternalError(TStringBuilder() << "TExecuteKqpPingRPC in state " << state << " received unexpected event "
79+
<< ev->GetTypeName() << Sprintf("(0x%08" PRIx32 ")", ev->GetTypeRewrite()));
80+
}
81+
82+
private:
83+
std::shared_ptr<TEvKqpProxyRequest> Request_;
84+
};
85+
86+
////////////////////////////////////////////////////////////////////////////////
87+
88+
using TEvSchemeCacheRequest = TGrpcRequestNoOperationCall<Debug::SchemeCacheRequest, Debug::SchemeCacheResponse>;
89+
90+
class TExecuteSchemeCachePingRPC : public TActorBootstrapped<TExecuteSchemeCachePingRPC> {
91+
public:
92+
static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
93+
return NKikimrServices::TActivity::OTHER;
94+
}
95+
96+
TExecuteSchemeCachePingRPC(TEvSchemeCacheRequest* request)
97+
: Request_(request)
98+
{}
99+
100+
void Bootstrap(const TActorContext &ctx) {
101+
this->Become(&TThis::StateWork);
102+
103+
Proceed(ctx);
104+
}
105+
106+
private:
107+
void StateWork(TAutoPtr<IEventHandle>& ev) {
108+
try {
109+
switch (ev->GetTypeRewrite()) {
110+
HFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, Handle);
111+
default:
112+
UnexpectedEvent(__func__, ev);
113+
}
114+
} catch (const yexception& ex) {
115+
InternalError(ex.what());
116+
}
117+
}
118+
119+
void Proceed(const TActorContext &ctx) {
120+
LOG_TRACE_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << " sending ping to SchemeCache");
121+
122+
auto* request = new TEvTxProxySchemeCache::TEvNavigateKeySet(new NSchemeCache::TSchemeCacheNavigate());
123+
if (!ctx.Send(MakeSchemeCacheID(), request)) {
124+
LOG_ERROR_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << " failed to send ping to SchemeCache");
125+
ReplyWithResult(StatusIds::INTERNAL_ERROR, ctx);
126+
}
127+
}
128+
129+
void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr&, const TActorContext& ctx) {
130+
LOG_TRACE_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << " got ping response from SchemeCache");
131+
ReplyWithResult(StatusIds::SUCCESS, ctx);
132+
}
133+
134+
private:
135+
void ReplyWithResult(StatusIds::StatusCode status, const TActorContext &ctx) {
136+
Request_->ReplyWithYdbStatus(status);
137+
Die(ctx);
138+
}
139+
140+
void InternalError(const TString& message) {
141+
ALOG_ERROR(NKikimrServices::RPC_REQUEST, "Internal error, message: " << message);
142+
ReplyWithResult(StatusIds::INTERNAL_ERROR, TActivationContext::AsActorContext());
143+
}
144+
145+
void UnexpectedEvent(const TString& state, TAutoPtr<NActors::IEventHandle>& ev) {
146+
InternalError(TStringBuilder() << "TExecuteSchemeCachePingRPC in state " << state <<
147+
" received unexpected event " << ev->GetTypeName() << Sprintf("(0x%08" PRIx32 ")", ev->GetTypeRewrite()));
148+
}
149+
150+
private:
151+
std::shared_ptr<TEvSchemeCacheRequest> Request_;
152+
};
153+
154+
////////////////////////////////////////////////////////////////////////////////
155+
156+
using TEvTxProxyRequest = TGrpcRequestNoOperationCall<Debug::TxProxyRequest, Debug::TxProxyResponse>;
157+
158+
class TExecuteTxProxyPingRPC : public TActorBootstrapped<TExecuteTxProxyPingRPC> {
159+
public:
160+
static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
161+
return NKikimrServices::TActivity::OTHER;
162+
}
163+
164+
TExecuteTxProxyPingRPC(TEvTxProxyRequest* request)
165+
: Request_(request)
166+
{}
167+
168+
void Bootstrap(const TActorContext &ctx) {
169+
this->Become(&TThis::StateWork);
170+
171+
Proceed(ctx);
172+
}
173+
174+
private:
175+
void StateWork(TAutoPtr<IEventHandle>& ev) {
176+
try {
177+
switch (ev->GetTypeRewrite()) {
178+
HFunc(TEvTxUserProxy::TEvAllocateTxIdResult, Handle);
179+
default:
180+
UnexpectedEvent(__func__, ev);
181+
}
182+
} catch (const yexception& ex) {
183+
InternalError(ex.what());
184+
}
185+
}
186+
187+
void Proceed(const TActorContext &ctx) {
188+
LOG_TRACE_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << " sending ping to TxProxy");
189+
if (!ctx.Send(MakeTxProxyID(), new TEvTxUserProxy::TEvAllocateTxId)) {
190+
LOG_ERROR_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << " failed to send ping to TxProxy");
191+
ReplyWithResult(StatusIds::INTERNAL_ERROR, ctx);
192+
}
193+
}
194+
195+
void Handle(TEvTxUserProxy::TEvAllocateTxIdResult::TPtr&, const TActorContext& ctx) {
196+
LOG_TRACE_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << " got ping response from TxProxy");
197+
ReplyWithResult(StatusIds::SUCCESS, ctx);
198+
}
199+
200+
private:
201+
void ReplyWithResult(StatusIds::StatusCode status, const TActorContext &ctx) {
202+
Request_->ReplyWithYdbStatus(status);
203+
Die(ctx);
204+
}
205+
206+
void InternalError(const TString& message) {
207+
ALOG_ERROR(NKikimrServices::RPC_REQUEST, "Internal error, message: " << message);
208+
ReplyWithResult(StatusIds::INTERNAL_ERROR, TActivationContext::AsActorContext());
209+
}
210+
211+
void UnexpectedEvent(const TString& state, TAutoPtr<NActors::IEventHandle>& ev) {
212+
InternalError(TStringBuilder() << "TExecuteTxProxyPingRPC in state " << state << " received unexpected event "
213+
<< ev->GetTypeName() << Sprintf("(0x%08" PRIx32 ")", ev->GetTypeRewrite()));
214+
}
215+
216+
private:
217+
std::shared_ptr<TEvTxProxyRequest> Request_;
218+
};
219+
220+
} // anonymous
221+
222+
////////////////////////////////////////////////////////////////////////////////
223+
224+
void DoGrpcProxyPing(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f) {
225+
// we are in the GRPC proxy already (or in the check actor in case of auth check),
226+
// thus ready to reply right here
227+
using TRequest = TGrpcRequestNoOperationCall<Debug::GrpcProxyRequest, Debug::GrpcProxyResponse>;
228+
TRequest* request = dynamic_cast<TRequest *>(p.get());
229+
Y_ABORT_UNLESS(request != nullptr, "Wrong using of TGRpcRequestWrapper in DoGrpcProxyPing");
230+
request->ReplyWithYdbStatus(StatusIds::SUCCESS);
231+
}
232+
233+
void DoKqpPing(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f) {
234+
auto* request = dynamic_cast<TEvKqpProxyRequest*>(p.release());
235+
Y_ABORT_UNLESS(request != nullptr, "Wrong using of TGRpcRequestWrapper in DoKqpPing");
236+
f.RegisterActor(new TExecuteKqpPingRPC(request));
237+
}
238+
239+
void DoSchemeCachePing(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f) {
240+
auto* request = dynamic_cast<TEvSchemeCacheRequest*>(p.release());
241+
Y_ABORT_UNLESS(request != nullptr, "Wrong using of TGRpcRequestWrapper in DoSchemeCachePing");
242+
f.RegisterActor(new TExecuteSchemeCachePingRPC(request));
243+
}
244+
245+
void DoTxProxyPing(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f) {
246+
auto* request = dynamic_cast<TEvTxProxyRequest*>(p.release());
247+
Y_ABORT_UNLESS(request != nullptr, "Wrong using of TGRpcRequestWrapper in DoTxProxyPing");
248+
f.RegisterActor(new TExecuteTxProxyPingRPC(request));
249+
}
250+
251+
} // namespace NKikimr::NGRpcService

‎ydb/core/grpc_services/service_debug.h‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
#pragma once
2+
3+
#include <memory>
4+
5+
namespace NKikimr::NGRpcService {
6+
7+
class IRequestOpCtx;
8+
class IRequestNoOpCtx;
9+
class IFacilityProvider;
10+
11+
void DoGrpcProxyPing(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f);
12+
void DoKqpPing(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f);
13+
void DoSchemeCachePing(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f);
14+
void DoTxProxyPing(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f);
15+
16+
} // namespace NKikimr::NGRpcService

‎ydb/core/grpc_services/ya.make‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@ SRCS(
6363
rpc_make_directory.cpp
6464
rpc_modify_permissions.cpp
6565
rpc_monitoring.cpp
66+
rpc_ping.cpp
6667
rpc_prepare_data_query.cpp
6768
rpc_rate_limiter_api.cpp
6869
rpc_read_columns.cpp

‎ydb/core/jaeger_tracing/request_discriminator.h‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,12 @@ enum class ERequestType: size_t {
7171
BSCONFIG_REPLACESTORAGECONFIG,
7272
BSCONFIG_FETCHSTORAGECONFIG,
7373

74+
PING_GRPC,
75+
PING_PROXY,
76+
PING_KQP,
77+
PING_SCHEME_CACHE,
78+
PING_TX_PROXY,
79+
7480
REQUEST_TYPES_CNT, // Add new types above this line
7581
};
7682

‎ydb/core/kqp/common/events/events.h‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -204,6 +204,12 @@ struct TEvKqp {
204204
Ydb::StatusIds::StatusCode Status;
205205
NYql::TIssues Issues;
206206
};
207+
208+
struct TEvProxyPingRequest : public TEventLocal<TEvProxyPingRequest, TKqpEvents::EvProxyPingRequest> {
209+
};
210+
211+
struct TEvProxyPingResponse : public TEventLocal<TEvProxyPingResponse, TKqpEvents::EvProxyPingResponse> {
212+
};
207213
};
208214

209215
} // namespace NKikimr::NKqp

‎ydb/core/kqp/common/simple/kqp_event_ids.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,8 @@ struct TKqpEvents {
4848
EvDelayedRequestError,
4949
EvBufferWrite,
5050
EvBufferWriteResult,
51+
EvProxyPingRequest,
52+
EvProxyPingResponse,
5153
};
5254

5355
static_assert (EvCompileInvalidateRequest + 1 == EvAbortExecution);

‎ydb/core/kqp/proxy_service/kqp_proxy_service.cpp‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -647,6 +647,10 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
647647
Send(ev->Sender, responseEv.Release(), 0, ev->Cookie);
648648
}
649649

650+
void Handle(TEvKqp::TEvProxyPingRequest::TPtr& ev) {
651+
Send(ev->Sender, new TEvKqp::TEvProxyPingResponse());
652+
}
653+
650654
void Handle(TEvKqp::TEvQueryRequest::TPtr& ev) {
651655
if (!DatabasesCache.SetDatabaseIdOrDefer(ev, static_cast<i32>(EDelayedRequestType::QueryRequest), ActorContext())) {
652656
return;
@@ -1349,6 +1353,7 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
13491353
hFunc(TEvents::TEvUndelivered, Handle);
13501354
hFunc(NConsole::TEvConfigsDispatcher::TEvSetConfigSubscriptionResponse, Handle);
13511355
hFunc(NConsole::TEvConsole::TEvConfigNotificationRequest, Handle);
1356+
hFunc(TEvKqp::TEvProxyPingRequest, Handle);
13521357
hFunc(TEvKqp::TEvQueryRequest, Handle);
13531358
hFunc(TEvKqp::TEvScriptRequest, Handle);
13541359
hFunc(TEvKqp::TEvCloseSessionRequest, Handle);

‎ydb/public/api/grpc/ya.make‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ SRCS(
1010
ydb_auth_v1.proto
1111
ydb_cms_v1.proto
1212
ydb_coordination_v1.proto
13+
ydb_debug_v1.proto
1314
ydb_discovery_v1.proto
1415
ydb_export_v1.proto
1516
ydb_import_v1.proto

‎ydb/public/api/grpc/ydb_debug_v1.proto‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
syntax = "proto3";
2+
3+
package Ydb.Debug.V1;
4+
option java_package = "com.yandex.ydb.debug.v1";
5+
6+
import "ydb/public/api/protos/ydb_debug.proto";
7+
8+
service DebugService {
9+
rpc PingPlainGrpc(Debug.PlainGrpcRequest) returns (Debug.PlainGrpcResponse);
10+
rpc PingGrpcProxy(Debug.GrpcProxyRequest) returns (Debug.GrpcProxyResponse);
11+
rpc PingKqpProxy(Debug.KqpProxyRequest) returns (Debug.KqpProxyResponse);
12+
rpc PingSchemeCache(Debug.SchemeCacheRequest) returns (Debug.SchemeCacheResponse);
13+
rpc PingTxProxy(Debug.TxProxyRequest) returns (Debug.TxProxyResponse);
14+
}

0 commit comments

Comments
 (0)