Add gRPC API for recommender (#8654)

pixcc · web-flow · commit 10c000388926 · 2024-09-12T15:35:53.000+03:00
diff --git a/ydb/core/base/hive.h b/ydb/core/base/hive.h
@@ -49,6 +49,7 @@ namespace NKikimr {
             EvUpdateTabletsObject,
             EvUpdateDomain,
             EvRequestTabletDistribution,
+            EvRequestScaleRecommendation,
 
             // replies
             EvBootTabletReply = EvBootTablet + 512,
@@ -84,6 +85,7 @@ namespace NKikimr {
             EvUpdateTabletsObjectReply,
             EvUpdateDomainReply,
             EvResponseTabletDistribution,
+            EvResponseScaleRecommendation,
 
             EvEnd
         };
@@ -876,6 +878,12 @@ namespace NKikimr {
 
         struct TEvResponseTabletDistribution : TEventPB<TEvResponseTabletDistribution,
             NKikimrHive::TEvResponseTabletDistribution, EvResponseTabletDistribution> {};
+
+        struct TEvRequestScaleRecommendation : TEventPB<TEvRequestScaleRecommendation,
+            NKikimrHive::TEvRequestScaleRecommendation, EvRequestScaleRecommendation> {};
+        
+        struct TEvResponseScaleRecommendation : TEventPB<TEvResponseScaleRecommendation,
+            NKikimrHive::TEvResponseScaleRecommendation, EvResponseScaleRecommendation> {};
     };
 
     IActor* CreateDefaultHive(const TActorId &tablet, TTabletStorageInfo *info);
diff --git a/ydb/core/grpc_services/rpc_get_scale_recommendation.cpp b/ydb/core/grpc_services/rpc_get_scale_recommendation.cpp
@@ -0,0 +1,132 @@
+#include "service_cms.h"
+
+#include <ydb/core/base/path.h>
+#include <ydb/core/base/tablet_pipe.h>
+#include <ydb/core/grpc_services/base/base.h>
+#include <ydb/core/grpc_services/rpc_request_base.h>
+#include <ydb/core/mind/hive/hive.h>
+#include <ydb/core/tx/scheme_cache/scheme_cache.h>
+
+#include <ydb/public/api/protos/ydb_cms.pb.h>
+
+namespace NKikimr::NGRpcService {
+
+using TEvGetScaleRecommendationRequest = TGrpcRequestNoOperationCall<Ydb::Cms::GetScaleRecommendationRequest, Ydb::Cms::GetScaleRecommendationResponse>;
+
+class TGetScaleRecommendationRPC : public TRpcRequestActor<TGetScaleRecommendationRPC, TEvGetScaleRecommendationRequest, false> {
+public:
+    using TRpcRequestActor::TRpcRequestActor;
+
+    void Bootstrap(const TActorContext&) {
+        ResolveDatabase(GetProtoRequest()->path());
+        this->Become(&TThis::StateWork);
+    }
+
+    STATEFN(StateWork) {
+        switch (ev->GetTypeRewrite()) {
+            hFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, Handle);
+            hFunc(TEvHive::TEvResponseScaleRecommendation, Handle);
+
+            hFunc(TEvTabletPipe::TEvClientConnected, Handle);
+            hFunc(TEvTabletPipe::TEvClientDestroyed, Handle);
+        }
+    }
+
+    void ResolveDatabase(const TString& databaseName) {
+        auto request = MakeHolder<NSchemeCache::TSchemeCacheNavigate>();
+        request->DatabaseName = databaseName;
+
+        auto& entry = request->ResultSet.emplace_back();
+        entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpPath;
+        entry.Path = NKikimr::SplitPath(databaseName);
+
+        this->Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(request.Release()));
+    }
+
+    void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
+        const auto& request = ev->Get()->Request;
+
+        if (request->ResultSet.empty()) {
+            return this->Reply(Ydb::StatusIds::SCHEME_ERROR, NKikimrIssues::TIssuesIds::GENERIC_RESOLVE_ERROR);
+        }
+
+        const auto& entry = request->ResultSet.front();
+
+        if (request->ErrorCount > 0) {
+            switch (entry.Status) {
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::Ok:
+                break;
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::AccessDenied:
+                return this->Reply(Ydb::StatusIds::UNAUTHORIZED, NKikimrIssues::TIssuesIds::ACCESS_DENIED);
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::RootUnknown:
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::PathErrorUnknown:
+                return this->Reply(Ydb::StatusIds::SCHEME_ERROR, NKikimrIssues::TIssuesIds::PATH_NOT_EXIST);
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::LookupError:
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::RedirectLookupError:
+                return this->Reply(Ydb::StatusIds::UNAVAILABLE, NKikimrIssues::TIssuesIds::RESOLVE_LOOKUP_ERROR);
+            default:
+                return this->Reply(Ydb::StatusIds::SCHEME_ERROR, NKikimrIssues::TIssuesIds::GENERIC_RESOLVE_ERROR);
+            }
+        }
+
+        if (!this->CheckAccess(CanonizePath(entry.Path), entry.SecurityObject, NACLib::GenericList)) {
+            return;
+        }
+
+        auto domainInfo = entry.DomainInfo;
+        if (!domainInfo || !domainInfo->Params.HasHive()) {
+            return this->Reply(Ydb::StatusIds::INTERNAL_ERROR, NKikimrIssues::TIssuesIds::GENERIC_RESOLVE_ERROR);
+        }
+
+        SendRequest(domainInfo->DomainKey, domainInfo->Params.GetHive());
+    }
+
+    void SendRequest(TPathId domainKey, ui64 hiveId) {
+        if (!PipeClient) {
+            NTabletPipe::TClientConfig config;
+            config.RetryPolicy = {.RetryLimitCount = 3};
+            PipeClient = this->RegisterWithSameMailbox(NTabletPipe::CreateClient(this->SelfId(), hiveId, config));
+        }
+
+        auto ev = std::make_unique<TEvHive::TEvRequestScaleRecommendation>();
+        ev->Record.MutableDomainKey()->SetSchemeShard(domainKey.OwnerId);
+        ev->Record.MutableDomainKey()->SetPathId(domainKey.LocalPathId);
+        NTabletPipe::SendData(this->SelfId(), PipeClient, ev.release());
+    }
+
+    void Handle(TEvHive::TEvResponseScaleRecommendation::TPtr& ev) {
+        TResponse response;
+        ui32 recommendedNodes = ev->Get()->Record.GetRecommendedNodes();
+        response.mutable_recommended_resources()->add_computational_units()->set_count(recommendedNodes);
+        return this->Reply(response);
+    }
+
+    void Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev) {
+        if (ev->Get()->Status != NKikimrProto::OK) {
+            DeliveryProblem();
+        }
+    }
+
+    void Handle(TEvTabletPipe::TEvClientDestroyed::TPtr&) {
+        DeliveryProblem();
+    }
+
+    void DeliveryProblem() {
+        this->Reply(Ydb::StatusIds::UNAVAILABLE);
+    }
+
+    void PassAway() override {
+        NTabletPipe::CloseAndForgetClient(this->SelfId(), PipeClient);
+        IActor::PassAway();
+    }
+
+private:
+    TActorId PipeClient;
+
+};
+
+void DoGetScaleRecommendationRequest(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f) {
+    f.RegisterActor(new TGetScaleRecommendationRPC(p.release()));
+}
+
+} // namespace NKikimr::NGRpcService
diff --git a/ydb/core/grpc_services/service_cms.h b/ydb/core/grpc_services/service_cms.h
@@ -6,6 +6,7 @@ namespace NKikimr {
 namespace NGRpcService {
 
 class IRequestOpCtx;
+class IRequestNoOpCtx;
 class IFacilityProvider;
 
 void DoCreateTenantRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
@@ -14,6 +15,7 @@ void DoGetTenantStatusRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityP
 void DoListTenantsRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
 void DoRemoveTenantRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
 void DoDescribeTenantOptionsRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+void DoGetScaleRecommendationRequest(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f);
 
 }
 }
diff --git a/ydb/core/grpc_services/ya.make b/ydb/core/grpc_services/ya.make
@@ -43,6 +43,7 @@ SRCS(
     rpc_fq_internal.cpp
     rpc_fq.cpp
     rpc_get_operation.cpp
+    rpc_get_scale_recommendation.cpp
     rpc_get_shard_locations.cpp
     rpc_import.cpp
     rpc_import_data.cpp
diff --git a/ydb/core/mind/hive/hive_impl.cpp b/ydb/core/mind/hive/hive_impl.cpp
@@ -3054,6 +3054,7 @@ void THive::ProcessEvent(std::unique_ptr<IEventHandle> event) {
         hFunc(TEvPrivate::TEvDeleteNode, Handle);
         hFunc(TEvHive::TEvRequestTabletDistribution, Handle);
         hFunc(TEvPrivate::TEvUpdateDataCenterFollowers, Handle);
+        hFunc(TEvHive::TEvRequestScaleRecommendation, Handle);
     }
 }
 
@@ -3157,6 +3158,7 @@ STFUNC(THive::StateWork) {
         fFunc(TEvPrivate::TEvDeleteNode::EventType, EnqueueIncomingEvent);
         fFunc(TEvHive::TEvRequestTabletDistribution::EventType, EnqueueIncomingEvent);
         fFunc(TEvPrivate::TEvUpdateDataCenterFollowers::EventType, EnqueueIncomingEvent);
+        fFunc(TEvHive::TEvRequestScaleRecommendation::EventType, EnqueueIncomingEvent);
         hFunc(TEvPrivate::TEvProcessIncomingEvent, Handle);
     default:
         if (!HandleDefaultEvents(ev, SelfId())) {
@@ -3458,6 +3460,11 @@ void THive::Handle(TEvPrivate::TEvUpdateDataCenterFollowers::TPtr& ev) {
     Execute(CreateUpdateDcFollowers(ev->Get()->DataCenter));
 }
 
+void THive::Handle(TEvHive::TEvRequestScaleRecommendation::TPtr& ev) {
+    auto response = std::make_unique<TEvHive::TEvResponseScaleRecommendation>();
+    Send(ev->Sender, response.release());
+}
+
 TVector<TNodeId> THive::GetNodesForWhiteboardBroadcast(size_t maxNodesToReturn) {
     TVector<TNodeId> nodes;
     TNodeId selfNodeId = SelfId().NodeId();
diff --git a/ydb/core/mind/hive/hive_impl.h b/ydb/core/mind/hive/hive_impl.h
@@ -578,6 +578,7 @@ class THive : public TActor<THive>, public TTabletExecutedFlat, public THiveShar
     void Handle(TEvPrivate::TEvDeleteNode::TPtr& ev);
     void Handle(TEvHive::TEvRequestTabletDistribution::TPtr& ev);
     void Handle(TEvPrivate::TEvUpdateDataCenterFollowers::TPtr& ev);
+    void Handle(TEvHive::TEvRequestScaleRecommendation::TPtr& ev);
 
 protected:
     void RestartPipeTx(ui64 tabletId);
diff --git a/ydb/core/protos/hive.proto b/ydb/core/protos/hive.proto
@@ -589,3 +589,11 @@ message TEvResponseTabletDistribution {
     }
     repeated TNode Nodes = 1;
 }
+
+message TEvRequestScaleRecommendation {
+    optional NKikimrSubDomains.TDomainKey DomainKey = 1;
+}
+
+message TEvResponseScaleRecommendation {
+    optional uint32 RecommendedNodes = 1;
+}
diff --git a/ydb/public/api/grpc/ydb_cms_v1.proto b/ydb/public/api/grpc/ydb_cms_v1.proto
@@ -27,4 +27,7 @@ service CmsService {
 
     // Describe supported database options.
     rpc DescribeDatabaseOptions(Cms.DescribeDatabaseOptionsRequest) returns (Cms.DescribeDatabaseOptionsResponse);
+
+    // Get resources scale recommendation for database.
+    rpc GetScaleRecommendation(Cms.GetScaleRecommendationRequest) returns (Cms.GetScaleRecommendationResponse);
 }
diff --git a/ydb/public/api/protos/ydb_cms.proto b/ydb/public/api/protos/ydb_cms.proto
@@ -4,7 +4,9 @@ option cc_enable_arenas = true;
 package Ydb.Cms;
 option java_package = "com.yandex.ydb.cms";
 
+import "ydb/public/api/protos/ydb_issue_message.proto";
 import "ydb/public/api/protos/ydb_operation.proto";
+import "ydb/public/api/protos/ydb_status_codes.proto";
 
 // A set of uniform storage units.
 // Single storage unit can be thought of as a reserved part of a RAID.
@@ -104,6 +106,23 @@ message DatabaseQuotas {
     repeated StorageQuotas storage_quotas = 6;
 }
 
+// A policy that is used for resource scale recommendation. If multiple are used,
+// recommender combines them to recommend the largest scale.
+message ScaleRecommenderPolicy {
+    // Policy that tracks metric and reactively recommend to adjust resources scale
+    // to keep metric close to the specified target value.
+    message TargetTrackingPolicy {
+        oneof target {
+            // A percentage of compute resources' average CPU utilization.
+            uint32 average_cpu_utilization_percent = 1; 
+        }
+    }
+
+    oneof policy {
+        TargetTrackingPolicy target_tracking_policy = 1;
+    }
+}
+
 // Request to create a new database. For successfull creation
 // specified database shouldn't exist. At least one storage
 // unit should be requested for the database.
@@ -129,6 +148,8 @@ message CreateDatabaseRequest {
     string idempotency_key = 9;
     // Optional quotas for the database
     DatabaseQuotas database_quotas = 10;
+    // Optional scale recommender policies
+    repeated ScaleRecommenderPolicy scale_recommender_policies = 11;
 }
 
 message CreateDatabaseResponse {
@@ -179,6 +200,8 @@ message GetDatabaseStatusResult {
     SchemaOperationQuotas schema_operation_quotas = 9;
     // Current quotas for the database
     DatabaseQuotas database_quotas = 10;
+    // Current scale recommender policies
+    repeated ScaleRecommenderPolicy scale_recommender_policies = 11;
 }
 
 // Change resources allocated for database.
@@ -207,6 +230,8 @@ message AlterDatabaseRequest {
     DatabaseQuotas database_quotas = 11;
     // Alter attributes. Leave the value blank to drop an attribute.
     map<string, string> alter_attributes = 12;
+    // Alter scale recommender policies.
+    repeated ScaleRecommenderPolicy scale_recommender_policies = 13;
 }
 
 message AlterDatabaseResponse {
@@ -271,3 +296,16 @@ message DescribeDatabaseOptionsResult {
     repeated AvailabilityZoneDescription availability_zones = 2;
     repeated ComputationalUnitDescription computational_units = 3;
 }
+
+// Get resources scale recommendation for database.
+message GetScaleRecommendationRequest {
+    // Required. Full path to database's home dir.
+    string path = 1;
+}
+
+message GetScaleRecommendationResponse {
+    StatusIds.StatusCode status = 1;
+    repeated Ydb.Issue.IssueMessage issues = 2;
+    // Recommended resources scale to be allocated for database.
+    Resources recommended_resources = 3;
+}
diff --git a/ydb/services/cms/grpc_service.cpp b/ydb/services/cms/grpc_service.cpp
@@ -14,23 +14,24 @@ void TGRpcCmsService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) {
 #ifdef ADD_REQUEST
 #error ADD_REQUEST macro already defined
 #endif
-#define ADD_REQUEST(NAME, CB) \
+#define ADD_REQUEST(NAME, CB, TCALL) \
     MakeIntrusive<TGRpcRequest<Cms::NAME##Request, Cms::NAME##Response, TGRpcCmsService>>          \
         (this, &Service_, CQ_,                                                                     \
             [this](NYdbGrpc::IRequestContextBase *ctx) {                                              \
                 NGRpcService::ReportGrpcReqToMon(*ActorSystem_, ctx->GetPeer());                   \
                 ActorSystem_->Send(GRpcRequestProxyId_,                                            \
-                    new TGrpcRequestOperationCall<Cms::NAME##Request, Cms::NAME##Response>         \
+                    new TCALL<Cms::NAME##Request, Cms::NAME##Response>         \
                         (ctx, &CB, TRequestAuxSettings{RLSWITCH(TRateLimiterMode::Rps), nullptr}));          \
             }, &Cms::V1::CmsService::AsyncService::Request ## NAME,                             \
             #NAME, logger, getCounterBlock("cms", #NAME))->Run();
 
-    ADD_REQUEST(CreateDatabase, DoCreateTenantRequest)
-    ADD_REQUEST(AlterDatabase, DoAlterTenantRequest)
-    ADD_REQUEST(GetDatabaseStatus, DoGetTenantStatusRequest)
-    ADD_REQUEST(ListDatabases, DoListTenantsRequest)
-    ADD_REQUEST(RemoveDatabase, DoRemoveTenantRequest)
-    ADD_REQUEST(DescribeDatabaseOptions, DoDescribeTenantOptionsRequest)
+    ADD_REQUEST(CreateDatabase, DoCreateTenantRequest, TGrpcRequestOperationCall)
+    ADD_REQUEST(AlterDatabase, DoAlterTenantRequest, TGrpcRequestOperationCall)
+    ADD_REQUEST(GetDatabaseStatus, DoGetTenantStatusRequest, TGrpcRequestOperationCall)
+    ADD_REQUEST(ListDatabases, DoListTenantsRequest, TGrpcRequestOperationCall)
+    ADD_REQUEST(RemoveDatabase, DoRemoveTenantRequest, TGrpcRequestOperationCall)
+    ADD_REQUEST(DescribeDatabaseOptions, DoDescribeTenantOptionsRequest, TGrpcRequestOperationCall)
+    ADD_REQUEST(GetScaleRecommendation, DoGetScaleRecommendationRequest, TGrpcRequestNoOperationCall)
 
 #undef ADD_REQUEST
 }

Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,7 @@ namespace NKikimr {`
`6`	`6`	`namespace NGRpcService {`
`7`	`7`
`8`	`8`	`class IRequestOpCtx;`
	`9`	`+class IRequestNoOpCtx;`
`9`	`10`	`class IFacilityProvider;`
`10`	`11`
`11`	`12`	`void DoCreateTenantRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);`
`@@ -14,6 +15,7 @@ void DoGetTenantStatusRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityP`
`14`	`15`	`void DoListTenantsRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);`
`15`	`16`	`void DoRemoveTenantRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);`
`16`	`17`	`void DoDescribeTenantOptionsRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);`
	`18`	`+void DoGetScaleRecommendationRequest(std::unique_ptr<IRequestNoOpCtx> p, const IFacilityProvider& f);`
`17`	`19`
`18`	`20`	`}`
`19`	`21`	`}`
Original file line number	Diff line number	Diff line change
`@@ -589,3 +589,11 @@ message TEvResponseTabletDistribution {`
`589`	`589`	`}`
`590`	`590`	`repeated TNode Nodes = 1;`
`591`	`591`	`}`
	`592`	`+`
	`593`	`+message TEvRequestScaleRecommendation {`
	`594`	`+ optional NKikimrSubDomains.TDomainKey DomainKey = 1;`
	`595`	`+}`
	`596`	`+`
	`597`	`+message TEvResponseScaleRecommendation {`
	`598`	`+ optional uint32 RecommendedNodes = 1;`
	`599`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -27,4 +27,7 @@ service CmsService {`
`27`	`27`
`28`	`28`	`// Describe supported database options.`
`29`	`29`	`rpc DescribeDatabaseOptions(Cms.DescribeDatabaseOptionsRequest) returns (Cms.DescribeDatabaseOptionsResponse);`
	`30`	`+`
	`31`	`+ // Get resources scale recommendation for database.`
	`32`	`+ rpc GetScaleRecommendation(Cms.GetScaleRecommendationRequest) returns (Cms.GetScaleRecommendationResponse);`
`30`	`33`	`}`