Skip to content

Commit 4ab7fba

Browse files
ydbotydbot
authored
stable-25-3: cherry-pick commit d7cd656, commit bf51ce2, commit b3e280a (#29477)
2 parents cc64f5f + 1e5c934 commit 4ab7fba

File tree

20 files changed

+179
-60
lines changed

20 files changed

+179
-60
lines changed

‎ydb/core/blobstorage/bridge/syncer/syncer.cpp‎

Lines changed: 31 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -6,11 +6,14 @@
66
namespace NKikimr::NBridge {
77

88
TSyncerActor::TSyncerActor(TIntrusivePtr<TBlobStorageGroupInfo> info, TGroupId sourceGroupId, TGroupId targetGroupId,
9-
std::shared_ptr<TSyncerDataStats> syncerDataStats)
9+
std::shared_ptr<TSyncerDataStats> syncerDataStats, TReplQuoter::TPtr syncRateQuoter,
10+
TBlobStorageGroupType sourceGroupType)
1011
: Info(std::move(info))
1112
, SourceGroupId(sourceGroupId)
1213
, TargetGroupId(targetGroupId)
1314
, SyncerDataStats(std::move(syncerDataStats))
15+
, SyncRateQuoter(std::move(syncRateQuoter))
16+
, SourceGroupType(sourceGroupType)
1417
{
1518
Y_ABORT_UNLESS(Info);
1619
Y_ABORT_UNLESS(Info->IsBridged());
@@ -368,7 +371,8 @@ namespace NKikimr::NBridge {
368371
return false;
369372
}
370373

371-
void TSyncerActor::IssueQuery(bool toTargetGroup, std::unique_ptr<IEventBase> ev, TQueryPayload payload) {
374+
void TSyncerActor::IssueQuery(bool toTargetGroup, std::unique_ptr<IEventBase> ev, TQueryPayload payload,
375+
ui64 quoterBytes) {
372376
switch (ev->Type()) {
373377
#define MSG(TYPE) \
374378
case TEvBlobStorage::TYPE: { \
@@ -399,11 +403,21 @@ namespace NKikimr::NBridge {
399403
std::unique_ptr<IEventHandle> handle(CreateEventForBSProxy(SelfId(),
400404
toTargetGroup ? TargetGroupId : SourceGroupId, ev.release(), cookie));
401405

406+
const TMonotonic now = TActivationContext::Monotonic();
407+
const TDuration timeout = SyncRateQuoter && quoterBytes
408+
? SyncRateQuoter->Take(now, quoterBytes)
409+
: TDuration::Zero();
410+
const TMonotonic timestamp = now + timeout;
411+
402412
if (QueriesInFlight < MaxQueriesInFlight) {
403-
TActivationContext::Send(handle.release());
413+
if (now < timestamp) {
414+
TActivationContext::Schedule(timestamp, handle.release());
415+
} else {
416+
TActivationContext::Send(handle.release());
417+
}
404418
++QueriesInFlight;
405419
} else {
406-
PendingQueries.push_back(std::move(handle));
420+
PendingQueries.emplace_back(std::move(handle), timestamp);
407421
}
408422
}
409423

@@ -479,8 +493,9 @@ namespace NKikimr::NBridge {
479493
++SyncerDataStats->BlobsDone;
480494
} else if (r.Status == NKikimrProto::NODATA) {
481495
// we have to query this blob and do full rewrite -- there was no data for it
496+
const ui64 quoterBytes = r.Id.BlobSize() * SourceGroupType.TotalPartCount() / SourceGroupType.DataParts();
482497
IssueQuery(false, std::make_unique<TEvBlobStorage::TEvGet>(r.Id, 0, 0, TInstant::Max(),
483-
NKikimrBlobStorage::FastRead));
498+
NKikimrBlobStorage::FastRead), {}, quoterBytes);
484499
} else if (r.Status == NKikimrProto::ERROR) {
485500
SyncerDataStats->BytesError += r.Id.BlobSize();
486501
++SyncerDataStats->BlobsError;
@@ -512,7 +527,13 @@ namespace NKikimr::NBridge {
512527
--QueriesInFlight;
513528
} else {
514529
Y_ABORT_UNLESS(QueriesInFlight == MaxQueriesInFlight);
515-
TActivationContext::Send(PendingQueries.front().release());
530+
TMonotonic now = TActivationContext::Monotonic();
531+
auto& [handle, timestamp] = PendingQueries.front();
532+
if (now < timestamp) {
533+
TActivationContext::Schedule(timestamp, handle.release());
534+
} else {
535+
TActivationContext::Send(handle.release());
536+
}
516537
PendingQueries.pop_front();
517538
}
518539

@@ -601,8 +622,10 @@ namespace NKikimr::NBridge {
601622
)
602623

603624
IActor *CreateSyncerActor(TIntrusivePtr<TBlobStorageGroupInfo> info, TGroupId sourceGroupId, TGroupId targetGroupId,
604-
std::shared_ptr<TSyncerDataStats> syncerDataStats) {
605-
return new TSyncerActor(std::move(info), sourceGroupId, targetGroupId, std::move(syncerDataStats));
625+
std::shared_ptr<TSyncerDataStats> syncerDataStats, TReplQuoter::TPtr syncRateQuoter,
626+
TBlobStorageGroupType sourceGroupType) {
627+
return new TSyncerActor(std::move(info), sourceGroupId, targetGroupId, std::move(syncerDataStats),
628+
std::move(syncRateQuoter), sourceGroupType);
606629
}
607630

608631
} // NKikimr::NBridge

‎ydb/core/blobstorage/bridge/syncer/syncer.h‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
#include "defs.h"
44

55
#include <ydb/core/blobstorage/groupinfo/blobstorage_groupinfo.h>
6+
#include <ydb/core/blobstorage/vdisk/repl/repl_quoter.h>
67

78
namespace NKikimr::NBridge {
89

@@ -16,6 +17,7 @@ namespace NKikimr::NBridge {
1617
};
1718

1819
IActor *CreateSyncerActor(TIntrusivePtr<TBlobStorageGroupInfo> info, TGroupId sourceGroupId, TGroupId targetGroupId,
19-
std::shared_ptr<TSyncerDataStats> syncerDataStats);
20+
std::shared_ptr<TSyncerDataStats> syncerDataStats, TReplQuoter::TPtr syncRateQuoter,
21+
TBlobStorageGroupType sourceGroupType);
2022

2123
} // NKikimr::NBridge

‎ydb/core/blobstorage/bridge/syncer/syncer_impl.h‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,13 @@ namespace NKikimr::NBridge {
2424
bool Finished = false;
2525
ui32 Step = 0;
2626
std::deque<TLogoBlobID> RestoreQueue;
27+
TReplQuoter::TPtr SyncRateQuoter;
28+
const TBlobStorageGroupType SourceGroupType;
2729

2830
public:
2931
TSyncerActor(TIntrusivePtr<TBlobStorageGroupInfo> info, TGroupId sourceGroupId, TGroupId targetGroupId,
30-
std::shared_ptr<TSyncerDataStats> syncerDataStats);
32+
std::shared_ptr<TSyncerDataStats> syncerDataStats, TReplQuoter::TPtr syncRateQuoter,
33+
TBlobStorageGroupType sourceGroupType);
3134

3235
void Bootstrap();
3336
void BeginNextStep();
@@ -50,7 +53,7 @@ namespace NKikimr::NBridge {
5053
const ui32 MaxQueriesInFlight = 16;
5154
ui32 QueriesInFlight = 0;
5255
THashMap<ui64, TQueryPayload> Payloads;
53-
std::deque<std::unique_ptr<IEventHandle>> PendingQueries;
56+
std::deque<std::tuple<std::unique_ptr<IEventHandle>, TMonotonic>> PendingQueries;
5457
ui64 NextCookie = 1;
5558

5659
bool Errors = false;
@@ -63,7 +66,8 @@ namespace NKikimr::NBridge {
6366
bool DoMergeEntities(std::deque<T>& source, std::deque<T>& target, bool sourceFinished, bool targetFinished,
6467
TCallback&& merge, std::optional<TKey>& lastMerged);
6568

66-
void IssueQuery(bool toTargetGroup, std::unique_ptr<IEventBase> ev, TQueryPayload queryPayload = {});
69+
void IssueQuery(bool toTargetGroup, std::unique_ptr<IEventBase> ev, TQueryPayload queryPayload = {},
70+
ui64 quoterBytes = 0);
6771
void Handle(TEvBlobStorage::TEvBlockResult::TPtr ev);
6872
void Handle(TEvBlobStorage::TEvCollectGarbageResult::TPtr ev);
6973
void Handle(TEvBlobStorage::TEvPutResult::TPtr ev);

‎ydb/core/blobstorage/dsproxy/dsproxy_state.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -187,7 +187,7 @@ namespace NKikimr {
187187
}
188188

189189
TString details = TStringBuilder() << " GroupId# " << GroupId
190-
<< "UnconfiguredStateTs# " << UnconfiguredStateTs
190+
<< " UnconfiguredStateTs# " << UnconfiguredStateTs
191191
<< " UnconfiguredStateReason# " << UnconfiguredStateReasonStr(UnconfiguredStateReason);
192192

193193
LOG_ERROR_S(*TlsActivationContext, NKikimrServices::BS_PROXY,
@@ -217,7 +217,7 @@ namespace NKikimr {
217217
return;
218218
}
219219
TString details = TStringBuilder() << " GroupId# " << GroupId
220-
<< "EstablishingSessionsStateTs# " << EstablishingSessionsStateTs
220+
<< " EstablishingSessionsStateTs# " << EstablishingSessionsStateTs
221221
<< " NumUnconnectedDisks# " << NumUnconnectedDisks;
222222
LOG_ERROR_S(*TlsActivationContext, NKikimrServices::BS_PROXY,
223223
"StateEstablishingSessions Wakeup TIMEOUT Marker# DSP12 " << details);

‎ydb/core/blobstorage/nodewarden/distconf_invoke_storage_config.cpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -476,6 +476,7 @@ namespace NKikimr::NStorage {
476476

477477
case EControllerOp::OTHER:
478478
record.SetOperation(NKikimrBlobStorage::TEvControllerDistconfRequest::ValidateConfig);
479+
record.MutableStorageConfig()->PackFrom(ProposedStorageConfig);
479480
break;
480481

481482
case EControllerOp::UNSET:

‎ydb/core/blobstorage/nodewarden/node_warden_group.cpp‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,16 @@ namespace NKikimr::NStorage {
114114
const auto [it, _] = Groups.try_emplace(groupId);
115115
TGroupRecord& group = it->second;
116116
group.MaxKnownGeneration = Max(group.MaxKnownGeneration, generation);
117+
if (newGroup) {
118+
if (newGroup->HasErasureSpecies()) {
119+
const auto erasure = static_cast<TBlobStorageGroupType::EErasureSpecies>(newGroup->GetErasureSpecies());
120+
Y_DEBUG_ABORT_UNLESS(!group.GType || group.GType->GetErasure() == erasure);
121+
group.GType.emplace(erasure);
122+
} else {
123+
Y_ABORT_UNLESS(newGroup->RingsSize() == 0); // ensure no VDisks in group
124+
group.GType.emplace(TBlobStorageGroupType::ErasureNone);
125+
}
126+
}
117127

118128
// forget pending queries
119129
if (fromController) {
@@ -380,8 +390,16 @@ namespace NKikimr::NStorage {
380390
if (startNew) {
381391
syncer.BridgeProxyGroupGeneration = syncer.PendingBridgeProxyGroupGeneration;
382392
syncer.SyncerDataStats = std::make_unique<NBridge::TSyncerDataStats>();
393+
394+
TBlobStorageGroupType sourceGroupType(TBlobStorageGroupType::ErasureNone);
395+
if (const auto it = Groups.find(syncer.SourceGroupId.GetRawId()); it != Groups.end() && it->second.GType) {
396+
sourceGroupType = *it->second.GType;
397+
} else {
398+
Y_DEBUG_ABORT("can't obtain source group type");
399+
}
400+
383401
syncer.ActorId = Register(NBridge::CreateSyncerActor(group.Info, syncer.SourceGroupId, syncer.TargetGroupId,
384-
syncer.SyncerDataStats));
402+
syncer.SyncerDataStats, SyncRateQuoter, sourceGroupType));
385403
syncer.Finished = false;
386404
syncer.ErrorReason.reset();
387405
++syncer.NumStart;

‎ydb/core/blobstorage/nodewarden/node_warden_impl.cpp‎

Lines changed: 26 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -456,6 +456,9 @@ void TNodeWarden::Bootstrap() {
456456
actorSystem->RegisterLocalService(MakeBlobStorageSyncBrokerID(), Register(
457457
CreateSyncBrokerActor(MaxInProgressSyncCount)));
458458

459+
// create bridge syncer rate quoter
460+
SyncRateQuoter = std::make_shared<TReplQuoter>(Cfg->BlobStorageConfig.GetBridgeSyncRateBytesPerSecond());
461+
459462
// determine if we are running in 'mock' mode
460463
EnableProxyMock = Cfg->BlobStorageConfig.GetServiceSet().GetEnableProxyMock();
461464

@@ -1403,6 +1406,10 @@ bool NKikimr::NStorage::DeriveStorageConfig(const NKikimrConfig::TAppConfig& app
14031406
const auto& bsFrom = appConfig.GetBlobStorageConfig();
14041407
auto *bsTo = config->MutableBlobStorageConfig();
14051408

1409+
const auto hasStaticGroupInfo = [](const NKikimrBlobStorage::TNodeWardenServiceSet& ss) {
1410+
return ss.PDisksSize() && ss.VDisksSize() && ss.GroupsSize();
1411+
};
1412+
14061413
if (bsFrom.HasServiceSet()) {
14071414
const auto& ssFrom = bsFrom.GetServiceSet();
14081415
auto *ssTo = bsTo->MutableServiceSet();
@@ -1419,10 +1426,6 @@ bool NKikimr::NStorage::DeriveStorageConfig(const NKikimrConfig::TAppConfig& app
14191426
ssTo->ClearReplBrokerConfig();
14201427
}
14211428

1422-
const auto hasStaticGroupInfo = [](const NKikimrBlobStorage::TNodeWardenServiceSet& ss) {
1423-
return ss.PDisksSize() && ss.VDisksSize() && ss.GroupsSize();
1424-
};
1425-
14261429
// update static group information unless distconf is enabled
14271430
if (!hasStaticGroupInfo(ssFrom) && config->GetSelfManagementConfig().GetEnabled()) {
14281431
// distconf enabled, keep it as is
@@ -1556,6 +1559,25 @@ bool NKikimr::NStorage::DeriveStorageConfig(const NKikimrConfig::TAppConfig& app
15561559
bsTo->ClearBscSettings();
15571560
}
15581561

1562+
// copy PDiskConfig from DefineHostConfig/DefineBox if this section is managed automatically
1563+
if (!hasStaticGroupInfo(bsFrom.GetServiceSet()) && config->GetSelfManagementConfig().GetEnabled()) {
1564+
THashMap<std::tuple<ui32, TString>, NKikimrBlobStorage::TPDiskConfig> pdiskConfigs;
1565+
auto callback = [&](const auto& node, const auto& drive) {
1566+
if (drive.HasPDiskConfig()) {
1567+
pdiskConfigs.emplace(std::make_tuple(node.GetNodeId(), drive.GetPath()), drive.GetPDiskConfig());
1568+
}
1569+
};
1570+
EnumerateConfigDrives(*config, 0, callback, nullptr, true);
1571+
for (auto& pdisk : *bsTo->MutableServiceSet()->MutablePDisks()) {
1572+
const auto key = std::make_tuple(pdisk.GetNodeID(), pdisk.GetPath());
1573+
if (const auto it = pdiskConfigs.find(key); it != pdiskConfigs.end()) {
1574+
pdisk.MutablePDiskConfig()->CopyFrom(it->second);
1575+
} else {
1576+
pdisk.ClearPDiskConfig();
1577+
}
1578+
}
1579+
}
1580+
15591581
// copy nameservice-related things
15601582
if (!appConfig.HasNameserviceConfig()) {
15611583
*errorReason = "origin config missing mandatory NameserviceConfig section";

‎ydb/core/blobstorage/nodewarden/node_warden_impl.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -529,6 +529,7 @@ namespace NKikimr::NStorage {
529529

530530
struct TGroupRecord {
531531
TIntrusivePtr<TBlobStorageGroupInfo> Info; // current group info
532+
std::optional<TBlobStorageGroupType> GType;
532533
ui32 MaxKnownGeneration = 0; // maximum seen generation
533534
std::optional<NKikimrBlobStorage::TGroupInfo> Group; // group info as a protobuf
534535
NKikimrBlobStorage::TGroupInfo EncryptionParams; // latest encryption parameters; set only when encryption enabled; overlay in respect to Group
@@ -680,6 +681,8 @@ namespace NKikimr::NStorage {
680681

681682
std::set<TWorkingSyncer> WorkingSyncers;
682683

684+
TReplQuoter::TPtr SyncRateQuoter;
685+
683686
void ApplyWorkingSyncers(const NKikimrBlobStorage::TEvControllerNodeServiceSetUpdate& update);
684687
void StartSyncerIfNeeded(TWorkingSyncer& syncer);
685688
void Handle(TAutoPtr<TEventHandle<TEvNodeWardenNotifySyncerFinished>> ev);

‎ydb/core/blobstorage/nodewarden/node_warden_resource.cpp‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,8 @@ void TNodeWarden::Handle(TEvNodeWardenStorageConfig::TPtr ev) {
9696
BridgeInfo = std::move(msg->BridgeInfo);
9797

9898
if (StorageConfig->HasBlobStorageConfig()) {
99-
if (const auto& bsConfig = StorageConfig->GetBlobStorageConfig(); bsConfig.HasServiceSet()) {
99+
const auto& bsConfig = StorageConfig->GetBlobStorageConfig();
100+
if (bsConfig.HasServiceSet()) {
100101
const NKikimrBlobStorage::TNodeWardenServiceSet *proposed = nullptr;
101102
if (const auto& proposedConfig = ev->Get()->ProposedConfig) {
102103
Y_VERIFY_S(StorageConfig->GetGeneration() < proposedConfig->GetGeneration(),
@@ -109,6 +110,7 @@ void TNodeWarden::Handle(TEvNodeWardenStorageConfig::TPtr ev) {
109110
}
110111
ApplyStorageConfig(bsConfig.GetServiceSet(), proposed);
111112
}
113+
SyncRateQuoter->UpdateBytesPerSecond(bsConfig.GetBridgeSyncRateBytesPerSecond());
112114
}
113115

114116
if (StorageConfig->HasStateStorageConfig() && StorageConfig->HasStateStorageBoardConfig() && StorageConfig->HasSchemeBoardConfig()) {

‎ydb/core/blobstorage/ut_blobstorage/lib/activity.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -166,7 +166,7 @@ class TActivityActor : public TActorBootstrapped<TActivityActor> {
166166
TString buffer = GenerateBuffer();
167167
const TLogoBlobID id(TabletId, Generation, Step++, 0, buffer.size(), 0);
168168
LOG_DEBUG_S(*TlsActivationContext, NActorsServices::TEST, Prefix << "sending TEvPut Id# " << id);
169-
SendToProxy(Quoter.Take(TActivationContext::Now(), 1), new TEvBlobStorage::TEvPut(id, buffer, TInstant::Max()));
169+
SendToProxy(Quoter.Take(TActivationContext::Monotonic(), 1), new TEvBlobStorage::TEvPut(id, buffer, TInstant::Max()));
170170
Inflight.emplace(id, std::move(buffer));
171171
}
172172
if (!NumWritesRemaining && Inflight.empty()) {

0 commit comments

Comments
 (0)