You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by aw...@apache.org on 2021/09/24 05:57:40 UTC
[kudu] branch master updated: [mini-cluster] allow using aliases
instead of addresses in EMC
This is an automated email from the ASF dual-hosted git repository.
awong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push:
new acf6de8 [mini-cluster] allow using aliases instead of addresses in EMC
acf6de8 is described below
commit acf6de866c878dc2f09fb5bd7cccf0312375e7c7
Author: Andrew Wong <aw...@cloudera.com>
AuthorDate: Wed Sep 15 19:36:16 2021 -0700
[mini-cluster] allow using aliases instead of addresses in EMC
This patch adds a configuration to ExternalMiniClusters to allow masters
and tablet servers to use and broadcast an alias instead of a bound
address. This is useful to simulate the default deployment where the
configuration of --rpc_bind_addresses results in servers broadcasting
their FQDN instead of a bound address.
This entails a couple things:
- The EMC may now use the --dns_addr_resolution_override flag to map
aliases to the reserved addresses.
- Previously we used Sockaddrs internally and converted them to
HostPorts to pass around to various functions for remote callers to
resolve (e.g. TServers pass these to Masters and Masters would
resolve the HostPort to communicate with TServers). Since the
HostPorts need to be resolvable remotely, wildcards would be
substituted with the FQDN. In ExternalMiniClusters, we don't bind to
the wildcard, but we still need to allow a server to broadcast its
alias in lieu of an FQDN. A new --host_for_test is added for this
purpose; all ExternalMasters and ExternalTabletServers set this when
the configured to use aliases.
- This also refactors our current Sockaddr-to-HostPort conversions to:
1) fetch HostPorts directly at call-sites, rather than getting
Sockaddrs and doing a conversion, and
2) also account for --host_for_test.
NOTE: this mini-cluster aliasing is compatible with the existing
--rpc_listen_on_unix_domain_socket flag. Client processes need only set
--host_for_test, and whatever server has the same --host_for_test will
treat that client as local and communicate over a UNIX socket.
A test is added that exercises this new functionality.
Change-Id: I09a22636748a13f282406119b52021184d92a76f
Reviewed-on: http://gerrit.cloudera.org:8080/17850
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Tested-by: Kudu Jenkins
---
src/kudu/integration-tests/CMakeLists.txt | 1 +
src/kudu/integration-tests/dns_alias-itest.cc | 184 +++++++++++++++++++++++++
src/kudu/master/master.cc | 10 +-
src/kudu/master/master_path_handlers.cc | 16 +--
src/kudu/master/sys_catalog.cc | 10 +-
src/kudu/mini-cluster/external_mini_cluster.cc | 38 ++++-
src/kudu/mini-cluster/external_mini_cluster.h | 9 ++
src/kudu/server/rpc_server.cc | 12 ++
src/kudu/server/rpc_server.h | 3 +
src/kudu/server/server_base.cc | 20 +--
src/kudu/server/webserver.cc | 12 ++
src/kudu/server/webserver.h | 3 +
src/kudu/tserver/heartbeater.cc | 11 +-
src/kudu/tserver/ts_tablet_manager.cc | 10 +-
src/kudu/util/net/net_util-test.cc | 17 +--
src/kudu/util/net/net_util.cc | 43 ++++--
src/kudu/util/net/net_util.h | 9 +-
17 files changed, 350 insertions(+), 58 deletions(-)
diff --git a/src/kudu/integration-tests/CMakeLists.txt b/src/kudu/integration-tests/CMakeLists.txt
index f7208c3..c687930 100644
--- a/src/kudu/integration-tests/CMakeLists.txt
+++ b/src/kudu/integration-tests/CMakeLists.txt
@@ -78,6 +78,7 @@ ADD_KUDU_TEST(delete_table-itest NUM_SHARDS 8 PROCESSORS 4)
ADD_KUDU_TEST(delete_tablet-itest PROCESSORS 2)
ADD_KUDU_TEST(disk_failure-itest PROCESSORS 2)
ADD_KUDU_TEST(disk_reservation-itest)
+ADD_KUDU_TEST(dns_alias-itest)
ADD_KUDU_TEST(exactly_once_writes-itest)
ADD_KUDU_TEST(flex_partitioning-itest TIMEOUT 1800 NUM_SHARDS 8 PROCESSORS 2)
ADD_KUDU_TEST(full_stack-insert-scan-test RUN_SERIAL true)
diff --git a/src/kudu/integration-tests/dns_alias-itest.cc b/src/kudu/integration-tests/dns_alias-itest.cc
new file mode 100644
index 0000000..0e92cc3
--- /dev/null
+++ b/src/kudu/integration-tests/dns_alias-itest.cc
@@ -0,0 +1,184 @@
+// 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 <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <gflags/gflags_declare.h>
+#include <gtest/gtest.h>
+
+#include "kudu/client/client.h"
+#include "kudu/gutil/stl_util.h"
+#include "kudu/gutil/strings/split.h"
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/integration-tests/cluster_itest_util.h"
+#include "kudu/integration-tests/test_workload.h"
+#include "kudu/mini-cluster/external_mini_cluster.h"
+#include "kudu/util/metrics.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/net/net_util.h"
+#include "kudu/util/net/sockaddr.h"
+#include "kudu/util/status.h"
+#include "kudu/util/test_macros.h"
+#include "kudu/util/test_util.h"
+
+DECLARE_bool(client_use_unix_domain_sockets);
+DECLARE_string(dns_addr_resolution_override);
+DECLARE_string(host_for_tests);
+
+METRIC_DECLARE_counter(rpc_connections_accepted_unix_domain_socket);
+METRIC_DECLARE_entity(server);
+
+using kudu::client::KuduClient;
+using kudu::client::KuduTabletServer;
+using kudu::cluster::ExternalMiniCluster;
+using kudu::cluster::ExternalMiniClusterOptions;
+using std::string;
+using std::unique_ptr;
+using std::vector;
+using strings::Split;
+using strings::Substitute;
+
+namespace kudu {
+namespace itest {
+
+namespace {
+constexpr const char* kTServerHostPrefix = "tserver.host";
+constexpr const char* kMasterHostPrefix = "master.host";
+} // anonymous namespace
+
+class DnsAliasITest : public KuduTest {
+ public:
+ void SetUp() override {
+ SetUpCluster();
+ }
+
+ void SetUpCluster(vector<string> extra_master_flags = {},
+ vector<string> extra_tserver_flags = {}) {
+ ExternalMiniClusterOptions opts;
+ opts.num_masters = 3;
+ opts.num_tablet_servers = 3;
+ opts.extra_master_flags = std::move(extra_master_flags);
+ opts.extra_tserver_flags = std::move(extra_tserver_flags);
+ opts.master_alias_prefix = kMasterHostPrefix;
+ opts.tserver_alias_prefix = kTServerHostPrefix;
+ cluster_.reset(new ExternalMiniCluster(std::move(opts)));
+ ASSERT_OK(cluster_->Start());
+
+ FLAGS_dns_addr_resolution_override = cluster_->dns_overrides();
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
+ }
+
+ void TearDown() override {
+ NO_FATALS(cluster_->AssertNoCrashes());
+ }
+ protected:
+ unique_ptr<ExternalMiniCluster> cluster_;
+ client::sp::shared_ptr<KuduClient> client_;
+};
+
+TEST_F(DnsAliasITest, TestBasic) {
+ // Based on the mini-cluster setup, the client should report the aliases.
+ auto master_addrs_str = client_->GetMasterAddresses();
+ vector<string> master_addrs = Split(master_addrs_str, ",");
+ ASSERT_EQ(cluster_->num_masters(), master_addrs.size()) << master_addrs_str;
+ for (const auto& master_addr : master_addrs) {
+ ASSERT_STR_CONTAINS(master_addr, kMasterHostPrefix);
+ // Try resolving a numeric IP. This should fail, since the returned values
+ // should be aliased.
+ Sockaddr addr;
+ Status s = addr.ParseString(master_addr, 0);
+ ASSERT_FALSE(s.ok());
+ }
+
+ vector<KuduTabletServer*> tservers;
+ ElementDeleter deleter(&tservers);
+ ASSERT_OK(client_->ListTabletServers(&tservers));
+ for (const auto* tserver : tservers) {
+ ASSERT_STR_CONTAINS(tserver->hostname(), kTServerHostPrefix);
+ // Try resolving a numeric IP. This should fail, since the returned values
+ // should be aliased.
+ Sockaddr addr;
+ Status s = addr.ParseString(tserver->hostname(), 0);
+ ASSERT_FALSE(s.ok());
+ }
+
+ // Running a test worload should succeed. Have the workload perform both
+ // scans and writes to exercise the aliasing codepaths of each.
+ TestWorkload w(cluster_.get());
+ w.set_num_write_threads(1);
+ w.set_num_read_threads(3);
+ w.set_num_replicas(3);
+ w.Setup();
+ w.Start();
+ while (w.rows_inserted() < 10) {
+ SleepFor(MonoDelta::FromMilliseconds(10));
+ }
+ w.StopAndJoin();
+}
+
+class DnsAliasWithUnixSocketsITest : public DnsAliasITest {
+ public:
+ void SetUp() override {
+ // Configure --host_for_tests in this process so the test client will think
+ // it's local to a tserver.
+ FLAGS_host_for_tests = Substitute("$0.$1", kTServerHostPrefix, kTServerIdxWithLocalClient);
+ FLAGS_client_use_unix_domain_sockets = true;
+ SetUpCluster({ "--rpc_listen_on_unix_domain_socket=true" },
+ { "--rpc_listen_on_unix_domain_socket=true" });
+ }
+ protected:
+ const int kTServerIdxWithLocalClient = 0;
+};
+
+TEST_F(DnsAliasWithUnixSocketsITest, TestBasic) {
+ TestWorkload w(cluster_.get());
+ w.set_num_write_threads(1);
+ w.set_num_read_threads(3);
+ w.set_num_replicas(3);
+ w.Setup();
+ w.Start();
+ while (w.rows_inserted() < 10) {
+ SleepFor(MonoDelta::FromMilliseconds(10));
+ }
+ w.StopAndJoin();
+ for (int i = 0; i < cluster_->num_tablet_servers(); i++) {
+ int64_t unix_connections = 0;
+
+ // Curl doesn't know about our DNS aliasing, so resolve the address and
+ // fetch the metric from the proper address.
+ vector<Sockaddr> addrs;
+ ASSERT_OK(cluster_->tablet_server(i)->bound_http_hostport().ResolveAddresses(&addrs));
+ ASSERT_EQ(1, addrs.size());
+ const auto& addr = addrs[0];
+ ASSERT_OK(GetInt64Metric(HostPort(addr.host(), addr.port()),
+ &METRIC_ENTITY_server, nullptr,
+ &METRIC_rpc_connections_accepted_unix_domain_socket,
+ "value", &unix_connections));
+ if (i == kTServerIdxWithLocalClient) {
+ ASSERT_LT(0, unix_connections);
+ } else {
+ ASSERT_EQ(0, unix_connections);
+ }
+ }
+}
+
+} // namespace itest
+} // namespace kudu
diff --git a/src/kudu/master/master.cc b/src/kudu/master/master.cc
index a9b9538..1c9800a 100644
--- a/src/kudu/master/master.cc
+++ b/src/kudu/master/master.cc
@@ -28,6 +28,7 @@
#include <glog/logging.h>
#include "kudu/cfile/block_cache.h"
+#include "kudu/common/common.pb.h"
#include "kudu/common/wire_protocol.h"
#include "kudu/common/wire_protocol.pb.h"
#include "kudu/consensus/metadata.pb.h"
@@ -380,10 +381,11 @@ Status Master::InitMasterRegistration() {
CHECK(!registration_initialized_.load());
ServerRegistrationPB reg;
- vector<Sockaddr> rpc_addrs;
- RETURN_NOT_OK_PREPEND(rpc_server()->GetAdvertisedAddresses(&rpc_addrs),
- "Couldn't get RPC addresses");
- RETURN_NOT_OK(AddHostPortPBs(rpc_addrs, reg.mutable_rpc_addresses()));
+ vector<HostPort> hps;
+ RETURN_NOT_OK(rpc_server()->GetAdvertisedHostPorts(&hps));
+ for (const auto& hp : hps) {
+ *reg.add_rpc_addresses() = HostPortToPB(hp);
+ }
if (web_server()) {
vector<Sockaddr> http_addrs;
diff --git a/src/kudu/master/master_path_handlers.cc b/src/kudu/master/master_path_handlers.cc
index 9012bec..9de3632 100644
--- a/src/kudu/master/master_path_handlers.cc
+++ b/src/kudu/master/master_path_handlers.cc
@@ -42,6 +42,7 @@
#include "kudu/consensus/metadata.pb.h"
#include "kudu/consensus/quorum_util.h"
#include "kudu/gutil/map-util.h"
+#include "kudu/gutil/port.h"
#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/stringprintf.h"
#include "kudu/gutil/strings/ascii_ctype.h"
@@ -58,6 +59,7 @@
#include "kudu/master/ts_descriptor.h"
#include "kudu/master/ts_manager.h"
#include "kudu/server/monitored_task.h"
+#include "kudu/server/rpc_server.h"
#include "kudu/server/webui_util.h"
#include "kudu/tablet/metadata.pb.h"
#include "kudu/util/cow_object.h"
@@ -66,7 +68,6 @@
#include "kudu/util/metrics.h"
#include "kudu/util/monotime.h"
#include "kudu/util/net/net_util.h"
-#include "kudu/util/net/sockaddr.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/string_case.h"
#include "kudu/util/url-coding.h"
@@ -846,14 +847,13 @@ string MasterPathHandlers::MasterAddrsToCsv() const {
}
return JoinElements(all_addresses, ",");
}
- Sockaddr addr = master_->first_rpc_address();
- HostPort hp;
- s = HostPortFromSockaddrReplaceWildcard(addr, &hp);
- if (s.ok()) {
- return hp.ToString();
+ vector<HostPort> hps;
+ s = master_->rpc_server()->GetBoundHostPorts(&hps);
+ if (PREDICT_FALSE(!s.ok())) {
+ LOG(WARNING) << "Unable to determine proper local hostname: " << s.ToString();
+ return string();
}
- LOG(WARNING) << "Unable to determine proper local hostname: " << s.ToString();
- return addr.ToString();
+ return hps[0].ToString();
}
Status MasterPathHandlers::GetLeaderMasterHttpAddr(string* leader_http_addr) const {
diff --git a/src/kudu/master/sys_catalog.cc b/src/kudu/master/sys_catalog.cc
index f6946e4..72ae70f 100644
--- a/src/kudu/master/sys_catalog.cc
+++ b/src/kudu/master/sys_catalog.cc
@@ -63,6 +63,7 @@
#include "kudu/master/master_options.h"
#include "kudu/rpc/result_tracker.h"
#include "kudu/security/token.pb.h"
+#include "kudu/server/rpc_server.h"
#include "kudu/tablet/metadata.pb.h"
#include "kudu/tablet/ops/op.h"
#include "kudu/tablet/ops/write_op.h"
@@ -80,7 +81,6 @@
#include "kudu/util/metrics.h"
#include "kudu/util/monotime.h"
#include "kudu/util/net/net_util.h"
-#include "kudu/util/net/sockaddr.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/slice.h"
@@ -1123,10 +1123,10 @@ Status SysCatalogTable::VisitTablets(TabletVisitor* visitor) {
void SysCatalogTable::InitLocalRaftPeerPB() {
local_peer_pb_.set_permanent_uuid(master_->fs_manager()->uuid());
- Sockaddr addr = master_->first_rpc_address();
- HostPort hp;
- CHECK_OK(HostPortFromSockaddrReplaceWildcard(addr, &hp));
- *local_peer_pb_.mutable_last_known_addr() = HostPortToPB(hp);
+ vector<HostPort> hps;
+ CHECK_OK(master_->rpc_server()->GetBoundHostPorts(&hps));
+ CHECK(!hps.empty());
+ *local_peer_pb_.mutable_last_known_addr() = HostPortToPB(hps[0]);
}
} // namespace master
diff --git a/src/kudu/mini-cluster/external_mini_cluster.cc b/src/kudu/mini-cluster/external_mini_cluster.cc
index 63baa74..d20355d 100644
--- a/src/kudu/mini-cluster/external_mini_cluster.cc
+++ b/src/kudu/mini-cluster/external_mini_cluster.cc
@@ -102,6 +102,7 @@ using strings::Substitute;
typedef ListTabletsResponsePB::StatusAndSchemaPB StatusAndSchemaPB;
DECLARE_string(block_manager);
+DECLARE_string(dns_addr_resolution_override);
DEFINE_bool(perf_record, false,
"Whether to run \"perf record --call-graph fp\" on each daemon in the cluster");
@@ -171,6 +172,27 @@ Status ExternalMiniCluster::HandleOptions() {
opts_.block_manager_type = FLAGS_block_manager;
}
+ vector<string> host_mappings;
+ if (!opts_.tserver_alias_prefix.empty()) {
+ for (int i = 0; i < opts_.num_tablet_servers; i++) {
+ host_mappings.emplace_back(Substitute("$0.$1=$2", opts_.tserver_alias_prefix, i,
+ GetBindIpForTabletServer(i)));
+ }
+ }
+ if (!opts_.master_alias_prefix.empty()) {
+ for (int i = 0; i < opts_.num_masters; i++) {
+ host_mappings.emplace_back(Substitute("$0.$1=$2", opts_.master_alias_prefix, i,
+ GetBindIpForMaster(i)));
+ }
+ }
+ if (!host_mappings.empty()) {
+ dns_overrides_ = JoinStrings(host_mappings, ",");
+ opts_.extra_master_flags.emplace_back(
+ Substitute("--dns_addr_resolution_override=$0", dns_overrides_));
+ opts_.extra_tserver_flags.emplace_back(
+ Substitute("--dns_addr_resolution_override=$0", dns_overrides_));
+ }
+
return Status::OK();
}
@@ -222,6 +244,8 @@ Status ExternalMiniCluster::Start() {
CHECK(tablet_servers_.empty()) << "Tablet servers are not empty (size: "
<< tablet_servers_.size() << "). Maybe you meant Restart()?";
RETURN_NOT_OK(HandleOptions());
+ gflags::FlagSaver saver;
+ FLAGS_dns_addr_resolution_override = dns_overrides_;
RETURN_NOT_OK_PREPEND(
rpc::MessengerBuilder("minicluster-messenger")
@@ -499,7 +523,10 @@ Status ExternalMiniCluster::StartMasters() {
"failed to reserve master socket address");
Sockaddr addr;
RETURN_NOT_OK(reserved_socket->GetSocketAddress(&addr));
- master_rpc_addrs.emplace_back(addr.host(), addr.port());
+ master_rpc_addrs.emplace_back(
+ opts_.master_alias_prefix.empty() ?
+ addr.host() : Substitute("$0.$1", opts_.master_alias_prefix, i),
+ addr.port());
reserved_sockets.emplace_back(std::move(reserved_socket));
}
}
@@ -552,6 +579,11 @@ Status ExternalMiniCluster::AddTabletServer() {
auto flags = SubstituteInFlags(opts_.extra_tserver_flags, idx);
copy(flags.begin(), flags.end(), std::back_inserter(extra_flags));
opts.extra_flags = extra_flags;
+ if (!opts_.tserver_alias_prefix.empty()) {
+ opts.extra_flags.emplace_back(
+ Substitute("--host_for_tests=$0.$1",
+ opts_.tserver_alias_prefix, tablet_servers_.size()));
+ }
opts.start_process_timeout = opts_.start_process_timeout;
opts.rpc_bind_address = HostPort(bind_host, 0);
opts.logtostderr = opts_.logtostderr;
@@ -617,6 +649,10 @@ Status ExternalMiniCluster::CreateMaster(const vector<HostPort>& master_rpc_addr
JoinPathSegments(cluster_root(),
"ranger-client")));
}
+ if (!opts_.master_alias_prefix.empty()) {
+ flags.emplace_back(Substitute("--host_for_tests=$0.$1",
+ opts_.master_alias_prefix, idx));
+ }
// Add custom master flags.
copy(opts_.extra_master_flags.begin(), opts_.extra_master_flags.end(),
std::back_inserter(flags));
diff --git a/src/kudu/mini-cluster/external_mini_cluster.h b/src/kudu/mini-cluster/external_mini_cluster.h
index 79d5b04..0a2e830 100644
--- a/src/kudu/mini-cluster/external_mini_cluster.h
+++ b/src/kudu/mini-cluster/external_mini_cluster.h
@@ -250,6 +250,9 @@ struct ExternalMiniClusterOptions {
// Default: BuiltinNtpConfigMode::ALL_SERVERS
BuiltinNtpConfigMode ntp_config_mode;
#endif // #if !defined(NO_CHRONY) ...
+
+ std::string master_alias_prefix;
+ std::string tserver_alias_prefix;
};
// A mini-cluster made up of subprocesses running each of the daemons
@@ -498,6 +501,10 @@ class ExternalMiniCluster : public MiniCluster {
// This helps keep the state of the actual cluster in sync with the state in ExternalMiniCluster.
Status RemoveMaster(const HostPort& hp);
+ const std::string& dns_overrides() const {
+ return dns_overrides_;
+ }
+
private:
Status StartMasters();
@@ -531,6 +538,8 @@ class ExternalMiniCluster : public MiniCluster {
std::shared_ptr<rpc::Messenger> messenger_;
+ std::string dns_overrides_;
+
DISALLOW_COPY_AND_ASSIGN(ExternalMiniCluster);
};
diff --git a/src/kudu/server/rpc_server.cc b/src/kudu/server/rpc_server.cc
index 7e476fe..c2842f2 100644
--- a/src/kudu/server/rpc_server.cc
+++ b/src/kudu/server/rpc_server.cc
@@ -247,6 +247,12 @@ Status RpcServer::GetBoundAddresses(vector<Sockaddr>* addresses) const {
return Status::OK();
}
+Status RpcServer::GetBoundHostPorts(vector<HostPort>* hostports) const {
+ vector<Sockaddr> addrs;
+ RETURN_NOT_OK_PREPEND(GetBoundAddresses(&addrs), "could not get bound RPC addresses");
+ return HostPortsFromAddrs(addrs, hostports);
+}
+
Status RpcServer::GetAdvertisedAddresses(vector<Sockaddr>* addresses) const {
if (server_state_ != BOUND &&
server_state_ != STARTED) {
@@ -259,6 +265,12 @@ Status RpcServer::GetAdvertisedAddresses(vector<Sockaddr>* addresses) const {
return Status::OK();
}
+Status RpcServer::GetAdvertisedHostPorts(vector<HostPort>* hostports) const {
+ vector<Sockaddr> addrs;
+ RETURN_NOT_OK_PREPEND(GetAdvertisedAddresses(&addrs), "could not get bound RPC addresses");
+ return HostPortsFromAddrs(addrs, hostports);
+}
+
const rpc::ServicePool* RpcServer::service_pool(const string& service_name) const {
return down_cast<rpc::ServicePool*>(messenger_->rpc_service(service_name).get());
}
diff --git a/src/kudu/server/rpc_server.h b/src/kudu/server/rpc_server.h
index 314b542..c23e97e 100644
--- a/src/kudu/server/rpc_server.h
+++ b/src/kudu/server/rpc_server.h
@@ -34,6 +34,7 @@
template <class T> class scoped_refptr;
namespace kudu {
+class HostPort;
namespace rpc {
class AcceptorPool;
@@ -86,10 +87,12 @@ class RpcServer {
// Return the addresses that this server has successfully
// bound to. Requires that the server has been Start()ed.
Status GetBoundAddresses(std::vector<Sockaddr>* addresses) const WARN_UNUSED_RESULT;
+ Status GetBoundHostPorts(std::vector<HostPort>* hostports) const WARN_UNUSED_RESULT;
// Return the addresses that this server is advertising externally
// to the world. Requires that the server has been Start()ed.
Status GetAdvertisedAddresses(std::vector<Sockaddr>* addresses) const WARN_UNUSED_RESULT;
+ Status GetAdvertisedHostPorts(std::vector<HostPort>* hostports) const WARN_UNUSED_RESULT;
const rpc::ServicePool* service_pool(const std::string& service_name) const;
diff --git a/src/kudu/server/server_base.cc b/src/kudu/server/server_base.cc
index e255076..43dc69e 100644
--- a/src/kudu/server/server_base.cc
+++ b/src/kudu/server/server_base.cc
@@ -693,26 +693,18 @@ Status ServerBase::GetStatusPB(ServerStatusPB* status) const {
// RPC ports
{
- vector<Sockaddr> addrs;
- RETURN_NOT_OK_PREPEND(rpc_server_->GetBoundAddresses(&addrs),
- "could not get bound RPC addresses");
- for (const Sockaddr& addr : addrs) {
- HostPort hp;
- RETURN_NOT_OK_PREPEND(HostPortFromSockaddrReplaceWildcard(addr, &hp),
- "could not get RPC hostport");
+ vector<HostPort> hps;
+ RETURN_NOT_OK(rpc_server_->GetBoundHostPorts(&hps));
+ for (const auto& hp : hps) {
*status->add_bound_rpc_addresses() = HostPortToPB(hp);
}
}
// HTTP ports
if (web_server_) {
- vector<Sockaddr> addrs;
- RETURN_NOT_OK_PREPEND(web_server_->GetBoundAddresses(&addrs),
- "could not get bound web addresses");
- for (const Sockaddr& addr : addrs) {
- HostPort hp;
- RETURN_NOT_OK_PREPEND(HostPortFromSockaddrReplaceWildcard(addr, &hp),
- "could not get web hostport");
+ vector<HostPort> hps;
+ RETURN_NOT_OK(web_server_->GetBoundHostPorts(&hps));
+ for (const auto& hp : hps) {
*status->add_bound_http_addresses() = HostPortToPB(hp);
}
}
diff --git a/src/kudu/server/webserver.cc b/src/kudu/server/webserver.cc
index dcd9082..518d08a 100644
--- a/src/kudu/server/webserver.cc
+++ b/src/kudu/server/webserver.cc
@@ -421,6 +421,12 @@ Status Webserver::GetBoundAddresses(std::vector<Sockaddr>* addrs) const {
return Status::OK();
}
+Status Webserver::GetBoundHostPorts(std::vector<HostPort>* hostports) const {
+ vector<Sockaddr> addrs;
+ RETURN_NOT_OK_PREPEND(GetBoundAddresses(&addrs), "could not get bound webserver addresses");
+ return HostPortsFromAddrs(addrs, hostports);
+}
+
Status Webserver::GetAdvertisedAddresses(vector<Sockaddr>* addresses) const {
if (!context_) {
return Status::ServiceUnavailable("Not started");
@@ -432,6 +438,12 @@ Status Webserver::GetAdvertisedAddresses(vector<Sockaddr>* addresses) const {
return Status::OK();
}
+Status Webserver::GetAdvertisedHostPorts(vector<HostPort>* hostports) const {
+ vector<Sockaddr> addrs;
+ RETURN_NOT_OK_PREPEND(GetAdvertisedAddresses(&addrs), "could not get bound webserver addresses");
+ return HostPortsFromAddrs(addrs, hostports);
+}
+
int Webserver::LogMessageCallbackStatic(const struct sq_connection* /*connection*/,
const char* message) {
if (message != nullptr) {
diff --git a/src/kudu/server/webserver.h b/src/kudu/server/webserver.h
index 42ca903..8eddfcd 100644
--- a/src/kudu/server/webserver.h
+++ b/src/kudu/server/webserver.h
@@ -35,6 +35,7 @@
namespace kudu {
class EasyJson;
+class HostPort;
// Wrapper class for the Mongoose web server library. Clients may register callback
// methods which produce output for a given URL path
@@ -56,10 +57,12 @@ class Webserver : public WebCallbackRegistry {
// Return the addresses that this server has successfully
// bound to. Requires that the server has been Start()ed.
Status GetBoundAddresses(std::vector<Sockaddr>* addrs) const WARN_UNUSED_RESULT;
+ Status GetBoundHostPorts(std::vector<HostPort>* hostports) const WARN_UNUSED_RESULT;
// Return the addresses that this server is advertising externally
// to the world. Requires that the server has been Start()ed.
Status GetAdvertisedAddresses(std::vector<Sockaddr>* addresses) const WARN_UNUSED_RESULT;
+ Status GetAdvertisedHostPorts(std::vector<HostPort>* hostports) const WARN_UNUSED_RESULT;
// Register a route 'path' to be rendered via template.
// The appropriate template to use is determined by 'path'.
diff --git a/src/kudu/tserver/heartbeater.cc b/src/kudu/tserver/heartbeater.cc
index 713a10f..7126cc5 100644
--- a/src/kudu/tserver/heartbeater.cc
+++ b/src/kudu/tserver/heartbeater.cc
@@ -35,6 +35,7 @@
#include <google/protobuf/stubs/common.h>
#include <google/protobuf/stubs/port.h>
+#include "kudu/common/common.pb.h"
#include "kudu/common/wire_protocol.h"
#include "kudu/common/wire_protocol.pb.h"
#include "kudu/consensus/replica_management.pb.h"
@@ -335,10 +336,16 @@ void Heartbeater::Thread::SetupCommonField(master::TSToMasterCommonPB* common) {
Status Heartbeater::Thread::SetupRegistration(ServerRegistrationPB* reg) {
reg->Clear();
+ vector<HostPort> hps;
+ RETURN_NOT_OK(server_->rpc_server()->GetAdvertisedHostPorts(&hps));
+ for (const auto& hp : hps) {
+ auto* pb = reg->add_rpc_addresses();
+ pb->set_host(hp.host());
+ pb->set_port(hp.port());
+ }
+ // Now fetch any UNIX domain sockets.
vector<Sockaddr> addrs;
RETURN_NOT_OK(CHECK_NOTNULL(server_->rpc_server())->GetAdvertisedAddresses(&addrs));
- RETURN_NOT_OK_PREPEND(AddHostPortPBs(addrs, reg->mutable_rpc_addresses()),
- "Failed to add RPC addresses to registration");
auto unix_socket_it = std::find_if(addrs.begin(), addrs.end(),
[](const Sockaddr& addr) {
return addr.is_unix();
diff --git a/src/kudu/tserver/ts_tablet_manager.cc b/src/kudu/tserver/ts_tablet_manager.cc
index 5554365..c3b745d 100644
--- a/src/kudu/tserver/ts_tablet_manager.cc
+++ b/src/kudu/tserver/ts_tablet_manager.cc
@@ -51,6 +51,7 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/master/master.pb.h"
#include "kudu/rpc/result_tracker.h"
+#include "kudu/server/rpc_server.h"
#include "kudu/tablet/metadata.pb.h"
#include "kudu/tablet/tablet.h"
#include "kudu/tablet/tablet_bootstrap.h"
@@ -68,7 +69,6 @@
#include "kudu/util/logging.h"
#include "kudu/util/monotime.h"
#include "kudu/util/net/net_util.h"
-#include "kudu/util/net/sockaddr.h"
#include "kudu/util/scoped_cleanup.h"
#include "kudu/util/stopwatch.h"
#include "kudu/util/threadpool.h"
@@ -1498,10 +1498,10 @@ TabletNumByDimensionMap TSTabletManager::GetNumLiveTabletsByDimension() const {
void TSTabletManager::InitLocalRaftPeerPB() {
DCHECK_EQ(state(), MANAGER_INITIALIZING);
local_peer_pb_.set_permanent_uuid(fs_manager_->uuid());
- Sockaddr addr = server_->first_rpc_address();
- HostPort hp;
- CHECK_OK(HostPortFromSockaddrReplaceWildcard(addr, &hp));
- *local_peer_pb_.mutable_last_known_addr() = HostPortToPB(hp);
+ vector<HostPort> hps;
+ CHECK_OK(server_->rpc_server()->GetBoundHostPorts(&hps));
+ DCHECK(!hps.empty());
+ *local_peer_pb_.mutable_last_known_addr() = HostPortToPB(hps[0]);
}
void TSTabletManager::TxnStalenessTrackerTask() {
diff --git a/src/kudu/util/net/net_util-test.cc b/src/kudu/util/net/net_util-test.cc
index c409b2b..89094bb 100644
--- a/src/kudu/util/net/net_util-test.cc
+++ b/src/kudu/util/net/net_util-test.cc
@@ -212,18 +212,19 @@ TEST_F(NetUtilTest, TestWithinNetwork) {
TEST_F(NetUtilTest, TestReverseLookup) {
string host;
Sockaddr addr;
- HostPort hp;
+ vector<HostPort> hps;
ASSERT_OK(addr.ParseString("0.0.0.0:12345", 0));
EXPECT_EQ(12345, addr.port());
- ASSERT_OK(HostPortFromSockaddrReplaceWildcard(addr, &hp));
- EXPECT_NE("0.0.0.0", hp.host());
- EXPECT_NE("", hp.host());
- EXPECT_EQ(12345, hp.port());
+ ASSERT_OK(HostPortsFromAddrs({ addr }, &hps));
+ EXPECT_NE("0.0.0.0", hps[0].host());
+ EXPECT_NE("", hps[0].host());
+ EXPECT_EQ(12345, hps[0].port());
+ hps.clear();
ASSERT_OK(addr.ParseString("127.0.0.1:12345", 0));
- ASSERT_OK(HostPortFromSockaddrReplaceWildcard(addr, &hp));
- EXPECT_EQ("127.0.0.1", hp.host());
- EXPECT_EQ(12345, hp.port());
+ ASSERT_OK(HostPortsFromAddrs({ addr }, &hps));
+ EXPECT_EQ("127.0.0.1", hps[0].host());
+ EXPECT_EQ(12345, hps[0].port());
}
TEST_F(NetUtilTest, TestLsof) {
diff --git a/src/kudu/util/net/net_util.cc b/src/kudu/util/net/net_util.cc
index aba828e..f587041 100644
--- a/src/kudu/util/net/net_util.cc
+++ b/src/kudu/util/net/net_util.cc
@@ -78,6 +78,9 @@ DEFINE_string(dns_addr_resolution_override, "",
"is expected to be a socket address with no port.");
TAG_FLAG(dns_addr_resolution_override, hidden);
+DEFINE_string(host_for_tests, "", "Host to use when resolving a given server's locally bound or "
+ "advertised addresses.");
+
using std::function;
using std::string;
using std::unordered_set;
@@ -124,6 +127,23 @@ Status GetAddrInfo(const string& hostname,
return Status::NetworkError(err_msg, gai_strerror(rc));
}
+// Converts the given Sockaddr into a HostPort, substituting the FQDN
+// in the case that the provided address is the wildcard.
+//
+// In the case of other addresses, the returned HostPort will contain just the
+// stringified form of the IP.
+Status HostPortFromSockaddrReplaceWildcard(const Sockaddr& addr, HostPort* hp) {
+ string host;
+ if (!FLAGS_host_for_tests.empty() || addr.IsWildcard()) {
+ RETURN_NOT_OK(GetFQDN(&host));
+ } else {
+ host = addr.host();
+ }
+ hp->set_host(host);
+ hp->set_port(addr.port());
+ return Status::OK();
+}
+
} // anonymous namespace
HostPort::HostPort()
@@ -407,6 +427,10 @@ Status ParseAddressList(const std::string& addr_list,
Status GetHostname(string* hostname) {
TRACE_EVENT0("net", "GetHostname");
+ if (!FLAGS_host_for_tests.empty()) {
+ *hostname = FLAGS_host_for_tests;
+ return Status::OK();
+ }
char name[HOST_NAME_MAX];
if (gethostname(name, HOST_NAME_MAX) != 0) {
const int err = errno;
@@ -460,6 +484,9 @@ Status GetFQDN(string* hostname) {
TRACE_EVENT0("net", "GetFQDN");
// Start with the non-qualified hostname
RETURN_NOT_OK(GetHostname(hostname));
+ if (!FLAGS_host_for_tests.empty()) {
+ return Status::OK();
+ }
struct addrinfo hints;
memset(&hints, 0, sizeof(hints));
@@ -493,15 +520,15 @@ Status SockaddrFromHostPort(const HostPort& host_port, Sockaddr* addr) {
return Status::OK();
}
-Status HostPortFromSockaddrReplaceWildcard(const Sockaddr& addr, HostPort* hp) {
- string host;
- if (addr.IsWildcard()) {
- RETURN_NOT_OK(GetFQDN(&host));
- } else {
- host = addr.host();
+Status HostPortsFromAddrs(const vector<Sockaddr>& addrs, vector<HostPort>* hps) {
+ DCHECK(!addrs.empty());
+ for (const auto& addr : addrs) {
+ if (!addr.is_ip()) continue;
+ HostPort hp;
+ RETURN_NOT_OK_PREPEND(HostPortFromSockaddrReplaceWildcard(addr, &hp),
+ "could not get RPC hostport");
+ hps->emplace_back(std::move(hp));
}
- hp->set_host(host);
- hp->set_port(addr.port());
return Status::OK();
}
diff --git a/src/kudu/util/net/net_util.h b/src/kudu/util/net/net_util.h
index d29ca97..ea77bc6 100644
--- a/src/kudu/util/net/net_util.h
+++ b/src/kudu/util/net/net_util.h
@@ -190,12 +190,15 @@ Status GetFQDN(std::string* hostname);
// list and logs a message in verbose mode.
Status SockaddrFromHostPort(const HostPort& host_port, Sockaddr* addr);
-// Converts the given Sockaddr into a HostPort, substituting the FQDN
-// in the case that the provided address is the wildcard.
+// Converts the given list of Sockaddrs into a list of HostPorts that can be
+// accessed from other machines, i.e. wildcards are replaced with the FQDN, the
+// --host_for_tests gflag is honored with the expectation that 'addrs' is the
+// list of locally bound or advertised addresses.
//
// In the case of other addresses, the returned HostPort will contain just the
// stringified form of the IP.
-Status HostPortFromSockaddrReplaceWildcard(const Sockaddr& addr, HostPort* hp);
+Status HostPortsFromAddrs(const std::vector<Sockaddr>& addrs,
+ std::vector<HostPort>* hps);
// Try to run 'lsof' to determine which process is preventing binding to
// the given 'addr'. If pids can be determined, outputs full 'ps' and 'pstree'