optimize nodes handler (#8111)

Author: adameat

ydb/core/base/board_lookup.cpp

@@ -25,6 +25,7 @@ namespace NKikimr {
 class TBoardLookupActor : public TActorBootstrapped<TBoardLookupActor> {
     const TString Path;
     const TActorId Owner;
+    const ui64 Cookie;
     const EBoardLookupMode Mode;
     const bool Subscriber;
     TBoardRetrySettings BoardRetrySettings;
@@ -111,12 +112,12 @@ class TBoardLookupActor : public TActorBootstrapped<TBoardLookupActor> {
     void NotAvailable() {
         if (CurrentStateFunc() != &TThis::StateSubscribe) {
             Send(Owner, new TEvStateStorage::TEvBoardInfo(
-                TEvStateStorage::TEvBoardInfo::EStatus::NotAvailable, Path));
+                TEvStateStorage::TEvBoardInfo::EStatus::NotAvailable, Path), 0, Cookie);
         } else {
             Send(Owner,
                 new TEvStateStorage::TEvBoardInfoUpdate(
                     TEvStateStorage::TEvBoardInfo::EStatus::NotAvailable, Path
-                )
+                ), 0, Cookie
             );
         }
         return PassAway();
@@ -129,7 +130,7 @@ class TBoardLookupActor : public TActorBootstrapped<TBoardLookupActor> {
                 auto reply = MakeHolder<TEvStateStorage::TEvBoardInfo>(
                     TEvStateStorage::TEvBoardInfo::EStatus::Ok, Path);
                 reply->InfoEntries = std::move(Info);
-                Send(Owner, std::move(reply));
+                Send(Owner, std::move(reply), 0, Cookie);
                 if (Subscriber) {
                     Become(&TThis::StateSubscribe);
                     return;
@@ -240,7 +241,7 @@ class TBoardLookupActor : public TActorBootstrapped<TBoardLookupActor> {
                 auto reply = MakeHolder<TEvStateStorage::TEvBoardInfoUpdate>(
                     TEvStateStorage::TEvBoardInfo::EStatus::Ok, Path);
                 reply->Updates = { { oid, std::move(update.value()) } };
-                Send(Owner, std::move(reply));
+                Send(Owner, std::move(reply), 0, Cookie);
             }
         } else {
             if (info.GetDropped()) {
@@ -308,7 +309,7 @@ class TBoardLookupActor : public TActorBootstrapped<TBoardLookupActor> {
                 auto reply = MakeHolder<TEvStateStorage::TEvBoardInfoUpdate>(
                     TEvStateStorage::TEvBoardInfo::EStatus::Ok, Path);
                 reply->Updates = std::move(updates);
-                Send(Owner, std::move(reply));
+                Send(Owner, std::move(reply), 0, Cookie);
             }
         }
 
@@ -484,7 +485,7 @@ class TBoardLookupActor : public TActorBootstrapped<TBoardLookupActor> {
             auto reply = MakeHolder<TEvStateStorage::TEvBoardInfoUpdate>(
                 TEvStateStorage::TEvBoardInfo::EStatus::Ok, Path);
             reply->Updates = std::move(updates);
-            Send(Owner, std::move(reply));
+            Send(Owner, std::move(reply), 0, Cookie);
         }
     }
 
@@ -495,9 +496,10 @@ class TBoardLookupActor : public TActorBootstrapped<TBoardLookupActor> {
 
     TBoardLookupActor(
         const TString &path, TActorId owner, EBoardLookupMode mode,
-        TBoardRetrySettings boardRetrySettings)
+        TBoardRetrySettings boardRetrySettings, ui64 cookie = 0)
         : Path(path)
         , Owner(owner)
+        , Cookie(cookie)
         , Mode(mode)
         , Subscriber(Mode == EBoardLookupMode::Subscription)
         , BoardRetrySettings(std::move(boardRetrySettings))
@@ -545,8 +547,8 @@ class TBoardLookupActor : public TActorBootstrapped<TBoardLookupActor> {
 
 IActor* CreateBoardLookupActor(
         const TString &path, const TActorId &owner, EBoardLookupMode mode,
-        TBoardRetrySettings boardRetrySettings) {
-    return new TBoardLookupActor(path, owner, mode, std::move(boardRetrySettings));
+        TBoardRetrySettings boardRetrySettings, ui64 cookie) {
+    return new TBoardLookupActor(path, owner, mode, std::move(boardRetrySettings), cookie);
 }
 
 }

ydb/core/base/statestorage.h

@@ -547,7 +547,7 @@ IActor* CreateStateStorageBoardReplica(const TIntrusivePtr<TStateStorageInfo> &,
 IActor* CreateSchemeBoardReplica(const TIntrusivePtr<TStateStorageInfo>&, ui32);
 IActor* CreateBoardLookupActor(
     const TString &path, const TActorId &owner, EBoardLookupMode mode,
-    TBoardRetrySettings boardRetrySettings = {});
+    TBoardRetrySettings boardRetrySettings = {}, ui64 cookie = 0);
 IActor* CreateBoardPublishActor(
     const TString &path, const TString &payload, const TActorId &owner, ui32 ttlMs, bool reg,
     TBoardRetrySettings boardRetrySettings = {});

ydb/core/mon/async_http_mon.cpp

@@ -554,10 +554,26 @@ class THttpMonServiceNodeRequest : public TActorBootstrapped<THttpMonServiceNode
         }
     }
 
+    TString RewriteWithForwardedFromNode(const TString& response) {
+        NHttp::THttpParser<NHttp::THttpRequest, NHttp::TSocketBuffer> parser(response);
+
+        NHttp::THeadersBuilder headers(parser.Headers);
+        headers.Set("X-Forwarded-From-Node", TStringBuilder() << Event->Sender.NodeId());
+
+        NHttp::THttpRenderer<NHttp::THttpRequest, NHttp::TSocketBuffer> renderer;
+        renderer.InitRequest(parser.Method, parser.URL, parser.Protocol, parser.Version);
+        renderer.Set(headers);
+        if (parser.HaveBody()) {
+            renderer.SetBody(parser.Body); // it shouldn't be here, 30x with a body is a bad idea
+        }
+        renderer.Finish();
+        return renderer.AsString();
+    }
+
     void Bootstrap() {
         NHttp::THttpConfig::SocketAddressType address;
         FromProto(address, Event->Get()->Record.GetAddress());
-        NHttp::THttpIncomingRequestPtr request = new NHttp::THttpIncomingRequest(Event->Get()->Record.GetHttpRequest(), Endpoint, address);
+        NHttp::THttpIncomingRequestPtr request = new NHttp::THttpIncomingRequest(RewriteWithForwardedFromNode(Event->Get()->Record.GetHttpRequest()), Endpoint, address);
         TStringBuilder prefix;
         prefix << "/node/" << TActivationContext::ActorSystem()->NodeId;
         if (request->URL.SkipPrefix(prefix)) {

ydb/core/node_whiteboard/node_whiteboard.h

@@ -375,12 +375,6 @@ struct TEvWhiteboard{
 
         TEvSystemStateUpdate(const TNodeLocation& systemLocation) {
             systemLocation.Serialize(Record.MutableLocation(), false);
-            const auto& x = systemLocation.GetLegacyValue();
-            auto *pb = Record.MutableSystemLocation();
-            pb->SetDataCenter(x.DataCenter);
-            pb->SetRoom(x.Room);
-            pb->SetRack(x.Rack);
-            pb->SetBody(x.Body);
         }
 
         TEvSystemStateUpdate(const NKikimrWhiteboard::TSystemStateInfo& systemStateInfo) {
@@ -503,5 +497,38 @@ inline TActorId MakeNodeWhiteboardServiceId(ui32 node) {
 
 IActor* CreateNodeWhiteboardService();
 
-} // NTabletState
+template<typename TRequestType>
+struct WhiteboardResponse {};
+
+template<>
+struct WhiteboardResponse<TEvWhiteboard::TEvTabletStateRequest> {
+    using Type = TEvWhiteboard::TEvTabletStateResponse;
+};
+
+template<>
+struct WhiteboardResponse<TEvWhiteboard::TEvPDiskStateRequest> {
+    using Type = TEvWhiteboard::TEvPDiskStateResponse;
+};
+
+template<>
+struct WhiteboardResponse<TEvWhiteboard::TEvVDiskStateRequest> {
+    using Type = TEvWhiteboard::TEvVDiskStateResponse;
+};
+
+template<>
+struct WhiteboardResponse<TEvWhiteboard::TEvSystemStateRequest> {
+    using Type = TEvWhiteboard::TEvSystemStateResponse;
+};
+
+template<>
+struct WhiteboardResponse<TEvWhiteboard::TEvBSGroupStateRequest> {
+    using Type = TEvWhiteboard::TEvBSGroupStateResponse;
+};
+
+template<>
+struct WhiteboardResponse<TEvWhiteboard::TEvNodeStateRequest> {
+    using Type = TEvWhiteboard::TEvNodeStateResponse;
+};
+
+} // NNodeWhiteboard
 } // NKikimr

ydb/core/testlib/test_client.cpp

@@ -606,7 +606,12 @@ namespace Tests {
 
         NKikimrBlobStorage::TDefineHostConfig hostConfig;
         hostConfig.SetHostConfigId(nodeId);
-        TString path = TStringBuilder() << Runtime->GetTempDir() << "pdisk_1.dat";
+        TString path;
+        if (Settings->UseSectorMap) {
+            path ="SectorMap:test-client[:2000]";
+        } else {
+            path = TStringBuilder() << Runtime->GetTempDir() << "pdisk_1.dat";
+        }
         hostConfig.AddDrive()->SetPath(path);
         if (Settings->Verbose) {
             Cerr << "test_client.cpp: SetPath # " << path << Endl;

ydb/core/testlib/test_client.h

@@ -159,6 +159,7 @@ namespace Tests {
         bool InitializeFederatedQuerySetupFactory = false;
         TString ServerCertFilePath;
         bool Verbose = true;
+        bool UseSectorMap = false;
 
         std::function<IActor*(const TTicketParserSettings&)> CreateTicketParser = NKikimr::CreateTicketParser;
         std::shared_ptr<TGrpcServiceFactory> GrpcServiceFactory;
@@ -211,6 +212,7 @@ namespace Tests {
         TServerSettings& SetYtGateway(NYql::IYtGateway::TPtr ytGateway) { YtGateway = std::move(ytGateway); return *this; }
         TServerSettings& SetInitializeFederatedQuerySetupFactory(bool value) { InitializeFederatedQuerySetupFactory = value; return *this; }
         TServerSettings& SetVerbose(bool value) { Verbose = value; return *this; }
+        TServerSettings& SetUseSectorMap(bool value) { UseSectorMap = value; return *this; }
         TServerSettings& SetPersQueueGetReadSessionsInfoWorkerFactory(
             std::shared_ptr<NKikimr::NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> factory
         ) {

ydb/core/viewer/json_handlers_viewer.cpp

@@ -243,7 +243,7 @@ void InitViewerHealthCheckJsonHandler(TJsonHandlers& handlers) {
 }
 
 void InitViewerNodesJsonHandler(TJsonHandlers& handlers) {
-    handlers.AddHandler("/viewer/nodes", new TJsonHandler<TJsonNodes>(TJsonNodes::GetSwagger()));
+    handlers.AddHandler("/viewer/nodes", new TJsonHandler<TJsonNodes>(TJsonNodes::GetSwagger()), 2);
 }
 
 void InitViewerACLJsonHandler(TJsonHandlers &jsonHandlers) {

ydb/core/viewer/json_pipe_req.cpp

@@ -15,6 +15,12 @@ TViewerPipeClient::~TViewerPipeClient() = default;
 
 TViewerPipeClient::TViewerPipeClient() = default;
 
+TViewerPipeClient::TViewerPipeClient(NWilson::TTraceId traceId) {
+    if (traceId) {
+        Span = {TComponentTracingLevels::THttp::TopLevel, std::move(traceId), "viewer", NWilson::EFlags::AUTO_END};
+    }
+}
+
 TViewerPipeClient::TViewerPipeClient(IViewer* viewer, NMon::TEvHttpInfo::TPtr& ev)
     : Viewer(viewer)
     , Event(ev)
@@ -133,6 +139,49 @@ void TViewerPipeClient::RequestHiveStorageStats(NNodeWhiteboard::TTabletId hiveI
     SendRequestToPipe(pipeClient, request.Release(), hiveId);
 }
 
+TViewerPipeClient::TRequestResponse<TEvViewer::TEvViewerResponse> TViewerPipeClient::MakeViewerRequest(TNodeId nodeId, TEvViewer::TEvViewerRequest* ev, ui32 flags) {
+    TActorId viewerServiceId = MakeViewerID(nodeId);
+    TRequestResponse<TEvViewer::TEvViewerResponse> response(Span.CreateChild(TComponentTracingLevels::THttp::Detailed, TypeName(*ev)));
+    if (response.Span) {
+        response.Span.Attribute("target_node_id", nodeId);
+        TStringBuilder askFor;
+        askFor << ev->Record.GetLocation().NodeIdSize() << " nodes (";
+        for (size_t i = 0; i < std::min<size_t>(ev->Record.GetLocation().NodeIdSize(), 16); ++i) {
+            if (i) {
+                askFor << ", ";
+            }
+            askFor << ev->Record.GetLocation().GetNodeId(i);
+        }
+        if (ev->Record.GetLocation().NodeIdSize() > 16) {
+            askFor << ", ...";
+        }
+        askFor << ")";
+        response.Span.Attribute("ask_for", askFor);
+        switch (ev->Record.Request_case()) {
+            case NKikimrViewer::TEvViewerRequest::kTabletRequest:
+                response.Span.Attribute("request_type", "TabletRequest");
+                break;
+            case NKikimrViewer::TEvViewerRequest::kSystemRequest:
+                response.Span.Attribute("request_type", "SystemRequest");
+                break;
+            case NKikimrViewer::TEvViewerRequest::kQueryRequest:
+                response.Span.Attribute("request_type", "QueryRequest");
+                break;
+            case NKikimrViewer::TEvViewerRequest::kRenderRequest:
+                response.Span.Attribute("request_type", "RenderRequest");
+                break;
+            case NKikimrViewer::TEvViewerRequest::kAutocompleteRequest:
+                response.Span.Attribute("request_type", "AutocompleteRequest");
+                break;
+            default:
+                response.Span.Attribute("request_type", ::ToString(static_cast<int>(ev->Record.Request_case())));
+                break;
+        }
+    }
+    SendRequest(viewerServiceId, ev, flags, nodeId, response.Span.GetTraceId());
+    return response;
+}
+
 TViewerPipeClient::TRequestResponse<TEvHive::TEvResponseHiveDomainStats> TViewerPipeClient::MakeRequestHiveDomainStats(NNodeWhiteboard::TTabletId hiveId) {
     TActorId pipeClient = ConnectTabletPipe(hiveId);
     THolder<TEvHive::TEvRequestHiveDomainStats> request = MakeHolder<TEvHive::TEvRequestHiveDomainStats>();
@@ -157,6 +206,16 @@ TViewerPipeClient::TRequestResponse<TEvHive::TEvResponseHiveStorageStats> TViewe
     return response;
 }
 
+TViewerPipeClient::TRequestResponse<TEvHive::TEvResponseHiveNodeStats> TViewerPipeClient::MakeRequestHiveNodeStats(TTabletId hiveId, TEvHive::TEvRequestHiveNodeStats* request) {
+    TActorId pipeClient = ConnectTabletPipe(hiveId);
+    auto response = MakeRequestToPipe<TEvHive::TEvResponseHiveNodeStats>(pipeClient, request, hiveId);
+    if (response.Span) {
+        auto hive_id = "#" + ::ToString(hiveId);
+        response.Span.Attribute("hive_id", hive_id);
+    }
+    return response;
+}
+
 TViewerPipeClient::TRequestResponse<TEvViewer::TEvViewerResponse> TViewerPipeClient::MakeRequestViewer(TNodeId nodeId, TEvViewer::TEvViewerRequest* request, ui32 flags) {
     auto requestType = request->Record.GetRequestCase();
     auto response = MakeRequest<TEvViewer::TEvViewerResponse>(MakeViewerID(nodeId), request, flags, nodeId);
@@ -415,6 +474,18 @@ void TViewerPipeClient::RequestStateStorageEndpointsLookup(const TString& path)
     ++Requests;
 }
 
+TViewerPipeClient::TRequestResponse<TEvStateStorage::TEvBoardInfo> TViewerPipeClient::MakeRequestStateStorageEndpointsLookup(const TString& path, ui64 cookie) {
+    TRequestResponse<TEvStateStorage::TEvBoardInfo> response(Span.CreateChild(TComponentTracingLevels::THttp::Detailed, "BoardLookupActor"));
+    RegisterWithSameMailbox(CreateBoardLookupActor(MakeEndpointsBoardPath(path),
+                                                   SelfId(),
+                                                   EBoardLookupMode::Second, {}, cookie));
+    if (response.Span) {
+        response.Span.Attribute("path", path);
+    }
+    ++Requests;
+    return response;
+}
+
 void TViewerPipeClient::RequestStateStorageMetadataCacheEndpointsLookup(const TString& path) {
     if (!AppData()->DomainsInfo->Domain) {
         return;
@@ -519,6 +590,9 @@ TString TViewerPipeClient::MakeForward(const std::vector<ui32>& nodes) {
 }
 
 void TViewerPipeClient::RequestDone(ui32 requests) {
+    if (requests == 0) {
+        return;
+    }
     Requests -= requests;
     if (!DelayedRequests.empty()) {
         SendDelayedRequests();
@@ -535,13 +609,56 @@ void TViewerPipeClient::Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev) {
     }
 }
 
+void TViewerPipeClient::HandleResolveDatabase(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
+    if (ev->Get()->Request->ResultSet.size() == 1 && ev->Get()->Request->ResultSet.begin()->Status == NSchemeCache::TSchemeCacheNavigate::EStatus::Ok) {
+        TSchemeCacheNavigate::TEntry& entry(ev->Get()->Request->ResultSet.front());
+        if (entry.DomainInfo) {
+            if (entry.DomainInfo->ResourcesDomainKey && entry.DomainInfo->DomainKey != entry.DomainInfo->ResourcesDomainKey) {
+                RequestSchemeCacheNavigate(TPathId(entry.DomainInfo->ResourcesDomainKey));
+            } else {
+                RequestStateStorageEndpointsLookup(CanonizePath(entry.Path));
+            }
+        }
+    } else {
+        ReplyAndPassAway(GetHTTPBADREQUEST("text/plain", "Failed to resolve database"));
+    }
+}
+
+void TViewerPipeClient::HandleResolveDatabase(TEvStateStorage::TEvBoardInfo::TPtr& ev) {
+    ReplyAndPassAway(MakeForward(GetNodesFromBoardReply(ev)));
+}
+
 void TViewerPipeClient::HandleTimeout() {
     ReplyAndPassAway(GetHTTPGATEWAYTIMEOUT());
 }
 
+STATEFN(TViewerPipeClient::StateResolveDatabase) {
+    switch (ev->GetTypeRewrite()) {
+        hFunc(TEvStateStorage::TEvBoardInfo, HandleResolveDatabase);
+        hFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, HandleResolveDatabase);
+        cFunc(TEvents::TEvWakeup::EventType, HandleTimeout);
+    }
+}
+
+void TViewerPipeClient::RedirectToDatabase(const TString& database) {
+    RequestSchemeCacheNavigate(database);
+    Become(&TViewerPipeClient::StateResolveDatabase, TDuration::MilliSeconds(1000), new TEvents::TEvWakeup());
+}
+
 void TViewerPipeClient::PassAway() {
+    std::sort(SubscriptionNodeIds.begin(), SubscriptionNodeIds.end());
+    SubscriptionNodeIds.erase(std::unique(SubscriptionNodeIds.begin(), SubscriptionNodeIds.end()), SubscriptionNodeIds.end());
+    for (TNodeId nodeId : SubscriptionNodeIds) {
+        Send(TActivationContext::InterconnectProxy(nodeId), new TEvents::TEvUnsubscribe());
+    }
     ClosePipes();
     TBase::PassAway();
 }
 
+void TViewerPipeClient::AddEvent(const TString& name) {
+    if (Span) {
+        Span.Event(name);
+    }
+}
+
 }