[Blockstore] preventing the rejection of a significant fraction of Di…

…sk Agents in the cluster (#2746)

cloud/blockstore/config/storage.proto

@@ -1077,4 +1077,10 @@ message TStorageServiceConfig
 
     // Disables the calculation of the FullPlacementGroups counter.
     optional bool DisableFullPlacementGroupCountCalculation = 395;
+
+    // The percentage of agents that is acceptable to reject at the start.
+    // If the number of agents that did not reconnect is higher than this
+    // percentage, then the rejection of such agents does not occur - we assume
+    // a connectivity failure in the cluster.
+    optional double DiskRegistryInitialAgentRejectionThreshold = 396;
 }

cloud/blockstore/libs/diagnostics/critical_events.h

@@ -66,6 +66,7 @@ namespace NCloud::NBlockStore {
     xxx(DiskRegistryOccupiedDeviceConfigurationHasChanged)                     \
     xxx(MirroredDiskChecksumMismatchUponRead)                                  \
     xxx(DiskRegistryWrongMigratedDeviceOwnership)                              \
+    xxx(DiskRegistryInitialAgentRejectionThresholdExceeded)                    \
 // BLOCKSTORE_CRITICAL_EVENTS
 
 #define BLOCKSTORE_IMPOSSIBLE_EVENTS(xxx)                                      \

cloud/blockstore/libs/storage/core/config.cpp

@@ -520,8 +520,7 @@ TDuration MSeconds(ui32 value)
                                                                                \
     xxx(EncryptionAtRestForDiskRegistryBasedDisksEnabled, bool,    false      )\
     xxx(DisableFullPlacementGroupCountCalculation,        bool,    false      )\
-
-
+    xxx(DiskRegistryInitialAgentRejectionThreshold,       double,    50       )\
 // BLOCKSTORE_STORAGE_CONFIG_RW
 
 #define BLOCKSTORE_STORAGE_CONFIG(xxx)                                         \

cloud/blockstore/libs/storage/core/config.h

@@ -621,6 +621,7 @@ class TStorageConfig
     [[nodiscard]] bool GetEncryptionAtRestForDiskRegistryBasedDisksEnabled() const;
 
     [[nodiscard]] bool GetDisableFullPlacementGroupCountCalculation() const;
+    [[nodiscard]] double GetDiskRegistryInitialAgentRejectionThreshold() const;
 };
 
 ui64 GetAllocationUnit(

cloud/blockstore/libs/storage/disk_registry/disk_registry_actor.cpp

@@ -76,7 +76,7 @@ void TDiskRegistryActor::ScheduleMakeBackup(
         - (ctx.Now() - lastBackupTs);
 
     LOG_DEBUG_S(ctx, TBlockStoreComponents::DISK_REGISTRY,
-        "Schedule backup at " << backupPeriod.ToDeadLine());
+        "Schedule backup at " << backupPeriod.ToDeadLine(ctx.Now()));
 
     TString hostPrefix = Config->GetDiskRegistryCountersHost();
     if (!hostPrefix.empty()) {
@@ -103,7 +103,7 @@ void TDiskRegistryActor::ScheduleCleanup(const TActorContext& ctx)
     const auto recyclingPeriod = Config->GetNonReplicatedDiskRecyclingPeriod();
 
     LOG_DEBUG_S(ctx, TBlockStoreComponents::DISK_REGISTRY,
-        "Schedule cleanup at " << recyclingPeriod.ToDeadLine());
+        "Schedule cleanup at " << recyclingPeriod.ToDeadLine(ctx.Now()));
 
     auto request = std::make_unique<TEvDiskRegistryPrivate::TEvCleanupDisksRequest>();
 
@@ -434,14 +434,7 @@ void TDiskRegistryActor::ScheduleRejectAgent(
     TString agentId,
     ui64 seqNo)
 {
-    auto timeout = Config->GetNonReplicatedAgentMaxTimeout();
-
-    // if there is no State it means that this is our initial agent rejection
-    // phase that happens during the LoadState tx => we should use max timeout
-    // in order not to reject all agents at once
-    if (State) {
-        timeout = State->GetRejectAgentTimeout(ctx.Now(), agentId);
-    }
+    auto timeout = State->GetRejectAgentTimeout(ctx.Now(), agentId);
 
     if (!timeout) {
         return;
@@ -458,12 +451,66 @@ void TDiskRegistryActor::ScheduleRejectAgent(
     ctx.Schedule(deadline, request.release());
 }
 
+void TDiskRegistryActor::ProcessInitialAgentRejectionPhase(
+    const TActorContext& ctx)
+{
+    LOG_INFO_S(
+        ctx,
+        TBlockStoreComponents::DISK_REGISTRY,
+        "Process the initial agents rejection phase");
+
+    ui32 expectedToBeOnline = 0;
+    TVector<TString> agentsToReject;
+
+    for (const auto& agent: State->GetAgents()) {
+        if (agent.GetState() == NProto::AGENT_STATE_UNAVAILABLE) {
+            continue;
+        }
+        ++expectedToBeOnline;
+
+        const auto& agentId = agent.GetAgentId();
+
+        if (!AgentRegInfo.contains(agentId)) {
+            agentsToReject.push_back(agentId);
+        }
+    }
+
+    if (agentsToReject.empty() || !expectedToBeOnline) {
+        return;
+    }
+
+    const double k =
+        100.0 * static_cast<double>(agentsToReject.size()) / expectedToBeOnline;
+
+    if (k > Config->GetDiskRegistryInitialAgentRejectionThreshold()) {
+        ReportDiskRegistryInitialAgentRejectionThresholdExceeded(
+            TStringBuilder()
+            << "Too many agents haven't reconnected: " << agentsToReject.size()
+            << "/" << expectedToBeOnline);
+        return;
+    }
+
+    for (const auto& agentId: agentsToReject) {
+        NCloud::Send(
+            ctx,
+            ctx.SelfID,
+            std::make_unique<TEvDiskRegistryPrivate::TEvAgentConnectionLost>(
+                agentId,
+                0));
+    }
+}
+
 void TDiskRegistryActor::HandleAgentConnectionLost(
     const TEvDiskRegistryPrivate::TEvAgentConnectionLost::TPtr& ev,
     const TActorContext& ctx)
 {
     auto* msg = ev->Get();
 
+    if (msg->AgentId.empty()) {
+        ProcessInitialAgentRejectionPhase(ctx);
+        return;
+    }
+
     auto it = AgentRegInfo.find(msg->AgentId);
     if (it != AgentRegInfo.end() && msg->SeqNo < it->second.SeqNo) {
         LOG_DEBUG_S(ctx, TBlockStoreComponents::DISK_REGISTRY,

cloud/blockstore/libs/storage/disk_registry/disk_registry_actor.h

@@ -310,6 +310,8 @@ class TDiskRegistryActor final
 
     void InitializeState(TDiskRegistryStateSnapshot snapshot);
 
+    void ProcessInitialAgentRejectionPhase(const NActors::TActorContext& ctx);
+
 private:
     STFUNC(StateBoot);
     STFUNC(StateInit);

cloud/blockstore/libs/storage/disk_registry/disk_registry_actor_loadstate.cpp

@@ -169,15 +169,21 @@ void TDiskRegistryActor::CompleteLoadState(
     // resend pending requests
     SendPendingRequests(ctx, PendingRequests);
 
-    for (const auto& agent: args.Snapshot.Agents) {
-        if (agent.GetState() != NProto::AGENT_STATE_UNAVAILABLE) {
-            // this event will be scheduled using NonReplicatedAgentMaxTimeout
-            ScheduleRejectAgent(ctx, agent.GetAgentId(), 0);
-        }
-    }
-
     InitializeState(std::move(args.Snapshot));
 
+    if (TDuration timeout = Config->GetNonReplicatedAgentMaxTimeout()) {
+        const auto deadline = timeout.ToDeadLine(ctx.Now());
+
+        LOG_INFO_S(
+            ctx,
+            TBlockStoreComponents::DISK_REGISTRY,
+            "Schedule the initial agents rejection phase to " << deadline);
+
+        auto request =
+            std::make_unique<TEvDiskRegistryPrivate::TEvAgentConnectionLost>();
+        ctx.Schedule(deadline, request.release());
+    }
+
     SecureErase(ctx);
 
     ScheduleCleanup(ctx);

cloud/blockstore/libs/storage/disk_registry/disk_registry_private.h

@@ -492,7 +492,9 @@ struct TEvDiskRegistryPrivate
     struct TAgentConnectionLost
     {
         TString AgentId;
-        ui64 SeqNo;
+        ui64 SeqNo = 0;
+
+        TAgentConnectionLost() = default;
 
         TAgentConnectionLost(
                 TString agentId,