Live migration to the new leader volume.

Author: drbasic

cloud/blockstore/apps/client/lib/command.cpp

@@ -351,7 +351,8 @@ NProto::TMountVolumeResponse TCommand::MountVolume(
         VolumeStats,
         endpoint,
         ClientConfig,
-        sessionConfig);
+        sessionConfig,
+        ISessionSwitcherWeakPtr());
 
     auto response = SafeExecute<NProto::TMountVolumeResponse>([&] {
         return WaitFor(session->MountVolume());

cloud/blockstore/apps/client/lib/destroy_volume.cpp

@@ -51,7 +51,7 @@ class TDestroyVolumeCommand final
         } else {
             request->SetDiskId(DiskId);
             request->SetSync(Sync);
-
+/*
             STORAGE_WARN("Waiting for confirmation");
             output << "Confirm disk destruction by typing disk id to stdin" << Endl;
             TString confirmation;
@@ -64,6 +64,7 @@ class TDestroyVolumeCommand final
 
                 return false;
             }
+*/
         }
 
         STORAGE_DEBUG("Sending DestroyVolume request");

cloud/blockstore/libs/client/session.cpp

@@ -8,6 +8,8 @@
 #include <cloud/blockstore/libs/service/context.h>
 #include <cloud/blockstore/libs/service/request_helpers.h>
 #include <cloud/blockstore/libs/service/service.h>
+#include <cloud/blockstore/libs/storage/model/volume_label.h>
+
 #include <cloud/storage/core/libs/common/error.h>
 #include <cloud/storage/core/libs/common/scheduler.h>
 #include <cloud/storage/core/libs/common/timer.h>
@@ -122,6 +124,7 @@ class TSession final
     const IBlockStorePtr Client;
     const TClientAppConfigPtr Config;
     TSessionConfig SessionConfig;
+    ISessionSwitcherWeakPtr SessionSwitcherPtr;
 
     TLog Log;
     TSessionInfo SessionInfo;
@@ -135,7 +138,8 @@ class TSession final
         IVolumeStatsPtr volumeStats,
         IBlockStorePtr client,
         TClientAppConfigPtr config,
-        const TSessionConfig& sessionConfig);
+        const TSessionConfig& sessionConfig,
+        ISessionSwitcherWeakPtr sessionSwitcherPtr);
 
     ui32 GetMaxTransfer() const override;
 
@@ -251,7 +255,8 @@ TSession::TSession(
         IVolumeStatsPtr volumeStats,
         IBlockStorePtr client,
         TClientAppConfigPtr config,
-        const TSessionConfig& sessionConfig)
+        const TSessionConfig& sessionConfig,
+        ISessionSwitcherWeakPtr sessionSwitcherPtr)
     : Timer(std::move(timer))
     , Scheduler(std::move(scheduler))
     , Logging(std::move(logging))
@@ -260,6 +265,7 @@ TSession::TSession(
     , Client(std::move(client))
     , Config(std::move(config))
     , SessionConfig(sessionConfig)
+    , SessionSwitcherPtr(std::move(sessionSwitcherPtr))
     , Log(Logging->CreateLog("BLOCKSTORE_CLIENT"))
 {}
 
@@ -629,6 +635,18 @@ void TSession::ProcessMountResponse(
                     SessionConfig.InstanceId)
                 << " complete request");
 
+            if (response.GetVolume().GetPrincipalDiskId()) {
+                Y_DEBUG_ABORT_UNLESS(
+                    response.GetVolume().GetPrincipalDiskId() ==
+                    NStorage::GetNextDiskId(response.GetVolume().GetDiskId()));
+
+                if (auto switcher = SessionSwitcherPtr.lock()) {
+                    switcher->SwitchSession(
+                        response.GetVolume().GetDiskId(),
+                        response.GetVolume().GetPrincipalDiskId());
+                }
+            }
+
             SessionInfo.MountState = EMountState::MountCompleted;
             SessionInfo.BlockSize = response.GetVolume().GetBlockSize();
             SessionInfo.SessionId = response.GetSessionId();
@@ -964,6 +982,15 @@ void TSession::HandleResponse(
 
         ForceVolumeRemount(sessionId);
         HandleRequest<T>(std::move(callContext), std::move(request), response);
+
+        if (HasProtoFlag(errorFlags, NProto::EF_OUTDATED_VOLUME)) {
+            if (auto switcher = SessionSwitcherPtr.lock()) {
+                switcher->SwitchSession(
+                    SessionConfig.DiskId,
+                    NStorage::GetNextDiskId(SessionConfig.DiskId));
+            }
+        }
+
         return;
     }
 
@@ -1020,7 +1047,8 @@ ISessionPtr CreateSession(
     IVolumeStatsPtr volumeStats,
     IBlockStorePtr client,
     TClientAppConfigPtr config,
-    const TSessionConfig& sessionConfig)
+    const TSessionConfig& sessionConfig,
+    ISessionSwitcherWeakPtr sessionSwitcherPtr)
 {
     return std::make_shared<TSession>(
         std::move(timer),
@@ -1030,7 +1058,8 @@ ISessionPtr CreateSession(
         std::move(volumeStats),
         std::move(client),
         std::move(config),
-        sessionConfig);
+        sessionConfig,
+        std::move(sessionSwitcherPtr));
 }
 
 }   // namespace NCloud::NBlockStore::NClient

cloud/blockstore/libs/client/session.h

@@ -43,6 +43,18 @@ struct ISession
 
 ////////////////////////////////////////////////////////////////////////////////
 
+struct ISessionSwitcher
+{
+    virtual ~ISessionSwitcher() = default;
+
+    virtual void SwitchSession(
+        const TString& diskId,
+        const TString& newDiskId) = 0;
+};
+using ISessionSwitcherWeakPtr = std::weak_ptr<ISessionSwitcher>;
+
+////////////////////////////////////////////////////////////////////////////////
+
 struct TSessionConfig
 {
     TString DiskId;
@@ -73,6 +85,7 @@ ISessionPtr CreateSession(
     IVolumeStatsPtr volumeStats,
     IBlockStorePtr client,
     TClientAppConfigPtr config,
-    const TSessionConfig& sessionConfig);
+    const TSessionConfig& sessionConfig,
+    ISessionSwitcherWeakPtr sessionSwitcherPtr);
 
 }   // namespace NCloud::NBlockStore::NClient