Skip to content

Commit e3eb4ca

Browse files
serbel324serbel324
committed
Add UT for group cache, improve SentToWarden for local messages (#22541)
1 parent 4960407 commit e3eb4ca

File tree

7 files changed

+187
-18
lines changed

7 files changed

+187
-18
lines changed

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

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -166,20 +166,14 @@ void TNodeWarden::ApplyStateStorageConfig(const NKikimrBlobStorage::TStorageConf
166166
}
167167

168168
// apply updates for the state storage proxy
169-
<<<<<<< HEAD
170169
#define FETCH_CONFIG(PART, PREFIX, PROTO) \
171-
Y_ABORT_UNLESS(StorageConfig.Has##PROTO##Config()); \
170+
Y_ABORT_UNLESS(StorageConfig->Has##PROTO##Config()); \
172171
char PART##Prefix[TActorId::MaxServiceIDLength] = PREFIX; \
173-
TIntrusivePtr<TStateStorageInfo> PART##Info = BuildStateStorageInfo(PART##Prefix, StorageConfig.Get##PROTO##Config());
172+
TIntrusivePtr<TStateStorageInfo> PART##Info = BuildStateStorageInfo(PART##Prefix, StorageConfig->Get##PROTO##Config());
174173

175174
FETCH_CONFIG(stateStorage, "ssr", StateStorage)
176175
FETCH_CONFIG(board, "ssb", StateStorageBoard)
177176
FETCH_CONFIG(schemeBoard, "sbr", SchemeBoard)
178-
=======
179-
#define FETCH_CONFIG(PART, PROTO) \
180-
Y_ABORT_UNLESS(StorageConfig->Has##PROTO##Config()); \
181-
TIntrusivePtr<TStateStorageInfo> PART##Info = Build##PROTO##Info(StorageConfig->Get##PROTO##Config());
182-
>>>>>>> a3fdda7cfea (Pass TStorageConfig through pointer to prevent copying when doing subscription fan-out (#18765))
183177

184178
STLOG(PRI_DEBUG, BS_NODE, NW52, "ApplyStateStorageConfig",
185179
(StateStorageConfig, StorageConfig->GetStateStorageConfig()),

‎ydb/core/blobstorage/ut_blobstorage/bsc_cache.cpp

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
#include <ydb/core/blobstorage/ut_blobstorage/lib/env.h>
2+
#include <ydb/core/blobstorage/ut_blobstorage/ut_helpers.h>
3+
4+
Y_UNIT_TEST_SUITE(GroupConfigurationPropagation) {
5+
6+
struct TTestCtx : public TTestCtxBase {
7+
TTestCtx(const TBlobStorageGroupType& erasure)
8+
: TTestCtxBase(TEnvironmentSetup::TSettings{
9+
.NodeCount = erasure.BlobSubgroupSize() * 2,
10+
.Erasure = erasure,
11+
.ControllerNodeId = erasure.BlobSubgroupSize() * 2,
12+
})
13+
{}
14+
15+
static bool BlockBscResponses(ui32, std::unique_ptr<IEventHandle>& ev) {
16+
if (ev->GetTypeRewrite() == TEvBlobStorage::TEvControllerNodeServiceSetUpdate::EventType) {
17+
return false;
18+
}
19+
20+
return true;
21+
}
22+
23+
void AllocateEdgeActorOnNodeWOConfig() {
24+
std::set<ui32> nodesWOConfig = GetComplementNodeSet(NodesWithConfig);
25+
UNIT_ASSERT(!NodesWithConfig.empty());
26+
AllocateEdgeActorOnSpecificNode(*nodesWOConfig.begin());
27+
NodeWithProxy = Edge.NodeId();
28+
}
29+
30+
void CheckStatus() {
31+
AllocateEdgeActorOnNodeWOConfig();
32+
auto res = GetGroupStatus(GroupId, WaitTime);
33+
UNIT_ASSERT(res);
34+
UNIT_ASSERT_VALUES_EQUAL_C(res->Get()->Status, NKikimrProto::OK, res->Get()->ErrorReason);
35+
}
36+
37+
void Setup() {
38+
Initialize();
39+
NodesWithConfig = GetNodesWithVDisks();
40+
NodesWithConfig.insert(Env->Settings.ControllerNodeId);
41+
AllocateEdgeActorOnNodeWOConfig();
42+
UNIT_ASSERT(NodeWithProxy != Env->Settings.ControllerNodeId);
43+
Env->Sim(TDuration::Minutes(10));
44+
Env->Runtime->FilterFunction = BlockBscResponses;
45+
}
46+
47+
void StopNode() {
48+
Env->StopNode(NodeWithProxy);
49+
}
50+
51+
void StartNode() {
52+
Env->StartNode(NodeWithProxy);
53+
AllocateEdgeActorOnNodeWOConfig();
54+
}
55+
56+
void RestartNode() {
57+
StopNode();
58+
StartNode();
59+
}
60+
61+
void UpdateGroupConfiguration() {
62+
NKikimrBlobStorage::TConfigRequest request;
63+
auto* reassign = request.AddCommand()->MutableReassignGroupDisk();
64+
for (const auto& slot : BaseConfig.GetVSlot()) {
65+
if (slot.GetGroupId() == GroupId) {
66+
reassign->SetGroupId(slot.GetGroupId());
67+
reassign->SetGroupGeneration(slot.GetGroupGeneration());
68+
reassign->SetFailRealmIdx(slot.GetFailRealmIdx());
69+
reassign->SetFailDomainIdx(slot.GetFailDomainIdx());
70+
reassign->SetVDiskIdx(slot.GetVDiskIdx());
71+
break;
72+
}
73+
}
74+
auto response = Env->Invoke(request);
75+
BaseConfig = Env->FetchBaseConfig();
76+
77+
std::set<ui32> nodesWithVDisks = GetNodesWithVDisks();
78+
NodesWithConfig.insert(nodesWithVDisks.begin(), nodesWithVDisks.end());
79+
}
80+
81+
const TDuration WaitTime = TDuration::Seconds(1);
82+
ui32 NodeWithProxy;
83+
std::set<ui32> NodesWithConfig;
84+
};
85+
86+
Y_UNIT_TEST(Simple) {
87+
TTestCtx ctx(TBlobStorageGroupType::Erasure4Plus2Block);
88+
ctx.Setup();
89+
ctx.CheckStatus();
90+
}
91+
92+
Y_UNIT_TEST(Reassign) {
93+
TTestCtx ctx(TBlobStorageGroupType::Erasure4Plus2Block);
94+
ctx.Setup();
95+
ctx.UpdateGroupConfiguration();
96+
ctx.CheckStatus();
97+
}
98+
99+
Y_UNIT_TEST(NodeRestart) {
100+
TTestCtx ctx(TBlobStorageGroupType::Erasure4Plus2Block);
101+
ctx.Setup();
102+
ctx.RestartNode();
103+
ctx.CheckStatus();
104+
}
105+
106+
Y_UNIT_TEST(NodeRestartAndUpdate) {
107+
TTestCtx ctx(TBlobStorageGroupType::Erasure4Plus2Block);
108+
ctx.Setup();
109+
ctx.StopNode();
110+
ctx.UpdateGroupConfiguration();
111+
ctx.StartNode();
112+
ctx.CheckStatus();
113+
}
114+
115+
Y_UNIT_TEST(BscRestart) {
116+
TTestCtx ctx(TBlobStorageGroupType::Erasure4Plus2Block);
117+
ctx.Setup();
118+
ctx.Env->RestartNode(ctx.Env->Settings.ControllerNodeId);
119+
ctx.CheckStatus();
120+
}
121+
122+
}

‎ydb/core/blobstorage/ut_blobstorage/ut_helpers.h

Lines changed: 38 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -278,19 +278,34 @@ struct TTestCtxBase {
278278
GroupId = group.GetGroupId();
279279
}
280280

281-
void AllocateEdgeActor() {
282-
Edge = Env->Runtime->AllocateEdgeActor(NodeCount);
281+
void AllocateEdgeActor(bool findNodeWithoutVDisks = false) {
282+
ui32 chosenNodeId = 0;
283+
if (!findNodeWithoutVDisks) {
284+
chosenNodeId = NodeCount;
285+
} else {
286+
std::set<ui32> nodesWOVDisks = GetComplementNodeSet(GetNodesWithVDisks());
287+
chosenNodeId = *nodesWOVDisks.begin();
288+
}
289+
290+
Y_VERIFY_S(chosenNodeId != 0, "No available nodes to allocate");
291+
Edge = Env->Runtime->AllocateEdgeActor(chosenNodeId);
292+
}
293+
294+
void AllocateEdgeActorOnSpecificNode(ui32 nodeId) {
295+
Edge = Env->Runtime->AllocateEdgeActor(nodeId);
283296
}
284297

285298
void FetchBaseConfig() {
286299
BaseConfig = Env->FetchBaseConfig();
287300
}
288301

289-
TAutoPtr<TEventHandle<TEvBlobStorage::TEvStatusResult>> GetGroupStatus(ui32 groupId) {
302+
TAutoPtr<TEventHandle<TEvBlobStorage::TEvStatusResult>> GetGroupStatus(ui32 groupId,
303+
TDuration waitTime = TDuration::Max()) {
304+
TInstant ts = (waitTime == TDuration::Max()) ? TInstant::Max() : Env->Now() + waitTime;
290305
Env->Runtime->WrapInActorContext(Edge, [&] {
291-
SendToBSProxy(Edge, groupId, new TEvBlobStorage::TEvStatus(TInstant::Max()));
306+
SendToBSProxy(Edge, groupId, new TEvBlobStorage::TEvStatus(ts));
292307
});
293-
return Env->WaitForEdgeActorEvent<TEvBlobStorage::TEvStatusResult>(Edge, false, TInstant::Max());
308+
return Env->WaitForEdgeActorEvent<TEvBlobStorage::TEvStatusResult>(Edge, false, ts);
294309
}
295310

296311
virtual void Initialize() {
@@ -389,6 +404,24 @@ struct TTestCtxBase {
389404
return blobs;
390405
}
391406

407+
std::set<ui32> GetNodesWithVDisks() {
408+
std::set<ui32> res;
409+
for (const auto& vslot : BaseConfig.GetVSlot()) {
410+
res.insert(vslot.GetVSlotId().GetNodeId());
411+
}
412+
return res;
413+
}
414+
415+
std::set<ui32> GetComplementNodeSet(std::set<ui32> nodes) {
416+
std::set<ui32> res;
417+
for (ui32 nodeId = 1; nodeId <= NodeCount; ++nodeId) {
418+
if (!nodes.contains(nodeId)) {
419+
res.insert(nodeId);
420+
}
421+
}
422+
return res;
423+
}
424+
392425
public:
393426
ui32 NodeCount;
394427
TBlobStorageGroupType Erasure;

‎ydb/core/blobstorage/ut_blobstorage/ya.make

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ SRCS(
1717
acceleration.cpp
1818
assimilation.cpp
1919
block_race.cpp
20+
bsc_cache.cpp
2021
counting_events.cpp
2122
deadlines.cpp
2223
decommit_3dc.cpp

‎ydb/core/mind/bscontroller/config.cpp

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ namespace NKikimr::NBsController {
4040

4141
if (CacheUpdate.KeyValuePairsSize()) {
4242
State.Outbox.emplace_back(Self->SelfId().NodeId(), std::make_unique<NStorage::TEvNodeWardenUpdateCache>(
43-
std::move(CacheUpdate)), 0);
43+
std::move(CacheUpdate)), 0, true);
4444
}
4545
}
4646

@@ -708,8 +708,12 @@ namespace NKikimr::NBsController {
708708
}
709709

710710
ui64 TBlobStorageController::TConfigState::ApplyConfigUpdates() {
711-
for (auto& [nodeId, ev, cookie] : Outbox) {
712-
Self.SendToWarden(nodeId, std::move(ev), cookie);
711+
for (TOutgoingMessage& msg : Outbox) {
712+
if (msg.ToLocalWarden) {
713+
Self.Send(MakeBlobStorageNodeWardenID(Self.SelfId().NodeId()), std::move(msg.Event), 0, msg.Cookie);
714+
} else {
715+
Self.SendToWarden(msg.NodeId, std::move(msg.Event), msg.Cookie);
716+
}
713717
}
714718
for (auto& ev : StatProcessorOutbox) {
715719
Self.SelfId().Send(Self.StatProcessorActorId, ev.release());

‎ydb/core/mind/bscontroller/config.h

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,22 @@ namespace NKikimr {
9191
THashSet<TPDiskId> PDisksToRemove;
9292

9393
// outgoing messages
94-
std::deque<std::tuple<TNodeId, std::unique_ptr<IEventBase>, ui64>> Outbox;
94+
struct TOutgoingMessage {
95+
TNodeId NodeId;
96+
std::unique_ptr<IEventBase> Event;
97+
ui64 Cookie;
98+
bool ToLocalWarden;
99+
100+
TOutgoingMessage(TNodeId nodeId, std::unique_ptr<IEventBase>&& event,
101+
ui64 cookie, bool toLocalWarden = false)
102+
: NodeId(nodeId)
103+
, Event(std::forward<std::unique_ptr<IEventBase>>(event))
104+
, Cookie(cookie)
105+
, ToLocalWarden(toLocalWarden)
106+
{}
107+
};
108+
109+
std::deque<TOutgoingMessage> Outbox;
95110
std::deque<std::unique_ptr<IEventBase>> StatProcessorOutbox;
96111
std::deque<std::unique_ptr<IEventBase>> NodeWhiteboardOutbox;
97112
THolder<TEvControllerUpdateSelfHealInfo> UpdateSelfHealInfoMsg;

‎ydb/core/mind/bscontroller/register_node.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -663,7 +663,7 @@ void TBlobStorageController::EraseKnownDrivesOnDisconnected(TNodeInfo *nodeInfo)
663663

664664
void TBlobStorageController::SendToWarden(TNodeId nodeId, std::unique_ptr<IEventBase> ev, ui64 cookie) {
665665
Y_ABORT_UNLESS(nodeId);
666-
if (auto *node = FindNode(nodeId); node && node->ConnectedServerId) {
666+
if (TNodeInfo* node = FindNode(nodeId); node && node->ConnectedServerId) {
667667
auto h = std::make_unique<IEventHandle>(MakeBlobStorageNodeWardenID(nodeId), SelfId(), ev.release(), 0, cookie);
668668
if (node->InterconnectSessionId) {
669669
h->Rewrite(TEvInterconnect::EvForward, node->InterconnectSessionId);

0 commit comments

Comments
 (0)