acelyc111 commented on code in PR #1658:
URL:
https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1519658474
##########
src/meta/test/state_sync_test.cpp:
##########
@@ -61,13 +62,15 @@ namespace dsn {
namespace replication {
class meta_options;
-static void random_assign_partition_config(std::shared_ptr<app_state> &app,
- const std::vector<dsn::rpc_address>
&server_list,
- int max_replica_count)
+static void random_assign_partition_config(
+ std::shared_ptr<app_state> &app,
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> &server_list,
Review Comment:
TODO: use a single host_port is enough, then use dns resolver to resolve it.
##########
src/meta/test/meta_bulk_load_service_test.cpp:
##########
@@ -506,6 +522,11 @@ class bulk_load_service_test : public meta_test_base
const rpc_address SECONDARY1 = rpc_address::from_ip_port("127.0.0.1",
10085);
Review Comment:
Resolve from the hosts below.
##########
src/meta/duplication/meta_duplication_service.cpp:
##########
@@ -427,6 +429,10 @@ void
meta_duplication_service::check_follower_app_if_create_completed(
p.primary = rpc_address::from_ip_port("127.0.0.1",
34801);
p.secondaries.emplace_back(rpc_address::from_ip_port("127.0.0.2", 34801));
p.secondaries.emplace_back(rpc_address::from_ip_port("127.0.0.3", 34801));
+ p.__set_hp_primary(host_port("localhost", 34801));
+ p.__set_hp_secondaries(std::vector<host_port>());
+ p.hp_secondaries.emplace_back(host_port("localhost",
34802));
+ p.hp_secondaries.emplace_back(host_port("localhost",
34803));
Review Comment:
TODO: Should be the same to line 430~431
##########
src/common/replication_other_types.h:
##########
@@ -49,43 +51,54 @@ typedef int64_t decree;
#define invalid_offset (-1LL)
#define invalid_signature 0
-inline bool is_primary(const partition_configuration &pc, const rpc_address
&node)
+inline bool is_primary(const partition_configuration &pc, const host_port
&node)
{
- return !node.is_invalid() && pc.primary == node;
+ return !node.is_invalid() && pc.hp_primary == node;
}
-inline bool is_secondary(const partition_configuration &pc, const rpc_address
&node)
+inline bool is_secondary(const partition_configuration &pc, const host_port
&node)
{
return !node.is_invalid() &&
- std::find(pc.secondaries.begin(), pc.secondaries.end(), node) !=
pc.secondaries.end();
+ std::find(pc.hp_secondaries.begin(), pc.hp_secondaries.end(), node)
!=
+ pc.hp_secondaries.end();
}
-inline bool is_member(const partition_configuration &pc, const rpc_address
&node)
+inline bool is_member(const partition_configuration &pc, const host_port &node)
{
return is_primary(pc, node) || is_secondary(pc, node);
}
inline bool is_partition_config_equal(const partition_configuration &pc1,
const partition_configuration &pc2)
{
// secondaries no need to be same order
- for (const rpc_address &addr : pc1.secondaries)
+ for (const host_port &addr : pc1.hp_secondaries)
if (!is_secondary(pc2, addr))
return false;
// last_drops is not considered into equality check
return pc1.ballot == pc2.ballot && pc1.pid == pc2.pid &&
pc1.max_replica_count == pc2.max_replica_count && pc1.primary ==
pc2.primary &&
- pc1.secondaries.size() == pc2.secondaries.size() &&
+ pc1.hp_primary == pc2.hp_primary && pc1.secondaries.size() ==
pc2.secondaries.size() &&
+ pc1.hp_secondaries.size() == pc2.hp_secondaries.size() &&
pc1.last_committed_decree == pc2.last_committed_decree;
}
class replica_helper
{
public:
- static bool remove_node(::dsn::rpc_address node,
- /*inout*/ std::vector<::dsn::rpc_address>
&nodeList);
+ template <typename T>
+ static bool remove_node(const T node,
Review Comment:
TODO: make it as an util function
##########
src/meta/server_state.cpp:
##########
@@ -570,6 +573,9 @@ dsn::error_code
server_state::sync_apps_from_remote_storage()
const blob &value)
mutable {
if (ec == ERR_OK) {
partition_configuration pc;
+ pc.__isset.hp_secondaries = true;
Review Comment:
TODO: when upgrade from old versions, these fields are not filled.
##########
src/meta/meta_server_failure_detector.cpp:
##########
@@ -281,50 +295,61 @@ bool
meta_server_failure_detector::update_stability_stat(const fd::beacon_msg &b
void meta_server_failure_detector::on_ping(const fd::beacon_msg &beacon,
rpc_replier<fd::beacon_ack> &reply)
{
- fd::beacon_ack ack;
- ack.time = beacon.time;
- ack.this_node = beacon.to_addr;
- ack.allowed = true;
+ host_port hp_from_node, hp_to_node;
+ GET_HOST_PORT(beacon, from_node, hp_from_node);
+ GET_HOST_PORT(beacon, to_node, hp_to_node);
if (beacon.__isset.start_time && !update_stability_stat(beacon)) {
- LOG_WARNING("{} is unstable, don't response to it's beacon",
beacon.from_addr);
+ LOG_WARNING(
+ "{}({}) is unstable, don't response to it's beacon",
beacon.from_node, hp_from_node);
Review Comment:
TODO: hostname fisrt
##########
src/meta/test/state_sync_test.cpp:
##########
@@ -381,10 +392,14 @@ void meta_service_test_app::construct_apps_test()
std::shared_ptr<meta_service> svc(new meta_service());
- std::vector<dsn::rpc_address> nodes;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> nodes;
std::string hint_message;
generate_node_list(nodes, 1, 1);
- svc->_state->construct_apps({resp}, nodes, hint_message);
+ std::vector<dsn::host_port> hps;
Review Comment:
TODO: optimize schema like "std::vector<dsn::host_port>"
##########
src/meta/test/misc/misc.cpp:
##########
@@ -382,19 +423,21 @@ void migration_check_and_apply(app_mapper &apps,
dsn::partition_configuration &pc =
the_app->partitions[proposal->gpid.get_partition_index()];
- CHECK(!pc.primary.is_invalid(), "");
- CHECK_EQ(pc.secondaries.size(), 2);
- for (auto &addr : pc.secondaries) {
- CHECK(!addr.is_invalid(), "");
+ CHECK(!pc.hp_primary.is_invalid(), "");
+ CHECK_EQ(pc.hp_secondaries.size(), 2);
+ for (auto &host_port : pc.hp_secondaries) {
Review Comment:
TODO: host_port -> hp
##########
src/replica/split/replica_split_manager.cpp:
##########
@@ -614,10 +618,12 @@ void replica_split_manager::child_notify_catch_up() // on
child partition
request->parent_gpid = _replica->_split_states.parent_gpid;
request->child_gpid = get_gpid();
request->child_ballot = get_ballot();
- request->child_address = _stub->_primary_address;
+ request->child = _stub->primary_address();
+ request->__set_hp_child(_stub->primary_host_port());
- LOG_INFO_PREFIX("send notification to primary parent[{}@{}], ballot={}",
+ LOG_INFO_PREFIX("send notification to primary parent[{}@{}({})],
ballot={}",
_replica->_split_states.parent_gpid,
+ _replica->_config.hp_primary,
Review Comment:
TODO: unify logs like this to a macro:
```
#define LOG_HOST_PORT_AND_IP(obj, field) \
fmt::format("{}({})", obj.field, obj.hp_##field)
```
##########
src/replica/replica_context.cpp:
##########
@@ -124,19 +125,21 @@ void
primary_context::get_replica_config(partition_status::type st,
{
config.pid = membership.pid;
config.primary = membership.primary;
+ config.__set_hp_primary(membership.hp_primary);
config.ballot = membership.ballot;
config.status = st;
config.learner_signature = learner_signature;
}
-bool primary_context::check_exist(::dsn::rpc_address node,
partition_status::type st)
+bool primary_context::check_exist(::dsn::host_port node,
partition_status::type st)
{
switch (st) {
case partition_status::PS_PRIMARY:
- return membership.primary == node;
+ return membership.hp_primary == node;
case partition_status::PS_SECONDARY:
- return std::find(membership.secondaries.begin(),
membership.secondaries.end(), node) !=
- membership.secondaries.end();
+ return std::find(membership.hp_secondaries.begin(),
Review Comment:
TODO: optimize this schema "std::find(...)"
##########
src/meta/test/meta_duplication_service_test.cpp:
##########
@@ -547,9 +551,10 @@ TEST_F(meta_duplication_service_test, duplication_sync)
// generate all primaries on node[0]
for (partition_configuration &pc : app->partitions) {
pc.ballot = random32(1, 10000);
- pc.primary = server_nodes[0];
- pc.secondaries.push_back(server_nodes[1]);
- pc.secondaries.push_back(server_nodes[2]);
+ pc.primary = addr;
+ pc.__set_hp_primary(server_nodes[0]);
+ pc.hp_secondaries.push_back(server_nodes[1]);
Review Comment:
Fill secondaries.
##########
src/meta/test/meta_test_base.cpp:
##########
@@ -161,12 +161,15 @@ std::vector<rpc_address>
meta_test_base::ensure_enough_alive_nodes(int min_node_
return nodes;
}
- nodes = generate_node_list(min_node_count);
+ auto node_pairs = generate_node_list(min_node_count);
+ for (const auto &p : node_pairs) {
+ nodes.emplace_back(p.first);
Review Comment:
TODO: optimize it!
##########
src/client/partition_resolver_simple.cpp:
##########
@@ -137,12 +138,12 @@ void
partition_resolver_simple::clear_all_pending_requests()
void partition_resolver_simple::on_timeout(request_context_ptr &&rc) const
{
- end_request(std::move(rc), ERR_TIMEOUT, rpc_address(), true);
+ end_request(std::move(rc), ERR_TIMEOUT, host_port(), true);
}
void partition_resolver_simple::end_request(request_context_ptr &&request,
error_code err,
- rpc_address addr,
+ host_port hp,
Review Comment:
TODO: Use constant reference
##########
src/meta/test/misc/misc.h:
##########
@@ -64,18 +65,22 @@ uint32_t random32(uint32_t min, uint32_t max);
// Generates a random number [min_count, max_count] of node addresses
// each node is given a random port value in range of [min_count, max_count]
-void generate_node_list(/*out*/ std::vector<dsn::rpc_address> &output_list,
- int min_count,
- int max_count);
+void generate_node_list(
+ /*out*/ std::vector<std::pair<dsn::host_port, dsn::rpc_address>>
&output_list,
+ int min_count,
+ int max_count);
// Generates `size` of node addresses, each with port value in range
[start_port, start_port + size]
-inline std::vector<dsn::rpc_address> generate_node_list(size_t size, int
start_port = 12321)
+inline std::vector<std::pair<dsn::host_port, dsn::rpc_address>>
+generate_node_list(size_t size, int start_port = 12321)
{
- std::vector<dsn::rpc_address> result;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> result;
result.resize(size);
- for (int i = 0; i < size; ++i)
- result[i] =
+ for (int i = 0; i < size; ++i) {
+ result[i].first = dsn::host_port("localhost",
static_cast<uint16_t>(start_port + i + 1));
Review Comment:
TODO: optimize this.
##########
idl/meta_admin.thrift:
##########
@@ -278,8 +282,9 @@ struct query_app_manual_compact_response
struct node_info
{
- 1:node_status status = node_status.NS_INVALID;
- 2:dsn.rpc_address address;
+ 1:node_status status = node_status.NS_INVALID;
+ 2:dsn.rpc_address address;
+ 3:optional dsn.host_port hp_address;
Review Comment:
TODO: address -> node, hp_address -> hp_node
##########
src/client/replication_ddl_client.cpp:
##########
@@ -469,8 +502,10 @@ dsn::error_code replication_ddl_client::list_nodes(
return resp.err;
}
- for (const dsn::replication::node_info &n : resp.infos) {
- nodes[n.address] = n.status;
+ for (const auto &n : resp.infos) {
+ host_port hp;
+ GET_HOST_PORT(n, address, hp);
Review Comment:
TODO: rename vars in node_info, address -> node, hp_address -> hp_node
##########
src/runtime/rpc/rpc_engine.cpp:
##########
@@ -149,6 +150,8 @@ bool rpc_client_matcher::on_recv_reply(network *net,
uint64_t key, message_ex *r
case GRPC_TO_LEADER:
if
(req->server_address.group_address()->is_update_leader_automatically()) {
req->server_address.group_address()->set_leader(addr);
+ req->server_host_port.group_host_port()->set_leader(
+ host_port::from_address(addr));
Review Comment:
TODO: can it resolve a host_port from reply in line 136?
##########
src/replica/bulk_load/test/replica_bulk_loader_test.cpp:
##########
@@ -413,6 +421,9 @@ class replica_bulk_loader_test : public replica_test_base
rpc_address PRIMARY = rpc_address::from_ip_port("127.0.0.2", 34801);
rpc_address SECONDARY = rpc_address::from_ip_port("127.0.0.3", 34801);
rpc_address SECONDARY2 = rpc_address::from_ip_port("127.0.0.4", 34801);
+ const host_port PRIMARY_HP = host_port("localhost", 34801);
Review Comment:
TODO: use the resolved IPs from the hosts
##########
src/runtime/rpc/rpc_host_port.cpp:
##########
@@ -209,4 +216,13 @@ error_s
host_port::resolve_addresses(std::vector<rpc_address> &addresses) const
return error_s::ok();
}
+void host_port::fill_host_ports_from_addresses(const std::vector<rpc_address>
&addr_v,
Review Comment:
TODO: make it as a util func
##########
src/failure_detector/failure_detector.cpp:
##########
@@ -302,28 +304,28 @@ void failure_detector::check_all_records()
}
}
-void failure_detector::add_allow_list(::dsn::rpc_address node)
+void failure_detector::add_allow_list(::dsn::host_port node)
{
zauto_lock l(_lock);
_allow_list.insert(node);
}
-bool failure_detector::remove_from_allow_list(::dsn::rpc_address node)
+bool failure_detector::remove_from_allow_list(::dsn::host_port node)
{
zauto_lock l(_lock);
return _allow_list.erase(node) > 0;
}
-void failure_detector::set_allow_list(const std::vector<std::string>
&replica_addrs)
+void failure_detector::set_allow_list(const std::vector<std::string>
&replica_hps)
{
CHECK(!_is_started, "FD is already started, the allow list should really
not be modified");
- std::vector<rpc_address> nodes;
- for (const auto &addr : replica_addrs) {
- const auto node = dsn::rpc_address::from_host_port(addr);
+ std::vector<host_port> nodes;
+ for (auto &hp : replica_hps) {
Review Comment:
TODO: search all to use `const auto &`
##########
src/runtime/rpc/serialization.h:
##########
@@ -101,4 +104,7 @@ inline void unmarshall(dsn::message_ex *msg, /*out*/ T &val)
unmarshall(reader, val,
(dsn_msg_serialize_format)msg->header->context.u.serialize_format);
}
+template <>
+inline void unmarshall(dsn::message_ex *msg, /*out*/ partition_configuration
&val);
Review Comment:
TODO: remove it?
##########
idl/bulk_load.thrift:
##########
@@ -89,15 +89,16 @@ struct partition_bulk_load_state
// meta server -> replica server
struct bulk_load_request
{
- 1:dsn.gpid pid;
- 2:string app_name;
- 3:dsn.rpc_address primary_addr;
- 4:string remote_provider_name;
- 5:string cluster_name;
- 6:i64 ballot;
- 7:bulk_load_status meta_bulk_load_status;
- 8:bool query_bulk_load_metadata;
- 9:string remote_root_path;
+ 1:dsn.gpid pid;
+ 2:string app_name;
+ 3:dsn.rpc_address primary_addr;
+ 4:string remote_provider_name;
+ 5:string cluster_name;
+ 6:i64 ballot;
+ 7:bulk_load_status meta_bulk_load_status;
+ 8:bool query_bulk_load_metadata;
+ 9:string remote_root_path;
+ 10:optional dsn.host_port hp_primary;
Review Comment:
TODO: primary_addr -> primary
##########
src/common/replication_other_types.h:
##########
@@ -49,43 +51,54 @@ typedef int64_t decree;
#define invalid_offset (-1LL)
#define invalid_signature 0
-inline bool is_primary(const partition_configuration &pc, const rpc_address
&node)
+inline bool is_primary(const partition_configuration &pc, const host_port
&node)
{
- return !node.is_invalid() && pc.primary == node;
+ return !node.is_invalid() && pc.hp_primary == node;
}
-inline bool is_secondary(const partition_configuration &pc, const rpc_address
&node)
+inline bool is_secondary(const partition_configuration &pc, const host_port
&node)
{
return !node.is_invalid() &&
- std::find(pc.secondaries.begin(), pc.secondaries.end(), node) !=
pc.secondaries.end();
+ std::find(pc.hp_secondaries.begin(), pc.hp_secondaries.end(), node)
!=
+ pc.hp_secondaries.end();
}
-inline bool is_member(const partition_configuration &pc, const rpc_address
&node)
+inline bool is_member(const partition_configuration &pc, const host_port &node)
{
return is_primary(pc, node) || is_secondary(pc, node);
}
inline bool is_partition_config_equal(const partition_configuration &pc1,
const partition_configuration &pc2)
{
// secondaries no need to be same order
- for (const rpc_address &addr : pc1.secondaries)
+ for (const host_port &addr : pc1.hp_secondaries)
Review Comment:
TODO: const auto ...
##########
src/meta/test/meta_partition_guardian_test.cpp:
##########
@@ -195,8 +204,12 @@ void meta_partition_guardian_test::cure_test()
ASSERT_TRUE(state->spin_wait_staging(20));
svc->_started = true;
- std::vector<dsn::rpc_address> nodes;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> nodes;
Review Comment:
TODO: duplicate in many places.
##########
src/meta/server_state.cpp:
##########
@@ -1850,34 +1954,40 @@ void
server_state::downgrade_secondary_to_inactive(std::shared_ptr<app_state> &a
void server_state::downgrade_stateless_nodes(std::shared_ptr<app_state> &app,
int pidx,
- const rpc_address &address)
+ const host_port &address)
{
std::shared_ptr<configuration_update_request> req =
std::make_shared<configuration_update_request>();
req->info = *app;
req->type = config_type::CT_REMOVE;
- req->host_node = address;
+ req->host_node = dsn::dns_resolver::instance().resolve_address(address);
req->node.set_invalid();
+ req->hp_node.reset();
req->config = app->partitions[pidx];
config_context &cc = app->helpers->contexts[pidx];
partition_configuration &pc = req->config;
unsigned i = 0;
- for (; i < pc.secondaries.size(); ++i) {
- if (pc.secondaries[i] == address) {
+ for (; i < pc.hp_secondaries.size(); ++i) {
+ if (pc.hp_secondaries[i] == address) {
req->node = pc.last_drops[i];
+ req->__set_hp_node(pc.hp_last_drops[i]);
break;
}
}
CHECK(!req->node.is_invalid(), "invalid node address, address = {}",
req->node);
// remove host_node & node from secondaries/last_drops, as it will be sync
to remote storage
- for (++i; i < pc.secondaries.size(); ++i) {
+ for (++i; i < pc.hp_secondaries.size(); ++i) {
Review Comment:
TODO: check the size of hp_secondaries and secondaries equals.
##########
src/meta/server_state.cpp:
##########
@@ -1443,39 +1472,84 @@ void server_state::request_check(const
partition_configuration &old,
switch (request.type) {
case config_type::CT_ASSIGN_PRIMARY:
- CHECK_NE(old.primary, request.node);
- CHECK(std::find(old.secondaries.begin(), old.secondaries.end(),
request.node) ==
- old.secondaries.end(),
- "");
+ if (request.__isset.hp_node) {
+ CHECK_NE(old.hp_primary, request.hp_node);
+ CHECK(std::find(old.hp_secondaries.begin(),
Review Comment:
TODO: Add an util function
##########
src/meta/test/meta_partition_guardian_test.cpp:
##########
@@ -859,69 +897,109 @@ void meta_partition_guardian_test::from_proposal_test()
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 2: test invalid proposal: invalid target" << std::endl;
- cpa2 =
- new_proposal_action(dsn::rpc_address(), node_list[0],
config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(dsn::rpc_address(),
Review Comment:
TODO: reduce the parameters, remove rpc_addresses
##########
src/failure_detector/test/failure_detector.cpp:
##########
@@ -236,7 +241,8 @@ class test_master : public service_app
utils::split_args(args[2].c_str(), ports, ',');
for (auto &port : ports) {
rpc_address addr(network::get_local_ipv4(), std::stoi(port));
- _master_fd->add_allow_list(addr);
+ const auto hp = ::dsn::host_port::from_address(addr);
Review Comment:
TODO: use local host directly.
##########
src/meta/test/misc/misc.cpp:
##########
@@ -58,12 +61,15 @@ uint32_t random32(uint32_t min, uint32_t max)
return res + min;
}
-void generate_node_list(std::vector<dsn::rpc_address> &output_list, int
min_count, int max_count)
+void generate_node_list(std::vector<std::pair<dsn::host_port,
dsn::rpc_address>> &output_list,
+ int min_count,
+ int max_count)
{
const auto count = random32(min_count, max_count);
output_list.resize(count);
- for (auto i = 0; i < count; ++i) {
- output_list[i] = dsn::rpc_address::from_ip_port("127.0.0.1", i + 1);
+ for (int i = 0; i < count; ++i) {
+ output_list[i] = std::make_pair(dsn::host_port("localhost", i + 1),
Review Comment:
TODO: use resolved IP
##########
src/failure_detector/test/failure_detector.cpp:
##########
@@ -652,19 +658,21 @@ TEST(fd, update_stability)
dsn::rpc_replier<beacon_ack> r(create_fake_rpc_response());
beacon_msg msg;
- msg.from_addr = rpc_address::from_host_port("localhost", 123);
- msg.to_addr = rpc_address::from_host_port("localhost", MPORT_START);
+ msg.from_node = rpc_address::from_host_port("localhost", 123);
Review Comment:
TODO: resolve and reuse the hosts
##########
src/meta/test/meta_partition_guardian_test.cpp:
##########
@@ -859,69 +897,109 @@ void meta_partition_guardian_test::from_proposal_test()
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 2: test invalid proposal: invalid target" << std::endl;
- cpa2 =
- new_proposal_action(dsn::rpc_address(), node_list[0],
config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(dsn::rpc_address(),
+ nodes_list[0].second,
+ dsn::host_port(),
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 3: test invalid proposal: invalid node" << std::endl;
- cpa2 =
- new_proposal_action(node_list[0], dsn::rpc_address(),
config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ dsn::rpc_address(),
+ nodes_list[0].first,
+ dsn::host_port(),
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 4: test invalid proposal: dead target" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0],
config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- get_node_state(nodes, node_list[0], false)->set_alive(false);
+ get_node_state(nodes, nodes_list[0].first, false)->set_alive(false);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
- get_node_state(nodes, node_list[0], false)->set_alive(true);
+ get_node_state(nodes, nodes_list[0].first, false)->set_alive(true);
std::cerr << "Case 5: test invalid proposal: dead node" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1],
config_type::CT_ADD_SECONDARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- get_node_state(nodes, node_list[1], false)->set_alive(false);
+ get_node_state(nodes, nodes_list[1].first, false)->set_alive(false);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
- get_node_state(nodes, node_list[1], false)->set_alive(true);
+ get_node_state(nodes, nodes_list[1].first, false)->set_alive(true);
std::cerr << "Case 6: test invalid proposal: already have priamry but
assign" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0],
config_type::CT_ASSIGN_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_ASSIGN_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[1];
+ pc.primary = nodes_list[1].second;
+ pc.__set_hp_primary(nodes_list[1].first);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 7: test invalid proposal: upgrade non-secondary" <<
std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0],
config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
pc.primary.set_invalid();
+ pc.hp_primary.reset();
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 8: test invalid proposal: add exist secondary" <<
std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1],
config_type::CT_ADD_SECONDARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[0];
- pc.secondaries = {node_list[1]};
+ pc.primary = nodes_list[1].second;
+ pc.__set_hp_primary(nodes_list[1].first);
+ pc.secondaries = {nodes_list[1].second};
+ pc.__set_hp_secondaries({nodes_list[1].first});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 9: test invalid proposal: downgrade non member" <<
std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1],
config_type::CT_REMOVE);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_REMOVE);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[0];
- pc.secondaries.clear();
+ pc.primary = nodes_list[0].second;
+ pc.__set_hp_primary(nodes_list[0].first);
+ pc.hp_secondaries.clear();
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 10: test abnormal learning detect" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1],
config_type::CT_ADD_SECONDARY);
- pc.primary = node_list[0];
- pc.secondaries.clear();
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
+ pc.primary = nodes_list[0].second;
+ pc.__set_hp_primary(nodes_list[0].first);
+ pc.hp_secondaries.clear();
Review Comment:
TODO: secondaries.clear()
##########
src/meta/test/meta_partition_guardian_test.cpp:
##########
@@ -859,69 +897,109 @@ void meta_partition_guardian_test::from_proposal_test()
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 2: test invalid proposal: invalid target" << std::endl;
- cpa2 =
- new_proposal_action(dsn::rpc_address(), node_list[0],
config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(dsn::rpc_address(),
+ nodes_list[0].second,
+ dsn::host_port(),
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 3: test invalid proposal: invalid node" << std::endl;
- cpa2 =
- new_proposal_action(node_list[0], dsn::rpc_address(),
config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ dsn::rpc_address(),
+ nodes_list[0].first,
+ dsn::host_port(),
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 4: test invalid proposal: dead target" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0],
config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- get_node_state(nodes, node_list[0], false)->set_alive(false);
+ get_node_state(nodes, nodes_list[0].first, false)->set_alive(false);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
- get_node_state(nodes, node_list[0], false)->set_alive(true);
+ get_node_state(nodes, nodes_list[0].first, false)->set_alive(true);
std::cerr << "Case 5: test invalid proposal: dead node" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1],
config_type::CT_ADD_SECONDARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- get_node_state(nodes, node_list[1], false)->set_alive(false);
+ get_node_state(nodes, nodes_list[1].first, false)->set_alive(false);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
- get_node_state(nodes, node_list[1], false)->set_alive(true);
+ get_node_state(nodes, nodes_list[1].first, false)->set_alive(true);
std::cerr << "Case 6: test invalid proposal: already have priamry but
assign" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0],
config_type::CT_ASSIGN_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_ASSIGN_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[1];
+ pc.primary = nodes_list[1].second;
+ pc.__set_hp_primary(nodes_list[1].first);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 7: test invalid proposal: upgrade non-secondary" <<
std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0],
config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
pc.primary.set_invalid();
+ pc.hp_primary.reset();
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 8: test invalid proposal: add exist secondary" <<
std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1],
config_type::CT_ADD_SECONDARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[0];
- pc.secondaries = {node_list[1]};
+ pc.primary = nodes_list[1].second;
+ pc.__set_hp_primary(nodes_list[1].first);
+ pc.secondaries = {nodes_list[1].second};
+ pc.__set_hp_secondaries({nodes_list[1].first});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 9: test invalid proposal: downgrade non member" <<
std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1],
config_type::CT_REMOVE);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_REMOVE);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[0];
- pc.secondaries.clear();
+ pc.primary = nodes_list[0].second;
+ pc.__set_hp_primary(nodes_list[0].first);
Review Comment:
TODO: secondaries.clear()
##########
src/meta/test/meta_split_service_test.cpp:
##########
@@ -377,7 +378,8 @@ class meta_split_service_test : public meta_test_base
const int32_t PARENT_BALLOT = 3;
const int32_t PARENT_INDEX = 0;
const int32_t CHILD_INDEX = 4;
- const rpc_address NODE = rpc_address::from_ip_port("127.0.0.1", 10086);
+ const host_port NODE = host_port("localhost", 10086);
+ const rpc_address NODE_ADDR = rpc_address::from_host_port("127.0.0.1",
10086);
Review Comment:
TODO: use the resolved IP.
##########
src/meta/test/update_configuration_test.cpp:
##########
@@ -238,52 +247,54 @@ void meta_service_test_app::update_configuration_test()
ss->_all_apps.emplace(1, app);
- std::vector<dsn::rpc_address> nodes;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> nodes;
Review Comment:
TODO: optimize schema like "std::vector<std::pair<dsn::host_port,
dsn::rpc_address>>".
##########
src/redis_protocol/proxy_lib/proxy_layer.h:
##########
@@ -79,8 +79,9 @@ class proxy_session : public
std::enable_shared_from_this<proxy_session>
// when get message from raw parser, request & response of
"dsn::message_ex*" are not in couple.
// we need to backup one request to create a response struct.
dsn::message_ex *_backup_one_request;
- // the client address for which this session served
- dsn::rpc_address _remote_address;
+ // the client for which this session served
Review Comment:
TODO: add host_port
##########
src/nfs/nfs_client_impl.cpp:
##########
@@ -347,14 +355,17 @@ void nfs_client_impl::end_copy(::dsn::error_code err,
METRIC_VAR_INCREMENT(nfs_client_copy_failed_requests);
if (!fc->user_req->is_finished) {
+ host_port hp = fc->user_req->file_size_req.hp_source;
Review Comment:
TODO: unify style
##########
src/replica/replica_learn.cpp:
##########
@@ -1096,19 +1142,21 @@ void replica::on_copy_remote_state_completed(error_code
err,
auto start_ts = dsn_now_ns();
err = apply_learned_state_from_private_log(lstate);
if (err == ERR_OK) {
- LOG_INFO_PREFIX("on_copy_remote_state_completed[{:#018x}]:
learnee = {}, "
+ LOG_INFO_PREFIX("on_copy_remote_state_completed[{:#018x}]:
learnee = {}({}), "
"learn_duration = {} ms, apply_log_duration =
{} ns, apply learned "
"state from private log succeed,
app_committed_decree = {}",
req.signature,
+ resp.config.hp_primary,
resp.config.primary,
_potential_secondary_states.duration_ms(),
dsn_now_ns() - start_ts,
_app->last_committed_decree());
} else {
- LOG_ERROR_PREFIX("on_copy_remote_state_completed[{:#018x}]:
learnee = {}, "
+ LOG_ERROR_PREFIX("on_copy_remote_state_completed[{:#018x}]:
learnee = {}({}), "
"learn_duration = {} ms, apply_log_duration =
{} ns, apply "
"learned state from private log failed, err =
{}",
req.signature,
+ resp.config.hp_primary,
Review Comment:
TODO: reverse the host and IP order
##########
src/replica/split/test/replica_split_test.cpp:
##########
@@ -525,9 +535,12 @@ class replica_split_test : public replica_test_base
const int32_t APP_ID = 2;
const int32_t OLD_PARTITION_COUNT = 8;
const int32_t NEW_PARTITION_COUNT = 16;
- const rpc_address PRIMARY = rpc_address::from_ip_port("127.0.0.1", 18230);
- const rpc_address SECONDARY = rpc_address::from_ip_port("127.0.0.2",
10058);
- const rpc_address SECONDARY2 = rpc_address::from_ip_port("127.0.0.3",
10805);
+ const host_port PRIMARY = host_port("localhost", 18230);
Review Comment:
TODO: use resolve ones
##########
src/replica/duplication/replica_follower.h:
##########
@@ -78,9 +79,12 @@ class replica_follower : replica_base
std::string master_replica_name()
{
std::string app_info = fmt::format("{}.{}", _master_cluster_name,
_master_app_name);
- if (_master_replica_config.primary != rpc_address::s_invalid_address) {
- return fmt::format(
- "{}({}|{})", app_info, _master_replica_config.primary,
_master_replica_config.pid);
+ if (_master_replica_config.hp_primary !=
host_port::s_invalid_host_port) {
Review Comment:
TODO: s_invalid_host_port vs is_invalid()
##########
src/runtime/rpc/network.sim.h:
##########
@@ -91,6 +92,7 @@ class sim_network_provider : public
connection_oriented_network
virtual error_code start(rpc_channel channel, int port, bool client_only);
virtual ::dsn::rpc_address address() { return _address; }
Review Comment:
TODO: const & xxx() const
##########
src/replica/replica_config.cpp:
##########
@@ -257,15 +267,20 @@ void
replica::downgrade_to_secondary_on_primary(configuration_update_request &pr
return;
CHECK_EQ(proposal.config.pid, _primary_states.membership.pid);
- CHECK_EQ(proposal.config.primary, _primary_states.membership.primary);
- CHECK(proposal.config.secondaries ==
_primary_states.membership.secondaries, "");
- CHECK_EQ(proposal.node, proposal.config.primary);
+ CHECK_EQ(proposal.config.hp_primary,
_primary_states.membership.hp_primary);
+ CHECK(proposal.config.hp_secondaries ==
_primary_states.membership.hp_secondaries, "");
+ CHECK_EQ(proposal.hp_node, proposal.config.hp_primary);
proposal.config.primary.set_invalid();
+ proposal.config.__set_hp_primary(host_port());
proposal.config.secondaries.push_back(proposal.node);
+ if (!proposal.config.__isset.hp_secondaries) {
Review Comment:
TODO: refactor this schema one by one
1. find out new newly added hp_xxx fields
2. check the usage
##########
src/replica/test/replica_test.cpp:
##########
@@ -271,6 +271,7 @@ TEST_P(replica_test, write_size_limited)
auto write_request = dsn::message_ex::create_request(RPC_TEST);
auto cleanup = dsn::defer([=]() { delete write_request; });
+ header.context.u.is_forwarded = false;
Review Comment:
TODO: why change it?
##########
src/runtime/rpc/rpc_engine.cpp:
##########
@@ -177,6 +180,8 @@ bool rpc_client_matcher::on_recv_reply(network *net,
uint64_t key, message_ex *r
req->server_address.group_address()->is_update_leader_automatically()) {
req->server_address.group_address()->set_leader(
reply->header->from_address);
+ req->server_host_port.group_host_port()->set_leader(
+
host_port::from_address(reply->header->from_address));
Review Comment:
TODO: add a from_host_port to header?
##########
src/replica/storage/simple_kv/simple_kv.app.example.h:
##########
@@ -45,8 +45,8 @@ class simple_kv_client_app : public ::dsn::service_app
return ::dsn::ERR_INVALID_PARAMETERS;
printf("%s %s %s\n", args[1].c_str(), args[2].c_str(),
args[3].c_str());
- const auto meta = rpc_address::from_host_port(args[2]);
- _simple_kv_client.reset(new simple_kv_client(args[1].c_str(), {meta},
args[3].c_str()));
+ const auto hp = host_port::from_string(args[2].c_str());
Review Comment:
TODO: remove c_str
##########
src/runtime/rpc/rpc_host_port.cpp:
##########
@@ -42,8 +43,11 @@ const host_port host_port::s_invalid_host_port;
host_port::host_port(std::string host, uint16_t port)
: _host(std::move(host)), _port(port), _type(HOST_TYPE_IPV4)
{
- // ipv4_from_host may be slow, just call it in DEBUG version.
- DCHECK_OK(rpc_address::ipv4_from_host(_host, nullptr), "invalid hostname:
{}", _host);
+ // Solve the problem of not translating "0.0.0.0"
+ if (_host != "0.0.0.0") {
Review Comment:
TODO: check it
##########
src/runtime/rpc/rpc_host_port.cpp:
##########
@@ -169,11 +173,14 @@ error_s
host_port::resolve_addresses(std::vector<rpc_address> &addresses) const
__builtin_unreachable();
}
- // 1. Try to resolve hostname in the form of "localhost:80" or
"192.168.0.1:8080".
- const auto rpc_addr = rpc_address::from_ip_port(this->to_string());
- if (rpc_addr) {
- addresses.emplace_back(rpc_addr);
- return error_s::ok();
+ // 1. Try to resolve hostname in the form of "192.168.0.1:8080".
+ uint32_t ip_addr;
+ if (inet_pton(AF_INET, this->_host.c_str(), &ip_addr)) {
Review Comment:
TODO: check it
##########
src/runtime/rpc/rpc_host_port.h:
##########
@@ -67,6 +77,8 @@ class host_port
[[nodiscard]] bool is_invalid() const { return _type == HOST_TYPE_INVALID;
}
+ operator bool() const { return !is_invalid(); }
Review Comment:
TODO: consider to remove it
##########
src/runtime/rpc/rpc_engine.cpp:
##########
@@ -517,9 +522,11 @@ error_code rpc_engine::start(const service_app_spec &aspec)
_local_primary_address = _client_nets[NET_HDR_DSN][0]->address();
_local_primary_address.set_port(aspec.ports.size() > 0 ?
*aspec.ports.begin() : aspec.id);
+ _local_primary_host_port = host_port::from_address(_local_primary_address);
Review Comment:
TODO: Is it equal to _client_nets[NET_HDR_DSN][0]->host_port() ?
##########
src/test/function_test/recovery/test_recovery.cpp:
##########
@@ -71,12 +70,15 @@ class recovery_test : public test_util
// cluster has only one meta server, while "onebox" means the cluster has
3 meta servers.
recovery_test() : test_util(std::map<std::string, std::string>(),
"single_master_cluster") {}
- std::vector<dsn::rpc_address> get_rpc_address_list(const
std::vector<uint16_t> &ports)
+ std::vector<dsn::host_port> get_rpc_host_port_list(const std::vector<int>
ports)
{
- std::vector<dsn::rpc_address> result;
+ std::vector<dsn::host_port> result;
result.reserve(ports.size());
for (const auto &port : ports) {
-
result.emplace_back(dsn::rpc_address(global_env::instance()._host_ip, port));
+ char hostname[1024];
+ gethostname(hostname, 1024);
Review Comment:
TODO: change _host_ip and use it
##########
src/shell/command_utils.cpp:
##########
@@ -22,29 +22,29 @@
#include "client/replication_ddl_client.h"
#include "command_executor.h"
#include "meta_admin_types.h"
-#include "runtime/rpc/rpc_address.h"
+#include "runtime/rpc/rpc_host_port.h"
#include "utils/error_code.h"
bool validate_ip(shell_context *sc,
const std::string &ip_str,
- dsn::rpc_address &target_address,
+ dsn::host_port &target_hp,
std::string &err_info)
{
- target_address = dsn::rpc_address::from_ip_port(ip_str);
- if (!target_address) {
- err_info = fmt::format("invalid ip:port={}, can't transform it into
rpc_address", ip_str);
+ target_hp = dsn::host_port::from_string(ip_str);
Review Comment:
TODO: use rpc_address or change the func name to validate_host_port
##########
src/test/function_test/utils/test_util.cpp:
##########
@@ -89,6 +89,7 @@ void test_util::SetUp()
ddl_client_ = std::make_shared<replication_ddl_client>(meta_list_);
ASSERT_TRUE(ddl_client_ != nullptr);
ddl_client_->set_max_wait_app_ready_secs(120);
+ ddl_client_->set_meta_servers_leader();
Review Comment:
TODO: is it necessary?
##########
src/server/config.min.ini:
##########
@@ -43,7 +43,7 @@
logging_start_level = LOG_LEVEL_INFO
[network]
- primary_interface = lo
+ primary_interface = @LOCAL_HOSTNAME@
Review Comment:
TODO: support not change configs when upgrade version
##########
src/shell/commands/table_management.cpp:
##########
@@ -378,11 +377,11 @@ bool app_disk(command_executor *e, shell_context *sc,
arguments args)
}
std::stringstream oss;
std::string hostname;
- std::string ip = p.primary.to_string();
+ const auto &ip = p.hp_primary.to_string();
Review Comment:
TODO: fix the resolve_ip
##########
src/server/pegasus_server_impl_init.cpp:
##########
@@ -629,7 +629,7 @@
pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
METRIC_VAR_INIT_replica(rdb_bloom_filter_point_lookup_positives),
METRIC_VAR_INIT_replica(rdb_bloom_filter_point_lookup_true_positives)
{
- _primary_address = dsn::rpc_address(dsn_primary_address()).to_string();
+ _primary_address = dsn_primary_host_port().to_string();
Review Comment:
TODO: _primary_address -> _primary_host_port
##########
src/shell/commands/rebalance.cpp:
##########
@@ -112,12 +114,12 @@ bool propose(command_executor *e, shell_context *sc,
arguments args)
proposal_type += optarg;
break;
case 't':
- target = dsn::rpc_address::from_host_port(optarg);
- PRINT_AND_RETURN_FALSE_IF_NOT(target, "parse {} as target_address
failed\n", optarg);
+ target = dsn::host_port::from_string(optarg);
Review Comment:
TODO: check both hostname and IP are accepted
##########
src/shell/commands/node_management.cpp:
##########
@@ -666,9 +666,9 @@ bool remote_command(command_executor *e, shell_context *sc,
arguments args)
node_desc &n = node_list[i];
std::string hostname;
if (resolve_ip) {
- dsn::utils::hostname_from_ip_port(n.address.to_string(),
&hostname);
+ hostname =
dsn::dns_resolver::instance().resolve_address(n.hp).to_string();
Review Comment:
TODO: fix the reversed
##########
src/shell/commands/detect_hotkey.cpp:
##########
@@ -100,10 +100,10 @@ bool detect_hotkey(command_executor *e, shell_context
*sc, arguments args)
return false;
}
- dsn::rpc_address target_address;
+ dsn::host_port target_hp;
std::string err_info;
std::string ip_str = cmd({"-d", "--address"}).str();
- if (!validate_ip(sc, ip_str, target_address, err_info)) {
+ if (!validate_ip(sc, ip_str, target_hp, err_info)) {
Review Comment:
TODO: it means only host_port is accepted?
##########
src/shell/commands/rebalance.cpp:
##########
@@ -194,20 +201,31 @@ bool balance(command_executor *e, shell_context *sc,
arguments args)
std::vector<configuration_proposal_action> &actions = request.action_list;
actions.reserve(4);
+ const auto &from_addr =
dsn::dns_resolver::instance().resolve_address(from);
+ const auto &to_addr = dsn::dns_resolver::instance().resolve_address(to);
if (balance_type == "move_pri") {
+ actions.emplace_back(new_proposal_action(
Review Comment:
TODO: reduce parameters
##########
src/test/function_test/config.ini:
##########
@@ -75,8 +75,8 @@ rpc_timeout_milliseconds = 5000
lb_interval_ms = 3000
[pegasus.clusters]
-onebox = 127.0.0.1:34601,127.0.0.1:34602,127.0.0.1:34603
-single_master_cluster = 127.0.0.1:34601
+onebox = @LOCAL_HOSTNAME@:34601,@LOCAL_HOSTNAME@:34602,@LOCAL_HOSTNAME@:34603
Review Comment:
TODO: check ip is acceptable as well
##########
src/runtime/test/rpc.cpp:
##########
@@ -224,8 +225,8 @@ TEST(core, group_address_no_response_2)
TEST(core, send_to_invalid_address)
Review Comment:
TODO: add test to send to invalid hp group
##########
src/runtime/rpc/asio_net_provider.cpp:
##########
@@ -356,7 +370,23 @@ void asio_udp_provider::do_receive()
return;
}
+ if (msg->header->from_address != remote_addr) {
Review Comment:
TODO: check it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
