feat(disk_balance): close origin replica and update replica dir

src/common/fs_manager.cpp

@@ -37,6 +37,7 @@
 #include <dsn/utility/filesystem.h>
 #include <thread>
 #include <dsn/dist/fmt_logging.h>
+#include <dsn/utility/fail_point.h>
 
 namespace dsn {
 namespace replication {
@@ -76,6 +77,7 @@ unsigned dir_node::remove(const gpid &pid)
 
 void dir_node::update_disk_stat()
 {
+    FAIL_POINT_INJECT_F("mock_dir_node", [](string_view) -> void { return; });
     dsn::utils::filesystem::disk_space_info info;
     if (dsn::utils::filesystem::get_disk_space_info(full_dir, info)) {
         disk_capacity_mb = info.capacity / 1024 / 1024;

src/replica/replica.cpp

@@ -372,10 +372,13 @@ bool replica::verbose_commit_log() const { return _stub->_verbose_commit_log; }
 
 void replica::close()
 {
-    dassert(status() == partition_status::PS_ERROR || status() == partition_status::PS_INACTIVE,
-            "%s: invalid state %s when calling replica::close",
-            name(),
-            enum_to_string(status()));
+    dassert_replica(status() == partition_status::PS_ERROR ||
+                        status() == partition_status::PS_INACTIVE ||
+                        _disk_migrator->status() >= disk_migration_status::MOVED,
+                    "invalid state(partition_status={}, migration_status={}) when calling "
+                    "replica close",
+                    enum_to_string(status()),
+                    enum_to_string(_disk_migrator->status()));
 
     uint64_t start_time = dsn_now_ms();
 
@@ -422,6 +425,13 @@ void replica::close()
         }
     }
 
+    if (_disk_migrator->status() == disk_migration_status::MOVED) {
+        // this will update disk_migration_status::MOVED->disk_migration_status::CLOSED
+        _disk_migrator->update_replica_dir();
+    } else if (_disk_migrator->status() == disk_migration_status::CLOSED) {
+        _disk_migrator.reset();
+    }
+
     _counter_private_log_size.clear();
 
     // duplication_impl may have ongoing tasks.

src/replica/replica_disk_migrator.cpp

@@ -30,6 +30,7 @@ namespace dsn {
 namespace replication {
 
 const std::string replica_disk_migrator::kReplicaDirTempSuffix = ".disk.balance.tmp";
+const std::string replica_disk_migrator::kReplicaDirOriginSuffix = ".disk.balance.ori";
 const std::string replica_disk_migrator::kDataDirFolder = "data/rdb/";
 const std::string replica_disk_migrator::kAppInfo = ".app-info";
 
@@ -58,7 +59,7 @@ void replica_disk_migrator::on_migrate_replica(replica_disk_migrate_rpc rpc)
                 enum_to_string(disk_migration_status::IDLE),
                 enum_to_string(status()));
 
-            const auto request = rpc.request();
+            const auto &request = rpc.request();
             tasking::enqueue(LPC_REPLICATION_LONG_COMMON, _replica->tracker(), [=]() {
                 migrate_replica(request);
             });
@@ -171,14 +172,11 @@ bool replica_disk_migrator::check_migration_args(replica_disk_migrate_rpc rpc)
 // THREAD_POOL_REPLICATION_LONG
 void replica_disk_migrator::migrate_replica(const replica_disk_migrate_request &req)
 {
-    if (status() != disk_migration_status::MOVING) {
-        derror_replica("disk migration(origin={}, target={}), err = Invalid migration status({})",
-                       req.origin_disk,
-                       req.target_disk,
-                       enum_to_string(status()));
-        reset_status();
-        return;
-    }
+    dassert_replica(status() == disk_migration_status::MOVING,
+                    "disk migration(origin={}, target={}), err = Invalid migration status({})",
+                    req.origin_disk,
+                    req.target_disk,
+                    enum_to_string(status()));
 
     if (init_target_dir(req) && migrate_replica_checkpoint(req) && migrate_replica_app_info(req)) {
         _status = disk_migration_status::MOVED;
@@ -191,7 +189,7 @@ void replica_disk_migrator::migrate_replica(const replica_disk_migrate_request &
                        enum_to_string(status()),
                        _replica->dir());
 
-        close_current_replica();
+        close_current_replica(req);
     }
 }
 
@@ -310,12 +308,56 @@ bool replica_disk_migrator::migrate_replica_app_info(const replica_disk_migrate_
     return true;
 }
 
-// TODO(jiashuo1)
 // THREAD_POOL_REPLICATION_LONG
-void replica_disk_migrator::close_current_replica() {}
+dsn::task_ptr replica_disk_migrator::close_current_replica(const replica_disk_migrate_request &req)
+{
+    if (_replica->status() != partition_status::type::PS_SECONDARY) {
+        derror_replica("migrate request(origin={}, target={}), err = Invalid partition status({})",
+                       req.origin_disk,
+                       req.target_disk,
+                       enum_to_string(_replica->status()));
+        reset_status();
+        utils::filesystem::remove_path(_target_replica_dir);
+        return nullptr;
+    }
 
-// TODO(jiashuo1)
-// run in replica::close_replica of THREAD_POOL_REPLICATION_LONG
-void replica_disk_migrator::update_replica_dir() {}
+    return _replica->_stub->begin_close_replica(_replica);
+}
+
+// run in replica->close_replica() of THREAD_POOL_REPLICATION_LONG
+void replica_disk_migrator::update_replica_dir()
+{
+    // origin_tmp_dir: /root/origin/gpid.app_type.disk.balance.ori
+    std::string origin_temp_dir = fmt::format("{}{}", _replica->dir(), kReplicaDirOriginSuffix);
+    if (!dsn::utils::filesystem::rename_path(_replica->dir(), origin_temp_dir)) {
+        reset_status();
+        utils::filesystem::remove_path(_target_replica_dir);
+        return;
+    }
+
+    std::string target_temp_dir = _target_replica_dir;
+    // update _target_replica_dir /root/gpid.app_type.disk.balance.tmp/ to
+    // /root/target/gpid.app_type/
+    boost::replace_first(_target_replica_dir, kReplicaDirTempSuffix, "");
+    if (!dsn::utils::filesystem::rename_path(target_temp_dir, _target_replica_dir)) {
+        reset_status();
+        // rename failed, delete tmp dir and revert origin dir
+        utils::filesystem::remove_path(target_temp_dir);
+        dsn::utils::filesystem::rename_path(origin_temp_dir, _replica->dir());
+        return;
+    }
+
+    _replica->get_replica_stub()->_fs_manager.remove_replica(get_gpid());
+    _replica->get_replica_stub()->_fs_manager.add_replica(get_gpid(), _target_replica_dir);
+    _replica->get_replica_stub()->update_disk_holding_replicas();
+
+    _status = disk_migration_status::CLOSED;
+    ddebug_replica("disk replica migration move data from origin dir({}) to new dir({}) "
+                   "succeed, update status from {}=>{}",
+                   _replica->dir(),
+                   _target_replica_dir,
+                   enum_to_string(disk_migration_status::MOVED),
+                   enum_to_string(status()));
+}
 } // namespace replication
 } // namespace dsn

src/replica/replica_disk_migrator.h

@@ -44,13 +44,14 @@ class replica_disk_migrator : replica_base
     bool migrate_replica_checkpoint(const replica_disk_migrate_request &req);
     bool migrate_replica_app_info(const replica_disk_migrate_request &req);
 
-    void close_current_replica();
+    dsn::task_ptr close_current_replica(const replica_disk_migrate_request &req);
     void update_replica_dir();
 
     void reset_status() { _status = disk_migration_status::IDLE; }
 
 private:
     const static std::string kReplicaDirTempSuffix;
+    const static std::string kReplicaDirOriginSuffix;
     const static std::string kDataDirFolder;
     const static std::string kAppInfo;
 
@@ -61,6 +62,7 @@ class replica_disk_migrator : replica_base
     disk_migration_status::type _status{disk_migration_status::IDLE};
 
     friend class replica;
+    friend class replica_stub;
     friend class replica_disk_migrate_test;
 };
 

src/replica/replica_init.cpp

@@ -33,6 +33,7 @@
 #include <dsn/utility/filesystem.h>
 #include <dsn/dist/replication/replication_app_base.h>
 #include <dsn/dist/fmt_logging.h>
+#include <dsn/utility/fail_point.h>
 
 namespace dsn {
 namespace replication {
@@ -124,6 +125,8 @@ error_code replica::initialize_on_load()
 
 /*static*/ replica *replica::load(replica_stub *stub, const char *dir)
 {
+    FAIL_POINT_INJECT_F("mock_replica_open", [&](string_view) -> replica * { return nullptr; });
+
     char splitters[] = {'\\', '/', 0};
     std::string name = utils::get_last_component(std::string(dir), splitters);
     if (name == "") {