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("update_disk_stat", [](string_view) {});
     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/common/replication_common.cpp

@@ -76,10 +76,8 @@ replication_options::replication_options()
     checkpoint_max_interval_hours = 2;
 
     gc_disabled = false;
-    gc_interval_ms = 30 * 1000;                             // 30 seconds
-    gc_memory_replica_interval_ms = 10 * 60 * 1000;         // 10 minutes
-    gc_disk_error_replica_interval_seconds = 7 * 24 * 3600; // 1 week
-    gc_disk_garbage_replica_interval_seconds = 24 * 3600;   // 1 day
+    gc_interval_ms = 30 * 1000;                     // 30 seconds
+    gc_memory_replica_interval_ms = 10 * 60 * 1000; // 10 minutes
 
     disk_stat_disabled = false;
     disk_stat_interval_seconds = 600;
@@ -357,18 +355,6 @@ void replication_options::initialize()
         gc_memory_replica_interval_ms,
         "after closing a healthy replica (due to LB), the replica will remain in memory for this "
         "long (ms) for quick recover");
-    gc_disk_error_replica_interval_seconds =
-        (int)dsn_config_get_value_uint64("replication",
-                                         "gc_disk_error_replica_interval_seconds",
-                                         gc_disk_error_replica_interval_seconds,
-                                         "error replica are deleted after they have been closed "
-                                         "and lasted on disk this long (seconds)");
-    gc_disk_garbage_replica_interval_seconds =
-        (int)dsn_config_get_value_uint64("replication",
-                                         "gc_disk_garbage_replica_interval_seconds",
-                                         gc_disk_garbage_replica_interval_seconds,
-                                         "garbage replica are deleted after they have been closed "
-                                         "and lasted on disk this long (seconds)");
 
     disk_stat_disabled = dsn_config_get_value_bool(
         "replication", "disk_stat_disabled", disk_stat_disabled, "whether to disable disk stat");

src/common/replication_common.h

@@ -80,8 +80,6 @@ class replication_options
     bool gc_disabled;
     int32_t gc_interval_ms;
     int32_t gc_memory_replica_interval_ms;
-    int32_t gc_disk_error_replica_interval_seconds;
-    int32_t gc_disk_garbage_replica_interval_seconds;
 
     bool disk_stat_disabled;
     int32_t disk_stat_interval_seconds;

src/replica/disk_cleaner.cpp

@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <dsn/utility/flags.h>
+#include <dsn/utility/filesystem.h>
+#include <dsn/dist/fmt_logging.h>
+#include <dsn/c/api_layer1.h>
+
+#include "disk_cleaner.h"
+
+namespace dsn {
+namespace replication {
+
+DSN_DEFINE_uint64("replication",
+                  gc_disk_error_replica_interval_seconds,
+                  7 * 24 * 3600 /*7day*/,
+                  "disk migration generate tmp dir gc interval, the folder name suffix is '.tmp' ");
+DSN_DEFINE_uint64("replication",
+                  gc_disk_garbage_replica_interval_seconds,
+                  24 * 3600 /*1day*/,
+                  "disk migration origin dir gc interval, the folder name suffix is '.ori' ");
+
+DSN_DEFINE_uint64("replication",
+                  gc_disk_migration_tmp_replica_interval_seconds,
+                  24 * 3600 /*1day*/,
+                  "disk migration generate tmp dir gc interval, the folder name suffix is '.tmp' ");
+DSN_DEFINE_uint64("replication",
+                  gc_disk_migration_origin_replica_interval_seconds,
+                  7 * 24 * 3600 /*7day*/,
+                  "disk migration origin dir gc interval, the folder name suffix is '.ori' ");
+
+const std::string kFolderSuffixErr = ".err";
+const std::string kFolderSuffixGar = ".gar";
+const std::string kFolderSuffixBak = ".bak";
+const std::string kFolderSuffixOri = ".ori";
+const std::string kFolderSuffixTmp = ".tmp";
+
+error_s disk_remove_useless_dirs(const std::vector<std::string> &data_dirs,
+                                 /*output*/ disk_cleaning_report &report)
+{
+    std::vector<std::string> sub_list;
+    for (auto &dir : data_dirs) {
+        std::vector<std::string> tmp_list;
+        if (!dsn::utils::filesystem::get_subdirectories(dir, tmp_list, false)) {
+            dwarn_f("gc_disk: failed to get subdirectories in {}", dir);
+            return error_s::make(ERR_OBJECT_NOT_FOUND, "failed to get subdirectories");
+        }
+        sub_list.insert(sub_list.end(), tmp_list.begin(), tmp_list.end());
+    }
+    for (auto &fpath : sub_list) {
+        auto name = dsn::utils::filesystem::get_file_name(fpath);
+        if (!is_data_dir_removable(name)) {
+            continue;
+        }
+        std::string folder_suffix = name.substr(name.length() - 4);
+
+        time_t mt;
+        if (!dsn::utils::filesystem::last_write_time(fpath, mt)) {
+            dwarn_f("gc_disk: failed to get last write time of {}", fpath);
+            continue;
+        }
+
+        auto last_write_time = (uint64_t)mt;
+        uint64_t current_time_ms = dsn_now_ms();
+        uint64_t remove_interval_seconds = current_time_ms / 1000;
+
+        // don't delete ".bak" directory because it is backed by administrator.
+        if (folder_suffix == kFolderSuffixErr) {
+            report.error_replica_count++;
+            remove_interval_seconds = FLAGS_gc_disk_error_replica_interval_seconds;
+        } else if (folder_suffix == kFolderSuffixGar) {
+            report.garbage_replica_count++;
+            remove_interval_seconds = FLAGS_gc_disk_garbage_replica_interval_seconds;
+        } else if (folder_suffix == kFolderSuffixTmp) {
+            report.disk_migrate_tmp_count++;
+            remove_interval_seconds = FLAGS_gc_disk_migration_tmp_replica_interval_seconds;
+        } else if (folder_suffix == kFolderSuffixOri) {
+            report.disk_migrate_origin_count++;
+            remove_interval_seconds = FLAGS_gc_disk_migration_origin_replica_interval_seconds;
+        }
+
+        if (last_write_time + remove_interval_seconds <= current_time_ms / 1000) {
+            if (!dsn::utils::filesystem::remove_path(fpath)) {
+                dwarn_f("gc_disk: failed to delete directory '{}', time_used_ms = {}",
+                        fpath,
+                        dsn_now_ms() - current_time_ms);
+            } else {
+                dwarn_f("gc_disk: replica_dir_op succeed to delete directory '{}'"
+                        ", time_used_ms = {}",
+                        fpath,
+                        dsn_now_ms() - current_time_ms);
+                report.remove_dir_count++;
+            }
+        } else {
+            ddebug_f("gc_disk: reserve directory '{}', wait_seconds = {}",
+                     fpath,
+                     last_write_time + remove_interval_seconds - current_time_ms / 1000);
+        }
+    }
+    return error_s::make(ERR_OK);
+}
+}
+}

src/replica/disk_cleaner.h

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#pragma once
+
+#include <vector>
+#include <dsn/utility/errors.h>
+#include <dsn/perf_counter/perf_counter_wrapper.h>
+
+namespace dsn {
+namespace replication {
+DSN_DECLARE_uint64(gc_disk_error_replica_interval_seconds);
+DSN_DECLARE_uint64(gc_disk_garbage_replica_interval_seconds);
+DSN_DECLARE_uint64(gc_disk_migration_tmp_replica_interval_seconds);
+DSN_DECLARE_uint64(gc_disk_migration_origin_replica_interval_seconds);
+
+// the invalid folder suffix, server will check disk folder and deal with them
+extern const std::string kFolderSuffixErr; // replica error dir
+extern const std::string kFolderSuffixGar; // replica closed and assign garbage dir
+extern const std::string kFolderSuffixBak; // replica backup dir which can be restored
+extern const std::string kFolderSuffixOri; // replica disk migration origin dir
+extern const std::string kFolderSuffixTmp; // replica disk migration temp dir
+
+struct disk_cleaning_report
+{
+    int remove_dir_count;
+
+    int garbage_replica_count;
+    int error_replica_count;
+    int disk_migrate_tmp_count;
+    int disk_migrate_origin_count;
+};
+
+/// Removes the useless data from data directories.
+error_s disk_remove_useless_dirs(const std::vector<std::string> &data_dirs,
+                                 /*output*/ disk_cleaning_report &report);
+
+inline bool is_data_dir_removable(const std::string &dir)
+{
+    if (dir.length() < 4) {
+        return false;
+    }
+    const std::string folder_suffix = dir.substr(dir.length() - 4);
+    return (folder_suffix == kFolderSuffixErr || folder_suffix == kFolderSuffixGar ||
+            folder_suffix == kFolderSuffixTmp || folder_suffix == kFolderSuffixOri);
+}
+
+// Note: ".bak" is invalid but not allow delete, because it can be backed by administrator.
+inline bool is_data_dir_invalid(const std::string &dir)
+{
+    if (dir.length() < 4) {
+        return false;
+    }
+    const std::string folder_suffix = dir.substr(dir.length() - 4);
+    return is_data_dir_removable(dir) || folder_suffix == kFolderSuffixBak;
+}
+}
+}

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.