Skip to content

Commit 4c04a4a

Browse files
SammyVimesSammyVimes
authored
Merge 2234ced into 8004b1a 
2 parents 8004b1a + 2234ced commit 4c04a4a

7 files changed

+104
-28
lines changed

‎ydb/apps/dstool/lib/dstool_cmd_cluster_workload_run.py

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ def add_options(p):
1818
p.add_argument('--enable-pdisk-encryption-keys-changes', action='store_true', help='Enable changes of PDisk encryption keys')
1919
p.add_argument('--enable-kill-tablets', action='store_true', help='Enable tablet killer')
2020
p.add_argument('--enable-kill-blob-depot', action='store_true', help='Enable BlobDepot killer')
21+
p.add_argument('--enable-restart-pdisks', action='store_true', help='Enable PDisk restarter')
2122
p.add_argument('--kill-signal', type=str, default='KILL', help='Kill signal to send to restart node')
2223

2324

@@ -144,6 +145,19 @@ def do_restart(node_id):
144145
if args.enable_pdisk_encryption_keys_changes:
145146
remove_old_pdisk_keys(pdisk_keys, pdisk_key_versions, node_id)
146147

148+
def do_restart_pdisk(node_id, pdisk_id):
149+
assert can_act_on_vslot(node_id, pdisk_id)
150+
request = common.kikimr_bsconfig.TConfigRequest(IgnoreDegradedGroupsChecks=True)
151+
cmd = request.Command.add().RestartPDisk
152+
cmd.TargetPDiskId.NodeId = node_id
153+
cmd.TargetPDiskId.PDiskId = pdisk_id
154+
try:
155+
response = common.invoke_bsc_request(request)
156+
except Exception as e:
157+
raise Exception('failed to perform restart request: %s' % e)
158+
if not response.Success:
159+
raise Exception('Unexpected error from BSC: %s' % response.ErrorDescription)
160+
147161
def do_evict(vslot_id):
148162
assert can_act_on_vslot(*vslot_id)
149163
try:
@@ -230,13 +244,16 @@ def do_kill_blob_depot():
230244
wipes = []
231245
readonlies = []
232246
unreadonlies = []
247+
pdisk_restarts = []
233248

234249
for vslot in base_config.VSlot:
235250
if common.is_dynamic_group(vslot.GroupId):
236251
vslot_id = common.get_vslot_id(vslot.VSlotId)
237252
vdisk_id = '[%08x:%d:%d:%d]' % (vslot.GroupId, vslot.FailRealmIdx, vslot.FailDomainIdx, vslot.VDiskIdx)
238253
if vslot_id in vslot_readonly and not args.disable_readonly:
239254
unreadonlies.append(('un-readonly vslot id: %s, vdisk id: %s' % (vslot_id, vdisk_id), (do_readonly, vslot, False)))
255+
if can_act_on_vslot(*vslot_id[:2]) and args.enable_restart_pdisks:
256+
pdisk_restarts.append(('restart pdisk node_id: %d, pdisk_id: %d' % vslot_id[:2], (do_restart_pdisk, *vslot_id[:2])))
240257
if can_act_on_vslot(*vslot_id) and (recent_restarts or args.disable_restarts):
241258
if not args.disable_evicts:
242259
evicts.append(('evict vslot id: %s, vdisk id: %s' % (vslot_id, vdisk_id), (do_evict, vslot_id)))
@@ -258,6 +275,8 @@ def pick(v):
258275
possible_actions.append(('readonly', (pick, readonlies)))
259276
if unreadonlies:
260277
possible_actions.append(('un-readonly', (pick, unreadonlies)))
278+
if pdisk_restarts:
279+
possible_actions.append(('restart-pdisk', (pick, pdisk_restarts)))
261280

262281
restarts = []
263282

‎ydb/apps/dstool/lib/dstool_cmd_group_virtual_create.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ def do(args):
6767
names_remaining.remove(group.VirtualGroupInfo.Name)
6868
elif group.VirtualGroupInfo.State == common.EVirtualGroupState.CREATE_FAILED:
6969
names_remaining.remove(group.VirtualGroupInfo.Name)
70-
errors.append(f'{group.VirtualGroupInfo.Name}: {group.ErrorReason}')
70+
errors.append(f'{group.VirtualGroupInfo.Name}: {group.VirtualGroupInfo.ErrorReason}')
7171

7272
if names_remaining:
7373
time.sleep(1)

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

Lines changed: 14 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -273,23 +273,30 @@ namespace NKikimr::NStorage {
273273
void TNodeWarden::DoRestartLocalPDisk(const NKikimrBlobStorage::TNodeWardenServiceSet::TPDisk& pdisk) {
274274
ui32 pdiskId = pdisk.GetPDiskID();
275275

276-
const TActorId actorId = MakeBlobStoragePDiskID(LocalNodeId, pdiskId);
276+
STLOG(PRI_NOTICE, BS_NODE, NW75, "DoRestartLocalPDisk", (PDiskId, pdiskId));
277+
278+
const auto [_, inserted] = PDiskRestartInFlight.emplace(pdiskId);
279+
280+
if (!inserted) {
281+
STLOG(PRI_NOTICE, BS_NODE, NW76, "Restart already in progress", (PDiskId, pdiskId));
282+
// Restart is already in progress.
283+
return;
284+
}
277285

278286
auto it = LocalPDisks.find(TPDiskKey(LocalNodeId, pdiskId));
279287
if (it == LocalPDisks.end()) {
288+
PDiskRestartInFlight.erase(pdiskId);
289+
290+
STLOG(PRI_NOTICE, BS_NODE, NW77, "Restart state carried from previous start, just starting", (PDiskId, pdiskId));
291+
280292
// This can happen if warden didn't handle pdisk's restart before node's restart.
281293
// In this case, PDisk has EntityStatus::RESTART instead of EntityStatus::INITIAL.
282294
StartLocalPDisk(pdisk);
283295
SendPDiskReport(pdiskId, NKikimrBlobStorage::TEvControllerNodeReport::PD_RESTARTED);
284296
return;
285297
}
286298

287-
const auto [_, inserted] = PDiskRestartInFlight.emplace(pdiskId);
288-
289-
if (!inserted) {
290-
// Restart is already in progress.
291-
return;
292-
}
299+
const TActorId actorId = MakeBlobStoragePDiskID(LocalNodeId, pdiskId);
293300

294301
TIntrusivePtr<TPDiskConfig> pdiskConfig = CreatePDiskConfig(it->second.Record);
295302

‎ydb/core/blobstorage/pdisk/blobstorage_pdisk_actor.cpp

Lines changed: 45 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,21 @@ namespace NPDisk {
3939

4040
LWTRACE_USING(BLOBSTORAGE_PROVIDER);
4141

42+
void CreatePDiskActor(
43+
TGenericExecutorThread& executorThread,
44+
const TIntrusivePtr<::NMonitoring::TDynamicCounters>& counters,
45+
const TIntrusivePtr<TPDiskConfig> &cfg,
46+
const NPDisk::TMainKey &mainKey,
47+
ui32 pDiskID, ui32 poolId, ui32 nodeId
48+
) {
49+
50+
TActorId actorId = executorThread.RegisterActor(CreatePDisk(cfg, mainKey, counters), TMailboxType::ReadAsFilled, poolId);
51+
52+
TActorId pDiskServiceId = MakeBlobStoragePDiskID(nodeId, pDiskID);
53+
54+
executorThread.ActorSystem->RegisterLocalService(pDiskServiceId, actorId);
55+
}
56+
4257
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
4358
// PDisk Actor
4459
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -993,29 +1008,48 @@ class TPDiskActor : public TActorBootstrapped<TPDiskActor> {
9931008
return;
9941009
}
9951010

1011+
if (PendingRestartResponse) {
1012+
PendingRestartResponse(restartAllowed, ev->Get()->Details);
1013+
PendingRestartResponse = {};
1014+
}
1015+
9961016
if (restartAllowed) {
997-
MainKey = ev->Get()->MainKey;
1017+
NPDisk::TMainKey newMainKey = ev->Get()->MainKey;
1018+
9981019
SecureWipeBuffer((ui8*)ev->Get()->MainKey.Keys.data(), sizeof(NPDisk::TKey) * ev->Get()->MainKey.Keys.size());
1020+
9991021
LOG_NOTICE_S(*TlsActivationContext, NKikimrServices::BS_PDISK, "PDiskId# " << PDisk->PDiskId
10001022
<< " Going to restart PDisk since recieved TEvAskWardenRestartPDiskResult");
10011023

1024+
const TActorIdentity& thisActorId = SelfId();
1025+
ui32 nodeId = thisActorId.NodeId();
1026+
ui32 poolId = thisActorId.PoolID();
1027+
ui32 pdiskId = PDisk->PDiskId;
1028+
10021029
PDisk->Stop();
10031030

1031+
TIntrusivePtr<TPDiskConfig> actorCfg = std::move(Cfg);
1032+
10041033
auto& newCfg = ev->Get()->Config;
1034+
10051035
if (newCfg) {
1006-
Y_VERIFY_S(Cfg->PDiskId == PDisk->PDiskId,
1007-
"New config's PDiskId# " << newCfg->PDiskId << " is not equal to real PDiskId# " << PDisk->PDiskId);
1008-
Cfg = std::move(newCfg);
1036+
Y_VERIFY_S(newCfg->PDiskId == pdiskId,
1037+
"New config's PDiskId# " << newCfg->PDiskId << " is not equal to real PDiskId# " << pdiskId);
1038+
1039+
actorCfg = std::move(newCfg);
10091040
}
10101041

1011-
StartPDiskThread();
1042+
const TActorContext& actorCtx = ActorContext();
10121043

1013-
Send(ev->Sender, new TEvBlobStorage::TEvNotifyWardenPDiskRestarted(PDisk->PDiskId));
1014-
}
1044+
auto& counters = AppData(actorCtx)->Counters;
10151045

1016-
if (PendingRestartResponse) {
1017-
PendingRestartResponse(restartAllowed, ev->Get()->Details);
1018-
PendingRestartResponse = {};
1046+
TGenericExecutorThread& executorThread = actorCtx.ExecutorThread;
1047+
1048+
PassAway();
1049+
1050+
CreatePDiskActor(executorThread, counters, actorCfg, newMainKey, pdiskId, poolId, nodeId);
1051+
1052+
Send(ev->Sender, new TEvBlobStorage::TEvNotifyWardenPDiskRestarted(pdiskId));
10191053
}
10201054
}
10211055

@@ -1290,10 +1324,7 @@ IActor* CreatePDisk(const TIntrusivePtr<TPDiskConfig> &cfg, const NPDisk::TMainK
12901324

12911325
void TRealPDiskServiceFactory::Create(const TActorContext &ctx, ui32 pDiskID,
12921326
const TIntrusivePtr<TPDiskConfig> &cfg, const NPDisk::TMainKey &mainKey, ui32 poolId, ui32 nodeId) {
1293-
TActorId actorId = ctx.ExecutorThread.RegisterActor(
1294-
CreatePDisk(cfg, mainKey, AppData(ctx)->Counters), TMailboxType::ReadAsFilled, poolId);
1295-
TActorId pDiskServiceId = MakeBlobStoragePDiskID(nodeId, pDiskID);
1296-
ctx.ExecutorThread.ActorSystem->RegisterLocalService(pDiskServiceId, actorId);
1327+
CreatePDiskActor(ctx.ExecutorThread, AppData(ctx)->Counters, cfg, mainKey, pDiskID, poolId, nodeId);
12971328
}
12981329

12991330
} // NKikimr

‎ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -305,6 +305,7 @@ void TPDisk::Stop() {
305305
LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
306306
<< " shutdown owner info# " << StartupOwnerInfo());
307307
}
308+
308309
BlockDevice->Stop();
309310

310311
// BlockDevice is stopped, the data will NOT hit the disk.

‎ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut.cpp

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -503,8 +503,9 @@ Y_UNIT_TEST_SUITE(TPDiskTest) {
503503
TVDiskMock vdisk(&testCtx);
504504
vdisk.InitFull();
505505
vdisk.SendEvLogSync();
506-
testCtx.Send(new TEvBlobStorage::TEvAskWardenRestartPDiskResult(testCtx.GetPDisk()->PDiskId, testCtx.MainKey, true, nullptr));
507-
const auto evInitRes = testCtx.Recv<TEvBlobStorage::TEvNotifyWardenPDiskRestarted>();
506+
507+
testCtx.StartPDiskRestart();
508+
508509
vdisk.InitFull();
509510
vdisk.SendEvLogSync();
510511
}
@@ -928,8 +929,7 @@ Y_UNIT_TEST_SUITE(TPDiskTest) {
928929
while (writeLog() == NKikimrProto::OK) {}
929930
UNIT_ASSERT_VALUES_EQUAL(writeLog(), NKikimrProto::OUT_OF_SPACE);
930931

931-
testCtx.Send(new TEvBlobStorage::TEvAskWardenRestartPDiskResult(testCtx.GetPDisk()->PDiskId, testCtx.MainKey, true, nullptr));
932-
const auto evInitRes = testCtx.Recv<TEvBlobStorage::TEvNotifyWardenPDiskRestarted>();
932+
testCtx.StartPDiskRestart();
933933

934934
vdisk.InitFull();
935935
vdisk.SendEvLogSync();
@@ -943,8 +943,8 @@ Y_UNIT_TEST_SUITE(PDiskCompatibilityInfo) {
943943
using TCurrent = NKikimrConfig::TCurrentCompatibilityInfo;
944944
THolder<NPDisk::TEvYardInitResult> RestartPDisk(TActorTestContext& testCtx, ui32 pdiskId, TVDiskMock& vdisk, TCurrent* newInfo) {
945945
TCompatibilityInfoTest::Reset(newInfo);
946-
testCtx.Send(new TEvBlobStorage::TEvAskWardenRestartPDiskResult(pdiskId, testCtx.MainKey, true, nullptr));
947-
testCtx.Recv<TEvBlobStorage::TEvNotifyWardenPDiskRestarted>();
946+
Y_UNUSED(pdiskId);
947+
testCtx.StartPDiskRestart();
948948
testCtx.Send(new NPDisk::TEvYardInit(vdisk.OwnerRound.fetch_add(1), vdisk.VDiskID, testCtx.TestCtx.PDiskGuid));
949949
return testCtx.Recv<NPDisk::TEvYardInitResult>();
950950
}

‎ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_env.h

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,9 +125,27 @@ struct TActorTestContext {
125125
new NPDisk::TEvYardControl(NPDisk::TEvYardControl::GetPDiskPointer, nullptr),
126126
NKikimrProto::OK);
127127
PDisk = reinterpret_cast<NPDisk::TPDisk*>(evControlRes->Cookie);
128+
129+
PDiskActor = PDisk->PDiskActor;
128130
}
129131
return PDisk;
130132
}
133+
134+
void StartPDiskRestart() {
135+
ui32 pdiskId = GetPDisk()->PDiskId;
136+
137+
Send(new TEvBlobStorage::TEvAskWardenRestartPDiskResult(pdiskId, MainKey, true, nullptr));
138+
const auto evInitRes = Recv<TEvBlobStorage::TEvNotifyWardenPDiskRestarted>();
139+
140+
if (!Settings.UsePDiskMock) {
141+
TActorId wellKnownPDiskActorId = MakeBlobStoragePDiskID(PDiskActor->NodeId(), pdiskId);
142+
143+
PDisk = nullptr;
144+
145+
// We will temporarily use well know pdisk actor id, because restarted pdisk actor id is not yet known.
146+
PDiskActor = wellKnownPDiskActorId;
147+
}
148+
}
131149

132150
template<typename T>
133151
auto SafeRunOnPDisk(T&& f) {

0 commit comments

Comments
 (0)