disk usage limit for CS compaction (in general slider limit) (#4864)

Author: ivanmorozov333

ydb/core/base/events.h

@@ -178,7 +178,8 @@ struct TKikimrEvents : TEvents {
         ES_REPLICATION_SERVICE,
         ES_BACKUP_SERVICE,
         ES_TX_BACKGROUND,
-        ES_SS_BG_TASKS
+        ES_SS_BG_TASKS,
+        ES_LIMITER
     };
 };
 

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -176,9 +176,12 @@
 #include <ydb/services/metadata/ds_table/service.h>
 #include <ydb/services/metadata/service.h>
 
-#include <ydb/core/tx/conveyor/usage/config.h>
 #include <ydb/core/tx/conveyor/service/service.h>
+#include <ydb/core/tx/conveyor/usage/config.h>
 #include <ydb/core/tx/conveyor/usage/service.h>
+#include <ydb/core/tx/limiter/service/service.h>
+#include <ydb/core/tx/limiter/usage/config.h>
+#include <ydb/core/tx/limiter/usage/service.h>
 
 #include <ydb/core/backup/controller/tablet.h>
 
@@ -2169,6 +2172,26 @@ void TKqpServiceInitializer::InitializeServices(NActors::TActorSystemSetup* setu
     }
 }
 
+TCompDiskLimiterInitializer::TCompDiskLimiterInitializer(const TKikimrRunConfig& runConfig)
+    : IKikimrServicesInitializer(runConfig) {
+}
+
+void TCompDiskLimiterInitializer::InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) {
+    NLimiter::TConfig serviceConfig;
+    Y_ABORT_UNLESS(serviceConfig.DeserializeFromProto<NLimiter::TCompDiskLimiterPolicy>(Config.GetCompDiskLimiterConfig()));
+
+    if (serviceConfig.IsEnabled()) {
+        TIntrusivePtr<::NMonitoring::TDynamicCounters> tabletGroup = GetServiceCounters(appData->Counters, "tablets");
+        TIntrusivePtr<::NMonitoring::TDynamicCounters> countersGroup = tabletGroup->GetSubgroup("type", "TX_COMP_DISK_LIMITER");
+
+        auto service = NLimiter::TCompDiskOperator::CreateService(serviceConfig, countersGroup);
+
+        setup->LocalServices.push_back(std::make_pair(
+            NLimiter::TCompDiskOperator::MakeServiceId(NodeId),
+            TActorSetupCmd(service, TMailboxType::HTSwap, appData->UserPoolId)));
+    }
+}
+
 TCompConveyorInitializer::TCompConveyorInitializer(const TKikimrRunConfig& runConfig)
     : IKikimrServicesInitializer(runConfig) {
 }

ydb/core/driver_lib/run/kikimr_services_initializers.h

@@ -391,6 +391,12 @@ class TKqpServiceInitializer : public IKikimrServicesInitializer {
     IGlobalObjectStorage& GlobalObjects;
 };
 
+class TCompDiskLimiterInitializer: public IKikimrServicesInitializer {
+public:
+    TCompDiskLimiterInitializer(const TKikimrRunConfig& runConfig);
+    void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
+};
+
 class TCompConveyorInitializer: public IKikimrServicesInitializer {
 public:
     TCompConveyorInitializer(const TKikimrRunConfig& runConfig);

ydb/core/driver_lib/run/run.cpp

@@ -1542,6 +1542,10 @@ TIntrusivePtr<TServiceInitializersList> TKikimrRunner::CreateServiceInitializers
         sil->AddServiceInitializer(new TExternalIndexInitializer(runConfig));
     }
 
+    if (serviceMask.EnableCompDiskLimiter) {
+        sil->AddServiceInitializer(new TCompDiskLimiterInitializer(runConfig));
+    }
+
     if (serviceMask.EnableScanConveyor) {
         sil->AddServiceInitializer(new TScanConveyorInitializer(runConfig));
     }

ydb/core/driver_lib/run/service_mask.h

@@ -77,6 +77,7 @@ union TBasicKikimrServicesMask {
 
         bool EnableDatabaseMetadataCache:1;
         bool EnableGraphService:1;
+        bool EnableCompDiskLimiter:1;
     };
 
     struct {

ydb/core/driver_lib/run/ya.make

@@ -112,6 +112,7 @@ PEERDIR(
     ydb/core/tx/columnshard
     ydb/core/tx/coordinator
     ydb/core/tx/conveyor/service
+    ydb/core/tx/limiter/service
     ydb/core/tx/datashard
     ydb/core/tx/long_tx_service
     ydb/core/tx/long_tx_service/public

ydb/core/protos/config.proto

@@ -595,6 +595,12 @@ message TConveyorConfig {
     optional double WorkersCountDouble = 5;
 }
 
+message TLimiterConfig {
+    optional bool Enabled = 1 [default = true];
+    optional uint64 Limit = 2;
+    optional uint64 PeriodMilliSeconds = 3 [default = 1000];
+}
+
 message TExternalIndexConfig {
     optional bool Enabled = 1 [default = true];
     optional TInternalRequestConfig RequestConfig = 2;
@@ -1841,6 +1847,7 @@ message TAppConfig {
     optional TS3ProxyResolverConfig S3ProxyResolverConfig = 76;
     optional TBackgroundCleaningConfig BackgroundCleaningConfig = 77;
     optional TBlobCacheConfig BlobCacheConfig = 78;
+    optional TLimiterConfig CompDiskLimiterConfig = 79;
 
     repeated TNamedConfig NamedConfigs = 100;
     optional string ClusterYamlConfig = 101;

ydb/core/tx/columnshard/columnshard__write_index.cpp

@@ -6,10 +6,34 @@
 #include "engines/changes/abstract/abstract.h"
 #include "engines/writer/compacted_blob_constructor.h"
 
+#include <ydb/core/tx/limiter/usage/abstract.h>
+#include <ydb/core/tx/limiter/usage/service.h>
+
 #include <ydb/library/actors/core/log.h>
 
 namespace NKikimr::NColumnShard {
 
+class TDiskResourcesRequest: public NLimiter::IResourceRequest {
+private:
+    using TBase = NLimiter::IResourceRequest;
+    std::shared_ptr<NOlap::TCompactedWriteController> WriteController;
+    const ui64 TabletId;
+
+private:
+    virtual void DoOnResourceAllocated() override {
+        NActors::TActivationContext::AsActorContext().Register(CreateWriteActor(TabletId, WriteController, TInstant::Max()));
+    }
+
+public:
+    TDiskResourcesRequest(const std::shared_ptr<NOlap::TCompactedWriteController>& writeController, const ui64 tabletId)
+        : TBase(writeController->GetWriteVolume())
+        , WriteController(writeController)
+        , TabletId(tabletId)
+    {
+
+    }
+};
+
 void TColumnShard::Handle(TEvPrivate::TEvWriteIndex::TPtr& ev, const TActorContext& ctx) {
     auto putStatus = ev->Get()->GetPutStatus();
 
@@ -32,7 +56,7 @@ void TColumnShard::Handle(TEvPrivate::TEvWriteIndex::TPtr& ev, const TActorConte
             if (*needDraftTransaction) {
                 Execute(new TTxWriteDraft(this, writeController));
             } else {
-                ctx.Register(CreateWriteActor(TabletID(), writeController, TInstant::Max()));
+                NLimiter::TCompDiskOperator::AskResource(std::make_shared<TDiskResourcesRequest>(writeController, TabletID()));
             }
         }
     } else {

ydb/core/tx/columnshard/engines/writer/compacted_blob_constructor.cpp

@@ -23,6 +23,7 @@ TCompactedWriteController::TCompactedWriteController(const TActorId& dstActor, T
         for (auto&& b : portionWithBlobs.GetBlobs()) {
             auto& task = AddWriteTask(TBlobWriteInfo::BuildWriteTask(b.GetBlob(), changes.MutableBlobsAction().GetWriting(b.GetOperator()->GetStorageId())));
             b.RegisterBlobId(portionWithBlobs, task.GetBlobId());
+            WriteVolume += b.GetSize();
         }
     }
 }

ydb/core/tx/columnshard/engines/writer/compacted_blob_constructor.h

@@ -13,11 +13,17 @@ class TCompactedWriteController : public NColumnShard::IWriteController {
 private:
     TAutoPtr<NColumnShard::TEvPrivate::TEvWriteIndex> WriteIndexEv;
     TActorId DstActor;
+    ui64 WriteVolume = 0;
+
 protected:
     void DoOnReadyResult(const NActors::TActorContext& ctx, const NColumnShard::TBlobPutResult::TPtr& putResult) override;
     virtual void DoAbort(const TString& reason) override;
+
 public:
     const TBlobsAction& GetBlobsAction();
+    ui64 GetWriteVolume() const {
+        return WriteVolume;
+    }
 
     TCompactedWriteController(const TActorId& dstActor, TAutoPtr<NColumnShard::TEvPrivate::TEvWriteIndex> writeEv);
     ~TCompactedWriteController();