src/common/duplication_common.cpp - incubator-pegasus - Git at Google

 // 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 "duplication_common.h"

 #include <nlohmann/json.hpp>
 #include <cstdint>
 #include <map>
 #include <utility>
 #include <vector>

 #include "common/common.h"
 #include "duplication_types.h"
 #include "nlohmann/detail/json_ref.hpp"
 #include "nlohmann/json_fwd.hpp"
 #include "utils/config_api.h"
 #include "utils/error_code.h"
 #include "utils/fmt_logging.h"
 #include "utils/singleton.h"
 #include "utils/time_utils.h"

 DSN_DEFINE_uint32(replication,
                   duplicate_log_batch_bytes,
                   4096,
                   "send mutation log batch bytes size per rpc");
 DSN_TAG_VARIABLE(duplicate_log_batch_bytes, FT_MUTABLE);

 // While many clusters are duplicated to a target cluster, we have to add many cluster
 // ids to the `*.ini` file of the target cluster, and the target cluster might be restarted
 // very frequently.
 //
 // This option is added so that only the target cluster id should be configured while
 // there is no need to add any other cluster id.
 DSN_DEFINE_bool(replication,
                 dup_ignore_other_cluster_ids,
                 false,
                 "Allow any other cluster id except myself to be ignored for duplication");

 namespace dsn {
 namespace replication {

 const std::string duplication_constants::kDuplicationCheckpointRootDir /*NOLINT*/ = "duplication";
 const std::string duplication_constants::kClustersSectionName /*NOLINT*/ = "pegasus.clusters";
 const std::string duplication_constants::kDuplicationEnvMasterClusterKey /*NOLINT*/ =
     "duplication.master_cluster";
 const std::string duplication_constants::kDuplicationEnvMasterMetasKey /*NOLINT*/ =
     "duplication.master_metas";
 const std::string duplication_constants::kDuplicationEnvMasterAppNameKey /*NOLINT*/ =
     "duplication.master_app_name";

 /*extern*/ const char *duplication_status_to_string(duplication_status::type status)
 {
     auto it = _duplication_status_VALUES_TO_NAMES.find(status);
     CHECK(it != _duplication_status_VALUES_TO_NAMES.end(),
           "unexpected type of duplication_status: {}",
           status);
     return it->second;
 }

 /*extern*/ const char *duplication_fail_mode_to_string(duplication_fail_mode::type fmode)
 {
     auto it = _duplication_fail_mode_VALUES_TO_NAMES.find(fmode);
     CHECK(it != _duplication_fail_mode_VALUES_TO_NAMES.end(),
           "unexpected type of duplication_fail_mode: {}",
           fmode);
     return it->second;
 }

 namespace internal {

 class duplication_group_registry : public utils::singleton<duplication_group_registry>
 {
 public:
     error_with<uint8_t> get_cluster_id(const std::string &cluster_name) const
     {
         if (cluster_name.empty()) {
             return error_s::make(ERR_INVALID_PARAMETERS, "cluster_name is empty");
         }
         if (_group.empty()) {
             return error_s::make(ERR_OBJECT_NOT_FOUND, "`duplication-group` is not configured");
         }

         auto it = _group.find(cluster_name);
         if (it == _group.end()) {
             return error_s::make(ERR_OBJECT_NOT_FOUND, "failed to get cluster id for ")
                    << cluster_name.data();
         }
         return it->second;
     }

     const std::set<uint8_t> &get_distinct_cluster_id_set() { return _distinct_cids; }

 private:
     duplication_group_registry()
     {
         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, "");
             CHECK(cluster_id < 128 && cluster_id > 0,
                   "cluster_id({}) for {} should be in [1, 127]",
                   cluster_id,
                   cluster);
             _group.emplace(cluster, static_cast<uint8_t>(cluster_id));
         }
         CHECK_EQ_MSG(clusters.size(),
                      _group.size(),
                      "there might be duplicate cluster_name in configuration");

         for (const auto &kv : _group) {
             _distinct_cids.insert(kv.second);
         }
         CHECK_EQ_MSG(_distinct_cids.size(),
                      _group.size(),
                      "there might be duplicate cluster_id in configuration");
     }
     ~duplication_group_registry() = default;

     std::map<std::string, uint8_t> _group;
     std::set<uint8_t> _distinct_cids;

     friend class utils::singleton<duplication_group_registry>;
 };

 } // namespace internal

 /*extern*/ error_with<uint8_t> get_duplication_cluster_id(const std::string &cluster_name)
 {
     return internal::duplication_group_registry::instance().get_cluster_id(cluster_name);
 }

 /*extern*/ uint8_t get_current_dup_cluster_id_or_default()
 {
     // Set cluster id to 0 as default if it is not configured, which means it would accept
     // writes from any cluster as long as the timestamp is larger.
     static const auto res = get_duplication_cluster_id(get_current_dup_cluster_name());
     static const uint8_t cluster_id = res.is_ok() ? res.get_value() : 0;
     return cluster_id;
 }

 /*extern*/ uint8_t get_current_dup_cluster_id()
 {
     static const uint8_t cluster_id =
         get_duplication_cluster_id(get_current_dup_cluster_name()).get_value();
     return cluster_id;
 }

 // 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)
 {
     char ts_buf[30];
     utils::time_ms_to_date_time(static_cast<uint64_t>(ent.create_ts), ts_buf, sizeof(ts_buf));
     nlohmann::json json{
         {"dupid", ent.dupid},
         {"create_ts", ts_buf},
         {"remote", ent.remote},
         {"status", duplication_status_to_string(ent.status)},
         {"fail_mode", duplication_fail_mode_to_string(ent.fail_mode)},
     };

     if (ent.__isset.progress) {
         nlohmann::json sub_json;
         for (const auto &p : ent.progress) {
             sub_json[std::to_string(p.first)] = p.second;
         }
         json["progress"] = sub_json;
     }

     if (ent.__isset.remote_app_name) {
         // remote_app_name is supported since v2.6.0, thus it won't be shown before v2.6.0.
         json["remote_app_name"] = ent.remote_app_name;
     }

     if (ent.__isset.remote_replica_count) {
         // remote_replica_count is supported since v2.6.0, thus it won't be shown before v2.6.0.
         json["remote_replica_count"] = ent.remote_replica_count;
     }

     return json;
 }

 /*extern*/ std::string duplication_entry_to_string(const duplication_entry &ent)
 {
     return duplication_entry_to_json(ent).dump();
 }

 /*extern*/ std::string duplication_query_response_to_string(const duplication_query_response &resp)
 {
     nlohmann::json json;
     int i = 1;
     for (const auto &ent : resp.entry_list) {
         json["appid"] = resp.appid;
         json[std::to_string(i)] = duplication_entry_to_json(ent);
         i++;
     }
     return json.dump();
 }

 /*extern*/ const std::set<uint8_t> &get_distinct_cluster_id_set()
 {
     return internal::duplication_group_registry::instance().get_distinct_cluster_id_set();
 }

 /*extern*/ bool is_dup_cluster_id_configured(uint8_t cluster_id)
 {
     if (cluster_id != get_current_dup_cluster_id()) {
         if (FLAGS_dup_ignore_other_cluster_ids) {
             return true;
         }
     }

     return get_distinct_cluster_id_set().find(cluster_id) != get_distinct_cluster_id_set().end();
 }

 } // namespace replication
 } // namespace dsn
	// 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 "duplication_common.h"

	#include <nlohmann/json.hpp>
	#include <cstdint>
	#include <map>
	#include <utility>
	#include <vector>

	#include "common/common.h"
	#include "duplication_types.h"
	#include "nlohmann/detail/json_ref.hpp"
	#include "nlohmann/json_fwd.hpp"
	#include "utils/config_api.h"
	#include "utils/error_code.h"
	#include "utils/fmt_logging.h"
	#include "utils/singleton.h"
	#include "utils/time_utils.h"

	DSN_DEFINE_uint32(replication,
	duplicate_log_batch_bytes,
	4096,
	"send mutation log batch bytes size per rpc");
	DSN_TAG_VARIABLE(duplicate_log_batch_bytes, FT_MUTABLE);

	// While many clusters are duplicated to a target cluster, we have to add many cluster
	// ids to the `*.ini` file of the target cluster, and the target cluster might be restarted
	// very frequently.
	//
	// This option is added so that only the target cluster id should be configured while
	// there is no need to add any other cluster id.
	DSN_DEFINE_bool(replication,
	dup_ignore_other_cluster_ids,
	false,
	"Allow any other cluster id except myself to be ignored for duplication");

	namespace dsn {
	namespace replication {

	const std::string duplication_constants::kDuplicationCheckpointRootDir /NOLINT/ = "duplication";
	const std::string duplication_constants::kClustersSectionName /NOLINT/ = "pegasus.clusters";
	const std::string duplication_constants::kDuplicationEnvMasterClusterKey /NOLINT/ =
	"duplication.master_cluster";
	const std::string duplication_constants::kDuplicationEnvMasterMetasKey /NOLINT/ =
	"duplication.master_metas";
	const std::string duplication_constants::kDuplicationEnvMasterAppNameKey /NOLINT/ =
	"duplication.master_app_name";

	/extern/ const char *duplication_status_to_string(duplication_status::type status)
	{
	auto it = _duplication_status_VALUES_TO_NAMES.find(status);
	CHECK(it != _duplication_status_VALUES_TO_NAMES.end(),
	"unexpected type of duplication_status: {}",
	status);
	return it->second;
	}

	/extern/ const char *duplication_fail_mode_to_string(duplication_fail_mode::type fmode)
	{
	auto it = _duplication_fail_mode_VALUES_TO_NAMES.find(fmode);
	CHECK(it != _duplication_fail_mode_VALUES_TO_NAMES.end(),
	"unexpected type of duplication_fail_mode: {}",
	fmode);
	return it->second;
	}

	namespace internal {

	class duplication_group_registry : public utils::singleton<duplication_group_registry>
	{
	public:
	error_with<uint8_t> get_cluster_id(const std::string &cluster_name) const
	{
	if (cluster_name.empty()) {
	return error_s::make(ERR_INVALID_PARAMETERS, "cluster_name is empty");
	}
	if (_group.empty()) {
	return error_s::make(ERR_OBJECT_NOT_FOUND, "`duplication-group` is not configured");
	}

	auto it = _group.find(cluster_name);
	if (it == _group.end()) {
	return error_s::make(ERR_OBJECT_NOT_FOUND, "failed to get cluster id for ")
	<< cluster_name.data();
	}
	return it->second;
	}

	const std::set<uint8_t> &get_distinct_cluster_id_set() { return _distinct_cids; }

	private:
	duplication_group_registry()
	{
	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, "");
	CHECK(cluster_id < 128 && cluster_id > 0,
	"cluster_id({}) for {} should be in [1, 127]",
	cluster_id,
	cluster);
	_group.emplace(cluster, static_cast<uint8_t>(cluster_id));
	}
	CHECK_EQ_MSG(clusters.size(),
	_group.size(),
	"there might be duplicate cluster_name in configuration");

	for (const auto &kv : _group) {
	_distinct_cids.insert(kv.second);
	}
	CHECK_EQ_MSG(_distinct_cids.size(),
	_group.size(),
	"there might be duplicate cluster_id in configuration");
	}
	~duplication_group_registry() = default;

	std::map<std::string, uint8_t> _group;
	std::set<uint8_t> _distinct_cids;

	friend class utils::singleton<duplication_group_registry>;
	};

	} // namespace internal

	/extern/ error_with<uint8_t> get_duplication_cluster_id(const std::string &cluster_name)
	{
	return internal::duplication_group_registry::instance().get_cluster_id(cluster_name);
	}

	/extern/ uint8_t get_current_dup_cluster_id_or_default()
	{
	// Set cluster id to 0 as default if it is not configured, which means it would accept
	// writes from any cluster as long as the timestamp is larger.
	static const auto res = get_duplication_cluster_id(get_current_dup_cluster_name());
	static const uint8_t cluster_id = res.is_ok() ? res.get_value() : 0;
	return cluster_id;
	}

	/extern/ uint8_t get_current_dup_cluster_id()
	{
	static const uint8_t cluster_id =
	get_duplication_cluster_id(get_current_dup_cluster_name()).get_value();
	return cluster_id;
	}

	// 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)
	{
	char ts_buf[30];
	utils::time_ms_to_date_time(static_cast<uint64_t>(ent.create_ts), ts_buf, sizeof(ts_buf));
	nlohmann::json json{
	{"dupid", ent.dupid},
	{"create_ts", ts_buf},
	{"remote", ent.remote},
	{"status", duplication_status_to_string(ent.status)},
	{"fail_mode", duplication_fail_mode_to_string(ent.fail_mode)},
	};

	if (ent.__isset.progress) {
	nlohmann::json sub_json;
	for (const auto &p : ent.progress) {
	sub_json[std::to_string(p.first)] = p.second;
	}
	json["progress"] = sub_json;
	}

	if (ent.__isset.remote_app_name) {
	// remote_app_name is supported since v2.6.0, thus it won't be shown before v2.6.0.
	json["remote_app_name"] = ent.remote_app_name;
	}

	if (ent.__isset.remote_replica_count) {
	// remote_replica_count is supported since v2.6.0, thus it won't be shown before v2.6.0.
	json["remote_replica_count"] = ent.remote_replica_count;
	}

	return json;
	}

	/extern/ std::string duplication_entry_to_string(const duplication_entry &ent)
	{
	return duplication_entry_to_json(ent).dump();
	}

	/extern/ std::string duplication_query_response_to_string(const duplication_query_response &resp)
	{
	nlohmann::json json;
	int i = 1;
	for (const auto &ent : resp.entry_list) {
	json["appid"] = resp.appid;
	json[std::to_string(i)] = duplication_entry_to_json(ent);
	i++;
	}
	return json.dump();
	}

	/extern/ const std::set<uint8_t> &get_distinct_cluster_id_set()
	{
	return internal::duplication_group_registry::instance().get_distinct_cluster_id_set();
	}

	/extern/ bool is_dup_cluster_id_configured(uint8_t cluster_id)
	{
	if (cluster_id != get_current_dup_cluster_id()) {
	if (FLAGS_dup_ignore_other_cluster_ids) {
	return true;
	}
	}

	return get_distinct_cluster_id_set().find(cluster_id) != get_distinct_cluster_id_set().end();
	}

	} // namespace replication
	} // namespace dsn