acelyc111 commented on code in PR #1464: URL: https://github.com/apache/incubator-pegasus/pull/1464#discussion_r1186204449
########## src/runtime/rpc/dns_resolver.h: ########## @@ -0,0 +1,54 @@ +/* + * 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 <unordered_map> +#include <vector> + +#include "runtime/rpc/rpc_address.h" +#include "runtime/rpc/rpc_host_port.h" +#include "utils/errors.h" +#include "utils/synchronize.h" + +namespace dsn { + +// This class provider way transfer host_port to rpc_address. Review Comment: ```suggestion // This class provide a way to resolve host_port to rpc_address. ``` ########## src/runtime/rpc/dns_resolver.h: ########## @@ -0,0 +1,54 @@ +/* + * 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 <unordered_map> +#include <vector> + +#include "runtime/rpc/rpc_address.h" +#include "runtime/rpc/rpc_host_port.h" +#include "utils/errors.h" +#include "utils/synchronize.h" + +namespace dsn { + +// This class provider way transfer host_port to rpc_address. +class dns_resolver +{ +public: + explicit dns_resolver() = default; + + void add_item(const host_port &hp, const rpc_address &addr); + + // Transfer host_port to unique rpc_address. Review Comment: ```suggestion // Resolve this host_port to an unique rpc_address. ``` ########## src/runtime/rpc/dns_resolver.cpp: ########## @@ -0,0 +1,107 @@ +/* + * 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 "runtime/rpc/dns_resolver.h" + +#include <utility> + +#include "runtime/rpc/group_address.h" +#include "runtime/rpc/group_host_port.h" +#include "utils/fmt_logging.h" + +namespace dsn { + +void dns_resolver::add_item(const host_port &hp, const rpc_address &addr) +{ + utils::auto_write_lock l(_lock); + _dsn_cache.insert(std::make_pair(hp, addr)); +} + +bool dns_resolver::get_cached_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + utils::auto_read_lock l(_lock); + const auto &found = _dsn_cache.find(hp); + if (found == _dsn_cache.end()) { + return false; + } + + addresses = {found->second}; + return true; +} + +error_s dns_resolver::resolve_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + CHECK(addresses.empty(), "invalid addresses, not empty"); + if (get_cached_addresses(hp, addresses)) { + return error_s::ok(); + } + + std::vector<rpc_address> resolved_addresses; + RETURN_NOT_OK(hp.resolve_addresses(resolved_addresses)); + + { + utils::auto_write_lock l(_lock); + if (resolved_addresses.size() > 1) { + LOG_WARNING("host_port '{}' resolves to {} different addresses, using {}.", Review Comment: How about logging all the addresses, it would be helpful for debuging. Tips: use `fmt::join(resolved_addresses, ",")` ########## src/runtime/rpc/dns_resolver.cpp: ########## @@ -0,0 +1,107 @@ +/* + * 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 "runtime/rpc/dns_resolver.h" + +#include <utility> + +#include "runtime/rpc/group_address.h" +#include "runtime/rpc/group_host_port.h" +#include "utils/fmt_logging.h" + +namespace dsn { + +void dns_resolver::add_item(const host_port &hp, const rpc_address &addr) +{ + utils::auto_write_lock l(_lock); + _dsn_cache.insert(std::make_pair(hp, addr)); +} + +bool dns_resolver::get_cached_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + utils::auto_read_lock l(_lock); + const auto &found = _dsn_cache.find(hp); + if (found == _dsn_cache.end()) { + return false; + } + + addresses = {found->second}; + return true; +} + +error_s dns_resolver::resolve_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + CHECK(addresses.empty(), "invalid addresses, not empty"); + if (get_cached_addresses(hp, addresses)) { + return error_s::ok(); + } + + std::vector<rpc_address> resolved_addresses; + RETURN_NOT_OK(hp.resolve_addresses(resolved_addresses)); + + { + utils::auto_write_lock l(_lock); + if (resolved_addresses.size() > 1) { + LOG_WARNING("host_port '{}' resolves to {} different addresses, using {}.", + hp, + resolved_addresses.size(), + resolved_addresses[0]); + } + _dsn_cache.insert(std::make_pair(hp, resolved_addresses[0])); + } + + addresses = std::move(resolved_addresses); + return error_s::ok(); +} + +rpc_address dns_resolver::resolve_address(const host_port &hp) +{ + switch (hp.type()) { + case HOST_TYPE_GROUP: { + rpc_address addr; + auto group_address = hp.group_host_port(); + addr.assign_group(group_address->name()); + + for (const auto &hp : group_address->members()) { + CHECK_TRUE(addr.group_address()->add(resolve_address(hp))); Review Comment: Add a CHECK to ensure `hp` is a HOST_TYPE_IPV4 type of host_port? ########## src/runtime/rpc/rpc_host_port.cpp: ########## @@ -113,4 +146,50 @@ void host_port::assign_group(const char *name) _group_host_port->add_ref(); } +error_s host_port::resolve_addresses(std::vector<rpc_address> &addresses) const +{ + CHECK(addresses.empty(), ""); + + switch (type()) { + case HOST_TYPE_INVALID: + return error_s::make(dsn::ERR_INVALID_STATE, "invalid host_port type: HOST_TYPE_INVALID"); + case HOST_TYPE_GROUP: + return error_s::make(dsn::ERR_INVALID_STATE, "invalid host_port type: HOST_TYPE_GROUP"); + case HOST_TYPE_IPV4: + break; + } Review Comment: How about? ```suggestion if (type() != HOST_TYPE_IPV4) { return error_s::make(dsn::ERR_INVALID_STATE, "invalid host_port type {}", type()); } ``` ########## src/runtime/rpc/rpc_host_port.cpp: ########## @@ -113,4 +146,50 @@ void host_port::assign_group(const char *name) _group_host_port->add_ref(); } +error_s host_port::resolve_addresses(std::vector<rpc_address> &addresses) const +{ + CHECK(addresses.empty(), ""); + + switch (type()) { + case HOST_TYPE_INVALID: + return error_s::make(dsn::ERR_INVALID_STATE, "invalid host_port type: HOST_TYPE_INVALID"); + case HOST_TYPE_GROUP: + return error_s::make(dsn::ERR_INVALID_STATE, "invalid host_port type: HOST_TYPE_GROUP"); + case HOST_TYPE_IPV4: + break; + } + + rpc_address rpc_addr; + // Transfer hostname like "local" & "192.168.0.1". Review Comment: ```suggestion // Resolve hostname like "localhost:80" or "192.168.0.1:8080". ``` from_string_ipv4 will fail if lack of port field, so add port in comments. ########## src/runtime/rpc/dns_resolver.cpp: ########## @@ -0,0 +1,107 @@ +/* + * 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 "runtime/rpc/dns_resolver.h" + +#include <utility> + +#include "runtime/rpc/group_address.h" +#include "runtime/rpc/group_host_port.h" +#include "utils/fmt_logging.h" + +namespace dsn { + +void dns_resolver::add_item(const host_port &hp, const rpc_address &addr) +{ + utils::auto_write_lock l(_lock); + _dsn_cache.insert(std::make_pair(hp, addr)); +} + +bool dns_resolver::get_cached_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + utils::auto_read_lock l(_lock); + const auto &found = _dsn_cache.find(hp); + if (found == _dsn_cache.end()) { + return false; + } + + addresses = {found->second}; + return true; +} + +error_s dns_resolver::resolve_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + CHECK(addresses.empty(), "invalid addresses, not empty"); + if (get_cached_addresses(hp, addresses)) { + return error_s::ok(); + } + + std::vector<rpc_address> resolved_addresses; + RETURN_NOT_OK(hp.resolve_addresses(resolved_addresses)); + + { + utils::auto_write_lock l(_lock); + if (resolved_addresses.size() > 1) { + LOG_WARNING("host_port '{}' resolves to {} different addresses, using {}.", + hp, + resolved_addresses.size(), + resolved_addresses[0]); + } + _dsn_cache.insert(std::make_pair(hp, resolved_addresses[0])); + } + + addresses = std::move(resolved_addresses); + return error_s::ok(); +} + +rpc_address dns_resolver::resolve_address(const host_port &hp) +{ + switch (hp.type()) { + case HOST_TYPE_GROUP: { + rpc_address addr; + auto group_address = hp.group_host_port(); + addr.assign_group(group_address->name()); + + for (const auto &hp : group_address->members()) { + CHECK_TRUE(addr.group_address()->add(resolve_address(hp))); + } + addr.group_address()->set_update_leader_automatically( + group_address->is_update_leader_automatically()); + addr.group_address()->set_leader(resolve_address(group_address->leader())); + return addr; + } + case HOST_TYPE_IPV4: { + std::vector<rpc_address> addresses; + CHECK_OK(resolve_addresses(hp, addresses), "host_port '{}' can not be resolved", hp); + CHECK(!addresses.empty(), "host_port '{}' can not be resolved to any address", hp); + + if (addresses.size() > 1) { + LOG_WARNING("host_port '{}' resolves to {} different addresses, using {}", Review Comment: ```suggestion LOG_WARNING("host_port '{}' resolves to {} different addresses, using the first one {}", ``` ########## src/runtime/rpc/rpc_host_port.h: ########## @@ -68,6 +70,10 @@ class host_port } void assign_group(const char *name); + // Transfer host_port to rpc_addresses. + // Trere could be multi rpc_address for one host_port. Review Comment: ```suggestion // Trere may be multiple rpc_addresses for one host_port. ``` ########## src/runtime/rpc/dns_resolver.cpp: ########## @@ -0,0 +1,107 @@ +/* + * 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 "runtime/rpc/dns_resolver.h" + +#include <utility> + +#include "runtime/rpc/group_address.h" +#include "runtime/rpc/group_host_port.h" +#include "utils/fmt_logging.h" + +namespace dsn { + +void dns_resolver::add_item(const host_port &hp, const rpc_address &addr) +{ + utils::auto_write_lock l(_lock); + _dsn_cache.insert(std::make_pair(hp, addr)); +} + +bool dns_resolver::get_cached_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + utils::auto_read_lock l(_lock); + const auto &found = _dsn_cache.find(hp); + if (found == _dsn_cache.end()) { + return false; + } + + addresses = {found->second}; + return true; +} + +error_s dns_resolver::resolve_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + CHECK(addresses.empty(), "invalid addresses, not empty"); + if (get_cached_addresses(hp, addresses)) { + return error_s::ok(); + } + + std::vector<rpc_address> resolved_addresses; + RETURN_NOT_OK(hp.resolve_addresses(resolved_addresses)); + + { + utils::auto_write_lock l(_lock); + if (resolved_addresses.size() > 1) { + LOG_WARNING("host_port '{}' resolves to {} different addresses, using {}.", Review Comment: ```suggestion LOG_WARNING("host_port '{}' resolves to {} different addresses, using the first one {}.", ``` ########## src/runtime/rpc/dns_resolver.cpp: ########## @@ -0,0 +1,107 @@ +/* + * 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 "runtime/rpc/dns_resolver.h" + +#include <utility> + +#include "runtime/rpc/group_address.h" +#include "runtime/rpc/group_host_port.h" +#include "utils/fmt_logging.h" + +namespace dsn { + +void dns_resolver::add_item(const host_port &hp, const rpc_address &addr) +{ + utils::auto_write_lock l(_lock); + _dsn_cache.insert(std::make_pair(hp, addr)); +} + +bool dns_resolver::get_cached_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + utils::auto_read_lock l(_lock); + const auto &found = _dsn_cache.find(hp); + if (found == _dsn_cache.end()) { + return false; + } + + addresses = {found->second}; + return true; +} + +error_s dns_resolver::resolve_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + CHECK(addresses.empty(), "invalid addresses, not empty"); + if (get_cached_addresses(hp, addresses)) { + return error_s::ok(); + } + + std::vector<rpc_address> resolved_addresses; + RETURN_NOT_OK(hp.resolve_addresses(resolved_addresses)); + Review Comment: Add a CHECK to ensure resolved_addresses is not empty? ########## src/runtime/rpc/dns_resolver.cpp: ########## @@ -0,0 +1,107 @@ +/* + * 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 "runtime/rpc/dns_resolver.h" + +#include <utility> + +#include "runtime/rpc/group_address.h" +#include "runtime/rpc/group_host_port.h" +#include "utils/fmt_logging.h" + +namespace dsn { + +void dns_resolver::add_item(const host_port &hp, const rpc_address &addr) +{ + utils::auto_write_lock l(_lock); + _dsn_cache.insert(std::make_pair(hp, addr)); +} + +bool dns_resolver::get_cached_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + utils::auto_read_lock l(_lock); + const auto &found = _dsn_cache.find(hp); + if (found == _dsn_cache.end()) { + return false; + } + + addresses = {found->second}; + return true; +} + +error_s dns_resolver::resolve_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + CHECK(addresses.empty(), "invalid addresses, not empty"); + if (get_cached_addresses(hp, addresses)) { + return error_s::ok(); + } + + std::vector<rpc_address> resolved_addresses; + RETURN_NOT_OK(hp.resolve_addresses(resolved_addresses)); + + { + utils::auto_write_lock l(_lock); + if (resolved_addresses.size() > 1) { + LOG_WARNING("host_port '{}' resolves to {} different addresses, using {}.", + hp, + resolved_addresses.size(), + resolved_addresses[0]); + } + _dsn_cache.insert(std::make_pair(hp, resolved_addresses[0])); + } + + addresses = std::move(resolved_addresses); + return error_s::ok(); +} + +rpc_address dns_resolver::resolve_address(const host_port &hp) +{ + switch (hp.type()) { + case HOST_TYPE_GROUP: { + rpc_address addr; + auto group_address = hp.group_host_port(); + addr.assign_group(group_address->name()); + + for (const auto &hp : group_address->members()) { + CHECK_TRUE(addr.group_address()->add(resolve_address(hp))); + } + addr.group_address()->set_update_leader_automatically( + group_address->is_update_leader_automatically()); + addr.group_address()->set_leader(resolve_address(group_address->leader())); + return addr; + } + case HOST_TYPE_IPV4: { + std::vector<rpc_address> addresses; + CHECK_OK(resolve_addresses(hp, addresses), "host_port '{}' can not be resolved", hp); + CHECK(!addresses.empty(), "host_port '{}' can not be resolved to any address", hp); + + if (addresses.size() > 1) { + LOG_WARNING("host_port '{}' resolves to {} different addresses, using {}", Review Comment: How about logging all the addresses, it would be helpful for debuging. Tips: use `fmt::join(addresses, ",")` ########## src/runtime/rpc/dns_resolver.cpp: ########## @@ -0,0 +1,107 @@ +/* + * 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 "runtime/rpc/dns_resolver.h" + +#include <utility> + +#include "runtime/rpc/group_address.h" +#include "runtime/rpc/group_host_port.h" +#include "utils/fmt_logging.h" + +namespace dsn { + +void dns_resolver::add_item(const host_port &hp, const rpc_address &addr) +{ + utils::auto_write_lock l(_lock); + _dsn_cache.insert(std::make_pair(hp, addr)); +} + +bool dns_resolver::get_cached_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + utils::auto_read_lock l(_lock); + const auto &found = _dsn_cache.find(hp); + if (found == _dsn_cache.end()) { + return false; + } + + addresses = {found->second}; + return true; +} + +error_s dns_resolver::resolve_addresses(const host_port &hp, std::vector<rpc_address> &addresses) +{ + CHECK(addresses.empty(), "invalid addresses, not empty"); + if (get_cached_addresses(hp, addresses)) { + return error_s::ok(); + } + + std::vector<rpc_address> resolved_addresses; + RETURN_NOT_OK(hp.resolve_addresses(resolved_addresses)); + + { + utils::auto_write_lock l(_lock); + if (resolved_addresses.size() > 1) { + LOG_WARNING("host_port '{}' resolves to {} different addresses, using {}.", + hp, + resolved_addresses.size(), + resolved_addresses[0]); + } + _dsn_cache.insert(std::make_pair(hp, resolved_addresses[0])); + } + + addresses = std::move(resolved_addresses); + return error_s::ok(); +} + +rpc_address dns_resolver::resolve_address(const host_port &hp) +{ + switch (hp.type()) { + case HOST_TYPE_GROUP: { + rpc_address addr; + auto group_address = hp.group_host_port(); + addr.assign_group(group_address->name()); + + for (const auto &hp : group_address->members()) { + CHECK_TRUE(addr.group_address()->add(resolve_address(hp))); + } + addr.group_address()->set_update_leader_automatically( + group_address->is_update_leader_automatically()); + addr.group_address()->set_leader(resolve_address(group_address->leader())); Review Comment: Add a CHECK to ensure group_address->leader() is a HOST_TYPE_IPV4 type of host_port? ########## src/runtime/rpc/rpc_host_port.h: ########## @@ -68,6 +70,10 @@ class host_port } void assign_group(const char *name); + // Transfer host_port to rpc_addresses. Review Comment: ```suggestion // Resolve host_port to rpc_addresses. ``` -- 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]
