feat: dynamic reload duplication config

src/base/CMakeLists.txt

@@ -30,5 +30,5 @@ set(MY_PROJ_SRC "")
 
 dsn_add_static_library()
 
-target_link_libraries(pegasus_base PUBLIC RocksDB::rocksdb sasl2 gssapi_krb5 krb5 dsn_replication_common)
+target_link_libraries(pegasus_base PUBLIC dsn_replication_common RocksDB::rocksdb sasl2 gssapi_krb5 krb5 dsn_replication_common)
 target_include_directories(pegasus_base PUBLIC "$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>")

src/common/duplication_common.cpp

@@ -18,6 +18,7 @@
 #include "duplication_common.h"
 
 #include <nlohmann/json.hpp>
+#include <cmath>
 #include <cstdint>
 #include <map>
 #include <utility>
@@ -28,6 +29,7 @@
 #include "nlohmann/detail/json_ref.hpp"
 #include "nlohmann/json_fwd.hpp"
 #include "utils/config_api.h"
+#include "utils/configuration.h"
 #include "utils/error_code.h"
 #include "utils/fmt_logging.h"
 #include "utils/singleton.h"
@@ -103,6 +105,59 @@ class duplication_group_registry : public utils::singleton<duplication_group_reg
     }
 
     const std::set<uint8_t> &get_distinct_cluster_id_set() { return _distinct_cids; }
+    error_with<uint8_t> reload_duplication_config(std::string config_file)
+    {
+        if (config_file.length() == 0) {
+            config_file = "config.ini";
+        }
+        const char *config_file_cstr = config_file.c_str();
+
+        std::map<std::string, uint8_t> new_group;
+        std::set<uint8_t> new_distinct_cids;
+        dsn::configuration old_config;
+
+        // reload default config.ini, user can point to another config file. update g_config here
+        if (!dsn_config_reload(config_file_cstr, nullptr, &old_config)) {
+            LOG_ERROR("Fail to reload config file {} \n", config_file_cstr);
+            return error_s::make(
+                ERR_OBJECT_NOT_FOUND,
+                " new `duplication-group` configured can not be read. Check your config.ini now");
+            ;
+        }
+
+        int influented_clusters = 0;
+
+        std::vector<std::string> clusters;
+        dsn_config_get_all_keys("duplication-group", clusters);
+        for (std::string &cluster : clusters) {
+            int64_t cluster_id =
+                dsn_config_get_value_int64("duplication-group", cluster.data(), 0, "");
+
+            // gns : do not dassert, just rolling it back and log error
+            if (cluster_id < 128 && cluster_id > 0) {
+                new_group.emplace(cluster, static_cast<uint8_t>(cluster_id));
+                new_distinct_cids.emplace(cluster_id);
+            } else {
+                LOG_ERROR(
+                    "cluster_id({}) for {} should be in [1, 127]", cluster_id, cluster.data());
+                // roll back cluster group and configuration
+                dsn_config_rollback(old_config);
+                return error_s::make(ERR_INVALID_PARAMETERS,
+                                     " new `duplication-group` configured invalid cluster id. "
+                                     "Check your config.ini now");
+            }
+        }
+
+        if (new_group.size() != _group.size()) {
+            influented_clusters = std::fabs(_group.size() - new_group.size());
+            LOG_DEBUG("There are {} influented lines after reloading the config file",
+                      influented_clusters);
+        }
+
+        swap(new_group, _group);
+        swap(new_distinct_cids, _distinct_cids);
+        return influented_clusters;
+    }
 
 private:
     duplication_group_registry()
@@ -161,6 +216,11 @@ class duplication_group_registry : public utils::singleton<duplication_group_reg
     return cluster_id;
 }
 
+/*extern*/ error_with<uint8_t> make_reloading_duplication_config(std::string &config_file)
+{
+    return internal::duplication_group_registry::instance().reload_duplication_config(config_file);
+}
+
 // TODO(wutao1): implement our C++ version of `TSimpleJSONProtocol` if there're
 //               more cases for converting thrift to JSON
 static nlohmann::json duplication_entry_to_json(const duplication_entry &ent)

src/common/duplication_common.h

@@ -62,6 +62,8 @@ inline bool is_duplication_status_invalid(duplication_status::type status)
 /// The returned cluster id of get_duplication_cluster_id("wuhan-mi-srv-ad") is 3.
 extern error_with<uint8_t> get_duplication_cluster_id(const std::string &cluster_name);
 
+extern error_with<uint8_t> make_reloading_duplication_config(std::string &config_file);
+
 extern uint8_t get_current_dup_cluster_id_or_default();
 
 extern uint8_t get_current_dup_cluster_id();

src/common/test/CMakeLists.txt

@@ -40,6 +40,8 @@ set(MY_BOOST_LIBS Boost::system Boost::filesystem)
 
 set(MY_BINPLACES
         config-test.ini
+        err-config-test.ini
+        new-config-test.ini
         run.sh
         )
 

src/common/test/duplication_common_test.cpp

@@ -24,12 +24,17 @@
  * THE SOFTWARE.
  */
 
-#include "common//duplication_common.h"
+#include "common/duplication_common.h"
 
 #include <cstdint>
+#include <memory>
+#include <vector>
 
+#include "common/replication_other_types.h"
 #include "gtest/gtest.h"
+#include "runtime/rpc/rpc_host_port.h"
 #include "test_util/test_util.h"
+#include "utils/config_api.h"
 #include "utils/error_code.h"
 #include "utils/flags.h"
 
@@ -39,6 +44,11 @@ DSN_DECLARE_bool(dup_ignore_other_cluster_ids);
 namespace dsn {
 namespace replication {
 
+std::string config_file = "config-test.ini";
+std::string unkown_file = "unknown.ini";
+std::string err_config_file = "err-config-test.ini";
+std::string new_config_file = "new-config-test.ini";
+
 TEST(duplication_common, get_duplication_cluster_id)
 {
     ASSERT_EQ(1, get_duplication_cluster_id("master-cluster").get_value());
@@ -82,5 +92,57 @@ TEST(duplication_common, dup_ignore_other_cluster_ids)
     }
 }
 
+TEST(duplication_common, reload_config_file)
+{
+    ASSERT_EQ(make_reloading_duplication_config(unkown_file).get_error().code(),
+              ERR_OBJECT_NOT_FOUND);
+    ASSERT_EQ(make_reloading_duplication_config(err_config_file).get_error().code(),
+              ERR_INVALID_PARAMETERS);
+}
+
+TEST(duplication_common, reload_get_duplication_cluster_id)
+{
+    make_reloading_duplication_config(config_file);
+    ASSERT_EQ(get_duplication_cluster_id("master-cluster").get_value(), 1);
+    ASSERT_EQ(get_duplication_cluster_id("slave-cluster").get_value(), 2);
+    ASSERT_EQ(get_duplication_cluster_id("strange-cluster").get_error().code(),
+              ERR_OBJECT_NOT_FOUND);
+
+    make_reloading_duplication_config(new_config_file);
+    ASSERT_EQ(get_duplication_cluster_id("master-cluster").get_value(), 1);
+    ASSERT_EQ(get_duplication_cluster_id("slave-cluster").get_value(), 2);
+    ASSERT_EQ(get_duplication_cluster_id("strange-cluster").get_value(), 3);
+}
+
+TEST(duplication_common, reload_get_meta_list)
+{
+    make_reloading_duplication_config(config_file);
+    replica_helper replica;
+    std::vector<dsn::host_port> addr_vec;
+    // replica.load_meta_servers(addr_vec,"pegasus.clusters","strange-cluster");
+    const char *strange_cluster_server_list =
+        dsn_config_get_value_string("pegasus.clusters", "strange-cluster", "", "");
+    dsn::replication::replica_helper::parse_server_list(strange_cluster_server_list, addr_vec);
+    ASSERT_EQ(addr_vec.size(), 0);
+    addr_vec.clear();
+
+    make_reloading_duplication_config(new_config_file);
+    const char *new_strange_cluster_server_list =
+        dsn_config_get_value_string("pegasus.clusters", "strange-cluster", "", "");
+    dsn::replication::replica_helper::parse_server_list(new_strange_cluster_server_list, addr_vec);
+    ASSERT_EQ(addr_vec.size(), 2);
+
+    std::string addr0 = addr_vec[0].to_string();
+    std::string addr1 = addr_vec[1].to_string();
+    ASSERT_EQ(addr0, "localhost.test1:37001");
+    ASSERT_EQ(addr1, "localhost.test2:37001");
+
+    addr_vec.clear();
+    const char *unkonw_cluster_server_list =
+        dsn_config_get_value_string("pegasus.clusters", "unknow-cluster", "", "");
+    dsn::replication::replica_helper::parse_server_list(unkonw_cluster_server_list, addr_vec);
+    ASSERT_EQ(addr_vec.size(), 0);
+}
+
 } // namespace replication
 } // namespace dsn

src/common/test/err-config-test.ini

@@ -0,0 +1,106 @@
+; The MIT License (MIT)
+;
+; Copyright (c) 2015 Microsoft Corporation
+;
+; -=- Robust Distributed System Nucleus (rDSN) -=-
+;
+; Permission is hereby granted, free of charge, to any person obtaining a copy
+; of this software and associated documentation files (the "Software"), to deal
+; in the Software without restriction, including without limitation the rights
+; to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+; copies of the Software, and to permit persons to whom the Software is
+; furnished to do so, subject to the following conditions:
+;
+; The above copyright notice and this permission notice shall be included in
+; all copies or substantial portions of the Software.
+;
+; THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+; IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+; FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+; AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+; LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+; OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+; THE SOFTWARE.
+
+[apps..default]
+run = true
+count = 1
+;network.client.RPC_CHANNEL_TCP = dsn::tools::sim_network_provider, 65536
+;network.client.RPC_CHANNEL_UDP = dsn::tools::sim_network_provider, 65536
+;network.server.0.RPC_CHANNEL_TCP = dsn::tools::sim_network_provider, 65536
+
+[apps.replica]
+type = replica
+run = true
+count = 1
+ports = 54321
+pools = THREAD_POOL_DEFAULT
+
+[core]
+;tool = simulator
+tool = nativerun
+
+;toollets = tracer, profiler
+;fault_injector
+pause_on_start = false
+cli_local = false
+cli_remote = false
+
+logging_start_level = LOG_LEVEL_DEBUG
+logging_factory_name = dsn::tools::simple_logger
+
+
+[tools.simple_logger]
+fast_flush = true
+short_header = false
+stderr_start_level = LOG_LEVEL_WARNING
+
+[tools.simulator]
+random_seed = 1465902258
+
+[tools.screen_logger]
+short_header = false
+
+[network]
+; how many network threads for network library (used by asio)
+io_service_worker_count = 2
+
+; specification for each thread pool
+[threadpool..default]
+worker_count = 4
+
+[threadpool.THREAD_POOL_DEFAULT]
+name = default
+partitioned = false
+max_input_queue_length = 1024
+worker_priority = THREAD_xPRIORITY_NORMAL
+worker_count = 2
+
+[threadpool.THREAD_POOL_REPLICATION]
+name = replica
+partitioned = true
+max_input_queue_length = 2560
+worker_priority = THREAD_xPRIORITY_NORMAL
+worker_count = 3
+
+[threadpool.THREAD_POOL_REPLICATION_LONG]
+name = replica_long
+
+[task..default]
+is_trace = true
+is_profile = true
+allow_inline = false
+rpc_call_channel = RPC_CHANNEL_TCP
+rpc_message_header_format = dsn
+rpc_timeout_milliseconds = 5000
+
+[replication]
+cluster_name = master-cluster
+
+[duplication-group]
+master-cluster = 1
+slave-cluster  = 2
+strange-cluster = 6657
+
+[pegasus.clusters]
+strange-cluster = localhost.test1:37001,localhost.test2:37001