Bandwidth recommendations for direct copy

cloud/blockstore/config/disk.proto

@@ -150,6 +150,22 @@ message TStorageDiscoveryConfig
 
 ////////////////////////////////////////////////////////////////////////////////
 
+message TDiskAgentThrottlingConfig
+{
+    // Host limits.
+    optional string InfraThrottlingConfigPath = 1;
+    optional uint32 DefaultNetworkMbitThroughput = 2;
+
+    // Fraction of network throughput utilized for migrations and shadow
+    // disk fill.
+    optional double DirectCopyBandwidthFraction = 3;
+
+    // Maximum bandwidth for one device in MiB/s.
+    optional uint64 MaxDeviceBandwidthMiB = 4;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
 message TDiskAgentConfig
 {
     optional bool Enabled = 1;
@@ -275,6 +291,9 @@ message TDiskAgentConfig
     }
 
     repeated TPathToSerialNumber PathToSerialNumberMapping = 37;
+
+    // Settings for traffic shaping.
+    optional TDiskAgentThrottlingConfig ThrottlingConfig = 38;
 }
 
 ////////////////////////////////////////////////////////////////////////////////

cloud/blockstore/libs/daemon/common/config_initializer.cpp

@@ -84,8 +84,8 @@ void TConfigInitializerCommon::InitDiskAgentConfig()
 
     DiskAgentConfig = std::make_shared<NStorage::TDiskAgentConfig>(
         std::move(diskAgentConfig),
-        Rack
-    );
+        Rack,
+        HostPerformanceProfile.NetworkMbitThroughput);
 }
 
 void TConfigInitializerCommon::InitRdmaConfig()

cloud/blockstore/libs/daemon/ydb/config_initializer.cpp

@@ -326,7 +326,8 @@ void TConfigInitializerYdb::ApplyDiskAgentConfig(const TString& text)
 
     DiskAgentConfig = std::make_shared<NStorage::TDiskAgentConfig>(
         std::move(config),
-        Rack);
+        Rack,
+        HostPerformanceProfile.NetworkMbitThroughput);
 }
 
 void TConfigInitializerYdb::ApplyDiskRegistryProxyConfig(const TString& text)

cloud/blockstore/libs/disk_agent/bootstrap.cpp

@@ -346,6 +346,7 @@ void TBootstrap::InitRdmaServer(NRdma::TRdmaConfig& config)
 
 bool TBootstrap::InitKikimrService()
 {
+    Configs->Log = Log;
     Configs->InitKikimrConfig();
     Configs->InitServerConfig();
     Configs->InitFeaturesConfig();

cloud/blockstore/libs/disk_agent/config_initializer.cpp

@@ -8,12 +8,12 @@
 #include <cloud/blockstore/libs/spdk/iface/config.h>
 #include <cloud/blockstore/libs/storage/core/config.h>
 #include <cloud/blockstore/libs/storage/disk_registry_proxy/model/config.h>
-
 #include <cloud/storage/core/libs/common/proto_helpers.h>
 #include <cloud/storage/core/libs/features/features_config.h>
 #include <cloud/storage/core/libs/kikimr/actorsystem.h>
 #include <cloud/storage/core/libs/version/version.h>
 
+#include <library/cpp/json/json_reader.h>
 #include <library/cpp/protobuf/util/pb_io.h>
 
 #include <util/datetime/base.h>
@@ -23,8 +23,54 @@
 
 namespace NCloud::NBlockStore::NServer {
 
+namespace {
+
 ////////////////////////////////////////////////////////////////////////////////
 
+std::optional<NJson::TJsonValue> ReadJsonFile(
+    TLog& Log,
+    const TString& filename)
+{
+    if (filename.empty()) {
+        return std::nullopt;
+    }
+
+    try {
+        TFileInput in(filename);
+        return NJson::ReadJsonTree(&in, true);
+    } catch (...) {
+        STORAGE_ERROR(
+            "Failed to read file: " << filename.Quote() << " with error: "
+                                    << CurrentExceptionMessage().c_str());
+        return std::nullopt;
+    }
+}
+
+ui32 ReadNetworkMbitThroughput(
+    TLog& Log,
+    const NProto::TDiskAgentConfig& diskAgentConfig)
+{
+    const auto& config = diskAgentConfig.GetThrottlingConfig();
+    ui32 networkThroughput = config.GetDefaultNetworkMbitThroughput();
+
+    if (auto json = ReadJsonFile(Log, config.GetInfraThrottlingConfigPath())) {
+        try {
+            if (auto* value = json->GetValueByPath("interfaces.[0].eth0.speed"))
+            {
+                networkThroughput = FromString<ui32>(value->GetStringSafe());
+            }
+        } catch (...) {
+            STORAGE_ERROR(
+                "Failed to read NetworkMbitThroughput. Error: "
+                << CurrentExceptionMessage().c_str());
+        }
+    }
+
+    return networkThroughput;
+}
+
+}   // namespace
+
 void TConfigInitializer::ApplyCMSConfigs(NKikimrConfig::TAppConfig cmsConfig)
 {
     if (cmsConfig.HasBlobStorageConfig()) {
@@ -167,10 +213,11 @@ void TConfigInitializer::InitDiskAgentConfig()
     SetupDiskAgentConfig(diskAgentConfig);
     ApplySpdkEnvConfig(diskAgentConfig.GetSpdkEnvConfig());
 
+    const ui32 networkMbitThroughput = ReadNetworkMbitThroughput(Log, diskAgentConfig);
     DiskAgentConfig = std::make_shared<NStorage::TDiskAgentConfig>(
         std::move(diskAgentConfig),
-        Rack
-    );
+        Rack,
+        networkMbitThroughput);
 }
 
 void TConfigInitializer::InitDiskRegistryProxyConfig()
@@ -428,9 +475,11 @@ void TConfigInitializer::ApplyDiskAgentConfig(const TString& text)
             DiskAgentConfig->GetStorageDiscoveryConfig());
     }
 
+    const ui32 networkMbitThroughput = ReadNetworkMbitThroughput(Log, config);
     DiskAgentConfig = std::make_shared<NStorage::TDiskAgentConfig>(
         std::move(config),
-        Rack);
+        Rack,
+        networkMbitThroughput);
 }
 
 void TConfigInitializer::ApplyDiskRegistryProxyConfig(const TString& text)

cloud/blockstore/libs/disk_agent/config_initializer.h

@@ -12,13 +12,15 @@
 #include <cloud/blockstore/libs/storage/core/config.h>
 #include <cloud/blockstore/libs/storage/disk_agent/model/config.h>
 #include <cloud/blockstore/libs/storage/disk_registry_proxy/model/public.h>
-#include <cloud/storage/core/libs/features/features_config.h>
 #include <cloud/storage/core/libs/coroutine/public.h>
+#include <cloud/storage/core/libs/features/features_config.h>
 
 #include <contrib/ydb/core/protos/auth.pb.h>
 #include <contrib/ydb/core/protos/blobstorage.pb.h>
 #include <contrib/ydb/core/protos/config.pb.h>
 
+#include <library/cpp/logger/log.h>
+
 namespace NCloud::NBlockStore::NServer {
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -38,6 +40,7 @@ struct TConfigInitializer
     NRdma::TRdmaConfigPtr RdmaConfig;
 
     TString Rack;
+    TLog Log;
 
     TConfigInitializer(TOptionsPtr options)
         : Options(std::move(options))

cloud/blockstore/libs/storage/disk_agent/actors/direct_copy_actor.cpp

@@ -31,10 +31,12 @@ auto PrepareRequest(
 TDirectCopyActor::TDirectCopyActor(
         const TActorId& source,
         TRequestInfoPtr requestInfo,
-        NProto::TDirectCopyBlocksRequest request)
+        NProto::TDirectCopyBlocksRequest request,
+        ui64 recommendedBandwidth)
     : Source(source)
     , RequestInfo(std::move(requestInfo))
     , Request(std::move(request))
+    , RecommendedBandwidth(recommendedBandwidth)
 {}
 
 void TDirectCopyActor::Bootstrap(const TActorContext& ctx)
@@ -88,6 +90,7 @@ void TDirectCopyActor::Done(const NActors::TActorContext& ctx)
     response->Record.SetAllZeroes(AllZeroes);
     response->Record.SetReadDuration(readDuration.MicroSeconds());
     response->Record.SetWriteDuration(writeDuration.MicroSeconds());
+    response->Record.SetRecommendedBandwidth(RecommendedBandwidth);
 
     NCloud::Reply(ctx, *RequestInfo, std::move(response));
 

cloud/blockstore/libs/storage/disk_agent/actors/direct_copy_actor.h

@@ -20,6 +20,7 @@ class TDirectCopyActor final
     const NActors::TActorId Source;
     const TRequestInfoPtr RequestInfo;
     const NProto::TDirectCopyBlocksRequest Request;
+    const ui64 RecommendedBandwidth;
 
     bool AllZeroes = false;
     TInstant ReadStartAt;
@@ -29,7 +30,8 @@ class TDirectCopyActor final
     TDirectCopyActor(
         const NActors::TActorId& source,
         TRequestInfoPtr requestInfo,
-        NProto::TDirectCopyBlocksRequest request);
+        NProto::TDirectCopyBlocksRequest request,
+        ui64 recommendedBandwidth);
 
     void Bootstrap(const NActors::TActorContext& ctx);
 

cloud/blockstore/libs/storage/disk_agent/disk_agent_actor.h

@@ -17,6 +17,7 @@
 #include <cloud/blockstore/libs/storage/core/config.h>
 #include <cloud/blockstore/libs/storage/core/pending_request.h>
 #include <cloud/blockstore/libs/storage/core/request_info.h>
+#include <cloud/blockstore/libs/storage/disk_agent/model/bandwidth_calculator.h>
 #include <cloud/blockstore/libs/storage/disk_agent/model/config.h>
 #include <cloud/blockstore/libs/storage/disk_agent/recent_blocks_tracker.h>
 
@@ -71,6 +72,8 @@ class TDiskAgentActor final
     // Pending WaitReady requests
     TDeque<TPendingRequest> PendingRequests;
 
+    TBandwidthCalculator BandwidthCalculator {*AgentConfig};
+
     ERegistrationState RegistrationState = ERegistrationState::NotStarted;
 
     NActors::TActorId StatsActor;

cloud/blockstore/libs/storage/disk_agent/disk_agent_actor_direct_copy.cpp

@@ -29,11 +29,16 @@ void TDiskAgentActor::HandleDirectCopyBlocks(
                           record.GetBlockCount()))
             .c_str());
 
+    ui64 recommendedBandwidth = BandwidthCalculator.RegisterRequest(
+        record.GetSourceDeviceUUID(),
+        ctx.Now());
+
     NCloud::Register<TDirectCopyActor>(
         ctx,
         SelfId(),
         CreateRequestInfo(ev->Sender, ev->Cookie, msg->CallContext),
-        std::move(record));
+        std::move(record),
+        recommendedBandwidth);
 }
 
 }   // namespace NCloud::NBlockStore::NStorage

cloud/blockstore/libs/storage/disk_agent/disk_agent_actor_ut.cpp

@@ -349,6 +349,10 @@ struct TCopyRangeFixture: public NUnitTest::TBaseFixture
         config.SetIOParserActorCount(0);
         config.SetBackend(NProto::DISK_AGENT_BACKEND_AIO);
 
+        auto* throttling = config.MutableThrottlingConfig();
+        throttling->SetDirectCopyBandwidthFraction(0.5);
+        throttling->SetDefaultNetworkMbitThroughput(800);
+
         for (const auto& memDevice: config.GetMemoryDevices()) {
             *config.AddMemoryDevices() = PrepareMemoryDevice(
                 memDevice.GetDeviceId(),
@@ -5665,6 +5669,60 @@ Y_UNIT_TEST_SUITE(TDiskAgentTest)
                 .MilliSeconds(),
             10);
     }
+
+    Y_UNIT_TEST_F(ShouldPerformDirectCopyAndCalcBandwidth, TCopyRangeFixture)
+    {
+        {   // The first disk agent has network bandwidth config. It should
+            // recommend the bandwidth.
+            auto request =
+                std::make_unique<TEvDiskAgent::TEvDirectCopyBlocksRequest>();
+            request->Record.MutableHeaders()->SetClientId(SourceClientId);
+
+            request->Record.SetSourceDeviceUUID("DA1-1");
+            request->Record.SetSourceStartIndex(SourceStartIndex);
+            request->Record.SetBlockSize(BlockSize);
+            request->Record.SetBlockCount(BlockCount);
+
+            request->Record.SetTargetNodeId(Runtime->GetNodeId(1));
+            request->Record.SetTargetClientId(TargetClientId);
+            request->Record.SetTargetDeviceUUID("DA2-1");
+            request->Record.SetTargetStartIndex(TargetStartIndex);
+
+            DiskAgent1->SendRequest(std::move(request));
+            auto response =
+                DiskAgent1
+                    ->RecvResponse<TEvDiskAgent::TEvDirectCopyBlocksResponse>();
+            UNIT_ASSERT(!HasError(response->GetError()));
+            UNIT_ASSERT_VALUES_EQUAL(
+                50_MB,
+                response->Record.GetRecommendedBandwidth());
+        }
+        {   // The second disk agent has no network bandwidth configuration. It
+            // shouldn't recommend the bandwidth.
+            auto request =
+                std::make_unique<TEvDiskAgent::TEvDirectCopyBlocksRequest>();
+            request->Record.MutableHeaders()->SetClientId(TargetClientId);
+
+            request->Record.SetSourceDeviceUUID("DA2-2");
+            request->Record.SetSourceStartIndex(SourceStartIndex);
+            request->Record.SetBlockSize(BlockSize);
+            request->Record.SetBlockCount(BlockCount);
+
+            request->Record.SetTargetNodeId(Runtime->GetNodeId(0));
+            request->Record.SetTargetClientId(ClientId);
+            request->Record.SetTargetDeviceUUID("DA1-1");
+            request->Record.SetTargetStartIndex(TargetStartIndex);
+
+            DiskAgent2->SendRequest(std::move(request));
+            auto response =
+                DiskAgent2
+                    ->RecvResponse<TEvDiskAgent::TEvDirectCopyBlocksResponse>();
+            UNIT_ASSERT(!HasError(response->GetError()));
+            UNIT_ASSERT_VALUES_EQUAL(
+                0,
+                response->Record.GetRecommendedBandwidth());
+        }
+    }
 }
 
 }   // namespace NCloud::NBlockStore::NStorage

cloud/blockstore/libs/storage/disk_agent/disk_agent_state_ut.cpp

@@ -114,7 +114,7 @@ auto CreateSpdkConfig()
         device.SetDeviceId("uuid-" + ToString(i + 1));
     }
 
-    return std::make_shared<TDiskAgentConfig>(std::move(config), "rack");
+    return std::make_shared<TDiskAgentConfig>(std::move(config), "rack", 1000);
 }
 
 struct TNullConfigParams
@@ -150,7 +150,7 @@ auto CreateNullConfig(TNullConfigParams params)
     config.SetCachedConfigPath(std::move(params.CachedConfigPath));
     config.SetCachedSessionsPath(std::move(params.CachedSessionsPath));
 
-    return std::make_shared<TDiskAgentConfig>(std::move(config), "rack");
+    return std::make_shared<TDiskAgentConfig>(std::move(config), "rack", 1000);
 }
 
 TStorageConfigPtr CreateStorageConfig()
@@ -736,7 +736,7 @@ Y_UNIT_TEST_SUITE(TDiskAgentStateTest)
                 size_t errors,
                 bool checkLockedDevices)
             {
-                auto config = std::make_shared<TDiskAgentConfig>(cfg, "rack");
+                auto config = std::make_shared<TDiskAgentConfig>(cfg, "rack", 1000);
 
                 TDiskAgentState state(
                     CreateStorageConfig(),
@@ -812,7 +812,7 @@ Y_UNIT_TEST_SUITE(TDiskAgentStateTest)
 
         TDiskAgentState state(
             CreateStorageConfig(),
-            std::make_shared<TDiskAgentConfig>(std::move(config), "rack"),
+            std::make_shared<TDiskAgentConfig>(std::move(config), "rack", 1000),
             nullptr,    // spdk
             CreateTestAllocator(),
             NServer::CreateNullStorageProvider(),