Straighten BSC <-> NW interoperation protocol

Author: alexvru

ydb/core/base/blobstorage.h

@@ -786,64 +786,35 @@ struct TEvBlobStorage {
  EvBunchOfEvents,
 
  // blobstorage controller interface
- // EvControllerReadSchemeString = EvPut + 11 * 512,
- // EvControllerReadDataString,
- EvControllerRegisterNode = EvPut + 11 * 512 + 2,
- EvControllerCreatePDisk,
- EvControllerCreateVDiskSlots,
- EvControllerCreateGroup,
- EvControllerSelectGroups,
- EvControllerGetGroup,
- EvControllerUpdateDiskStatus,
- EvControllerUpdateGroupsUsage, // Not used.
- EvControllerConfigRequest,
- EvControllerConfigResponse,
- EvControllerProposeRequest,
- EvControllerProposeResponse,
- EvControllerVDiskStatusSubscribeRequest,
- EvControllerVDiskStatusReport,
- EvControllerGroupStatusRequest,
- EvControllerGroupStatusResponse,
- EvControllerUpdateGroup,
- EvControllerUpdateFaultyDisks,
- EvControllerProposeGroupKey,
- EvControllerUpdateGroupLatencies, // Not used.
- EvControllerUpdateGroupStat,
- EvControllerNotifyGroupChange,
- EvControllerCommitGroupLatencies,
- EvControllerUpdateSelfHealInfo,
- EvControllerScrubQueryStartQuantum,
- EvControllerScrubQuantumFinished,
- EvControllerScrubReportQuantumInProgress,
- EvControllerUpdateNodeDrives,
- EvControllerGroupDecommittedNotify,
- EvControllerGroupDecommittedResponse,
- EvControllerGroupMetricsExchange,
-
- // EvControllerReadSchemeStringResult = EvPut + 12 * 512,
- // EvControllerReadDataStringResult,
- EvControllerNodeServiceSetUpdate = EvPut + 12 * 512 + 2,
- EvControllerCreatePDiskResult,
- EvControllerCreateVDiskSlotsResult,
- EvControllerCreateGroupResult,
- EvControllerSelectGroupsResult,
- EvRequestControllerInfo,
- EvResponseControllerInfo,
- EvControllerGroupReconfigureReplace, // Not used.
- EvControllerGroupReconfigureReplaceResult, // Not used.
- EvControllerGroupReconfigureWipe,
- EvControllerGroupReconfigureWipeResult,
- EvControllerNodeReport,
- EvControllerScrubStartQuantum,
-
- EvControllerMigrationPause,
- EvControllerMigrationContinue,
- EvControllerMigrationFinished,
- EvControllerMigrationBatch,
- EvControllerMigrationBatchRequest,
- EvControllerMigrationDone,
-
- EvControllerUpdateSystemViews,
+ EvControllerRegisterNode = 0x10031602,
+ EvControllerSelectGroups = 0x10031606,
+ EvControllerGetGroup = 0x10031607,
+ EvControllerUpdateDiskStatus = 0x10031608,
+ EvControllerConfigRequest = 0x1003160a,
+ EvControllerConfigResponse = 0x1003160b,
+ EvControllerProposeGroupKey = 0x10031614,
+ EvControllerUpdateGroupStat = 0x10031616,
+ EvControllerNotifyGroupChange = 0x10031617,
+ EvControllerCommitGroupLatencies = 0x10031618,
+ EvControllerUpdateSelfHealInfo = 0x10031619,
+ EvControllerScrubQueryStartQuantum = 0x1003161a,
+ EvControllerScrubQuantumFinished = 0x1003161b,
+ EvControllerScrubReportQuantumInProgress = 0x1003161c,
+ EvControllerUpdateNodeDrives = 0x1003161d,
+ EvControllerGroupDecommittedNotify = 0x1003161e,
+ EvControllerGroupDecommittedResponse = 0x1003161f,
+ EvControllerGroupMetricsExchange = 0x10031620,
+
+ // BSC interface result section
+ EvControllerNodeServiceSetUpdate = 0x10031802,
+ EvControllerSelectGroupsResult = 0x10031806,
+ EvRequestControllerInfo = 0x10031807,
+ EvResponseControllerInfo = 0x10031808,
+ EvControllerGroupReconfigureWipe = 0x1003180b,
+ EvControllerGroupReconfigureWipeResult = 0x1003180c,
+ EvControllerNodeReport = 0x1003180d,
+ EvControllerScrubStartQuantum = 0x1003180e,
+ EvControllerUpdateSystemViews = 0x10031815,
 
  // proxy - node controller interface
  EvConfigureProxy = EvPut + 13 * 512,

ydb/core/blobstorage/vdisk/common/blobstorage_event_filter.cpp

@@ -54,12 +54,6 @@ namespace NKikimr {
  TEvBlobStorage::EvControllerUpdateDiskStatus,
  TEvBlobStorage::EvControllerConfigRequest,
  TEvBlobStorage::EvControllerConfigResponse,
- TEvBlobStorage::EvControllerVDiskStatusSubscribeRequest,
- TEvBlobStorage::EvControllerVDiskStatusReport,
- TEvBlobStorage::EvControllerGroupStatusRequest,
- TEvBlobStorage::EvControllerGroupStatusResponse,
- TEvBlobStorage::EvControllerUpdateGroup,
- TEvBlobStorage::EvControllerUpdateFaultyDisks,
  TEvBlobStorage::EvControllerUpdateGroupStat,
 
  TEvBlobStorage::EvControllerSelectGroupsResult,

ydb/core/mind/bscontroller/config.cpp

@@ -17,22 +17,21 @@ namespace NKikimr::NBsController {
  , State(state)
  {}
 
- void Execute(std::deque<std::unique_ptr<IEventHandle>>& outbox) {
+ void Execute() {
  ApplyUpdates();
 
  for (auto &pair : Services) {
  const TNodeId &nodeId = pair.first;
 
- if (TNodeInfo *node = Self->FindNode(nodeId); node && node->ConnectedCount) {
- auto event = MakeHolder<TEvBlobStorage::TEvControllerNodeServiceSetUpdate>();
+ if (TNodeInfo *node = Self->FindNode(nodeId); node && node->ConnectedServerId) {
+ auto event = std::make_unique<TEvBlobStorage::TEvControllerNodeServiceSetUpdate>();
  auto& record = event->Record;
  pair.second.Swap(&record);
  record.SetStatus(NKikimrProto::OK);
  record.SetNodeID(nodeId);
  record.SetInstanceId(Self->InstanceId);
  record.SetAvailDomain(AppData()->DomainsInfo->GetDomain()->DomainUid);
- outbox.push_back(std::make_unique<IEventHandle>(MakeBlobStorageNodeWardenID(nodeId),
- Self->SelfId(), event.Release()));
+ State.Outbox.emplace_back(nodeId, std::move(event), 0);
  }
  }
  }
@@ -448,7 +447,7 @@ namespace NKikimr::NBsController {
  }
  }
 
- TNodeWardenUpdateNotifier(this, state).Execute(state.Outbox);
+ TNodeWardenUpdateNotifier(this, state).Execute();
 
  state.CheckConsistency();
  state.Commit();
@@ -460,7 +459,7 @@ namespace NKikimr::NBsController {
  }
 
  void TBlobStorageController::CommitSelfHealUpdates(TConfigState& state) {
- auto ev = MakeHolder<TEvControllerNotifyGroupChange>();
+ auto ev = std::make_unique<TEvControllerNotifyGroupChange>();
  auto sh = MakeHolder<TEvControllerUpdateSelfHealInfo>();
 
  for (auto&& [base, overlay] : state.Groups.Diff()) {
@@ -500,7 +499,7 @@ namespace NKikimr::NBsController {
  }
 
  if (ev->Created || ev->Deleted) {
- state.Outbox.push_back(std::make_unique<IEventHandle>(StatProcessorActorId, SelfId(), ev.Release()));
+ state.StatProcessorOutbox.push_back(std::move(ev));
  }
  if (sh->GroupsToUpdate) {
  FillInSelfHealGroups(*sh, &state);
@@ -590,8 +589,11 @@ namespace NKikimr::NBsController {
  }
 
  ui64 TBlobStorageController::TConfigState::ApplyConfigUpdates() {
- for (auto& msg : Outbox) {
- TActivationContext::Send(msg.release());
+ for (auto& [nodeId, ev, cookie] : Outbox) {
+ Self.SendToWarden(nodeId, std::move(ev), cookie);
+ }
+ for (auto& ev : StatProcessorOutbox) {
+ Self.SelfId().Send(Self.StatProcessorActorId, ev.release());
  }
 
  if (UpdateSelfHealInfoMsg) {

ydb/core/mind/bscontroller/config.h

@@ -91,7 +91,8 @@ namespace NKikimr {
  THashSet<TPDiskId> PDisksToRemove;
 
  // outgoing messages
- std::deque<std::unique_ptr<IEventHandle>> Outbox;
+ std::deque<std::tuple<TNodeId, std::unique_ptr<IEventBase>, ui64>> Outbox;
+ std::deque<std::unique_ptr<IEventBase>> StatProcessorOutbox;
  THolder<TEvControllerUpdateSelfHealInfo> UpdateSelfHealInfoMsg;
 
  // deferred callbacks

ydb/core/mind/bscontroller/get_group.cpp

@@ -4,7 +4,8 @@ namespace NKikimr::NBsController {
 
 class TBlobStorageController::TTxGetGroup : public TTransactionBase<TBlobStorageController> {
  TEvBlobStorage::TEvControllerGetGroup::TPtr Request;
- std::unique_ptr<IEventHandle> Response;
+ std::unique_ptr<TEvBlobStorage::TEvControllerNodeServiceSetUpdate> Response;
+ TNodeId NodeId = {};
 
 public:
  TTxGetGroup(TEvBlobStorage::TEvControllerGetGroup::TPtr& ev, TBlobStorageController *controller)
@@ -20,27 +21,34 @@ class TBlobStorageController::TTxGetGroup : public TTransactionBase<TBlobStorage
 
  STLOG(PRI_DEBUG, BS_CONTROLLER, BSCTXGG01, "Handle TEvControllerGetGroup", (Request, Request->Get()->Record));
 
+ NodeId = Request->Get()->Record.GetNodeID();
+
+ if (!Self->ValidateIncomingNodeWardenEvent(*Request)) {
+ Response = std::make_unique<TEvBlobStorage::TEvControllerNodeServiceSetUpdate>(NKikimrProto::ERROR, NodeId);
+ return true;
+ }
+
  const auto& v = Request->Get()->Record.GetGroupIDs();
  TSet<ui32> groupIDsToRead(v.begin(), v.end());
 
- const TNodeId nodeId = Request->Get()->Record.GetNodeID();
- auto res = std::make_unique<TEvBlobStorage::TEvControllerNodeServiceSetUpdate>(NKikimrProto::OK, nodeId);
- Self->ReadGroups(groupIDsToRead, true, res.get(), nodeId);
+ Response = std::make_unique<TEvBlobStorage::TEvControllerNodeServiceSetUpdate>(NKikimrProto::OK, NodeId);
+ Self->ReadGroups(groupIDsToRead, true, Response.get(), NodeId);
 
- auto& node = Self->GetNode(nodeId);
+ auto& node = Self->GetNode(NodeId);
  for (TGroupId groupId : v) {
  node.GroupsRequested.insert(groupId);
- Self->GroupToNode.emplace(groupId, nodeId);
+ Self->GroupToNode.emplace(groupId, NodeId);
  }
 
- Response = std::make_unique<IEventHandle>(nodeId ? MakeBlobStorageNodeWardenID(nodeId) : Request->Sender,
- Self->SelfId(), res.release());
-
  return true;
  }
 
  void Complete(const TActorContext&) override {
- TActivationContext::Send(Response.release());
+ if (NodeId) {
+ Self->SendToWarden(NodeId, std::move(Response), Request->Cookie);
+ } else {
+ Self->SendInReply(*Request, std::move(Response));
+ }
  }
 };
 

ydb/core/mind/bscontroller/group_reconfigure_wipe.cpp

@@ -1,93 +1,10 @@
 #include "impl.h"
 
-namespace NKikimr {
-namespace NBsController {
-
-class TBlobStorageController::TTxGroupReconfigureWipe : public TTransactionBase<TBlobStorageController> {
-protected:
- TEvBlobStorage::TEvControllerGroupReconfigureWipe::TPtr Event;
- NKikimrProto::EReplyStatus Status;
- TString ErrorReason;
- typedef TMap<TNodeId, THolder<TEvBlobStorage::TEvControllerNodeServiceSetUpdate>> TResultForNode;
- TResultForNode ResultForNode;
-
-public:
- TTxGroupReconfigureWipe(TEvBlobStorage::TEvControllerGroupReconfigureWipe::TPtr &ev,
- TBlobStorageController *controller)
- : TBase(controller)
- , Event(ev)
- , Status(NKikimrProto::OK)
- {}
-
- TTxType GetTxType() const override { return NBlobStorageController::TXTYPE_GROUP_RECONFIGURE_WIPE; }
-
- bool Execute(TTransactionContext &txc, const TActorContext&) override {
- STLOG(PRI_DEBUG, BS_CONTROLLER, BSCTXGRW01, "TTxGroupReconfigureWipe execute");
- NIceDb::TNiceDb db(txc.DB);
-
- Status = NKikimrProto::ERROR;
- ResultForNode.clear();
-
- auto& record = Event->Get()->Record;
- const TVSlotId id(record.GetVSlotId());
-
- if (TVSlotInfo *info = Self->FindVSlot(id); !info) {
- STLOG(PRI_ERROR, BS_CONTROLLER, BSCTXGRW02, "VSlot does not exist", (VSlotId, id));
- ErrorReason = TStringBuilder() << "VSlotId# " << id << " does not exist";
- } else if (info->Mood != TMood::Normal) { // Reply ERROR if From VSlot is improper mood
- STLOG(PRI_ERROR, BS_CONTROLLER, BSCTXGRW03, "VSlot is not in the 'Normal' mood", (VSlotId, id),
- (Mood, TMood::Name(TMood::EValue(info->Mood))));
- ErrorReason = TStringBuilder() << "VSlotId# " << id << " is in Mood# " << TMood::Name(TMood::EValue(info->Mood));
- } else {
- // TODO(cthulhu): Prohibit more than Max simultaneous reconfigurations on one group
-
- // Mark VSlot for wipe
- info->Mood = TMood::Wipe;
- db.Table<Schema::VSlot>().Key(id.GetKey()).Update<Schema::VSlot::Mood>(info->Mood);
-
- // Prepare results for nodes
- if (TNodeInfo *node = Self->FindNode(id.NodeId); node && node->ConnectedCount) {
- auto& msg = ResultForNode[id.NodeId];
- msg = MakeHolder<TEvBlobStorage::TEvControllerNodeServiceSetUpdate>(NKikimrProto::OK, id.NodeId);
- Self->ReadVSlot(*info, msg.Get());
- TSet<ui32> groupIDsToRead;
- groupIDsToRead.insert(info->GroupId);
- Self->ReadGroups(groupIDsToRead, false, msg.Get(), id.NodeId);
- for (const TGroupId groupId : groupIDsToRead) {
- STLOG(PRI_ERROR, BS_CONTROLLER, BSCTXGRW05, "No configuration for group", (GroupId, groupId));
- }
- }
-
- Status = NKikimrProto::OK;
- }
-
- return true;
- }
-
- void Complete(const TActorContext&) override {
- // Send results to nodes
- if (Status == NKikimrProto::OK) {
- for (auto& [nodeId, msg] : ResultForNode) {
- // TODO(cthulhu): Availability Domain !!!
- const auto& node = nodeId;
- const auto& record = msg->Record;
- STLOG(PRI_DEBUG, BS_CONTROLLER, BSCTXGRW07, "Sending update", (NodeId, node), (Message, record));
- TActivationContext::Send(new IEventHandle(MakeBlobStorageNodeWardenID(nodeId), Self->SelfId(), msg.Release()));
- }
- }
-
- // Respond
- auto response = MakeHolder<TEvBlobStorage::TEvControllerGroupReconfigureWipeResult>(Status, ErrorReason);
- auto& record = Event->Get()->Record;
- STLOG(Status == NKikimrProto::OK ? PRI_DEBUG : PRI_ERROR, BS_CONTROLLER, BSCTXGRW06, "TTxGroupReconfigureWipe complete",
- (Status, Status), (Request, record), (Response, response->Record));
- TActivationContext::Send(new IEventHandle(Event->Sender, Self->SelfId(), response.Release(), 0, Event->Cookie));
- }
-};
+namespace NKikimr::NBsController {
 
 void TBlobStorageController::Handle(TEvBlobStorage::TEvControllerGroupReconfigureWipe::TPtr& ev) {
- Execute(new TTxGroupReconfigureWipe(ev, this));
+ SendInReply(*ev, std::make_unique<TEvBlobStorage::TEvControllerGroupReconfigureWipeResult>(NKikimrProto::ERROR,
+ "unsupported operation"));
 }
 
 }
-}