Skip to content

KIKIMR-20523: VDisks balancing mvp #609

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
Mar 20, 2024
Merged

KIKIMR-20523: VDisks balancing mvp #609

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
aeb1b25
KIKIMR-20521: Add UseVDisksBalancing feature flag
Dec 15, 2023
45706d2
Move delete logic from barrier essense to handoff map
robdrynkin Feb 27, 2024
9564a91
Fix issues
robdrynkin Mar 4, 2024
202074e
Remove TActorContext
robdrynkin Mar 13, 2024
0c31ea1
Fix issues
robdrynkin Mar 13, 2024
4375e80
Remove old handoff map tests
robdrynkin Mar 13, 2024
12f8df1
Fix issue about TEvGenerationChanged and GInfo
robdrynkin Mar 15, 2024
92b2a70
Fix aquire/release repl token
robdrynkin Mar 15, 2024
7bb7d2a
BLOG_ -> STLOG
robdrynkin Mar 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions ydb/core/blobstorage/groupinfo/blobstorage_groupinfo.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -420,6 +420,10 @@ ui32 TBlobStorageGroupInfo::TTopology::GetIdxInSubgroup(const TVDiskIdShort& vdi
return BlobMapper->GetIdxInSubgroup(vdisk, hash);
}

bool TBlobStorageGroupInfo::TTopology::IsHandoff(const TVDiskIdShort& vdisk, ui32 hash) const {
return BlobMapper->GetIdxInSubgroup(vdisk, hash) >= GType.TotalPartCount();
}

TVDiskIdShort TBlobStorageGroupInfo::TTopology::GetVDiskInSubgroup(ui32 idxInSubgroup, ui32 hash) const {
return BlobMapper->GetVDiskInSubgroup(idxInSubgroup, hash);
}
Expand Down
2 changes: 2 additions & 0 deletions ydb/core/blobstorage/groupinfo/blobstorage_groupinfo.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,8 @@ class TBlobStorageGroupInfo : public TThrRefBase {
// function returns idxInSubgroup-th element of vdisks array from PickSubgroup
TVDiskIdShort GetVDiskInSubgroup(ui32 idxInSubgroup, ui32 hash) const;

bool IsHandoff(const TVDiskIdShort& vdisk, ui32 hash) const;


TFailRealmIterator FailRealmsBegin() const;
TFailRealmIterator FailRealmsEnd() const;
Expand Down
59 changes: 37 additions & 22 deletions ydb/core/blobstorage/ut_blobstorage/balancing.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ struct TTestEnv {
return data;
};

void SendPut(ui32 step, const TString& data, NKikimrProto::EReplyStatus expectedStatus) {
NKikimrProto::EReplyStatus SendPut(ui32 step, const TString& data) {
const TLogoBlobID id(1, 1, step, 0, data.size(), 0);
Cerr << "SEND TEvPut with key " << id.ToString() << Endl;
const TActorId sender = Env.Runtime->AllocateEdgeActor(GroupInfo->GetActorId(*RunningNodes.begin()).NodeId(), __FILE__, __LINE__);
Expand All @@ -69,8 +69,12 @@ struct TTestEnv {
SendToBSProxy(sender, GroupInfo->GroupID, ev.release());
});
auto res = Env.WaitForEdgeActorEvent<TEvBlobStorage::TEvPutResult>(sender, false);
UNIT_ASSERT_VALUES_EQUAL(res->Get()->Status, expectedStatus);
Cerr << "TEvPutResult: " << res->Get()->ToString() << Endl;
return res->Get()->Status;
};

void SendPut(ui32 step, const TString& data, NKikimrProto::EReplyStatus expectedStatus) {
UNIT_ASSERT_VALUES_EQUAL(SendPut(step, data), expectedStatus);
};

auto SendGet(ui32 step, ui32 dataSize, bool mustRestoreFirst=false) {
Expand Down Expand Up @@ -114,7 +118,6 @@ struct TTestEnv {
const TActorId sender = Env.Runtime->AllocateEdgeActor(GroupInfo->GetActorId(*RunningNodes.begin()).NodeId(), __FILE__, __LINE__);
TVector<ui32> partsRes;

Cerr << "Get request for vdisk " << position << Endl;
auto queueId = GetQueue(vDiskId);
Env.Runtime->WrapInActorContext(sender, [&] {
Env.Runtime->Send(new IEventHandle(queueId, sender, ev.release()));
Expand Down Expand Up @@ -244,13 +247,6 @@ struct TStopOneNodeTest {
Env->Sim(TDuration::Seconds(10));
Cerr << "Finish compaction 2" << Endl;

Cerr << "Start compaction 3" << Endl;
for (ui32 pos = 0; pos < Env->Settings.NodeCount; ++pos) {
Env->CompactVDisk(Env.GroupInfo->GetActorId(pos));
}
Env->Sim(TDuration::Seconds(10));
Cerr << "Finish compaction 3" << Endl;

Env.CheckPartsLocations(MakeLogoBlobId(step, data.size()));
UNIT_ASSERT_VALUES_EQUAL(Env.SendGet(step, data.size())->Get()->Responses[0].Buffer.ConvertToString(), data);
}
Expand All @@ -264,21 +260,23 @@ struct TRandomTest {
void RunTest() {
TVector<TString> data(Reserve(NumIters));

TVector<ui32> successfulSteps;

for (ui32 step = 0; step < NumIters; ++step) {
Cerr << step << Endl;
Cerr << "Step = " << step << Endl;
data.push_back(GenData(16 + random() % 4096));
auto blobId = MakeLogoBlobId(step, data.back().size());
auto locations = Env.GetExpectedPartsLocations(blobId);

if (Env.SendPut(step, data.back()) == NKikimrProto::OK) {
successfulSteps.push_back(step);
}

if (random() % 10 == 1 && Env.RunningNodes.size() + 2 > Env->Settings.NodeCount) {
ui32 nodeId = random() % Env->Settings.NodeCount;
Cerr << "Stop node " << nodeId << Endl;
Env.StopNode(nodeId);
ui32 pos = random() % Env->Settings.NodeCount;
Cerr << "Stop node " << pos << Endl;
Env.StopNode(pos);
Env->Sim(TDuration::Seconds(10));
}

Env.SendPut(step, data.back(), NKikimrProto::OK);

if (random() % 10 == 1) {
for (ui32 pos = 0; pos < Env->Settings.NodeCount; ++pos) {
if (!Env.RunningNodes.contains(pos)) {
Expand All @@ -293,6 +291,7 @@ struct TRandomTest {
if (random() % 50 == 1) {
ui32 pos = random() % Env->Settings.NodeCount;
if (Env.RunningNodes.contains(pos)) {
Cerr << "Compact vdisk " << pos << Endl;
Env->CompactVDisk(Env.GroupInfo->GetActorId(pos));
Env->Sim(TDuration::Seconds(10));
}
Expand All @@ -302,6 +301,7 @@ struct TRandomTest {
if (random() % 100 == 1) {
ui32 pos = random() % Env->Settings.NodeCount;
if (Env.RunningNodes.contains(pos)) {
Cerr << "Wipe node " << pos << Endl;
auto baseConfig = Env->FetchBaseConfig();
const auto& someVSlot = baseConfig.GetVSlot(pos);
const auto& loc = someVSlot.GetVSlotId();
Expand All @@ -313,14 +313,29 @@ struct TRandomTest {
}
}

UNIT_ASSERT(successfulSteps.size() > NumIters / 2);

Cerr << "Starting nodes" << Endl;
for (ui32 pos = 0; pos < Env->Settings.NodeCount; ++pos) {
Env.StartNode(pos);
}
Env->Sim(TDuration::Seconds(10));

Cerr << "Start compaction 1" << Endl;
for (ui32 pos = 0; pos < Env->Settings.NodeCount; ++pos) {
Env->CompactVDisk(Env.GroupInfo->GetActorId(pos));
}
Env->Sim(TDuration::Seconds(60));

Cerr << "Start compaction 2" << Endl;
for (ui32 pos = 0; pos < Env->Settings.NodeCount; ++pos) {
Env->CompactVDisk(Env.GroupInfo->GetActorId(pos));
}
Env->Sim(TDuration::Seconds(60));

Env->Sim(TDuration::Seconds(300));
Cerr << "Start checking" << Endl;
for (ui32 step = 0; step < NumIters; ++step) {
Cerr << step << Endl;
for (ui32 step: successfulSteps) {
Cerr << "step = " << step << Endl;
Env.CheckPartsLocations(MakeLogoBlobId(step, data[step].size()));
UNIT_ASSERT_VALUES_EQUAL(Env.SendGet(step, data[step].size())->Get()->Responses[0].Buffer.ConvertToString(), data[step]);
}
Expand All @@ -338,7 +353,7 @@ Y_UNIT_TEST_SUITE(VDiskBalancing) {
TStopOneNodeTest{TTestEnv(9, TBlobStorageGroupType::ErasureMirror3dc), GenData(100)}.RunTest();
}
Y_UNIT_TEST(TestStopOneNode_Block42_HugeBlob) {
TStopOneNodeTest{TTestEnv(8, TBlobStorageGroupType::Erasure4Plus2Block), GenData(521_KB)}.RunTest();
TStopOneNodeTest{TTestEnv(8, TBlobStorageGroupType::Erasure4Plus2Block), GenData(521_KB * 6)}.RunTest();
}
Y_UNIT_TEST(TestStopOneNode_Mirror3dc_HugeBlob) {
TStopOneNodeTest{TTestEnv(9, TBlobStorageGroupType::ErasureMirror3dc), GenData(521_KB)}.RunTest();
Expand Down
1 change: 1 addition & 0 deletions ydb/core/blobstorage/ut_blobstorage/lib/env.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -275,6 +275,7 @@ struct TEnvironmentSetup {
// NKikimrServices::LOCAL,
// NActorsServices::INTERCONNECT,
// NActorsServices::INTERCONNECT_SESSION,
// NKikimrServices::BS_VDISK_BALANCING,
};
for (const auto& comp : debug) {
Runtime->SetLogPriority(comp, NLog::PRI_DEBUG);
Expand Down
3 changes: 2 additions & 1 deletion ydb/core/blobstorage/ut_blobstorage/ut_balancing/ya.make
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
UNITTEST_FOR(ydb/core/blobstorage/ut_blobstorage)
SKIP_TEST(VDisks balancing is not implemented yet)

SIZE(MEDIUM)

FORK_SUBTESTS()

TIMEOUT(600)

SRCS(
Expand Down
1 change: 0 additions & 1 deletion ydb/core/blobstorage/ut_vdisk/lib/prepare.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,6 @@ TDefaultVDiskSetup::TDefaultVDiskSetup() {
cfg->SyncJobTimeout = TDuration::Seconds(20);
cfg->RunSyncer = true;
cfg->ReplTimeInterval = TDuration::Seconds(10);
cfg->RunHandoff = true;
cfg->SkeletonFrontQueueBackpressureCheckMsgId = false;
};
AddConfigModifier(modifier);
Expand Down
8 changes: 1 addition & 7 deletions ydb/core/blobstorage/ut_vdisk/lib/setup.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,6 @@ struct TFastVDiskSetupHndOff : public TFastVDiskSetup {
TFastVDiskSetupHndOff() {
auto modifier = [] (NKikimr::TVDiskConfig *cfg) {
cfg->HullCompLevelRateThreshold = 0.01; // to compact very few chunks from level 0
cfg->RunHandoff = false; // do not run handoff
};
AddConfigModifier(modifier);
}
Expand All @@ -82,12 +81,7 @@ struct TFastVDiskSetupCompacted : public TFastVDiskSetup {
};

struct TFastVDiskSetupCompactedHndOff : public TFastVDiskSetupCompacted {
TFastVDiskSetupCompactedHndOff() {
auto modifier = [] (NKikimr::TVDiskConfig *cfg) {
cfg->RunHandoff = false; // do not run handoff
};
AddConfigModifier(modifier);
}
TFastVDiskSetupCompactedHndOff() = default;
};

struct TFastCompactionGCNoSyncVDiskSetup : public TFastVDiskSetup {
Expand Down
69 changes: 0 additions & 69 deletions ydb/core/blobstorage/ut_vdisk/lib/test_repl.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -385,75 +385,6 @@ virtual void Scenario(const TActorContext &ctx) {
SYNC_TEST_END(TTestReplProxyKeepBits, TSyncTestWithSmallCommonDataset)


///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
SYNC_TEST_BEGIN(TTestHandoffMoveDel, TSyncTestBase)
virtual void Scenario(const TActorContext &ctx) {
TDataSnapshotPtr data(new TDataSnapshot(Conf->GroupInfo.Get()));
TString aaaa("aaaa");
TLogoBlobID id0(DefaultTestTabletId, 1, 322, 0, aaaa.size(), 0);
// [0:0:0:0:0] - main
// [0:0:0:1:1] - main
// [0:0:0:2:0] - main
// [0:0:0:3:1] - handoff

TAllVDisks::TVDiskInstance &vdisk00 = Conf->VDisks->Get(0);
TAllVDisks::TVDiskInstance &vdisk11 = Conf->VDisks->Get(3);
TAllVDisks::TVDiskInstance &vdisk20 = Conf->VDisks->Get(4);
TAllVDisks::TVDiskInstance &vdisk31 = Conf->VDisks->Get(7);
data->PutExact(vdisk00.VDiskID, vdisk00.ActorID, TLogoBlobID(id0, 1), aaaa); // main 0
data->PutExact(vdisk11.VDiskID, vdisk11.ActorID, TLogoBlobID(id0, 2), aaaa); // main 1
data->PutExact(vdisk31.VDiskID, vdisk31.ActorID, TLogoBlobID(id0, 3), aaaa); // handoff

// load data
SyncRunner->Run(ctx, CreateLoadDataSnapshot(SyncRunner->NotifyID(), Conf, data,
NKikimrBlobStorage::EPutHandleClass::TabletLog));
LOG_NOTICE(ctx, NActorsServices::TEST, " Data is loaded");

// wait for sync
SyncRunner->Run(ctx, CreateWaitForSync(SyncRunner->NotifyID(), Conf));
LOG_NOTICE(ctx, NActorsServices::TEST, " SYNC done");
// wait for compaction
SyncRunner->Run(ctx, CreateWaitForCompaction(SyncRunner->NotifyID(), Conf));
LOG_NOTICE(ctx, NActorsServices::TEST, " COMPACTION done");
// wait for sync
SyncRunner->Run(ctx, CreateWaitForSync(SyncRunner->NotifyID(), Conf));
LOG_NOTICE(ctx, NActorsServices::TEST, " SYNC done");
// wait for compaction
SyncRunner->Run(ctx, CreateWaitForCompaction(SyncRunner->NotifyID(), Conf));
LOG_NOTICE(ctx, NActorsServices::TEST, " COMPACTION done");
// wait for sync
SyncRunner->Run(ctx, CreateWaitForSync(SyncRunner->NotifyID(), Conf));
LOG_NOTICE(ctx, NActorsServices::TEST, " SYNC done");
// wait for compaction
SyncRunner->Run(ctx, CreateWaitForCompaction(SyncRunner->NotifyID(), Conf));
LOG_NOTICE(ctx, NActorsServices::TEST, " COMPACTION done");
// wait for sync
SyncRunner->Run(ctx, CreateWaitForSync(SyncRunner->NotifyID(), Conf));
LOG_NOTICE(ctx, NActorsServices::TEST, " SYNC done");
// wait for compaction
SyncRunner->Run(ctx, CreateWaitForCompaction(SyncRunner->NotifyID(), Conf));
LOG_NOTICE(ctx, NActorsServices::TEST, " COMPACTION done");

// check data
TDataSnapshotPtr result(new TDataSnapshot(Conf->GroupInfo.Get()));
#ifdef OPTIMIZE_SYNC
result->PutExact(vdisk00.VDiskID, vdisk00.ActorID, TLogoBlobID(id0, 1), aaaa, TIngress(0x24)); // main 0
result->PutExact(vdisk11.VDiskID, vdisk11.ActorID, TLogoBlobID(id0, 2), aaaa, TIngress(0x12)); // main 1
result->PutExact(vdisk20.VDiskID, vdisk20.ActorID, TLogoBlobID(id0, 3), aaaa, TIngress(0x9)); // main 2
result->PutExact(vdisk31.VDiskID, vdisk31.ActorID, TLogoBlobID(id0, 3), "", TIngress(0xc38)); // handoff
#else
result->PutExact(vdisk00.VDiskID, vdisk00.ActorID, TLogoBlobID(id0, 1), aaaa, TIngress(0xc3c)); // main 0 old: 0xc3c
result->PutExact(vdisk11.VDiskID, vdisk11.ActorID, TLogoBlobID(id0, 2), aaaa, TIngress(0xc3a)); // main 1 old: 0xc3a
result->PutExact(vdisk20.VDiskID, vdisk20.ActorID, TLogoBlobID(id0, 3), aaaa, TIngress(0xc39)); // main 2 old: 0xc39
result->PutExact(vdisk31.VDiskID, vdisk31.ActorID, TLogoBlobID(id0, 3), "", TIngress(0xc38)); // handoff
#endif
TSyncRunner::TReturnValue ret{0, 0};
while (ret.Status != 1) {
ret = SyncRunner->Run(ctx, CreateCheckDataSnapshot(SyncRunner->NotifyID(), Conf, result));
}
}
SYNC_TEST_END(TTestHandoffMoveDel, TSyncTestBase)

///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
SYNC_TEST_BEGIN(TTestCollectAllSimpleDataset, TSyncTestBase)
virtual void Scenario(const TActorContext &ctx) {
Expand Down
1 change: 0 additions & 1 deletion ydb/core/blobstorage/ut_vdisk/lib/test_repl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@ void operator ()(TConfiguration *conf); \

SIMPLE_CLASS_DEF_NO_PARAMS(TTestReplProxyData)
SIMPLE_CLASS_DEF_NO_PARAMS(TTestReplProxyKeepBits)
SIMPLE_CLASS_DEF_NO_PARAMS(TTestHandoffMoveDel)
SIMPLE_CLASS_DEF_NO_PARAMS(TTestCollectAllSimpleDataset)
SIMPLE_CLASS_DEF_NO_PARAMS(TTestStub)

Expand Down
2 changes: 1 addition & 1 deletion ydb/core/blobstorage/ut_vdisk/lib/test_synclog.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ class TSyncLogTestWriteActor : public TActorBootstrapped<TSyncLogTestWriteActor>
VDiskConfig = vDiskInstance.Cfg;
TestCtx->SelfVDiskId = groupInfo->GetVDiskId(VCtx->ShortSelfVDisk);

Db = MakeIntrusive<TDb>(VDiskConfig, groupInfo, VCtx);
Db = MakeIntrusive<TDb>(VDiskConfig, VCtx);

ctx.Send(VDiskConfig->BaseInfo.PDiskActorID,
new NPDisk::TEvYardInit(2, TestCtx->SelfVDiskId, VDiskConfig->BaseInfo.PDiskGuid));
Expand Down
10 changes: 0 additions & 10 deletions ydb/core/blobstorage/ut_vdisk/vdisk_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -412,16 +412,6 @@ Y_UNIT_TEST_SUITE(TBsVDiskBrokenPDisk) {
}
}

///////////////////////////////////////////////////////////////////////////////////////////////////////
// HANDOFF MOVE DEL
///////////////////////////////////////////////////////////////////////////////////////////////////////
Y_UNIT_TEST_SUITE(TBsVDiskHandoffMoveDel) {
Y_UNIT_TEST(HandoffMoveDel) {
TTestHandoffMoveDel test;
TestRun<TTestHandoffMoveDel, TFastVDiskSetupCompacted>(&test, TIMEOUT);
}
}

///////////////////////////////////////////////////////////////////////////////////////////////////////
// Huge
///////////////////////////////////////////////////////////////////////////////////////////////////////
Expand Down
6 changes: 0 additions & 6 deletions ydb/core/blobstorage/vdisk/anubis_osiris/blobstorage_anubis_osiris.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,12 +47,6 @@ namespace NKikimr {
return !IsAnubis();
}

// prepared data to insert to Hull Database
struct THullDbInsert {
TLogoBlobID Id;
TIngress Ingress;
};

// return data to insert to Hull Database, we create ingress according to whether this
// blob is Anubis or Osiris record
THullDbInsert PrepareInsert(const TBlobStorageGroupInfo::TTopology *top,
Expand Down
Loading