Skip to content

Commit e5a8411

Browse files
adameatadameat
authored
move redirects to base class (#8934)
1 parent 4a8eeda commit e5a8411

File tree

11 files changed

+274
-375
lines changed

11 files changed

+274
-375
lines changed

‎ydb/core/viewer/json_pipe_req.cpp‎

Lines changed: 98 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -93,26 +93,34 @@ void TViewerPipeClient::SendDelayedRequests() {
9393
}
9494
}
9595

96-
TPathId TViewerPipeClient::GetPathId(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
97-
if (ev->Get()->Request->ResultSet.size() == 1) {
98-
if (ev->Get()->Request->ResultSet.begin()->Self) {
99-
const auto& info = ev->Get()->Request->ResultSet.begin()->Self->Info;
96+
TPathId TViewerPipeClient::GetPathId(const TEvTxProxySchemeCache::TEvNavigateKeySetResult& ev) {
97+
if (ev.Request->ResultSet.size() == 1) {
98+
if (ev.Request->ResultSet.begin()->Self) {
99+
const auto& info = ev.Request->ResultSet.begin()->Self->Info;
100100
return TPathId(info.GetSchemeshardId(), info.GetPathId());
101101
}
102-
if (ev->Get()->Request->ResultSet.begin()->TableId) {
103-
return ev->Get()->Request->ResultSet.begin()->TableId.PathId;
102+
if (ev.Request->ResultSet.begin()->TableId) {
103+
return ev.Request->ResultSet.begin()->TableId.PathId;
104104
}
105105
}
106106
return {};
107107
}
108108

109-
TString TViewerPipeClient::GetPath(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
110-
if (ev->Get()->Request->ResultSet.size() == 1) {
111-
return CanonizePath(ev->Get()->Request->ResultSet.begin()->Path);
109+
TString TViewerPipeClient::GetPath(const TEvTxProxySchemeCache::TEvNavigateKeySetResult& ev) {
110+
if (ev.Request->ResultSet.size() == 1) {
111+
return CanonizePath(ev.Request->ResultSet.begin()->Path);
112112
}
113113
return {};
114114
}
115115

116+
TPathId TViewerPipeClient::GetPathId(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
117+
return GetPathId(*ev->Get());
118+
}
119+
120+
TString TViewerPipeClient::GetPath(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
121+
return GetPath(*ev->Get());
122+
}
123+
116124
bool TViewerPipeClient::IsSuccess(const std::unique_ptr<TEvTxProxySchemeCache::TEvNavigateKeySetResult>& ev) {
117125
return (ev->Request->ResultSet.size() == 1) && (ev->Request->ResultSet.begin()->Status == NSchemeCache::TSchemeCacheNavigate::EStatus::Ok);
118126
}
@@ -147,6 +155,23 @@ TString TViewerPipeClient::GetError(const std::unique_ptr<TEvTxProxySchemeCache:
147155
}
148156
}
149157

158+
bool TViewerPipeClient::IsSuccess(const std::unique_ptr<TEvStateStorage::TEvBoardInfo>& ev) {
159+
return ev->Status == TEvStateStorage::TEvBoardInfo::EStatus::Ok;
160+
}
161+
162+
TString TViewerPipeClient::GetError(const std::unique_ptr<TEvStateStorage::TEvBoardInfo>& ev) {
163+
switch (ev->Status) {
164+
case TEvStateStorage::TEvBoardInfo::EStatus::Unknown:
165+
return "Unknown";
166+
case TEvStateStorage::TEvBoardInfo::EStatus::Ok:
167+
return "Ok";
168+
case TEvStateStorage::TEvBoardInfo::EStatus::NotAvailable:
169+
return "NotAvailable";
170+
default:
171+
return ::ToString(static_cast<int>(ev->Status));
172+
}
173+
}
174+
150175
void TViewerPipeClient::RequestHiveDomainStats(NNodeWhiteboard::TTabletId hiveId) {
151176
TActorId pipeClient = ConnectTabletPipe(hiveId);
152177
THolder<TEvHive::TEvRequestHiveDomainStats> request = MakeHolder<TEvHive::TEvRequestHiveDomainStats>();
@@ -540,10 +565,10 @@ void TViewerPipeClient::RequestStateStorageMetadataCacheEndpointsLookup(const TS
540565
++Requests;
541566
}
542567

543-
std::vector<TNodeId> TViewerPipeClient::GetNodesFromBoardReply(TEvStateStorage::TEvBoardInfo::TPtr& ev) {
568+
std::vector<TNodeId> TViewerPipeClient::GetNodesFromBoardReply(const TEvStateStorage::TEvBoardInfo& ev) {
544569
std::vector<TNodeId> databaseNodes;
545-
if (ev->Get()->Status == TEvStateStorage::TEvBoardInfo::EStatus::Ok) {
546-
for (const auto& [actorId, infoEntry] : ev->Get()->InfoEntries) {
570+
if (ev.Status == TEvStateStorage::TEvBoardInfo::EStatus::Ok) {
571+
for (const auto& [actorId, infoEntry] : ev.InfoEntries) {
547572
databaseNodes.emplace_back(actorId.NodeId());
548573
}
549574
}
@@ -552,11 +577,20 @@ std::vector<TNodeId> TViewerPipeClient::GetNodesFromBoardReply(TEvStateStorage::
552577
return databaseNodes;
553578
}
554579

580+
std::vector<TNodeId> TViewerPipeClient::GetNodesFromBoardReply(TEvStateStorage::TEvBoardInfo::TPtr& ev) {
581+
return GetNodesFromBoardReply(*ev->Get());
582+
}
583+
555584
void TViewerPipeClient::InitConfig(const TCgiParameters& params) {
556585
Followers = FromStringWithDefault(params.Get("followers"), Followers);
557586
Metrics = FromStringWithDefault(params.Get("metrics"), Metrics);
558587
WithRetry = FromStringWithDefault(params.Get("with_retry"), WithRetry);
559588
MaxRequestsInFlight = FromStringWithDefault(params.Get("max_requests_in_flight"), MaxRequestsInFlight);
589+
Database = params.Get("database");
590+
if (!Database) {
591+
Database = params.Get("tenant");
592+
}
593+
Direct = FromStringWithDefault<bool>(params.Get("direct"), Direct);
560594
}
561595

562596
void TViewerPipeClient::InitConfig(const TRequestSettings& settings) {
@@ -653,23 +687,43 @@ void TViewerPipeClient::Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev) {
653687
}
654688
}
655689

690+
void TViewerPipeClient::HandleResolveResource(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
691+
ResourceNavigateResponse->Set(std::move(ev));
692+
if (ResourceNavigateResponse->IsOk()) {
693+
TSchemeCacheNavigate::TEntry& entry(ResourceNavigateResponse->Get()->Request->ResultSet.front());
694+
SharedDatabase = CanonizePath(entry.Path);
695+
if (SharedDatabase == AppData()->TenantName) {
696+
Direct = true;
697+
return Bootstrap(); // retry bootstrap without redirect this time
698+
}
699+
DatabaseBoardInfoResponse = MakeRequestStateStorageEndpointsLookup(SharedDatabase);
700+
} else {
701+
ReplyAndPassAway(GetHTTPBADREQUEST("text/plain", "Failed to resolve database - shared database not found"));
702+
}
703+
}
704+
656705
void TViewerPipeClient::HandleResolveDatabase(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
657-
if (ev->Get()->Request->ResultSet.size() == 1 && ev->Get()->Request->ResultSet.begin()->Status == NSchemeCache::TSchemeCacheNavigate::EStatus::Ok) {
658-
TSchemeCacheNavigate::TEntry& entry(ev->Get()->Request->ResultSet.front());
659-
if (entry.DomainInfo) {
660-
if (entry.DomainInfo->ResourcesDomainKey && entry.DomainInfo->DomainKey != entry.DomainInfo->ResourcesDomainKey) {
661-
RequestSchemeCacheNavigate(TPathId(entry.DomainInfo->ResourcesDomainKey));
662-
} else {
663-
RequestStateStorageEndpointsLookup(CanonizePath(entry.Path));
664-
}
706+
DatabaseNavigateResponse->Set(std::move(ev));
707+
if (DatabaseNavigateResponse->IsOk()) {
708+
TSchemeCacheNavigate::TEntry& entry(DatabaseNavigateResponse->Get()->Request->ResultSet.front());
709+
if (entry.DomainInfo && entry.DomainInfo->ResourcesDomainKey && entry.DomainInfo->DomainKey != entry.DomainInfo->ResourcesDomainKey) {
710+
ResourceNavigateResponse = MakeRequestSchemeCacheNavigate(TPathId(entry.DomainInfo->ResourcesDomainKey));
711+
Become(&TViewerPipeClient::StateResolveResource);
712+
return;
665713
}
714+
DatabaseBoardInfoResponse = MakeRequestStateStorageEndpointsLookup(CanonizePath(entry.Path));
666715
} else {
667-
ReplyAndPassAway(GetHTTPBADREQUEST("text/plain", "Failed to resolve database"));
716+
ReplyAndPassAway(GetHTTPBADREQUEST("text/plain", "Failed to resolve database - not found"));
668717
}
669718
}
670719

671-
void TViewerPipeClient::HandleResolveDatabase(TEvStateStorage::TEvBoardInfo::TPtr& ev) {
672-
ReplyAndPassAway(MakeForward(GetNodesFromBoardReply(ev)));
720+
void TViewerPipeClient::HandleResolve(TEvStateStorage::TEvBoardInfo::TPtr& ev) {
721+
DatabaseBoardInfoResponse->Set(std::move(ev));
722+
if (DatabaseBoardInfoResponse->IsOk()) {
723+
ReplyAndPassAway(MakeForward(GetNodesFromBoardReply(DatabaseBoardInfoResponse->GetRef())));
724+
} else {
725+
ReplyAndPassAway(GetHTTPBADREQUEST("text/plain", "Failed to resolve database - no nodes found"));
726+
}
673727
}
674728

675729
void TViewerPipeClient::HandleTimeout() {
@@ -678,15 +732,33 @@ void TViewerPipeClient::HandleTimeout() {
678732

679733
STATEFN(TViewerPipeClient::StateResolveDatabase) {
680734
switch (ev->GetTypeRewrite()) {
681-
hFunc(TEvStateStorage::TEvBoardInfo, HandleResolveDatabase);
735+
hFunc(TEvStateStorage::TEvBoardInfo, HandleResolve);
682736
hFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, HandleResolveDatabase);
683737
cFunc(TEvents::TEvWakeup::EventType, HandleTimeout);
684738
}
685739
}
686740

741+
STATEFN(TViewerPipeClient::StateResolveResource) {
742+
switch (ev->GetTypeRewrite()) {
743+
hFunc(TEvStateStorage::TEvBoardInfo, HandleResolve);
744+
hFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, HandleResolveResource);
745+
cFunc(TEvents::TEvWakeup::EventType, HandleTimeout);
746+
}
747+
}
748+
687749
void TViewerPipeClient::RedirectToDatabase(const TString& database) {
688-
RequestSchemeCacheNavigate(database);
689-
Become(&TViewerPipeClient::StateResolveDatabase, TDuration::MilliSeconds(1000), new TEvents::TEvWakeup());
750+
DatabaseNavigateResponse = MakeRequestSchemeCacheNavigate(database);
751+
Become(&TViewerPipeClient::StateResolveDatabase);
752+
}
753+
754+
bool TViewerPipeClient::NeedToRedirect() {
755+
Direct |= !Event->Get()->Request.GetHeader("X-Forwarded-From-Node").empty(); // we're already forwarding
756+
Direct |= (Database == AppData()->TenantName) || Database.empty(); // we're already on the right node or don't use database filter
757+
if (Database && !Direct) {
758+
RedirectToDatabase(Database); // to find some dynamic node and redirect query there
759+
return true;
760+
}
761+
return false;
690762
}
691763

692764
void TViewerPipeClient::PassAway() {

‎ydb/core/viewer/json_pipe_req.h‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,9 @@ class TViewerPipeClient : public TActorBootstrapped<TViewerPipeClient> {
3737
bool Followers = true;
3838
bool Metrics = true;
3939
bool WithRetry = true;
40+
TString Database;
41+
TString SharedDatabase;
42+
bool Direct = false;
4043
ui32 Requests = 0;
4144
ui32 MaxRequestsInFlight = 200;
4245
NWilson::TSpan Span;
@@ -158,6 +161,10 @@ class TViewerPipeClient : public TActorBootstrapped<TViewerPipeClient> {
158161
}
159162
};
160163

164+
std::optional<TRequestResponse<TEvTxProxySchemeCache::TEvNavigateKeySetResult>> DatabaseNavigateResponse;
165+
std::optional<TRequestResponse<TEvTxProxySchemeCache::TEvNavigateKeySetResult>> ResourceNavigateResponse;
166+
std::optional<TRequestResponse<TEvStateStorage::TEvBoardInfo>> DatabaseBoardInfoResponse;
167+
161168
NTabletPipe::TClientConfig GetPipeClientConfig();
162169

163170
~TViewerPipeClient();
@@ -211,12 +218,18 @@ class TViewerPipeClient : public TActorBootstrapped<TViewerPipeClient> {
211218
return MakeBSControllerID();
212219
}
213220

221+
static TPathId GetPathId(const TEvTxProxySchemeCache::TEvNavigateKeySetResult& ev);
222+
static TString GetPath(const TEvTxProxySchemeCache::TEvNavigateKeySetResult& ev);
223+
214224
static TPathId GetPathId(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev);
215225
static TString GetPath(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev);
216226

217227
static bool IsSuccess(const std::unique_ptr<TEvTxProxySchemeCache::TEvNavigateKeySetResult>& ev);
218228
static TString GetError(const std::unique_ptr<TEvTxProxySchemeCache::TEvNavigateKeySetResult>& ev);
219229

230+
static bool IsSuccess(const std::unique_ptr<TEvStateStorage::TEvBoardInfo>& ev);
231+
static TString GetError(const std::unique_ptr<TEvStateStorage::TEvBoardInfo>& ev);
232+
220233
TRequestResponse<TEvHive::TEvResponseHiveDomainStats> MakeRequestHiveDomainStats(TTabletId hiveId);
221234
TRequestResponse<TEvHive::TEvResponseHiveStorageStats> MakeRequestHiveStorageStats(TTabletId hiveId);
222235
TRequestResponse<TEvHive::TEvResponseHiveNodeStats> MakeRequestHiveNodeStats(TTabletId hiveId, TEvHive::TEvRequestHiveNodeStats* request);
@@ -252,6 +265,7 @@ class TViewerPipeClient : public TActorBootstrapped<TViewerPipeClient> {
252265
void RequestStateStorageMetadataCacheEndpointsLookup(const TString& path);
253266
TRequestResponse<TEvStateStorage::TEvBoardInfo> MakeRequestStateStorageEndpointsLookup(const TString& path, ui64 cookie = 0);
254267
std::vector<TNodeId> GetNodesFromBoardReply(TEvStateStorage::TEvBoardInfo::TPtr& ev);
268+
std::vector<TNodeId> GetNodesFromBoardReply(const TEvStateStorage::TEvBoardInfo& ev);
255269
void InitConfig(const TCgiParameters& params);
256270
void InitConfig(const TRequestSettings& settings);
257271
void ClosePipes();
@@ -285,9 +299,12 @@ class TViewerPipeClient : public TActorBootstrapped<TViewerPipeClient> {
285299
void AddEvent(const TString& name);
286300
void Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev);
287301
void HandleResolveDatabase(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev);
288-
void HandleResolveDatabase(TEvStateStorage::TEvBoardInfo::TPtr& ev);
302+
void HandleResolveResource(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev);
303+
void HandleResolve(TEvStateStorage::TEvBoardInfo::TPtr& ev);
289304
STATEFN(StateResolveDatabase);
305+
STATEFN(StateResolveResource);
290306
void RedirectToDatabase(const TString& database);
307+
bool NeedToRedirect();
291308
void HandleTimeout();
292309
void PassAway() override;
293310
};

‎ydb/core/viewer/storage_groups.h‎

Lines changed: 42 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,6 @@ class TStorageGroups : public TViewerPipeClient {
122122
using TFieldsType = std::bitset<+EGroupFields::COUNT>;
123123

124124
// Common
125-
std::optional<TRequestResponse<TEvTxProxySchemeCache::TEvNavigateKeySetResult>> DatabaseNavigateResult;
126125
std::unordered_map<TPathId, TRequestResponse<TEvTxProxySchemeCache::TEvNavigateKeySetResult>> NavigateKeySetResult;
127126
std::unordered_map<TPathId, TTabletId> PathId2HiveId;
128127
std::unordered_map<TTabletId, TRequestResponse<TEvHive::TEvResponseHiveStorageStats>> HiveStorageStats;
@@ -145,8 +144,6 @@ class TStorageGroups : public TViewerPipeClient {
145144
ui64 PDiskStateRequestsInFlight = 0;
146145

147146
ui32 Timeout = 0;
148-
TString Database;
149-
bool Direct = false;
150147
TString Filter;
151148
std::unordered_set<TString> DatabaseStoragePools;
152149
std::unordered_set<TString> FilterStoragePools;
@@ -610,18 +607,11 @@ class TStorageGroups : public TViewerPipeClient {
610607
: TBase(viewer, ev)
611608
{
612609
const auto& params(Event->Get()->Request.GetParams());
613-
InitConfig(params);
614610
Timeout = FromStringWithDefault<ui32>(params.Get("timeout"), 10000);
615-
Database = params.Get("tenant");
616-
if (Database.empty()) {
617-
Database = params.Get("database");
618-
}
619611
if (!Database.empty()) {
620612
FieldsRequired.set(+EGroupFields::PoolName);
621613
NeedFilter = true;
622614
}
623-
Direct = FromStringWithDefault<bool>(params.Get("direct"), Direct);
624-
625615
FieldsRequired.set(+EGroupFields::GroupId);
626616
TString filterStoragePool = params.Get("pool");
627617
if (!filterStoragePool.empty()) {
@@ -726,32 +716,35 @@ class TStorageGroups : public TViewerPipeClient {
726716

727717
public:
728718
void Bootstrap() override {
729-
Direct |= !TBase::Event->Get()->Request.GetHeader("X-Forwarded-From-Node").empty(); // we're already forwarding
730-
Direct |= (Database == AppData()->TenantName) || Database.empty(); // we're already on the right node or don't use database filter
731-
732-
if (Database && !Direct) {
733-
return RedirectToDatabase(Database); // to find some dynamic node and redirect query there
719+
if (TBase::NeedToRedirect()) {
720+
return;
721+
}
722+
if (Database) {
723+
if (!DatabaseNavigateResponse) {
724+
DatabaseNavigateResponse = MakeRequestSchemeCacheNavigate(Database, 0);
725+
} else {
726+
auto pathId = GetPathId(DatabaseNavigateResponse->GetRef());
727+
auto result = NavigateKeySetResult.emplace(pathId, std::move(*DatabaseNavigateResponse));
728+
ProcessNavigate(result.first->second, true);
729+
}
730+
}
731+
if (FallbackToWhiteboard) {
732+
RequestWhiteboard();
734733
} else {
735-
if (Database) {
736-
DatabaseNavigateResult = MakeRequestSchemeCacheNavigate(Database, 0);
734+
if (FieldsNeeded(FieldsBsGroups)) {
735+
GetGroupsResponse = RequestBSControllerGroups();
737736
}
738-
if (FallbackToWhiteboard) {
739-
RequestWhiteboard();
740-
} else {
741-
if (FieldsNeeded(FieldsBsGroups)) {
742-
GetGroupsResponse = RequestBSControllerGroups();
743-
}
744-
if (FieldsNeeded(FieldsBsPools)) {
745-
GetStoragePoolsResponse = RequestBSControllerPools();
746-
}
747-
if (FieldsNeeded(FieldsBsVSlots)) {
748-
GetVSlotsResponse = RequestBSControllerVSlots();
749-
}
750-
if (FieldsNeeded(FieldsBsPDisks)) {
751-
GetPDisksResponse = RequestBSControllerPDisks();
752-
}
737+
if (FieldsNeeded(FieldsBsPools)) {
738+
GetStoragePoolsResponse = RequestBSControllerPools();
739+
}
740+
if (FieldsNeeded(FieldsBsVSlots)) {
741+
GetVSlotsResponse = RequestBSControllerVSlots();
742+
}
743+
if (FieldsNeeded(FieldsBsPDisks)) {
744+
GetPDisksResponse = RequestBSControllerPDisks();
753745
}
754746
}
747+
755748
if (Requests == 0) {
756749
return ReplyAndPassAway();
757750
}
@@ -1100,19 +1093,7 @@ class TStorageGroups : public TViewerPipeClient {
11001093
}
11011094
}
11021095

1103-
void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
1104-
bool firstNavigate = (ev->Cookie == 0);
1105-
TPathId pathId = GetPathId(ev);
1106-
if (firstNavigate && DatabaseNavigateResult.has_value() && pathId) {
1107-
NavigateKeySetResult.emplace(pathId, std::move(*DatabaseNavigateResult));
1108-
}
1109-
auto itNavigateKeySetResult = NavigateKeySetResult.find(pathId);
1110-
if (itNavigateKeySetResult == NavigateKeySetResult.end()) {
1111-
BLOG_W("Invalid NavigateKeySetResult PathId: " << pathId << " Path: " << CanonizePath(ev->Get()->Request->ResultSet.begin()->Path));
1112-
return RequestDone();
1113-
}
1114-
auto& navigateResult(itNavigateKeySetResult->second);
1115-
navigateResult.Set(std::move(ev));
1096+
void ProcessNavigate(TRequestResponse<TEvTxProxySchemeCache::TEvNavigateKeySetResult>& navigateResult, bool firstNavigate) {
11161097
if (navigateResult.IsOk()) {
11171098
TString path = CanonizePath(navigateResult->Request->ResultSet.begin()->Path);
11181099
TIntrusiveConstPtr<TSchemeCacheNavigate::TDomainDescription> domainDescription = navigateResult->Request->ResultSet.begin()->DomainDescription;
@@ -1134,6 +1115,22 @@ class TStorageGroups : public TViewerPipeClient {
11341115
}
11351116
}
11361117
}
1118+
}
1119+
1120+
void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
1121+
bool firstNavigate = (ev->Cookie == 0);
1122+
TPathId pathId = GetPathId(ev);
1123+
if (firstNavigate && DatabaseNavigateResponse && pathId) {
1124+
NavigateKeySetResult.emplace(pathId, std::move(*DatabaseNavigateResponse));
1125+
}
1126+
auto itNavigateKeySetResult = NavigateKeySetResult.find(pathId);
1127+
if (itNavigateKeySetResult == NavigateKeySetResult.end()) {
1128+
BLOG_W("Invalid NavigateKeySetResult PathId: " << pathId << " Path: " << CanonizePath(ev->Get()->Request->ResultSet.begin()->Path));
1129+
return RequestDone();
1130+
}
1131+
auto& navigateResult(itNavigateKeySetResult->second);
1132+
navigateResult.Set(std::move(ev));
1133+
ProcessNavigate(navigateResult, firstNavigate);
11371134
RequestDone();
11381135
}
11391136

0 commit comments

Comments
 (0)