You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2020/05/06 16:30:14 UTC
[impala] branch master updated: IMPALA-9692 (part 1): Refactor
TBackendDescriptor to protobuf
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new e46bbf2 IMPALA-9692 (part 1): Refactor TBackendDescriptor to protobuf
e46bbf2 is described below
commit e46bbf29d43f1d3b118933d06c63ceb135beaf9c
Author: Thomas Tauber-Marshall <tm...@cloudera.com>
AuthorDate: Mon Mar 23 15:14:49 2020 -0700
IMPALA-9692 (part 1): Refactor TBackendDescriptor to protobuf
The new admission control service will be written in protobuf, so
there are various admission control related structures currently
stored in Thrift that it would be convenient to convert to protobuf,
to minimize the amount of converting back and forth that needs to be
done.
This patch converts TBackendDescriptor to protobuf. It isn't used
directly in any rpcs - we serialize it ourselves to send to the
statestore as a string, so no rpc definitions are affected.
This patch is just a refactor and doesn't contain any functional
changes.
Testing:
- Passed a full run of existing tests.
Change-Id: Ie7d1e373d9c87887144517ff6a4c2d5996aa88b8
Reviewed-on: http://gerrit.cloudera.org:8080/15822
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/common/global-types.h | 5 +
be/src/runtime/coordinator-backend-state.cc | 40 +++---
be/src/runtime/coordinator-backend-state.h | 8 +-
be/src/runtime/coordinator.cc | 36 +++--
be/src/runtime/coordinator.h | 12 +-
be/src/runtime/exec-env.cc | 12 +-
be/src/runtime/exec-env.h | 5 +-
be/src/scheduling/admission-controller-test.cc | 20 +--
be/src/scheduling/admission-controller.cc | 20 +--
be/src/scheduling/cluster-membership-mgr-test.cc | 22 ++--
be/src/scheduling/cluster-membership-mgr.cc | 154 +++++++++++-----------
be/src/scheduling/cluster-membership-mgr.h | 33 ++---
be/src/scheduling/cluster-membership-test-util.cc | 40 +++---
be/src/scheduling/cluster-membership-test-util.h | 12 +-
be/src/scheduling/executor-blacklist.cc | 34 ++---
be/src/scheduling/executor-blacklist.h | 20 ++-
be/src/scheduling/executor-group.cc | 69 +++++-----
be/src/scheduling/executor-group.h | 16 +--
be/src/scheduling/query-schedule.cc | 2 +-
be/src/scheduling/query-schedule.h | 11 +-
be/src/scheduling/scheduler-test-util.cc | 25 ++--
be/src/scheduling/scheduler.cc | 93 ++++++-------
be/src/scheduling/scheduler.h | 28 ++--
be/src/service/cancellation-work.h | 10 +-
be/src/service/impala-http-handler.cc | 41 +++---
be/src/service/impala-server.cc | 78 ++++++-----
be/src/service/impala-server.h | 16 +--
be/src/statestore/CMakeLists.txt | 4 +
be/src/util/container-util.h | 44 ++++++-
be/src/util/debug-util.cc | 7 +
be/src/util/debug-util.h | 2 +
be/src/util/network-util.cc | 19 +++
be/src/util/network-util.h | 18 ++-
be/src/util/uid-util.h | 8 ++
be/src/util/unique-id-hash.h | 15 +++
common/protobuf/CMakeLists.txt | 2 +-
common/protobuf/statestore_service.proto | 79 +++++++++++
common/thrift/StatestoreService.thrift | 56 --------
38 files changed, 621 insertions(+), 495 deletions(-)
diff --git a/be/src/common/global-types.h b/be/src/common/global-types.h
index a6a78c9..6546d76 100644
--- a/be/src/common/global-types.h
+++ b/be/src/common/global-types.h
@@ -26,6 +26,11 @@ class time_zone;
namespace impala {
+class UniqueIdPB;
+
+// Used to uniquely identify individual impalads.
+typedef UniqueIdPB BackendIdPB;
+
typedef cctz::time_zone Timezone;
// There is also a "proper" Timezone for UTC, but using nullptr makes some
diff --git a/be/src/runtime/coordinator-backend-state.cc b/be/src/runtime/coordinator-backend-state.cc
index 6be9c36..11060ad 100644
--- a/be/src/runtime/coordinator-backend-state.cc
+++ b/be/src/runtime/coordinator-backend-state.cc
@@ -73,15 +73,15 @@ Coordinator::BackendState::BackendState(const QuerySchedule& schedule,
state_idx_(state_idx),
filter_mode_(filter_mode),
backend_exec_params_(&exec_params),
- host_(backend_exec_params_->be_desc.address),
- krpc_host_(backend_exec_params_->be_desc.krpc_address),
+ host_(backend_exec_params_->be_desc.address()),
+ krpc_host_(backend_exec_params_->be_desc.krpc_address()),
query_ctx_(query_ctx),
query_id_(schedule.query_id()),
num_remaining_instances_(backend_exec_params_->instance_params.size()) {}
void Coordinator::BackendState::Init(const vector<FragmentStats*>& fragment_stats,
RuntimeProfile* host_profile_parent, ObjectPool* obj_pool) {
- host_profile_ = RuntimeProfile::Create(obj_pool, TNetworkAddressToString(host_));
+ host_profile_ = RuntimeProfile::Create(obj_pool, NetworkAddressPBToString(host_));
host_profile_parent->AddChild(host_profile_);
RuntimeProfile::Counter* admission_slots =
ADD_COUNTER(host_profile_, "AdmissionSlots", TUnit::UNIT);
@@ -92,7 +92,8 @@ void Coordinator::BackendState::Init(const vector<FragmentStats*>& fragment_stat
int prev_fragment_idx = -1;
for (const FInstanceExecParams* instance_params:
backend_exec_params_->instance_params) {
- DCHECK(host_ == instance_params->host); // all hosts must be the same
+ DCHECK_EQ(
+ FromNetworkAddressPB(host_), instance_params->host); // all hosts must be the same
int fragment_idx = instance_params->fragment().idx;
DCHECK_LT(fragment_idx, fragment_stats.size());
if (prev_fragment_idx != -1 && fragment_idx != prev_fragment_idx) {
@@ -238,8 +239,8 @@ void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options,
}
std::unique_ptr<ControlServiceProxy> proxy;
- Status get_proxy_status =
- ControlService::GetProxy(krpc_host_, host_.hostname, &proxy);
+ Status get_proxy_status = ControlService::GetProxy(
+ FromNetworkAddressPB(krpc_host_), host_.hostname(), &proxy);
if (!get_proxy_status.ok()) {
SetExecError(get_proxy_status, exec_status_barrier);
goto done;
@@ -303,8 +304,7 @@ void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options,
request.set_query_ctx_sidecar_idx(query_ctx_sidecar_idx);
VLOG_FILE << "making rpc: ExecQueryFInstances"
- << " host=" << TNetworkAddressToString(impalad_address())
- << " query_id=" << PrintId(query_id_);
+ << " host=" << impalad_address() << " query_id=" << PrintId(query_id_);
proxy->ExecQueryFInstancesAsync(request, &exec_response_, &exec_rpc_controller_,
std::bind(&Coordinator::BackendState::ExecCompleteCb, this, exec_status_barrier,
@@ -396,8 +396,7 @@ void Coordinator::BackendState::LogFirstInProgress(
for (Coordinator::BackendState* backend_state : backend_states) {
if (!backend_state->IsDone()) {
VLOG_QUERY << "query_id=" << PrintId(backend_state->query_id_)
- << ": first in-progress backend: "
- << TNetworkAddressToString(backend_state->impalad_address());
+ << ": first in-progress backend: " << backend_state->impalad_address();
break;
}
}
@@ -483,7 +482,7 @@ bool Coordinator::BackendState::ApplyExecStatusReport(
// Append the log messages from each update with the global state of the query
// execution
MergeErrorMaps(stateful_report.error_log(), &error_log_);
- VLOG_FILE << "host=" << TNetworkAddressToString(host_)
+ VLOG_FILE << "host=" << host_
<< " error log: " << PrintErrorMapToString(error_log_);
if (stateful_report.has_aux_error_info()) {
@@ -582,7 +581,8 @@ bool Coordinator::BackendState::Cancel() {
VLogForBackend("Sending CancelQueryFInstances rpc");
std::unique_ptr<ControlServiceProxy> proxy;
- Status get_proxy_status = ControlService::GetProxy(krpc_host_, host_.hostname, &proxy);
+ Status get_proxy_status = ControlService::GetProxy(
+ FromNetworkAddressPB(krpc_host_), host_.hostname(), &proxy);
if (!get_proxy_status.ok()) {
status_.MergeStatus(get_proxy_status);
VLogForBackend(Substitute("Could not get proxy: $0", get_proxy_status.msg().msg()));
@@ -624,13 +624,12 @@ void Coordinator::BackendState::PublishFilter(FilterState* state,
// If the backend is already done, it's not waiting for this filter, so we skip
// sending it in this case.
if (IsDone()) return;
- VLOG(2) << "PublishFilter filter_id=" << rpc_params.filter_id()
- << " backend=" << TNetworkAddressToString(host_);
+ VLOG(2) << "PublishFilter filter_id=" << rpc_params.filter_id() << " backend=" << host_;
Status status;
std::unique_ptr<DataStreamServiceProxy> proxy;
- Status get_proxy_status =
- DataStreamService::GetProxy(krpc_host_, host_.hostname, &proxy);
+ Status get_proxy_status = DataStreamService::GetProxy(
+ FromNetworkAddressPB(krpc_host_), host_.hostname(), &proxy);
if (!get_proxy_status.ok()) {
// Failing to send a filter is not a query-wide error - the remote fragment will
// continue regardless.
@@ -869,7 +868,7 @@ void Coordinator::BackendState::ToJson(Value* value, Document* document) {
value->AddMember("cpu_sys_s", resource_utilization.cpu_sys_ns / 1e9,
document->GetAllocator());
- string host = TNetworkAddressToString(impalad_address());
+ string host = NetworkAddressPBToString(impalad_address());
Value val(host.c_str(), document->GetAllocator());
value->AddMember("host", val, document->GetAllocator());
@@ -900,13 +899,14 @@ void Coordinator::BackendState::InstanceStatsToJson(Value* value, Document* docu
// impalad_address is not protected by lock_. The lifetime of the backend state is
// protected by Coordinator::lock_.
- Value val(TNetworkAddressToString(impalad_address()).c_str(), document->GetAllocator());
+ Value val(
+ NetworkAddressPBToString(impalad_address()).c_str(), document->GetAllocator());
value->AddMember("host", val, document->GetAllocator());
}
void Coordinator::BackendState::VLogForBackend(const string& msg) {
- VLOG_QUERY << "query_id=" << PrintId(query_id_)
- << " target backend=" << TNetworkAddressToString(krpc_host_) << ": " << msg;
+ VLOG_QUERY << "query_id=" << PrintId(query_id_) << " target backend=" << krpc_host_
+ << ": " << msg;
}
} // namespace impala
diff --git a/be/src/runtime/coordinator-backend-state.h b/be/src/runtime/coordinator-backend-state.h
index bc51b87..f9cab49 100644
--- a/be/src/runtime/coordinator-backend-state.h
+++ b/be/src/runtime/coordinator-backend-state.h
@@ -74,8 +74,8 @@ class Coordinator::BackendState {
/// The following are initialized in the constructor and always valid to call.
int state_idx() const { return state_idx_; }
const BackendExecParams* exec_params() const { return backend_exec_params_; }
- const TNetworkAddress& impalad_address() const { return host_; }
- const TNetworkAddress& krpc_impalad_address() const { return krpc_host_; }
+ const NetworkAddressPB& impalad_address() const { return host_; }
+ const NetworkAddressPB& krpc_impalad_address() const { return krpc_host_; }
/// Returns true if there are no fragment instances scheduled on this backend.
bool IsEmptyBackend() const { return backend_exec_params_->instance_params.empty(); }
/// Return true if execution at this backend is done.
@@ -309,9 +309,9 @@ class Coordinator::BackendState {
RuntimeProfile* host_profile_ = nullptr;
/// Thrift address of execution backend.
- TNetworkAddress host_;
+ NetworkAddressPB host_;
/// Krpc address of execution backend.
- TNetworkAddress krpc_host_;
+ NetworkAddressPB krpc_host_;
/// The query context of the Coordinator that owns this BackendState.
const TQueryCtx& query_ctx_;
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index e2d464c..935543a 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -496,10 +496,9 @@ Status Coordinator::StartBackendExec() {
// If Exec() rpc failed for a reason besides being aborted, blacklist the executor.
if (!backend_state->exec_rpc_status().ok()
&& !backend_state->exec_rpc_status().IsAborted()) {
- LOG(INFO) << "Blacklisting "
- << TNetworkAddressToString(backend_state->impalad_address())
+ LOG(INFO) << "Blacklisting " << backend_state->impalad_address()
<< " because an Exec() rpc to it failed.";
- const TBackendDescriptor& be_desc = backend_state->exec_params()->be_desc;
+ const BackendDescriptorPB& be_desc = backend_state->exec_params()->be_desc;
ExecEnv::GetInstance()->cluster_membership_mgr()->BlacklistExecutor(be_desc,
FromKuduStatus(backend_state->exec_rpc_status(), "Exec() rpc failed"));
}
@@ -535,13 +534,13 @@ Status Coordinator::FinishBackendStartup() {
Status backend_status = backend_state->GetStatus();
if (!backend_status.ok() && status.ok()) {
status = backend_status;
- error_hostname = backend_state->impalad_address().hostname;
+ error_hostname = backend_state->impalad_address().hostname();
}
if (backend_state->rpc_latency() > max_latency) {
// Find the backend that takes the most time to acknowledge to
// the ExecQueryFInstances() RPC.
max_latency = backend_state->rpc_latency();
- max_latency_host = TNetworkAddressToString(backend_state->impalad_address());
+ max_latency_host = NetworkAddressPBToString(backend_state->impalad_address());
}
latencies.Update(backend_state->rpc_latency());
// Mark backend complete if no fragment instances were assigned to it.
@@ -904,7 +903,7 @@ Status Coordinator::UpdateBackendExecStatus(const ReportExecStatusRequestPB& req
int pending_backends = backend_exec_complete_barrier_->pending();
if (pending_backends >= 1) {
VLOG_QUERY << "Backend completed:"
- << " host=" << TNetworkAddressToString(backend_state->impalad_address())
+ << " host=" << backend_state->impalad_address()
<< " remaining=" << pending_backends
<< " query_id=" << PrintId(query_id());
BackendState::LogFirstInProgress(backend_states_);
@@ -923,9 +922,9 @@ Status Coordinator::UpdateBackendExecStatus(const ReportExecStatusRequestPB& req
// Transition the status if we're not already in a terminal state. This won't block
// because either this transitions to an ERROR state or the query is already in
// a terminal state.
- discard_result(UpdateExecState(status,
- is_fragment_failure ? &failed_instance_id : nullptr,
- TNetworkAddressToString(backend_state->impalad_address())));
+ discard_result(
+ UpdateExecState(status, is_fragment_failure ? &failed_instance_id : nullptr,
+ NetworkAddressPBToString(backend_state->impalad_address())));
}
// We've applied all changes from the final status report - notify waiting threads.
discard_result(backend_exec_complete_barrier_->Notify());
@@ -966,8 +965,7 @@ void Coordinator::UpdateBlacklistWithAuxErrorInfo(
DCHECK(rpc_error_info.has_posix_error_code());
const NetworkAddressPB& dest_node = rpc_error_info.dest_node();
- auto dest_node_and_be_state =
- addr_to_backend_state_.find(FromNetworkAddressPB(dest_node));
+ auto dest_node_and_be_state = addr_to_backend_state_.find(dest_node);
// If the target address of the RPC is not known to the Coordinator, it cannot
// be blacklisted.
@@ -1008,7 +1006,7 @@ void Coordinator::UpdateBlacklistWithAuxErrorInfo(
}
}
-int64_t Coordinator::GetMaxBackendStateLagMs(TNetworkAddress* address) {
+int64_t Coordinator::GetMaxBackendStateLagMs(NetworkAddressPB* address) {
DCHECK(exec_rpcs_complete_.Load()) << "Exec() must be called first.";
DCHECK_GT(backend_states_.size(), 0);
int64_t current_time = BackendState::GenerateReportTimestamp();
@@ -1053,8 +1051,7 @@ void Coordinator::ComputeQuerySummary() {
for (BackendState* backend_state: backend_states_) {
ResourceUtilization utilization = backend_state->ComputeResourceUtilization();
total_utilization.Merge(utilization);
- string network_address = TNetworkAddressToString(
- backend_state->impalad_address());
+ string network_address = NetworkAddressPBToString(backend_state->impalad_address());
mem_info << network_address << "("
<< PrettyPrinter::Print(utilization.peak_per_host_mem_consumption,
TUnit::BYTES) << ") ";
@@ -1161,7 +1158,7 @@ void Coordinator::ReleaseBackendAdmissionControlResources(
DCHECK(admission_controller != nullptr);
vector<TNetworkAddress> host_addrs;
for (auto backend_state : backend_states) {
- host_addrs.push_back(backend_state->impalad_address());
+ host_addrs.push_back(FromNetworkAddressPB(backend_state->impalad_address()));
}
admission_controller->ReleaseQueryBackends(schedule_, host_addrs);
}
@@ -1175,11 +1172,12 @@ Coordinator::ResourceUtilization Coordinator::ComputeQueryResourceUtilization()
return query_resource_utilization;
}
-vector<TNetworkAddress> Coordinator::GetActiveBackends(
- const vector<TNetworkAddress>& candidates) {
+vector<NetworkAddressPB> Coordinator::GetActiveBackends(
+ const vector<NetworkAddressPB>& candidates) {
// Build set from vector so that runtime of this function is O(backend_states.size()).
- std::unordered_set<TNetworkAddress> candidate_set(candidates.begin(), candidates.end());
- vector<TNetworkAddress> result;
+ std::unordered_set<NetworkAddressPB> candidate_set(
+ candidates.begin(), candidates.end());
+ vector<NetworkAddressPB> result;
lock_guard<SpinLock> l(backend_states_init_lock_);
for (BackendState* backend_state : backend_states_) {
if (candidate_set.find(backend_state->impalad_address()) != candidate_set.end()
diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h
index de69e89..fc3a9f2 100644
--- a/be/src/runtime/coordinator.h
+++ b/be/src/runtime/coordinator.h
@@ -154,7 +154,7 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// for that backend. Only valid to call if Exec() has returned successfully. May return
/// 0, for example if all of the backends have already completed, in which case
/// 'address' will not be set.
- int64_t GetMaxBackendStateLagMs(TNetworkAddress* address);
+ int64_t GetMaxBackendStateLagMs(NetworkAddressPB* address);
/// Get cumulative profile aggregated over all fragments of the query.
/// This is a snapshot of the current state of execution and will change in
@@ -232,8 +232,8 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// Return the backends in 'candidates' that still have at least one fragment instance
/// executing on them. The returned backends may not be in the same order as the input.
- std::vector<TNetworkAddress> GetActiveBackends(
- const std::vector<TNetworkAddress>& candidates);
+ std::vector<NetworkAddressPB> GetActiveBackends(
+ const std::vector<NetworkAddressPB>& candidates);
private:
class BackendState;
@@ -257,12 +257,12 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// are non-nullptr and owned by obj_pool(). Populated by Exec()/InitBackendStates().
std::vector<BackendState*> backend_states_;
- /// A map from the TNetworkAddress of a backend to the BackendState running on the
- /// TNetworkAddress. All values are non-nullptr and owned by obj_pool(). The address
+ /// A map from the NetworkAddressPB of a backend to the BackendState running on the
+ /// NetworkAddressPB. All values are non-nullptr and owned by obj_pool(). The address
/// is the kRPC address (Coordinator::BackendState::krpc_impalad_address) of the
/// Backend. This map is distinct from QuerySchedule::per_backend_exec_params(),
/// which uses the Thrift address as the key rather than the kRPC address.
- boost::unordered_map<TNetworkAddress, BackendState*> addr_to_backend_state_;
+ boost::unordered_map<NetworkAddressPB, BackendState*> addr_to_backend_state_;
/// Protects the population of backend_states_ vector (not the BackendState objects).
/// Used when accessing backend_states_ if it's not guaranteed that
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index c6e7844..a25f1ca 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -195,10 +195,10 @@ void SendClusterMembershipToFrontend(
}
if (group) {
for (const auto& backend : group->GetAllExecutorDescriptors()) {
- if (backend.is_executor) {
+ if (backend.is_executor()) {
if (is_default_group) {
- update_req.hostnames.insert(backend.address.hostname);
- update_req.ip_addresses.insert(backend.ip_address);
+ update_req.hostnames.insert(backend.address().hostname());
+ update_req.ip_addresses.insert(backend.ip_address());
}
update_req.num_executors++;
}
@@ -253,7 +253,7 @@ ExecEnv::ExecEnv(int backend_port, int krpc_port,
rpc_metrics_(metrics_->GetOrCreateChildGroup("rpc")),
enable_webserver_(FLAGS_enable_webserver && webserver_port > 0),
configured_backend_address_(MakeNetworkAddress(FLAGS_hostname, backend_port)) {
- UUIDToTUniqueId(boost::uuids::random_generator()(), &backend_id_);
+ UUIDToUniqueIdPB(boost::uuids::random_generator()(), &backend_id_);
// Resolve hostname to IP address.
ABORT_IF_ERROR(HostnameToIpAddr(FLAGS_hostname, &ip_address_));
@@ -551,9 +551,9 @@ void ExecEnv::SetImpalaServer(ImpalaServer* server) {
});
cluster_membership_mgr_->RegisterUpdateCallbackFn(
[server](ClusterMembershipMgr::SnapshotPtr snapshot) {
- std::unordered_set<TBackendId> current_backend_set;
+ std::unordered_set<BackendIdPB> current_backend_set;
for (const auto& it : snapshot->current_backends) {
- current_backend_set.insert(it.second.backend_id);
+ current_backend_set.insert(it.second.backend_id());
}
server->CancelQueriesOnFailedBackends(current_backend_set);
});
diff --git a/be/src/runtime/exec-env.h b/be/src/runtime/exec-env.h
index c3133e7..0902ada 100644
--- a/be/src/runtime/exec-env.h
+++ b/be/src/runtime/exec-env.h
@@ -24,6 +24,7 @@
#include <boost/scoped_ptr.hpp>
// NOTE: try not to add more headers here: exec-env.h is included in many many files.
+#include "common/global-types.h"
#include "common/status.h"
#include "runtime/client-cache-types.h"
#include "util/hdfs-bulk-ops-defs.h" // For declaration of HdfsOpThreadPool
@@ -113,7 +114,7 @@ class ExecEnv {
/// StartServices() was successful.
TNetworkAddress GetThriftBackendAddress() const;
- const TBackendId& backend_id() const { return backend_id_; }
+ const BackendIdPB& backend_id() const { return backend_id_; }
KrpcDataStreamMgr* stream_mgr() { return stream_mgr_.get(); }
@@ -177,7 +178,7 @@ class ExecEnv {
private:
// Used to uniquely identify this impalad.
- TBackendId backend_id_;
+ BackendIdPB backend_id_;
boost::scoped_ptr<ObjectPool> obj_pool_;
boost::scoped_ptr<MetricGroup> metrics_;
diff --git a/be/src/scheduling/admission-controller-test.cc b/be/src/scheduling/admission-controller-test.cc
index cf36cb9..3c54061 100644
--- a/be/src/scheduling/admission-controller-test.cc
+++ b/be/src/scheduling/admission-controller-test.cc
@@ -129,14 +129,14 @@ class AdmissionControllerTest : public testing::Test {
BackendExecParams* backend_exec_params = pool_.Add(new BackendExecParams());
backend_exec_params->min_mem_reservation_bytes = min_mem_reservation_bytes;
backend_exec_params->slots_to_use = slots_to_use;
- backend_exec_params->be_desc.__set_admit_mem_limit(admit_mem_limit);
- backend_exec_params->be_desc.__set_admission_slots(slots_available);
- backend_exec_params->be_desc.__set_is_executor(true);
- backend_exec_params->be_desc.__set_address(host_addr);
+ backend_exec_params->be_desc.set_admit_mem_limit(admit_mem_limit);
+ backend_exec_params->be_desc.set_admission_slots(slots_available);
+ backend_exec_params->be_desc.set_is_executor(true);
+ *backend_exec_params->be_desc.mutable_address() = FromTNetworkAddress(host_addr);
if (i == 0) {
// Add first element as the coordinator.
backend_exec_params->is_coord_backend = true;
- backend_exec_params->be_desc.__set_is_executor(!is_dedicated_coord);
+ backend_exec_params->be_desc.set_is_executor(!is_dedicated_coord);
}
per_backend_exec_params->emplace(host_addr, *backend_exec_params);
}
@@ -925,8 +925,8 @@ TEST_F(AdmissionControllerTest, DedicatedCoordAdmissionChecks) {
BackendExecParams* coord_exec_params = pool_.Add(new BackendExecParams());
coord_exec_params->is_coord_backend = true;
coord_exec_params->thread_reservation = 1;
- coord_exec_params->be_desc.__set_admit_mem_limit(512 * MEGABYTE);
- coord_exec_params->be_desc.__set_is_executor(false);
+ coord_exec_params->be_desc.set_admit_mem_limit(512 * MEGABYTE);
+ coord_exec_params->be_desc.set_is_executor(false);
const string coord_host_name = Substitute("host$0", 1);
TNetworkAddress coord_addr = MakeNetworkAddress(coord_host_name, 25000);
const string coord_host = TNetworkAddressToString(coord_addr);
@@ -934,8 +934,8 @@ TEST_F(AdmissionControllerTest, DedicatedCoordAdmissionChecks) {
// Add executor backend.
BackendExecParams* backend_exec_params = pool_.Add(new BackendExecParams());
backend_exec_params->thread_reservation = 1;
- backend_exec_params->be_desc.__set_admit_mem_limit(GIGABYTE);
- backend_exec_params->be_desc.__set_is_executor(true);
+ backend_exec_params->be_desc.set_admit_mem_limit(GIGABYTE);
+ backend_exec_params->be_desc.set_is_executor(true);
const string exec_host_name = Substitute("host$0", 2);
TNetworkAddress exec_addr = MakeNetworkAddress(exec_host_name, 25000);
const string exec_host = TNetworkAddressToString(exec_addr);
@@ -979,7 +979,7 @@ TEST_F(AdmissionControllerTest, DedicatedCoordAdmissionChecks) {
// Test 2: coord's admit_mem_limit < executor's admit_mem_limit. Query rejected because
// coord's admit_mem_limit is less than mem_to_admit on the coord.
// Re-using previous QuerySchedule object.
- coord_exec_params->be_desc.__set_admit_mem_limit(100 * MEGABYTE);
+ coord_exec_params->be_desc.set_admit_mem_limit(100 * MEGABYTE);
(*per_backend_exec_params)[coord_addr] = *coord_exec_params;
query_schedule->set_per_backend_exec_params(*per_backend_exec_params);
ASSERT_TRUE(admission_controller->RejectForSchedule(
diff --git a/be/src/scheduling/admission-controller.cc b/be/src/scheduling/admission-controller.cc
index 96b6496..4844531 100644
--- a/be/src/scheduling/admission-controller.cc
+++ b/be/src/scheduling/admission-controller.cc
@@ -534,7 +534,7 @@ bool AdmissionController::HasAvailableMemResources(const QuerySchedule& schedule
for (const auto& entry : schedule.per_backend_exec_params()) {
const TNetworkAddress& host = entry.first;
const string host_id = TNetworkAddressToString(host);
- int64_t admit_mem_limit = entry.second.be_desc.admit_mem_limit;
+ int64_t admit_mem_limit = entry.second.be_desc.admit_mem_limit();
const HostStats& host_stats = host_stats_[host_id];
int64_t mem_reserved = host_stats.mem_reserved;
int64_t mem_admitted = host_stats.mem_admitted;
@@ -569,7 +569,7 @@ bool AdmissionController::HasAvailableSlots(const QuerySchedule& schedule,
for (const auto& entry : schedule.per_backend_exec_params()) {
const TNetworkAddress& host = entry.first;
const string host_id = TNetworkAddressToString(host);
- int64_t admission_slots = entry.second.be_desc.admission_slots;
+ int64_t admission_slots = entry.second.be_desc.admission_slots();
int64_t slots_in_use = host_stats_[host_id].slots_in_use;
VLOG_ROW << "Checking available slot on host=" << host_id
<< " slots_in_use=" << slots_in_use
@@ -694,9 +694,9 @@ bool AdmissionController::RejectForSchedule(const QuerySchedule& schedule,
for (const auto& e : schedule.per_backend_exec_params()) {
const BackendExecParams& bp = e.second;
// TODO(IMPALA-8757): Extend slot based admission to default executor group
- if (!default_group && bp.slots_to_use > bp.be_desc.admission_slots) {
+ if (!default_group && bp.slots_to_use > bp.be_desc.admission_slots()) {
*rejection_reason = Substitute(REASON_NOT_ENOUGH_SLOTS_ON_BACKEND, bp.slots_to_use,
- TNetworkAddressToString(bp.be_desc.address), bp.be_desc.admission_slots);
+ NetworkAddressPBToString(bp.be_desc.address()), bp.be_desc.admission_slots());
return true;
}
@@ -710,10 +710,10 @@ bool AdmissionController::RejectForSchedule(const QuerySchedule& schedule,
}
if (bp.is_coord_backend) {
coord_admit_mem_limit.first = &e.first;
- coord_admit_mem_limit.second = bp.be_desc.admit_mem_limit;
- } else if (bp.be_desc.admit_mem_limit < min_executor_admit_mem_limit.second) {
+ coord_admit_mem_limit.second = bp.be_desc.admit_mem_limit();
+ } else if (bp.be_desc.admit_mem_limit() < min_executor_admit_mem_limit.second) {
min_executor_admit_mem_limit.first = &e.first;
- min_executor_admit_mem_limit.second = bp.be_desc.admit_mem_limit;
+ min_executor_admit_mem_limit.second = bp.be_desc.admit_mem_limit();
}
}
@@ -1219,7 +1219,7 @@ Status AdmissionController::ComputeGroupSchedules(
LOG(WARNING) << queue_node->not_admitted_reason;
return Status::OK();
}
- const TBackendDescriptor& local_be_desc = *membership_snapshot->local_be_desc;
+ const BackendDescriptorPB& local_be_desc = *membership_snapshot->local_be_desc;
vector<const ExecutorGroup*> executor_groups;
GetExecutorGroupsForPool(
@@ -1981,8 +1981,8 @@ void AdmissionController::UpdateExecGroupMetricMap(
auto new_grp_it = snapshot->executor_groups.find(new_grp);
DCHECK(new_grp_it != snapshot->executor_groups.end());
ExecutorGroup group = new_grp_it->second;
- for (const TBackendDescriptor& be_desc : group.GetAllExecutorDescriptors()) {
- const string& host = TNetworkAddressToString(be_desc.address);
+ for (const BackendDescriptorPB& be_desc : group.GetAllExecutorDescriptors()) {
+ const string& host = NetworkAddressPBToString(be_desc.address());
auto stats = host_stats_.find(host);
if (stats != host_stats_.end()) {
currently_running = std::max(currently_running, stats->second.num_admitted);
diff --git a/be/src/scheduling/cluster-membership-mgr-test.cc b/be/src/scheduling/cluster-membership-mgr-test.cc
index acebe10..268ad26 100644
--- a/be/src/scheduling/cluster-membership-mgr-test.cc
+++ b/be/src/scheduling/cluster-membership-mgr-test.cc
@@ -72,14 +72,14 @@ class ClusterMembershipMgrTest : public testing::Test {
string backend_id;
std::unique_ptr<MetricGroup> metric_group;
std::unique_ptr<ClusterMembershipMgr> cmm;
- std::shared_ptr<TBackendDescriptor> desc;
+ std::shared_ptr<BackendDescriptorPB> desc;
};
/// The list of backends_ that owns all backends.
vector<unique_ptr<Backend>> backends_;
/// Various lists of pointers that point into elements of 'backends_'. These pointers
/// will stay valid when backend_ resizes. Backends can be in one of 5 states:
- /// - "Offline": A TBackendDescriptor has been created but has not been associated with
+ /// - "Offline": A BackendDescriptorPB has been created but has not been associated with
/// a ClusterMembershipMgr.
/// - "Starting": A ClusterMembershipMgr has been created, but the associated backend
/// descriptor is not running yet (no callback registered with the
@@ -166,8 +166,8 @@ class ClusterMembershipMgrTest : public testing::Test {
if (idx == -1) idx = backends_.size();
backends_.push_back(make_unique<Backend>());
auto& be = backends_.back();
- be->desc = make_shared<TBackendDescriptor>(MakeBackendDescriptor(idx));
- be->backend_id = be->desc->address.hostname;
+ be->desc = make_shared<BackendDescriptorPB>(MakeBackendDescriptor(idx));
+ be->backend_id = be->desc->address().hostname();
offline_.push_back(be.get());
return be.get();
}
@@ -213,7 +213,7 @@ class ClusterMembershipMgrTest : public testing::Test {
/// moves the backend from 'running_' to 'quiescing_'.
void QuiesceBackend(Backend* be) {
ASSERT_TRUE(IsInVector(be, running_));
- be->desc->__set_is_quiescing(true);
+ be->desc->set_is_quiescing(true);
TopicDeltas topic_deltas = Poll(be);
ASSERT_EQ(1, topic_deltas.size());
ASSERT_EQ(1, topic_deltas[0].topic_entries.size());
@@ -269,13 +269,13 @@ class ClusterMembershipMgrTest : public testing::Test {
/// It also serves as an example for how to craft statestore messages and pass them to
/// UpdaUpdateMembership().
TEST_F(ClusterMembershipMgrTest, TwoInstances) {
- auto b1 = make_shared<TBackendDescriptor>(MakeBackendDescriptor(1));
- auto b2 = make_shared<TBackendDescriptor>(MakeBackendDescriptor(2));
+ auto b1 = make_shared<BackendDescriptorPB>(MakeBackendDescriptor(1));
+ auto b2 = make_shared<BackendDescriptorPB>(MakeBackendDescriptor(2));
MetricGroup tmp_metrics1("test-metrics1");
MetricGroup tmp_metrics2("test-metrics2");
- ClusterMembershipMgr cmm1(b1->address.hostname, nullptr, &tmp_metrics1);
- ClusterMembershipMgr cmm2(b2->address.hostname, nullptr, &tmp_metrics2);
+ ClusterMembershipMgr cmm1(b1->address().hostname(), nullptr, &tmp_metrics1);
+ ClusterMembershipMgr cmm2(b2->address().hostname(), nullptr, &tmp_metrics2);
const Statestore::TopicId topic_id = Statestore::IMPALA_MEMBERSHIP_TOPIC;
StatestoreSubscriber::TopicDeltaMap topic_delta_map = {{topic_id, TTopicDelta()}};
@@ -317,12 +317,12 @@ TEST_F(ClusterMembershipMgrTest, TwoInstances) {
// Both managers now have the same state. Shutdown one of them and step through
// propagating the update.
- b1->is_quiescing = true;
+ b1->set_is_quiescing(true);
// Send an empty update to the 1st one to trigger propagation of the shutdown
returned_topic_deltas.clear();
topic_delta_map[topic_id] = empty_delta;
cmm1.UpdateMembership(topic_delta_map, &returned_topic_deltas);
- // The mgr will return its changed TBackendDescriptor
+ // The mgr will return its changed BackendDescriptorPB
ASSERT_EQ(1, returned_topic_deltas.size());
// It will also remove itself from the executor group (but not the current backends).
ASSERT_EQ(1, GetDefaultGroupSize(cmm1));
diff --git a/be/src/scheduling/cluster-membership-mgr.cc b/be/src/scheduling/cluster-membership-mgr.cc
index 90c57df..72a473d 100644
--- a/be/src/scheduling/cluster-membership-mgr.cc
+++ b/be/src/scheduling/cluster-membership-mgr.cc
@@ -28,15 +28,15 @@ using namespace impala;
/// Looks for an executor group with name 'name' in 'executor_groups' and returns it. If
/// the group doesn't exist yet, it creates a new one and inserts it into
/// 'executor_groups'.
-ExecutorGroup* FindOrInsertExecutorGroup(const TExecutorGroupDesc& group,
+ExecutorGroup* FindOrInsertExecutorGroup(const ExecutorGroupDescPB& group,
ClusterMembershipMgr::ExecutorGroups* executor_groups) {
- auto it = executor_groups->find(group.name);
+ auto it = executor_groups->find(group.name());
if (it != executor_groups->end()) {
- DCHECK_EQ(group.name, it->second.name());
+ DCHECK_EQ(group.name(), it->second.name());
return &it->second;
}
bool inserted;
- tie(it, inserted) = executor_groups->emplace(group.name, ExecutorGroup(group));
+ tie(it, inserted) = executor_groups->emplace(group.name(), ExecutorGroup(group));
DCHECK(inserted);
return &it->second;
}
@@ -54,7 +54,6 @@ ClusterMembershipMgr::ClusterMembershipMgr(
string local_backend_id, StatestoreSubscriber* subscriber, MetricGroup* metrics)
: current_membership_(std::make_shared<const Snapshot>()),
statestore_subscriber_(subscriber),
- thrift_serializer_(/* compact= */ false),
local_backend_id_(move(local_backend_id)) {
DCHECK(metrics != nullptr);
MetricGroup* metric_grp = metrics->GetOrCreateChildGroup("cluster-membership");
@@ -190,7 +189,7 @@ void ClusterMembershipMgr::UpdateMembership(
// Deleted item
if (item.deleted) {
if (new_backend_map->find(item.key) != new_backend_map->end()) {
- const TBackendDescriptor& be_desc = (*new_backend_map)[item.key];
+ const BackendDescriptorPB& be_desc = (*new_backend_map)[item.key];
bool blacklisted = new_blacklist->FindAndRemove(be_desc)
== ExecutorBlacklist::State::BLACKLISTED;
if (blacklisted) {
@@ -199,10 +198,10 @@ void ClusterMembershipMgr::UpdateMembership(
}
// If the backend was quiescing or was previously blacklisted, it will already
// have been removed from 'executor_groups'.
- if (be_desc.is_executor && !be_desc.is_quiescing && !blacklisted) {
- for (const auto& group : be_desc.executor_groups) {
- VLOG(1) << "Removing backend " << item.key << " from group " << group
- << " (deleted)";
+ if (be_desc.is_executor() && !be_desc.is_quiescing() && !blacklisted) {
+ for (const auto& group : be_desc.executor_groups()) {
+ VLOG(1) << "Removing backend " << item.key << " from group "
+ << group.DebugString() << " (deleted)";
FindOrInsertExecutorGroup(
group, new_executor_groups)->RemoveExecutor(be_desc);
}
@@ -213,38 +212,33 @@ void ClusterMembershipMgr::UpdateMembership(
}
// New or existing item
- TBackendDescriptor be_desc;
- // Benchmarks have suggested that this method can deserialize ~10m messages per
- // second, so no immediate need to consider optimization.
- uint32_t len = item.value.size();
- Status status = DeserializeThriftMsg(
- reinterpret_cast<const uint8_t*>(item.value.data()), &len, false, &be_desc);
- if (!status.ok()) {
+ BackendDescriptorPB be_desc;
+ bool success = be_desc.ParseFromString(item.value);
+ if (!success) {
LOG_EVERY_N(WARNING, 30) << "Error deserializing membership topic item with key: "
<< item.key;
continue;
}
- if (be_desc.ip_address.empty()) {
+ if (be_desc.ip_address().empty()) {
// Each backend resolves its own IP address and transmits it inside its backend
// descriptor as part of the statestore update. If it is empty, then either that
// code has been changed, or someone else is sending malformed packets.
LOG_EVERY_N(WARNING, 30) << "Ignoring subscription request with empty IP address "
- "from subscriber: " << TNetworkAddressToString(be_desc.address);
+ << "from subscriber: " << be_desc.address();
continue;
}
if (item.key == local_backend_id_) {
if (local_be_desc.get() == nullptr) {
LOG_EVERY_N(WARNING, 30) << "Another host registered itself with the local "
- << "backend id (" << item.key << "), but the local backend has not started "
- "yet. The offending address is: " << TNetworkAddressToString(be_desc.address);
- } else if (be_desc.address != local_be_desc->address) {
+ << "backend id (" << item.key << "), but the local backend has not started "
+ << "yet. The offending address is: " << be_desc.address();
+ } else if (be_desc.address() != local_be_desc->address()) {
// Someone else has registered this subscriber ID with a different address. We
// will try to re-register (i.e. overwrite their subscription), but there is
// likely a configuration problem.
LOG_EVERY_N(WARNING, 30) << "Duplicate subscriber registration from address: "
- << TNetworkAddressToString(be_desc.address) << " (we are: "
- << TNetworkAddressToString(local_be_desc->address) << ", backend id: "
- << item.key << ")";
+ << be_desc.address() << " (we are: " << local_be_desc->address()
+ << ", backend id: " << item.key << ")";
}
// We will always set the local backend explicitly below, so we ignore it here.
continue;
@@ -253,15 +247,15 @@ void ClusterMembershipMgr::UpdateMembership(
auto it = new_backend_map->find(item.key);
if (it != new_backend_map->end()) {
// Update
- TBackendDescriptor& existing = it->second;
+ BackendDescriptorPB& existing = it->second;
// Once a backend starts quiescing, it must stay in the quiescing state until it
// has been deleted from the cluster membership. Once a node starts quiescing, it
// can never transfer back to a running state.
- if (existing.is_quiescing) DCHECK(be_desc.is_quiescing);
+ if (existing.is_quiescing()) DCHECK(be_desc.is_quiescing());
// If the node starts quiescing
- if (be_desc.is_quiescing && !existing.is_quiescing && existing.is_executor) {
+ if (be_desc.is_quiescing() && !existing.is_quiescing() && existing.is_executor()) {
// If the backend starts quiescing and it is present in the blacklist, remove it
// from the blacklist. If the backend is present in the blacklist, there is no
// need to remove it from the executor group because it has already been removed
@@ -272,9 +266,9 @@ void ClusterMembershipMgr::UpdateMembership(
DCHECK(!IsBackendInExecutorGroups(be_desc, *new_executor_groups));
} else {
// Executor needs to be removed from its groups
- for (const auto& group : be_desc.executor_groups) {
- VLOG(1) << "Removing backend " << item.key << " from group " << group
- << " (quiescing)";
+ for (const auto& group : be_desc.executor_groups()) {
+ VLOG(1) << "Removing backend " << item.key << " from group "
+ << group.DebugString() << " (quiescing)";
FindOrInsertExecutorGroup(group, new_executor_groups)
->RemoveExecutor(be_desc);
}
@@ -284,9 +278,9 @@ void ClusterMembershipMgr::UpdateMembership(
} else {
// Create
new_backend_map->insert(make_pair(item.key, be_desc));
- if (!be_desc.is_quiescing && be_desc.is_executor) {
- for (const auto& group : be_desc.executor_groups) {
- VLOG(1) << "Adding backend " << item.key << " to group " << group;
+ if (!be_desc.is_quiescing() && be_desc.is_executor()) {
+ for (const auto& group : be_desc.executor_groups()) {
+ VLOG(1) << "Adding backend " << item.key << " to group " << group.DebugString();
FindOrInsertExecutorGroup(group, new_executor_groups)->AddExecutor(be_desc);
}
}
@@ -300,12 +294,12 @@ void ClusterMembershipMgr::UpdateMembership(
if (needs_blacklist_maintenance) {
// Add any backends that were removed from the blacklist and put on probation back
// into 'executor_groups'.
- std::list<TBackendDescriptor> probation_list;
+ std::list<BackendDescriptorPB> probation_list;
new_blacklist->Maintenance(&probation_list);
- for (const TBackendDescriptor& be_desc : probation_list) {
- for (const auto& group : be_desc.executor_groups) {
- VLOG(1) << "Adding backend " << TNetworkAddressToString(be_desc.address)
- << " to group " << group << " (passed blacklist timeout)";
+ for (const BackendDescriptorPB& be_desc : probation_list) {
+ for (const auto& group : be_desc.executor_groups()) {
+ VLOG(1) << "Adding backend " << be_desc.address() << " to group "
+ << group.DebugString() << " (passed blacklist timeout)";
FindOrInsertExecutorGroup(group, new_executor_groups)->AddExecutor(be_desc);
}
}
@@ -317,13 +311,13 @@ void ClusterMembershipMgr::UpdateMembership(
if (NeedsLocalBackendUpdate(*new_state, local_be_desc)) {
// We need to update both the new membership state and the statestore
(*new_backend_map)[local_backend_id_] = *local_be_desc;
- for (const auto& group : local_be_desc->executor_groups) {
- if (local_be_desc->is_quiescing) {
- VLOG(1) << "Removing local backend from group " << group;
+ for (const auto& group : local_be_desc->executor_groups()) {
+ if (local_be_desc->is_quiescing()) {
+ VLOG(1) << "Removing local backend from group " << group.DebugString();
FindOrInsertExecutorGroup(
group, new_executor_groups)->RemoveExecutor(*local_be_desc);
- } else if (local_be_desc->is_executor) {
- VLOG(1) << "Adding local backend to group " << group;
+ } else if (local_be_desc->is_executor()) {
+ VLOG(1) << "Adding local backend to group " << group.DebugString();
FindOrInsertExecutorGroup(
group, new_executor_groups)->AddExecutor(*local_be_desc);
}
@@ -347,14 +341,15 @@ void ClusterMembershipMgr::UpdateMembership(
}
void ClusterMembershipMgr::BlacklistExecutor(
- const TBackendDescriptor& be_desc, const Status& cause) {
+ const BackendDescriptorPB& be_desc, const Status& cause) {
DCHECK(!cause.ok());
if (!ExecutorBlacklist::BlacklistingEnabled()) return;
lock_guard<mutex> l(update_membership_lock_);
// Don't blacklist the local executor. Some queries may have root fragments that must be
// scheduled on the coordinator and will always fail if its blacklisted.
- if (be_desc.ip_address == current_membership_->local_be_desc->ip_address
- && be_desc.address.port == current_membership_->local_be_desc->address.port) {
+ if (be_desc.ip_address() == current_membership_->local_be_desc->ip_address()
+ && be_desc.address().port()
+ == current_membership_->local_be_desc->address().port()) {
return;
}
@@ -369,10 +364,10 @@ void ClusterMembershipMgr::BlacklistExecutor(
// Check the Snapshot that we'll be updating to see if the backend is present, to avoid
// copying the Snapshot if it isn't.
bool exists = false;
- for (const auto& group : be_desc.executor_groups) {
- auto it = base_snapshot->executor_groups.find(group.name);
+ for (const auto& group : be_desc.executor_groups()) {
+ auto it = base_snapshot->executor_groups.find(group.name());
if (it != base_snapshot->executor_groups.end()
- && it->second.LookUpBackendDesc(be_desc.address) != nullptr) {
+ && it->second.LookUpBackendDesc(be_desc.address()) != nullptr) {
exists = true;
break;
}
@@ -384,8 +379,7 @@ void ClusterMembershipMgr::BlacklistExecutor(
return;
}
- LOG(INFO) << "Blacklisting " << TNetworkAddressToString(be_desc.address) << ": "
- << cause;
+ LOG(INFO) << "Blacklisting " << be_desc.address() << ": " << cause;
std::shared_ptr<Snapshot> new_state;
if (recovering) {
@@ -397,9 +391,9 @@ void ClusterMembershipMgr::BlacklistExecutor(
}
ExecutorGroups* new_executor_groups = &(new_state->executor_groups);
- for (const auto& group : be_desc.executor_groups) {
- VLOG(1) << "Removing backend " << TNetworkAddressToString(be_desc.address)
- << " from group " << group << " (blacklisted)";
+ for (const auto& group : be_desc.executor_groups()) {
+ VLOG(1) << "Removing backend " << be_desc.address() << " from group "
+ << group.DebugString() << " (blacklisted)";
FindOrInsertExecutorGroup(group, new_executor_groups)->RemoveExecutor(be_desc);
}
@@ -431,7 +425,7 @@ void ClusterMembershipMgr::BlacklistExecutor(
}
void ClusterMembershipMgr::AddLocalBackendToStatestore(
- const TBackendDescriptor& local_be_desc,
+ const BackendDescriptorPB& local_be_desc,
vector<TTopicDelta>* subscriber_topic_updates) {
VLOG(1) << "Sending local backend to statestore";
@@ -445,10 +439,9 @@ void ClusterMembershipMgr::AddLocalBackendToStatestore(
TTopicItem& item = update.topic_entries.back();
item.key = local_backend_id_;
- Status status = thrift_serializer_.SerializeToString(&local_be_desc, &item.value);
- if (!status.ok()) {
- LOG(FATAL) << "Failed to serialize Impala backend descriptor for statestore topic:"
- << " " << status.GetDetail();
+ bool success = local_be_desc.SerializeToString(&item.value);
+ if (!success) {
+ LOG(FATAL) << "Failed to serialize Impala backend descriptor for statestore topic.";
subscriber_topic_updates->pop_back();
return;
}
@@ -476,15 +469,15 @@ bool ClusterMembershipMgr::NeedsLocalBackendUpdate(const Snapshot& state,
if (state.local_be_desc.get() == nullptr) return true;
auto it = state.current_backends.find(local_backend_id_);
if (it == state.current_backends.end()) return true;
- return it->second.is_quiescing != local_be_desc->is_quiescing;
+ return it->second.is_quiescing() != local_be_desc->is_quiescing();
}
bool ClusterMembershipMgr::CheckConsistency(const BackendIdMap& current_backends,
const ExecutorGroups& executor_groups, const ExecutorBlacklist& executor_blacklist) {
// Build a map of all backend descriptors
- std::unordered_map<TNetworkAddress, TBackendDescriptor> address_to_backend;
+ std::unordered_map<NetworkAddressPB, BackendDescriptorPB> address_to_backend;
for (const auto& it : current_backends) {
- address_to_backend.emplace(it.second.address, it.second);
+ address_to_backend.emplace(it.second.address(), it.second);
}
// Check groups against the map
@@ -492,38 +485,39 @@ bool ClusterMembershipMgr::CheckConsistency(const BackendIdMap& current_backends
const string& group_name = group_it.first;
const ExecutorGroup& group = group_it.second;
ExecutorGroup::Executors backends = group.GetAllExecutorDescriptors();
- for (const TBackendDescriptor& group_be : backends) {
- if (!group_be.is_executor) {
- LOG(WARNING) << "Backend " << group_be.address << " in group " << group_name
+ for (const BackendDescriptorPB& group_be : backends) {
+ if (!group_be.is_executor()) {
+ LOG(WARNING) << "Backend " << group_be.DebugString() << " in group " << group_name
<< " is not an executor";
return false;
}
- if (group_be.is_quiescing) {
- LOG(WARNING) << "Backend " << group_be.address << " in group " << group_name
+ if (group_be.is_quiescing()) {
+ LOG(WARNING) << "Backend " << group_be.DebugString() << " in group " << group_name
<< " is quiescing";
return false;
}
- auto current_be_it = address_to_backend.find(group_be.address);
+ auto current_be_it = address_to_backend.find(group_be.address());
if (current_be_it == address_to_backend.end()) {
- LOG(WARNING) << "Backend " << group_be.address << " is in group " << group_name
- << " but not in current set of backends";
+ LOG(WARNING) << "Backend " << group_be.DebugString() << " is in group "
+ << group_name << " but not in current set of backends";
return false;
}
- if (current_be_it->second.is_quiescing != group_be.is_quiescing) {
- LOG(WARNING) << "Backend " << group_be.address << " in group " << group_name
+ if (current_be_it->second.is_quiescing() != group_be.is_quiescing()) {
+ LOG(WARNING) << "Backend " << group_be.DebugString() << " in group " << group_name
<< " differs from backend in current set of backends: is_quiescing ("
- << current_be_it->second.is_quiescing << " != " << group_be.is_quiescing
+ << current_be_it->second.is_quiescing() << " != " << group_be.is_quiescing()
<< ")";
return false;
}
- if (current_be_it->second.is_executor != group_be.is_executor) {
- LOG(WARNING) << "Backend " << group_be.address << " in group " << group_name
+ if (current_be_it->second.is_executor() != group_be.is_executor()) {
+ LOG(WARNING) << "Backend " << group_be.DebugString() << " in group " << group_name
<< " differs from backend in current set of backends: is_executor ("
- << current_be_it->second.is_executor << " != " << group_be.is_executor << ")";
+ << current_be_it->second.is_executor() << " != " << group_be.is_executor()
+ << ")";
return false;
}
if (executor_blacklist.IsBlacklisted(group_be)) {
- LOG(WARNING) << "Backend " << group_be.address << " in group " << group_name
+ LOG(WARNING) << "Backend " << group_be.DebugString() << " in group " << group_name
<< " is blacklisted.";
return false;
}
@@ -551,9 +545,9 @@ void ClusterMembershipMgr::UpdateMetrics(const SnapshotPtr& new_state){
}
bool ClusterMembershipMgr::IsBackendInExecutorGroups(
- const TBackendDescriptor& be_desc, const ExecutorGroups& executor_groups) {
+ const BackendDescriptorPB& be_desc, const ExecutorGroups& executor_groups) {
for (const auto& executor_group : executor_groups) {
- if (executor_group.second.LookUpBackendDesc(be_desc.address) != nullptr) {
+ if (executor_group.second.LookUpBackendDesc(be_desc.address()) != nullptr) {
return true;
}
}
diff --git a/be/src/scheduling/cluster-membership-mgr.h b/be/src/scheduling/cluster-membership-mgr.h
index c2e3b5f..3954fd3 100644
--- a/be/src/scheduling/cluster-membership-mgr.h
+++ b/be/src/scheduling/cluster-membership-mgr.h
@@ -17,21 +17,17 @@
#pragma once
-#include <gtest/gtest_prod.h> // for FRIEND_TEST
+#include <functional>
+#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>
-#include <unordered_set>
#include <vector>
-#include "common/global-types.h"
#include "common/status.h"
-#include "gen-cpp/StatestoreService_types.h"
-#include "gutil/threading/thread_collision_warner.h"
#include "scheduling/executor-blacklist.h"
#include "scheduling/executor-group.h"
#include "statestore/statestore-subscriber.h"
-#include "util/container-util.h"
#include "util/metrics-fwd.h"
namespace impala {
@@ -75,10 +71,10 @@ class ClusterMembershipMgr {
public:
/// A immutable pointer to a backend descriptor. It is used to return a consistent,
/// immutable copy of a backend descriptor.
- typedef std::shared_ptr<const TBackendDescriptor> BeDescSharedPtr;
+ typedef std::shared_ptr<const BackendDescriptorPB> BeDescSharedPtr;
/// Maps statestore subscriber IDs to backend descriptors.
- typedef std::unordered_map<std::string, TBackendDescriptor> BackendIdMap;
+ typedef std::unordered_map<std::string, BackendDescriptorPB> BackendIdMap;
/// Maps executor group names to executor groups. For now, only a default group exists
/// and all executors are part of that group.
@@ -93,11 +89,12 @@ class ClusterMembershipMgr {
Snapshot(const Snapshot&) = default;
/// The current backend descriptor of the local backend.
BeDescSharedPtr local_be_desc;
- /// Map from unique backend ID to TBackendDescriptor for all known backends, including
- /// those that are quiescing or blacklisted. The {backend ID, TBackendDescriptor}
- /// pairs represent the IMPALA_MEMBERSHIP_TOPIC {key, value} pairs of known executors
- /// retrieved from the statestore. It's important to track both the backend ID as well
- /// as the TBackendDescriptor so we know what is being removed in a given update.
+ /// Map from unique backend ID to BackendDescriptorPB for all known backends,
+ /// including those that are quiescing or blacklisted. The {backend ID,
+ /// BackendDescriptorPB} pairs represent the IMPALA_MEMBERSHIP_TOPIC {key, value}
+ /// pairs of known executors retrieved from the statestore. It's important to track
+ /// both the backend ID as well as the BackendDescriptorPB so we know what is being
+ /// removed in a given update.
BackendIdMap current_backends;
/// A map of executor groups by their names. Only contains executors that are
/// available for scheduling queries on and not executors that are quiescing or
@@ -165,11 +162,11 @@ class ClusterMembershipMgr {
/// Adds the given backend to the local blacklist. Updates 'current_membership_' to
/// remove the backend from 'executor_groups' so that it will not be scheduled on.
/// 'cause' is an error status representing the reason the node was blacklisted.
- void BlacklistExecutor(const TBackendDescriptor& be_desc, const Status& cause);
+ void BlacklistExecutor(const BackendDescriptorPB& be_desc, const Status& cause);
private:
/// Serializes and adds the local backend descriptor to 'subscriber_topic_updates'.
- void AddLocalBackendToStatestore(const TBackendDescriptor& local_be_desc,
+ void AddLocalBackendToStatestore(const BackendDescriptorPB& local_be_desc,
std::vector<TTopicDelta>* subscriber_topic_updates);
/// Returns the local backend descriptor or nullptr if no local backend has been
@@ -205,7 +202,7 @@ class ClusterMembershipMgr {
/// Used to check the consistency of the ClusterMembershipMgr state. Currently, only
/// used in DCHECKs.
bool IsBackendInExecutorGroups(
- const TBackendDescriptor& be_desc, const ExecutorGroups& executor_groups);
+ const BackendDescriptorPB& be_desc, const ExecutorGroups& executor_groups);
/// Ensures that only one thread is processing a membership update at a time, either
/// from a statestore update or a blacklisting decision. Must be taken before any other
@@ -237,10 +234,6 @@ class ClusterMembershipMgr {
/// backends is fixed (only useful for tests). Thread-safe, not protected by a lock.
StatestoreSubscriber* statestore_subscriber_;
- /// Serializes TBackendDescriptors when creating topic updates. Not protected by a lock
- /// - only used in statestore thread.
- ThriftSerializer thrift_serializer_;
-
/// Unique - across the cluster - identifier for this impala backend. Used to validate
/// incoming backend descriptors and to register this backend with the statestore. Not
/// protected by a lock - only used in the statestore thread.
diff --git a/be/src/scheduling/cluster-membership-test-util.cc b/be/src/scheduling/cluster-membership-test-util.cc
index 0b99641..17d84c3 100644
--- a/be/src/scheduling/cluster-membership-test-util.cc
+++ b/be/src/scheduling/cluster-membership-test-util.cc
@@ -44,34 +44,34 @@ string HostIdxToIpAddr(int host_idx) {
return IP_PREFIX + suffix;
}
-TBackendDescriptor MakeBackendDescriptor(
- int idx, const TExecutorGroupDesc& group_desc, int port_offset) {
- TBackendDescriptor be_desc;
- be_desc.address.hostname = HostIdxToHostname(idx);
- be_desc.address.port = BACKEND_PORT + port_offset;
- be_desc.ip_address = HostIdxToIpAddr(idx);
+BackendDescriptorPB MakeBackendDescriptor(
+ int idx, const ExecutorGroupDescPB& group_desc, int port_offset) {
+ BackendDescriptorPB be_desc;
+ be_desc.mutable_address()->set_hostname(HostIdxToHostname(idx));
+ be_desc.mutable_address()->set_port(BACKEND_PORT + port_offset);
+ be_desc.set_ip_address(HostIdxToIpAddr(idx));
// krpc_address is always resolved
- be_desc.krpc_address.hostname = be_desc.ip_address;
- be_desc.krpc_address.port = KRPC_PORT + port_offset;
- be_desc.__set_is_coordinator(true);
- be_desc.__set_is_executor(true);
- be_desc.is_quiescing = false;
- be_desc.executor_groups.push_back(group_desc);
+ be_desc.mutable_krpc_address()->set_hostname(be_desc.ip_address());
+ be_desc.mutable_krpc_address()->set_port(KRPC_PORT + port_offset);
+ be_desc.set_is_coordinator(true);
+ be_desc.set_is_executor(true);
+ be_desc.set_is_quiescing(false);
+ *be_desc.add_executor_groups() = group_desc;
return be_desc;
}
-TBackendDescriptor MakeBackendDescriptor(
+BackendDescriptorPB MakeBackendDescriptor(
int idx, const ExecutorGroup& group, int port_offset) {
- TExecutorGroupDesc group_desc;
- group_desc.name = group.name();
- group_desc.min_size = group.min_size();
+ ExecutorGroupDescPB group_desc;
+ group_desc.set_name(group.name());
+ group_desc.set_min_size(group.min_size());
return MakeBackendDescriptor(idx, group_desc, port_offset);
}
-TBackendDescriptor MakeBackendDescriptor(int idx, int port_offset) {
- TExecutorGroupDesc group_desc;
- group_desc.name = ImpalaServer::DEFAULT_EXECUTOR_GROUP_NAME;
- group_desc.min_size = 1;
+BackendDescriptorPB MakeBackendDescriptor(int idx, int port_offset) {
+ ExecutorGroupDescPB group_desc;
+ group_desc.set_name(ImpalaServer::DEFAULT_EXECUTOR_GROUP_NAME);
+ group_desc.set_min_size(1);
return MakeBackendDescriptor(idx, group_desc, port_offset);
}
diff --git a/be/src/scheduling/cluster-membership-test-util.h b/be/src/scheduling/cluster-membership-test-util.h
index 3b282a7..7787178 100644
--- a/be/src/scheduling/cluster-membership-test-util.h
+++ b/be/src/scheduling/cluster-membership-test-util.h
@@ -17,11 +17,13 @@
#pragma once
-#include "gen-cpp/StatestoreService_types.h"
#include <string>
namespace impala {
+
+class BackendDescriptorPB;
class ExecutorGroup;
+class ExecutorGroupDescPB;
namespace test {
@@ -37,15 +39,15 @@ std::string HostIdxToIpAddr(int host_idx);
/// To create backends on the same host, an optional port offset can be specified.
///
/// Make a backend descriptor for group 'group_desc'.
-TBackendDescriptor MakeBackendDescriptor(
- int idx, const TExecutorGroupDesc& group_desc, int port_offset = 0);
+BackendDescriptorPB MakeBackendDescriptor(
+ int idx, const ExecutorGroupDescPB& group_desc, int port_offset = 0);
/// Make a backend descriptor for 'group'.
-TBackendDescriptor MakeBackendDescriptor(
+BackendDescriptorPB MakeBackendDescriptor(
int idx, const ExecutorGroup& group, int port_offset = 0);
/// Make a backend descriptor for the default executor group.
-TBackendDescriptor MakeBackendDescriptor(int idx, int port_offset = 0);
+BackendDescriptorPB MakeBackendDescriptor(int idx, int port_offset = 0);
} // end namespace test
} // end namespace impala
diff --git a/be/src/scheduling/executor-blacklist.cc b/be/src/scheduling/executor-blacklist.cc
index 206fb2b..fae1b20 100644
--- a/be/src/scheduling/executor-blacklist.cc
+++ b/be/src/scheduling/executor-blacklist.cc
@@ -37,10 +37,10 @@ bool ExecutorBlacklist::BlacklistingEnabled() {
}
void ExecutorBlacklist::Blacklist(
- const TBackendDescriptor& be_desc, const Status& cause) {
+ const BackendDescriptorPB& be_desc, const Status& cause) {
DCHECK(BlacklistingEnabled());
- DCHECK(!be_desc.ip_address.empty());
- vector<Entry>& be_descs = executor_list_[be_desc.ip_address];
+ DCHECK(!be_desc.ip_address().empty());
+ vector<Entry>& be_descs = executor_list_[be_desc.ip_address()];
auto it = find_if(be_descs.begin(), be_descs.end(),
std::bind(eqBePort, be_desc, std::placeholders::_1));
if (it != be_descs.end()) {
@@ -68,13 +68,13 @@ void ExecutorBlacklist::Blacklist(
// This executor was not already on the list, create a new Entry for it.
be_descs.emplace_back(be_desc, MonotonicMillis(), cause);
}
- VLOG(2) << "Blacklisted " << TNetworkAddressToString(be_desc.address)
- << ", current blacklist: " << DebugString();
+ VLOG(2) << "Blacklisted " << be_desc.address() << ", current blacklist: "
+ << DebugString();
}
ExecutorBlacklist::State ExecutorBlacklist::FindAndRemove(
- const TBackendDescriptor& be_desc) {
- auto be_descs_it = executor_list_.find(be_desc.ip_address);
+ const BackendDescriptorPB& be_desc) {
+ auto be_descs_it = executor_list_.find(be_desc.ip_address());
if (be_descs_it == executor_list_.end()) {
// Executor wasn't on the blacklist.
return NOT_BLACKLISTED;
@@ -111,7 +111,7 @@ bool ExecutorBlacklist::NeedsMaintenance() const {
return false;
}
-void ExecutorBlacklist::Maintenance(std::list<TBackendDescriptor>* probation_list) {
+void ExecutorBlacklist::Maintenance(std::list<BackendDescriptorPB>* probation_list) {
int64_t blacklist_timeout = GetBlacklistTimeoutMs();
int64_t probation_timeout = blacklist_timeout * PROBATION_TIMEOUT_MULTIPLIER;
int64_t now = MonotonicMillis();
@@ -123,7 +123,7 @@ void ExecutorBlacklist::Maintenance(std::list<TBackendDescriptor>* probation_lis
if (entry_it->state == State::BLACKLISTED) {
// Check if we can take it off the blacklist and put it on probation.
if (elapsed > blacklist_timeout * entry_it->num_consecutive_blacklistings) {
- LOG(INFO) << "Executor " << TNetworkAddressToString(entry_it->be_desc.address)
+ LOG(INFO) << "Executor " << entry_it->be_desc.address()
<< " passed the timeout and will be taken off the blacklist.";
probation_list->push_back(entry_it->be_desc);
entry_it->state = State::ON_PROBATION;
@@ -148,11 +148,11 @@ void ExecutorBlacklist::Maintenance(std::list<TBackendDescriptor>* probation_lis
}
bool ExecutorBlacklist::IsBlacklisted(
- const TBackendDescriptor& be_desc, Status* cause, int64_t* time_remaining_ms) const {
+ const BackendDescriptorPB& be_desc, Status* cause, int64_t* time_remaining_ms) const {
for (auto executor_it : executor_list_) {
- if (executor_it.first == be_desc.ip_address) {
+ if (executor_it.first == be_desc.ip_address()) {
for (auto entry_it : executor_it.second) {
- if (entry_it.be_desc.address.port == be_desc.address.port
+ if (entry_it.be_desc.address().port() == be_desc.address().port()
&& entry_it.state == State::BLACKLISTED) {
if (cause != nullptr) *cause = entry_it.cause;
int64_t elapsed_ms = MonotonicMillis() - entry_it.blacklist_time_ms;
@@ -172,7 +172,7 @@ std::string ExecutorBlacklist::BlacklistToString() const {
for (auto executor_it : executor_list_) {
for (auto entry_it : executor_it.second) {
if (entry_it.state == State::BLACKLISTED) {
- ss << TNetworkAddressToString(entry_it.be_desc.address) << " ";
+ ss << entry_it.be_desc.address() << " ";
}
}
}
@@ -184,7 +184,7 @@ std::string ExecutorBlacklist::DebugString() const {
ss << "ExecutorBlacklist[";
for (auto executor_it : executor_list_) {
for (auto entry_it : executor_it.second) {
- ss << TNetworkAddressToString(entry_it.be_desc.address) << " ("
+ ss << entry_it.be_desc.address() << " ("
<< (entry_it.state == BLACKLISTED ? "blacklisted" : "on probation") << ") ";
}
}
@@ -201,10 +201,10 @@ int64_t ExecutorBlacklist::GetBlacklistTimeoutMs() const {
}
bool ExecutorBlacklist::eqBePort(
- const TBackendDescriptor& be_desc, const Entry& existing) {
+ const BackendDescriptorPB& be_desc, const Entry& existing) {
// The IP addresses must already match, so it is sufficient to check the port.
- DCHECK_EQ(existing.be_desc.ip_address, be_desc.ip_address);
- return existing.be_desc.address.port == be_desc.address.port;
+ DCHECK_EQ(existing.be_desc.ip_address(), be_desc.ip_address());
+ return existing.be_desc.address().port() == be_desc.address().port();
}
} // namespace impala
diff --git a/be/src/scheduling/executor-blacklist.h b/be/src/scheduling/executor-blacklist.h
index 4d14b69..5731bb5 100644
--- a/be/src/scheduling/executor-blacklist.h
+++ b/be/src/scheduling/executor-blacklist.h
@@ -20,13 +20,11 @@
#include <unordered_map>
#include <vector>
-#include "gen-cpp/StatestoreService_types.h"
+#include "gen-cpp/statestore_service.pb.h"
#include "util/network-util.h"
namespace impala {
-class TBackendDescriptor;
-
/// Class to maintain a local blacklist of executors.
///
/// Executors can be added to the blacklist with Blacklist() and removed with
@@ -66,13 +64,13 @@ class ExecutorBlacklist {
/// was on probation, updates its entry in 'executor_list_' accordingly. Should only be
/// called if BlacklistingEnabled() is true. 'cause' is an error status indicating why
/// the executor is being blacklisted.
- void Blacklist(const TBackendDescriptor& be_desc, const Status& cause);
+ void Blacklist(const BackendDescriptorPB& be_desc, const Status& cause);
/// Removes an executor from the blacklist or probation, if it is in 'executor_list_'.
/// Does not put blacklisted executors on probation. Returns the executor's state prior
/// to this call. Will return 'BLACKLISTED' for an executor that has passed the
/// blacklist timeout if Maintenance() wasn't called since the timeout.
- State FindAndRemove(const TBackendDescriptor& be_desc);
+ State FindAndRemove(const BackendDescriptorPB& be_desc);
/// Returns true if there are executors that have passed the blacklist timeout and can
/// be removed from the blacklist. Note that this does not consider executors that can
@@ -86,12 +84,12 @@ class ExecutorBlacklist {
/// probation, and its descriptor will be returned in 'probation_list'. If an executor
/// has been on probation for longer than the probation timeout, it will be taken off
/// probation.
- void Maintenance(std::list<TBackendDescriptor>* probation_list);
+ void Maintenance(std::list<BackendDescriptorPB>* probation_list);
/// If 'be_desc' is blacklisted, sets 'cause' to the error Status that caused this
/// executor to be blacklisted, sets 'time_remaining_ms' to the amount of time the
/// executor has left on the blacklist, and returns true.
- bool IsBlacklisted(const TBackendDescriptor& be_desc, Status* cause = nullptr,
+ bool IsBlacklisted(const BackendDescriptorPB& be_desc, Status* cause = nullptr,
int64_t* time_remaining_ms = nullptr) const;
/// Returns a space-separated string of the addresses of executors that are currently
@@ -104,15 +102,15 @@ class ExecutorBlacklist {
private:
/// Info about an executor that is either blacklisted or on probabtion.
struct Entry {
- Entry(
- const TBackendDescriptor& be_desc, int64_t blacklist_time_ms, const Status& cause)
+ Entry(const BackendDescriptorPB& be_desc, int64_t blacklist_time_ms,
+ const Status& cause)
: be_desc(be_desc),
blacklist_time_ms(blacklist_time_ms),
state(State::BLACKLISTED),
num_consecutive_blacklistings(1),
cause(cause) {}
- TBackendDescriptor be_desc;
+ BackendDescriptorPB be_desc;
/// The UnixMillis() of the last time this executor was blacklisted.
int64_t blacklist_time_ms;
@@ -136,7 +134,7 @@ class ExecutorBlacklist {
/// Predicate that returns true if the port number in 'be_desc' matches that in
/// 'existing.be_desc'. Assumes that they have the same 'ip_address'. Used to do find()
/// on the values of 'executor_list_'.
- static bool eqBePort(const TBackendDescriptor& be_desc, const Entry& exiting);
+ static bool eqBePort(const BackendDescriptorPB& be_desc, const Entry& exiting);
/// Map from node ip address to a list of executors for that node that have been
/// blacklisted. Note that in normal operation there will be a single impalad per node
diff --git a/be/src/scheduling/executor-group.cc b/be/src/scheduling/executor-group.cc
index 5661154..2043b81 100644
--- a/be/src/scheduling/executor-group.cc
+++ b/be/src/scheduling/executor-group.cc
@@ -30,11 +30,11 @@ ExecutorGroup::ExecutorGroup(string name) : ExecutorGroup(name, 1) {}
ExecutorGroup::ExecutorGroup(string name, int64_t min_size)
: name_(name), min_size_(min_size), executor_ip_hash_ring_(NUM_HASH_RING_REPLICAS) {
- DCHECK_GT(min_size_, 0);
- }
+ DCHECK_GT(min_size_, 0);
+}
-ExecutorGroup::ExecutorGroup(const TExecutorGroupDesc& desc)
- : ExecutorGroup(desc.name, desc.min_size) {}
+ExecutorGroup::ExecutorGroup(const ExecutorGroupDescPB& desc)
+ : ExecutorGroup(desc.name(), desc.min_size()) {}
const ExecutorGroup::Executors& ExecutorGroup::GetExecutorsForHost(
const IpAddr& ip) const {
@@ -59,58 +59,58 @@ ExecutorGroup::Executors ExecutorGroup::GetAllExecutorDescriptors() const {
return executors;
}
-void ExecutorGroup::AddExecutor(const TBackendDescriptor& be_desc) {
+void ExecutorGroup::AddExecutor(const BackendDescriptorPB& be_desc) {
// be_desc.is_executor can be false for the local backend when scheduling queries to run
// on the coordinator host.
- DCHECK(!be_desc.ip_address.empty());
- Executors& be_descs = executor_map_[be_desc.ip_address];
- auto eq = [&be_desc](const TBackendDescriptor& existing) {
+ DCHECK(!be_desc.ip_address().empty());
+ Executors& be_descs = executor_map_[be_desc.ip_address()];
+ auto eq = [&be_desc](const BackendDescriptorPB& existing) {
// The IP addresses must already match, so it is sufficient to check the port.
- DCHECK_EQ(existing.ip_address, be_desc.ip_address);
- return existing.address.port == be_desc.address.port;
+ DCHECK_EQ(existing.ip_address(), be_desc.ip_address());
+ return existing.address().port() == be_desc.address().port();
};
if (find_if(be_descs.begin(), be_descs.end(), eq) != be_descs.end()) {
LOG(DFATAL) << "Tried to add existing backend to executor group: "
- << be_desc.krpc_address;
+ << be_desc.krpc_address();
return;
}
if (!CheckConsistencyOrWarn(be_desc)) {
LOG(WARNING) << "Ignoring inconsistent backend for executor group: "
- << be_desc.krpc_address;
+ << be_desc.krpc_address();
return;
}
if (be_descs.empty()) {
- executor_ip_hash_ring_.AddNode(be_desc.ip_address);
+ executor_ip_hash_ring_.AddNode(be_desc.ip_address());
}
be_descs.push_back(be_desc);
- executor_ip_map_[be_desc.address.hostname] = be_desc.ip_address;
+ executor_ip_map_[be_desc.address().hostname()] = be_desc.ip_address();
}
-void ExecutorGroup::RemoveExecutor(const TBackendDescriptor& be_desc) {
- auto be_descs_it = executor_map_.find(be_desc.ip_address);
+void ExecutorGroup::RemoveExecutor(const BackendDescriptorPB& be_desc) {
+ auto be_descs_it = executor_map_.find(be_desc.ip_address());
if (be_descs_it == executor_map_.end()) {
LOG(DFATAL) << "Tried to remove a backend from non-existing host: "
- << be_desc.krpc_address;
+ << be_desc.krpc_address();
return;
}
- auto eq = [&be_desc](const TBackendDescriptor& existing) {
+ auto eq = [&be_desc](const BackendDescriptorPB& existing) {
// The IP addresses must already match, so it is sufficient to check the port.
- DCHECK_EQ(existing.ip_address, be_desc.ip_address);
- return existing.address.port == be_desc.address.port;
+ DCHECK_EQ(existing.ip_address(), be_desc.ip_address());
+ return existing.address().port() == be_desc.address().port();
};
Executors& be_descs = be_descs_it->second;
auto remove_it = find_if(be_descs.begin(), be_descs.end(), eq);
if (remove_it == be_descs.end()) {
LOG(DFATAL) << "Tried to remove non-existing backend from per-host list: "
- << be_desc.krpc_address;
+ << be_desc.krpc_address();
return;
}
be_descs.erase(remove_it);
if (be_descs.empty()) {
executor_map_.erase(be_descs_it);
- executor_ip_map_.erase(be_desc.address.hostname);
- executor_ip_hash_ring_.RemoveNode(be_desc.ip_address);
+ executor_ip_map_.erase(be_desc.address().hostname());
+ executor_ip_hash_ring_.RemoveNode(be_desc.ip_address());
}
}
@@ -128,13 +128,13 @@ bool ExecutorGroup::LookUpExecutorIp(const Hostname& hostname, IpAddr* ip) const
return false;
}
-const TBackendDescriptor* ExecutorGroup::LookUpBackendDesc(
- const TNetworkAddress& host) const {
+const BackendDescriptorPB* ExecutorGroup::LookUpBackendDesc(
+ const NetworkAddressPB& host) const {
IpAddr ip;
- if (LookUpExecutorIp(host.hostname, &ip)) {
+ if (LookUpExecutorIp(host.hostname(), &ip)) {
const ExecutorGroup::Executors& be_list = GetExecutorsForHost(ip);
- for (const TBackendDescriptor& desc : be_list) {
- if (desc.address == host) return &desc;
+ for (const BackendDescriptorPB& desc : be_list) {
+ if (desc.address() == host) return &desc;
}
}
return nullptr;
@@ -156,15 +156,16 @@ bool ExecutorGroup::IsHealthy() const {
return true;
}
-bool ExecutorGroup::CheckConsistencyOrWarn(const TBackendDescriptor& be_desc) const {
+bool ExecutorGroup::CheckConsistencyOrWarn(const BackendDescriptorPB& be_desc) const {
// Check if the executor's group configuration matches this group.
- for (const TExecutorGroupDesc& desc : be_desc.executor_groups) {
- if (desc.name == name_) {
- if (desc.min_size == min_size_) {
+ for (const ExecutorGroupDescPB& desc : be_desc.executor_groups()) {
+ if (desc.name() == name_) {
+ if (desc.min_size() == min_size_) {
return true;
} else {
- LOG(WARNING) << "Backend " << be_desc << " is configured for executor group "
- << desc << " but group has minimum size " << min_size_;
+ LOG(WARNING) << "Backend " << be_desc.DebugString()
+ << " is configured for executor group " << desc.DebugString()
+ << " but group has minimum size " << min_size_;
return false;
}
}
diff --git a/be/src/scheduling/executor-group.h b/be/src/scheduling/executor-group.h
index f616505..9c47f14 100644
--- a/be/src/scheduling/executor-group.h
+++ b/be/src/scheduling/executor-group.h
@@ -20,8 +20,8 @@
#include <unordered_map>
#include <vector>
-#include "gen-cpp/StatestoreService_types.h"
#include "gen-cpp/Types_types.h"
+#include "gen-cpp/statestore_service.pb.h"
#include "scheduling/hash-ring.h"
#include "util/container-util.h"
#include "util/network-util.h"
@@ -47,11 +47,11 @@ class ExecutorGroup {
public:
explicit ExecutorGroup(std::string name);
explicit ExecutorGroup(std::string name, int64_t min_size);
- explicit ExecutorGroup(const TExecutorGroupDesc& desc);
+ explicit ExecutorGroup(const ExecutorGroupDescPB& desc);
ExecutorGroup(const ExecutorGroup& other) = default;
/// List of backends, in this case they're all executors.
- typedef std::vector<TBackendDescriptor> Executors;
+ typedef std::vector<BackendDescriptorPB> Executors;
typedef std::vector<IpAddr> IpAddrs;
/// Returns the list of executors on a particular host. The caller must make sure that
@@ -68,14 +68,14 @@ class ExecutorGroup {
/// Adds an executor to the group. If it already exists, it is ignored. Backend
/// descriptors are identified by their IP address and port. Backends that fail the
/// consistency check (CheckConsistencyOrWarn()) are ignored. Note that executors can be
- /// added to an executor group even if they don't have a matching TExecutorGroupDesc in
+ /// added to an executor group even if they don't have a matching ExecutorGroupDescPB in
/// their 'executor_groups' list. This is required when building the coordinator-only
/// group during scheduling.
- void AddExecutor(const TBackendDescriptor& be_desc);
+ void AddExecutor(const BackendDescriptorPB& be_desc);
/// Removes an executor from the group if it exists. Otherwise does nothing. Backend
/// descriptors are identified by their IP address and port.
- void RemoveExecutor(const TBackendDescriptor& be_desc);
+ void RemoveExecutor(const BackendDescriptorPB& be_desc);
/// Look up the IP address of 'hostname' in the internal executor maps and return
/// whether the lookup was successful. If 'hostname' itself is a valid IP address and is
@@ -89,7 +89,7 @@ class ExecutorGroup {
/// nullptr if it's not found. The returned descriptor should not be retained beyond the
/// lifetime of this ExecutorGroup and the caller must make sure that the group does not
/// change while it holds the pointer.
- const TBackendDescriptor* LookUpBackendDesc(const TNetworkAddress& host) const;
+ const BackendDescriptorPB* LookUpBackendDesc(const NetworkAddressPB& host) const;
/// Returns the hash ring associated with this executor group. It's owned by the group
/// and the caller must not hold a reference beyond the groups lifetime.
@@ -114,7 +114,7 @@ class ExecutorGroup {
/// Finds the first executor group in 'be_desc.executor_groups' and validates its target
/// size. Returns true if a group is found and its min size matches, or if no group is
/// found. Returns false and logs a warning otherwise.
- bool CheckConsistencyOrWarn(const TBackendDescriptor& be_desc) const;
+ bool CheckConsistencyOrWarn(const BackendDescriptorPB& be_desc) const;
const std::string name_;
diff --git a/be/src/scheduling/query-schedule.cc b/be/src/scheduling/query-schedule.cc
index 224792a..11bcb0c 100644
--- a/be/src/scheduling/query-schedule.cc
+++ b/be/src/scheduling/query-schedule.cc
@@ -275,7 +275,7 @@ bool QuerySchedule::UseDedicatedCoordEstimates() const {
for (auto& itr : per_backend_exec_params_) {
if (!itr.second.is_coord_backend) continue;
auto& coord = itr.second;
- bool is_dedicated_coord = !coord.be_desc.is_executor;
+ bool is_dedicated_coord = !coord.be_desc.is_executor();
bool only_coord_fragment_scheduled =
RequiresCoordinatorFragment() && coord.instance_params.size() == 1;
bool no_fragment_scheduled = coord.instance_params.size() == 0;
diff --git a/be/src/scheduling/query-schedule.h b/be/src/scheduling/query-schedule.h
index b2d5d65..f0ee98a 100644
--- a/be/src/scheduling/query-schedule.h
+++ b/be/src/scheduling/query-schedule.h
@@ -21,19 +21,16 @@
#include <vector>
#include <string>
#include <unordered_map>
-#include <boost/scoped_ptr.hpp>
#include "common/global-types.h"
-#include "common/status.h"
-#include "util/promise.h"
+#include "gen-cpp/Frontend_types.h"
+#include "gen-cpp/Types_types.h" // for TNetworkAddress
+#include "gen-cpp/statestore_service.pb.h"
#include "util/container-util.h"
#include "util/runtime-profile.h"
-#include "gen-cpp/Types_types.h" // for TNetworkAddress
-#include "gen-cpp/Frontend_types.h"
namespace impala {
-class Coordinator;
struct FragmentExecParams;
struct FInstanceExecParams;
@@ -51,7 +48,7 @@ typedef std::unordered_map<TNetworkAddress, PerNodeScanRanges>
/// via QuerySchedule::set_per_backend_exec_params(). Used as an input to
/// AdmissionController and a BackendState.
struct BackendExecParams {
- TBackendDescriptor be_desc;
+ BackendDescriptorPB be_desc;
/// The fragment instance params assigned to this backend. All instances of a
/// particular fragment are contiguous in this vector. Query lifetime;
diff --git a/be/src/scheduling/scheduler-test-util.cc b/be/src/scheduling/scheduler-test-util.cc
index 37ad463..132415f 100644
--- a/be/src/scheduling/scheduler-test-util.cc
+++ b/be/src/scheduling/scheduler-test-util.cc
@@ -93,16 +93,16 @@ const int64_t FileSplitGeneratorSpec::DEFAULT_FILE_SIZE = 1 << 22;
const int64_t FileSplitGeneratorSpec::DEFAULT_BLOCK_SIZE = 1 << 20;
ClusterMembershipMgr::BeDescSharedPtr BuildBackendDescriptor(const Host& host) {
- auto be_desc = make_shared<TBackendDescriptor>();
- be_desc->address.hostname = host.ip;
- be_desc->address.port = host.be_port;
- be_desc->ip_address = host.ip;
- be_desc->__set_is_coordinator(host.is_coordinator);
- be_desc->__set_is_executor(host.is_executor);
- be_desc->is_quiescing = false;
- be_desc->executor_groups.push_back(TExecutorGroupDesc());
- be_desc->executor_groups.back().name = ImpalaServer::DEFAULT_EXECUTOR_GROUP_NAME;
- be_desc->executor_groups.back().min_size = 1;
+ auto be_desc = make_shared<BackendDescriptorPB>();
+ be_desc->mutable_address()->set_hostname(host.ip);
+ be_desc->mutable_address()->set_port(host.be_port);
+ be_desc->set_ip_address(host.ip);
+ be_desc->set_is_coordinator(host.is_coordinator);
+ be_desc->set_is_executor(host.is_executor);
+ be_desc->set_is_quiescing(false);
+ ExecutorGroupDescPB* exec_desc = be_desc->add_executor_groups();
+ exec_desc->set_name(ImpalaServer::DEFAULT_EXECUTOR_GROUP_NAME);
+ exec_desc->set_min_size(1);
return be_desc;
}
@@ -713,9 +713,8 @@ void SchedulerWrapper::AddHostToTopicDelta(const Host& host, TTopicDelta* delta)
// Build topic item.
TTopicItem item;
item.key = host.ip;
- ThriftSerializer serializer(false);
- Status status = serializer.SerializeToString(be_desc.get(), &item.value);
- DCHECK(status.ok());
+ bool success = be_desc->SerializeToString(&item.value);
+ DCHECK(success);
// Add to topic delta.
delta->topic_entries.push_back(item);
diff --git a/be/src/scheduling/scheduler.cc b/be/src/scheduling/scheduler.cc
index 00af384..6c29db2 100644
--- a/be/src/scheduling/scheduler.cc
+++ b/be/src/scheduling/scheduler.cc
@@ -17,26 +17,28 @@
#include "scheduling/scheduler.h"
+#include <stdlib.h>
#include <algorithm>
+#include <limits>
#include <random>
#include <unordered_map>
#include <vector>
-#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string/join.hpp>
-#include <boost/bind.hpp>
-#include <boost/mem_fn.hpp>
#include <boost/unordered_set.hpp>
#include <gutil/strings/substitute.h>
#include "common/logging.h"
-#include "flatbuffers/flatbuffers.h"
-#include "gen-cpp/ImpalaInternalService_constants.h"
+#include "gen-cpp/CatalogObjects_generated.h"
+#include "gen-cpp/DataSinks_types.h"
+#include "gen-cpp/ErrorCodes_types.h"
+#include "gen-cpp/ImpalaInternalService_types.h"
#include "gen-cpp/Types_types.h"
-#include "runtime/exec-env.h"
+#include "gen-cpp/common.pb.h"
+#include "gen-cpp/statestore_service.pb.h"
+#include "scheduling/executor-group.h"
#include "scheduling/hash-ring.h"
-#include "statestore/statestore-subscriber.h"
#include "thirdparty/pcg-cpp-0.98/include/pcg_random.hpp"
-#include "util/container-util.h"
+#include "util/debug-util.h"
#include "util/flat_buffer.h"
#include "util/hash-util.h"
#include "util/metrics.h"
@@ -84,14 +86,15 @@ Scheduler::Scheduler(MetricGroup* metrics, RequestPoolService* request_pool_serv
}
}
-const TBackendDescriptor& Scheduler::LookUpBackendDesc(
+const BackendDescriptorPB& Scheduler::LookUpBackendDesc(
const ExecutorConfig& executor_config, const TNetworkAddress& host) {
- const TBackendDescriptor* desc = executor_config.group.LookUpBackendDesc(host);
+ const BackendDescriptorPB* desc =
+ executor_config.group.LookUpBackendDesc(FromTNetworkAddress(host));
if (desc == nullptr) {
// Local host may not be in executor_config's executor group if it's a dedicated
// coordinator, or if it is configured to be in a different executor group.
- const TBackendDescriptor& local_be_desc = executor_config.local_be_desc;
- DCHECK(host == local_be_desc.address);
+ const BackendDescriptorPB& local_be_desc = executor_config.local_be_desc;
+ DCHECK(host == FromNetworkAddressPB(local_be_desc.address()));
desc = &local_be_desc;
}
return *desc;
@@ -99,10 +102,10 @@ const TBackendDescriptor& Scheduler::LookUpBackendDesc(
TNetworkAddress Scheduler::LookUpKrpcHost(
const ExecutorConfig& executor_config, const TNetworkAddress& backend_host) {
- const TBackendDescriptor& backend_descriptor =
+ const BackendDescriptorPB& backend_descriptor =
LookUpBackendDesc(executor_config, backend_host);
- DCHECK(backend_descriptor.__isset.krpc_address);
- TNetworkAddress krpc_host = backend_descriptor.krpc_address;
+ DCHECK(backend_descriptor.has_krpc_address());
+ TNetworkAddress krpc_host = FromNetworkAddressPB(backend_descriptor.krpc_address());
DCHECK(IsResolvedAddress(krpc_host));
return krpc_host;
}
@@ -227,10 +230,10 @@ void Scheduler::ComputeFragmentExecParams(
dest.__set_fragment_instance_id(dest_params->instance_exec_params[i].instance_id);
const TNetworkAddress& host = dest_params->instance_exec_params[i].host;
dest.__set_thrift_backend(host);
- const TBackendDescriptor& desc = LookUpBackendDesc(executor_config, host);
- DCHECK(desc.__isset.krpc_address);
- DCHECK(IsResolvedAddress(desc.krpc_address));
- dest.__set_krpc_backend(desc.krpc_address);
+ const BackendDescriptorPB& desc = LookUpBackendDesc(executor_config, host);
+ DCHECK(desc.has_krpc_address());
+ DCHECK(IsResolvedAddress(desc.krpc_address()));
+ dest.__set_krpc_backend(FromNetworkAddressPB(desc.krpc_address()));
}
// enumerate senders consecutively;
@@ -273,17 +276,18 @@ void Scheduler::ComputeFragmentExecParams(const ExecutorConfig& executor_config,
// case 1: root fragment instance executed at coordinator
VLOG(3) << "Computing exec params for coordinator fragment "
<< fragment_params->fragment.display_name;
- const TBackendDescriptor& local_be_desc = executor_config.local_be_desc;
- const TNetworkAddress& coord = local_be_desc.address;
- DCHECK(local_be_desc.__isset.krpc_address);
- const TNetworkAddress& krpc_coord = local_be_desc.krpc_address;
+ const BackendDescriptorPB& local_be_desc = executor_config.local_be_desc;
+ const NetworkAddressPB& coord = local_be_desc.address();
+ DCHECK(local_be_desc.has_krpc_address());
+ const NetworkAddressPB& krpc_coord = local_be_desc.krpc_address();
DCHECK(IsResolvedAddress(krpc_coord));
// make sure that the coordinator instance ends up with instance idx 0
TUniqueId instance_id = fragment_params->is_coord_fragment
? schedule->query_id()
: schedule->GetNextInstanceId();
- fragment_params->instance_exec_params.emplace_back(
- instance_id, coord, krpc_coord, 0, *fragment_params);
+ fragment_params->instance_exec_params.emplace_back(instance_id,
+ FromNetworkAddressPB(coord), FromNetworkAddressPB(krpc_coord), 0,
+ *fragment_params);
FInstanceExecParams& instance_params = fragment_params->instance_exec_params.back();
// That instance gets all of the scan ranges, if there are any.
@@ -587,7 +591,7 @@ Status Scheduler::ComputeScanRangeAssignment(const ExecutorConfig& executor_conf
// TODO: Build this one from executor_group
ExecutorGroup coord_only_executor_group("coordinator-only-group");
- const TBackendDescriptor& local_be_desc = executor_config.local_be_desc;
+ const BackendDescriptorPB& local_be_desc = executor_config.local_be_desc;
coord_only_executor_group.AddExecutor(local_be_desc);
VLOG_QUERY << "Exec at coord is " << (exec_at_coord ? "true" : "false");
AssignmentCtx assignment_ctx(
@@ -605,7 +609,7 @@ Status Scheduler::ComputeScanRangeAssignment(const ExecutorConfig& executor_conf
// Select executor for the current scan range.
if (exec_at_coord) {
DCHECK(assignment_ctx.executor_group().LookUpExecutorIp(
- local_be_desc.address.hostname, nullptr));
+ local_be_desc.address().hostname(), nullptr));
assignment_ctx.RecordScanRangeAssignment(local_be_desc, node_id, host_list,
scan_range_locations, assignment);
} else {
@@ -668,7 +672,7 @@ Status Scheduler::ComputeScanRangeAssignment(const ExecutorConfig& executor_conf
const IpAddr* executor_ip = nullptr;
executor_ip = assignment_ctx.SelectExecutorFromCandidates(
executor_candidates, decide_local_assignment_by_rank);
- TBackendDescriptor executor;
+ BackendDescriptorPB executor;
assignment_ctx.SelectExecutorOnHost(*executor_ip, &executor);
assignment_ctx.RecordScanRangeAssignment(
executor, node_id, host_list, scan_range_locations, assignment);
@@ -699,7 +703,7 @@ Status Scheduler::ComputeScanRangeAssignment(const ExecutorConfig& executor_conf
} else {
executor_ip = assignment_ctx.SelectRemoteExecutor();
}
- TBackendDescriptor executor;
+ BackendDescriptorPB executor;
assignment_ctx.SelectExecutorOnHost(*executor_ip, &executor);
assignment_ctx.RecordScanRangeAssignment(
executor, node_id, host_list, *scan_range_locations, assignment);
@@ -749,7 +753,7 @@ std::vector<TPlanNodeId> Scheduler::FindScanNodes(const TPlan& plan) {
return FindNodes(plan, SCAN_NODE_TYPES);
}
-void Scheduler::GetScanHosts(const TBackendDescriptor& local_be_desc,
+void Scheduler::GetScanHosts(const BackendDescriptorPB& local_be_desc,
const vector<TPlanNodeId>& scan_ids, const FragmentExecParams& params,
vector<TNetworkAddress>* scan_hosts) {
for (const TPlanNodeId& scan_id : scan_ids) {
@@ -767,7 +771,7 @@ void Scheduler::GetScanHosts(const TBackendDescriptor& local_be_desc,
// TODO: we'll need to revisit this strategy once we can partition joins
// (in which case this fragment might be executing a right outer join
// with a large build table)
- scan_hosts->push_back(local_be_desc.address);
+ scan_hosts->push_back(FromNetworkAddressPB(local_be_desc.address()));
}
}
}
@@ -839,15 +843,16 @@ void Scheduler::ComputeBackendExecParams(
// This also ensures an entry always exists for the coordinator backend.
int64_t coord_min_reservation = 0;
- const TNetworkAddress& coord_addr = executor_config.local_be_desc.address;
- BackendExecParams& coord_be_params = per_backend_params[coord_addr];
+ const NetworkAddressPB& coord_addr = executor_config.local_be_desc.address();
+ BackendExecParams& coord_be_params =
+ per_backend_params[FromNetworkAddressPB(coord_addr)];
coord_be_params.is_coord_backend = true;
coord_min_reservation = coord_be_params.min_mem_reservation_bytes;
int64_t largest_min_reservation = 0;
for (auto& backend : per_backend_params) {
const TNetworkAddress& host = backend.first;
- const TBackendDescriptor be_desc = LookUpBackendDesc(executor_config, host);
+ const BackendDescriptorPB be_desc = LookUpBackendDesc(executor_config, host);
backend.second.be_desc = be_desc;
if (!backend.second.is_coord_backend) {
largest_min_reservation =
@@ -1015,7 +1020,7 @@ int Scheduler::AssignmentCtx::GetExecutorRank(const IpAddr& ip) const {
}
void Scheduler::AssignmentCtx::SelectExecutorOnHost(
- const IpAddr& executor_ip, TBackendDescriptor* executor) {
+ const IpAddr& executor_ip, BackendDescriptorPB* executor) {
DCHECK(executor_group_.LookUpExecutorIp(executor_ip, nullptr));
const ExecutorGroup::Executors& executors_on_host =
executor_group_.GetExecutorsForHost(executor_ip);
@@ -1027,10 +1032,10 @@ void Scheduler::AssignmentCtx::SelectExecutorOnHost(
next_executor_on_host =
FindOrInsert(&next_executor_per_host_, executor_ip, executors_on_host.begin());
auto eq = [next_executor_on_host](auto& elem) {
- const TBackendDescriptor& next_executor = **next_executor_on_host;
+ const BackendDescriptorPB& next_executor = **next_executor_on_host;
// The IP addresses must already match, so it is sufficient to check the port.
- DCHECK_EQ(next_executor.ip_address, elem.ip_address);
- return next_executor.address.port == elem.address.port;
+ DCHECK_EQ(next_executor.ip_address(), elem.ip_address());
+ return next_executor.address().port() == elem.address().port();
};
DCHECK(find_if(executors_on_host.begin(), executors_on_host.end(), eq)
!= executors_on_host.end());
@@ -1044,7 +1049,7 @@ void Scheduler::AssignmentCtx::SelectExecutorOnHost(
}
void Scheduler::AssignmentCtx::RecordScanRangeAssignment(
- const TBackendDescriptor& executor, PlanNodeId node_id,
+ const BackendDescriptorPB& executor, PlanNodeId node_id,
const vector<TNetworkAddress>& host_list,
const TScanRangeLocationList& scan_range_locations,
FragmentScanRangeAssignment* assignment) {
@@ -1058,7 +1063,8 @@ void Scheduler::AssignmentCtx::RecordScanRangeAssignment(
}
IpAddr executor_ip;
- bool ret = executor_group_.LookUpExecutorIp(executor.address.hostname, &executor_ip);
+ bool ret =
+ executor_group_.LookUpExecutorIp(executor.address().hostname(), &executor_ip);
DCHECK(ret);
DCHECK(!executor_ip.empty());
assignment_heap_.InsertOrUpdate(
@@ -1096,8 +1102,8 @@ void Scheduler::AssignmentCtx::RecordScanRangeAssignment(
if (!remote_read) total_local_assignments_->Increment(1);
}
- PerNodeScanRanges* scan_ranges =
- FindOrInsert(assignment, executor.address, PerNodeScanRanges());
+ PerNodeScanRanges* scan_ranges = FindOrInsert(
+ assignment, FromNetworkAddressPB(executor.address()), PerNodeScanRanges());
vector<TScanRangeParams>* scan_range_params_list =
FindOrInsert(scan_ranges, node_id, vector<TScanRangeParams>());
// Add scan range.
@@ -1109,8 +1115,7 @@ void Scheduler::AssignmentCtx::RecordScanRangeAssignment(
scan_range_params_list->push_back(scan_range_params);
if (VLOG_FILE_IS_ON) {
- VLOG_FILE << "Scheduler assignment to executor: "
- << TNetworkAddressToString(executor.address) << "("
+ VLOG_FILE << "Scheduler assignment to executor: " << executor.address() << "("
<< (remote_read ? "remote" : "local") << " selection)";
}
}
diff --git a/be/src/scheduling/scheduler.h b/be/src/scheduling/scheduler.h
index b5910c3..a620f6c 100644
--- a/be/src/scheduling/scheduler.h
+++ b/be/src/scheduling/scheduler.h
@@ -17,8 +17,6 @@
#pragma once
-#include <list>
-#include <mutex>
#include <string>
#include <vector>
#include <boost/heap/binomial_heap.hpp>
@@ -27,21 +25,23 @@
#include "common/global-types.h"
#include "common/status.h"
-#include "gen-cpp/CatalogObjects_generated.h"
#include "gen-cpp/PlanNodes_types.h"
-#include "gen-cpp/StatestoreService_types.h"
-#include "rapidjson/document.h"
-#include "rpc/thrift-util.h"
+#include "gen-cpp/Types_types.h"
#include "scheduling/executor-group.h"
#include "scheduling/query-schedule.h"
-#include "scheduling/request-pool-service.h"
-#include "statestore/statestore-subscriber.h"
#include "util/metrics-fwd.h"
#include "util/network-util.h"
#include "util/runtime-profile.h"
namespace impala {
+class BackendDescriptorPB;
+class MetricGroup;
+class RequestPoolService;
+class TPlanExecInfo;
+class TQueryOptions;
+class TScanRangeLocationList;
+
namespace test {
class SchedulerWrapper;
}
@@ -64,7 +64,7 @@ class Scheduler {
/// Current snapshot of executors to be used for scheduling a scan.
struct ExecutorConfig {
const ExecutorGroup& group;
- const TBackendDescriptor& local_be_desc;
+ const BackendDescriptorPB& local_be_desc;
};
/// Populates given query schedule and assigns fragments to hosts based on scan
@@ -178,15 +178,15 @@ class Scheduler {
const IpAddr* GetNextUnusedExecutorAndIncrement();
/// Pick an executor in round-robin fashion from multiple executors on a single host.
- void SelectExecutorOnHost(const IpAddr& executor_ip, TBackendDescriptor* executor);
+ void SelectExecutorOnHost(const IpAddr& executor_ip, BackendDescriptorPB* executor);
/// Build a new TScanRangeParams object and append it to the assignment list for the
/// tuple (executor, node_id) in 'assignment'. Also, update assignment_heap_ and
/// assignment_byte_counters_, increase the counters 'total_assignments_' and
/// 'total_local_assignments_'. 'scan_range_locations' contains information about the
/// scan range and its replica locations.
- void RecordScanRangeAssignment(const TBackendDescriptor& executor, PlanNodeId node_id,
- const vector<TNetworkAddress>& host_list,
+ void RecordScanRangeAssignment(const BackendDescriptorPB& executor,
+ PlanNodeId node_id, const vector<TNetworkAddress>& host_list,
const TScanRangeLocationList& scan_range_locations,
FragmentScanRangeAssignment* assignment);
@@ -260,7 +260,7 @@ class Scheduler {
/// Returns the backend descriptor corresponding to 'host' which could be a remote
/// backend or the local host itself. The returned descriptor should not be retained
/// beyond the lifetime of 'executor_config'.
- const TBackendDescriptor& LookUpBackendDesc(
+ const BackendDescriptorPB& LookUpBackendDesc(
const ExecutorConfig& executor_config, const TNetworkAddress& host);
/// Returns the KRPC host in 'executor_config' based on the thrift backend address
@@ -406,7 +406,7 @@ class Scheduler {
/// Add all hosts that the scans identified by 'scan_ids' are executed on to
/// 'scan_hosts'.
- void GetScanHosts(const TBackendDescriptor& local_be_desc,
+ void GetScanHosts(const BackendDescriptorPB& local_be_desc,
const std::vector<TPlanNodeId>& scan_ids, const FragmentExecParams& params,
std::vector<TNetworkAddress>* scan_hosts);
diff --git a/be/src/service/cancellation-work.h b/be/src/service/cancellation-work.h
index f4a8727..4bf2586 100644
--- a/be/src/service/cancellation-work.h
+++ b/be/src/service/cancellation-work.h
@@ -50,8 +50,8 @@ class CancellationWork {
}
// Construct a BACKEND_FAILURE CancellationWork instance.
- static CancellationWork BackendFailure(const TUniqueId& query_id,
- const std::vector<TNetworkAddress>& failed_backends) {
+ static CancellationWork BackendFailure(
+ const TUniqueId& query_id, const std::vector<NetworkAddressPB>& failed_backends) {
return CancellationWork(query_id, CancellationWorkCause::BACKEND_FAILED, Status::OK(),
failed_backends, false);
}
@@ -62,7 +62,7 @@ class CancellationWork {
DCHECK_ENUM_EQ(cause_, CancellationWorkCause::TERMINATED_BY_SERVER);
return error_;
}
- const std::vector<TNetworkAddress>& failed_backends() const {
+ const std::vector<NetworkAddressPB>& failed_backends() const {
DCHECK_ENUM_EQ(cause_, CancellationWorkCause::BACKEND_FAILED);
return failed_backends_;
}
@@ -70,7 +70,7 @@ class CancellationWork {
private:
CancellationWork(const TUniqueId& query_id, CancellationWorkCause cause,
- const Status& error, const std::vector<TNetworkAddress>& failed_backends,
+ const Status& error, const std::vector<NetworkAddressPB>& failed_backends,
bool unregister)
: query_id_(query_id),
cause_(cause),
@@ -93,7 +93,7 @@ class CancellationWork {
// If cause is BACKEND_FAILED, all of the backend that were detected to fail. Otherwise
// not used.
- std::vector<TNetworkAddress> failed_backends_;
+ std::vector<NetworkAddressPB> failed_backends_;
// If true, unregister the query after cancelling it.
bool unregister_;
diff --git a/be/src/service/impala-http-handler.cc b/be/src/service/impala-http-handler.cc
index cff8a69..aa27a69 100644
--- a/be/src/service/impala-http-handler.cc
+++ b/be/src/service/impala-http-handler.cc
@@ -508,7 +508,7 @@ void ImpalaHttpHandler::QueryStateHandler(const Webserver::WebRequest& req,
for (const ImpalaServer::QueryLocations::value_type& location :
server_->query_locations_) {
Value location_json(kObjectType);
- Value location_name(TNetworkAddressToString(location.second.address).c_str(),
+ Value location_name(NetworkAddressPBToString(location.second.address).c_str(),
document->GetAllocator());
location_json.AddMember("location", location_name, document->GetAllocator());
Value backend_id_str(PrintId(location.first).c_str(), document->GetAllocator());
@@ -946,33 +946,34 @@ void ImpalaHttpHandler::BackendsHandler(const Webserver::WebRequest& req,
cluster_membership_mgr->GetSnapshot();
DCHECK(membership_snapshot.get() != nullptr);
for (const auto& entry : membership_snapshot->current_backends) {
- TBackendDescriptor backend = entry.second;
+ BackendDescriptorPB backend = entry.second;
Value backend_obj(kObjectType);
- string address = TNetworkAddressToString(backend.address);
+ string address = NetworkAddressPBToString(backend.address());
Value str(address.c_str(), document->GetAllocator());
- Value krpc_address(
- TNetworkAddressToString(backend.krpc_address).c_str(), document->GetAllocator());
+ Value krpc_address(NetworkAddressPBToString(backend.krpc_address()).c_str(),
+ document->GetAllocator());
backend_obj.AddMember("address", str, document->GetAllocator());
backend_obj.AddMember("krpc_address", krpc_address, document->GetAllocator());
- Value backend_id_str(PrintId(backend.backend_id).c_str(), document->GetAllocator());
+ Value backend_id_str(PrintId(backend.backend_id()).c_str(), document->GetAllocator());
backend_obj.AddMember("backend_id", backend_id_str, document->GetAllocator());
string webserver_url =
- Substitute("$0://$1", backend.secure_webserver ? "https" : "http",
- TNetworkAddressToString(backend.debug_http_address));
+ Substitute("$0://$1", backend.secure_webserver() ? "https" : "http",
+ NetworkAddressPBToString(backend.debug_http_address()));
Value webserver_url_val(webserver_url.c_str(), document->GetAllocator());
backend_obj.AddMember("webserver_url", webserver_url_val, document->GetAllocator());
- backend_obj.AddMember("is_coordinator", backend.is_coordinator,
- document->GetAllocator());
- backend_obj.AddMember("is_executor", backend.is_executor, document->GetAllocator());
- backend_obj.AddMember("is_quiescing", backend.is_quiescing, document->GetAllocator());
+ backend_obj.AddMember(
+ "is_coordinator", backend.is_coordinator(), document->GetAllocator());
+ backend_obj.AddMember("is_executor", backend.is_executor(), document->GetAllocator());
+ backend_obj.AddMember(
+ "is_quiescing", backend.is_quiescing(), document->GetAllocator());
Status blacklist_cause;
int64_t blacklist_time_remaining_ms;
bool is_blacklisted = membership_snapshot->executor_blacklist.IsBlacklisted(
backend, &blacklist_cause, &blacklist_time_remaining_ms);
backend_obj.AddMember("is_blacklisted", is_blacklisted, document->GetAllocator());
- backend_obj.AddMember(
- "is_active", !is_blacklisted && !backend.is_quiescing, document->GetAllocator());
- if (backend.is_quiescing) {
+ backend_obj.AddMember("is_active", !is_blacklisted && !backend.is_quiescing(),
+ document->GetAllocator());
+ if (backend.is_quiescing()) {
// Backends cannot be both blacklisted and quiescing.
DCHECK(!is_blacklisted);
++num_quiescing_backends;
@@ -990,7 +991,7 @@ void ImpalaHttpHandler::BackendsHandler(const Webserver::WebRequest& req,
} else {
++num_active_backends;
}
- Value admit_mem_limit(PrettyPrinter::PrintBytes(backend.admit_mem_limit).c_str(),
+ Value admit_mem_limit(PrettyPrinter::PrintBytes(backend.admit_mem_limit()).c_str(),
document->GetAllocator());
backend_obj.AddMember("admit_mem_limit", admit_mem_limit, document->GetAllocator());
// If the host address does not exist in the 'host_stats', this would ensure that a
@@ -1001,14 +1002,16 @@ void ImpalaHttpHandler::BackendsHandler(const Webserver::WebRequest& req,
Value mem_admitted(PrettyPrinter::PrintBytes(
host_stats[address].mem_admitted).c_str(), document->GetAllocator());
backend_obj.AddMember("mem_admitted", mem_admitted, document->GetAllocator());
- backend_obj.AddMember("admission_slots", backend.admission_slots,
- document->GetAllocator());
+ backend_obj.AddMember(
+ "admission_slots", backend.admission_slots(), document->GetAllocator());
backend_obj.AddMember("num_admitted", host_stats[address].num_admitted,
document->GetAllocator());
backend_obj.AddMember("admission_slots_in_use", host_stats[address].slots_in_use,
document->GetAllocator());
vector<string> group_names;
- for (const auto& group : backend.executor_groups) group_names.push_back(group.name);
+ for (const auto& group : backend.executor_groups()) {
+ group_names.push_back(group.name());
+ }
Value executor_groups(JoinStrings(group_names, ", ").c_str(),
document->GetAllocator());
backend_obj.AddMember("executor_groups", executor_groups, document->GetAllocator());
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 4b420db..f57d776 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -307,8 +307,7 @@ DECLARE_bool(compact_catalog_topic);
namespace {
using namespace impala;
-vector<impala::TExecutorGroupDesc> GetExecutorGroups(const string& flag) {
- vector<impala::TExecutorGroupDesc> result;
+void SetExecutorGroups(const string& flag, BackendDescriptorPB* be_desc) {
vector<StringPiece> groups;
groups = Split(flag, ",", SkipEmpty());
if (groups.empty()) groups.push_back(ImpalaServer::DEFAULT_EXECUTOR_GROUP_NAME);
@@ -316,21 +315,20 @@ vector<impala::TExecutorGroupDesc> GetExecutorGroups(const string& flag) {
// Name and optional minimum group size are separated by ':'.
for (const StringPiece& group : groups) {
int colon_idx = group.find_first_of(':');
- TExecutorGroupDesc group_desc;
- group_desc.name = group.substr(0, colon_idx).as_string();
- group_desc.min_size = 1;
+ ExecutorGroupDescPB* group_desc = be_desc->add_executor_groups();
+ group_desc->set_name(group.substr(0, colon_idx).as_string());
if (colon_idx != StringPiece::npos) {
StringParser::ParseResult result;
- group_desc.min_size = StringParser::StringToInt<int64_t>(
- group.data() + colon_idx + 1, group.length() - colon_idx - 1, &result);
+ group_desc->set_min_size(StringParser::StringToInt<int64_t>(
+ group.data() + colon_idx + 1, group.length() - colon_idx - 1, &result));
if (result != StringParser::PARSE_SUCCESS) {
LOG(FATAL) << "Failed to parse minimum executor group size from group: "
<< group.ToString();
}
+ } else {
+ group_desc->set_min_size(1);
}
- result.push_back(group_desc);
}
- return result;
}
} // end anonymous namespace
@@ -1209,7 +1207,7 @@ void ImpalaServer::CloseClientRequestState(
// node to fail was last sender to an exchange, the coordinator will realise and
// fail the query at the same time the failure detection path does the same
// thing. They will harmlessly race to remove the query from this map.
- auto it = query_locations_.find(entry.second.be_desc.backend_id);
+ auto it = query_locations_.find(entry.second.be_desc.backend_id());
if (it != query_locations_.end()) {
it->second.query_ids.erase(request_state->query_id());
}
@@ -1453,7 +1451,7 @@ void ImpalaServer::CancelFromThreadPool(uint32_t thread_id,
case CancellationWorkCause::BACKEND_FAILED: {
// We only want to proceed with cancellation if the backends are still in use for
// the query.
- vector<TNetworkAddress> active_backends;
+ vector<NetworkAddressPB> active_backends;
Coordinator* coord = request_state->GetCoordinator();
if (coord == nullptr) {
// Query hasn't started yet - it still will run on all backends.
@@ -1468,7 +1466,7 @@ void ImpalaServer::CancelFromThreadPool(uint32_t thread_id,
}
stringstream msg;
for (int i = 0; i < active_backends.size(); ++i) {
- msg << TNetworkAddressToString(active_backends[i]);
+ msg << active_backends[i];
if (i + 1 != active_backends.size()) msg << ", ";
}
error = Status::Expected(TErrorCode::UNREACHABLE_IMPALADS, msg.str());
@@ -1745,11 +1743,11 @@ void ImpalaServer::RegisterQueryLocations(
if (!per_backend_params.empty()) {
lock_guard<mutex> l(query_locations_lock_);
for (const auto& entry : per_backend_params) {
- const TBackendId& backend_id = entry.second.be_desc.backend_id;
+ const BackendIdPB& backend_id = entry.second.be_desc.backend_id();
auto it = query_locations_.find(backend_id);
if (it == query_locations_.end()) {
query_locations_.emplace(
- backend_id, QueryLocationInfo(entry.second.be_desc.address, query_id));
+ backend_id, QueryLocationInfo(entry.second.be_desc.address(), query_id));
} else {
it->second.query_ids.insert(query_id);
}
@@ -1758,13 +1756,13 @@ void ImpalaServer::RegisterQueryLocations(
}
void ImpalaServer::CancelQueriesOnFailedBackends(
- const std::unordered_set<TBackendId>& current_membership) {
+ const std::unordered_set<BackendIdPB>& current_membership) {
// Maps from query id (to be cancelled) to a list of failed Impalads that are
// the cause of the cancellation. Note that we don't need to use TBackendIds as a single
// query can't be scheduled on two backends with the same TNetworkAddress so there's no
// ambiguity, and passing the TNetworkAddresses into the CancellationWork makes them
// available for generating a user-friendly error message.
- map<TUniqueId, vector<TNetworkAddress>> queries_to_cancel;
+ map<TUniqueId, vector<NetworkAddressPB>> queries_to_cancel;
{
// Build a list of queries that are running on failed hosts (as evidenced by their
// absence from the membership list).
@@ -1795,7 +1793,7 @@ void ImpalaServer::CancelQueriesOnFailedBackends(
for (const auto& cancellation_entry : queries_to_cancel) {
stringstream backends_ss;
for (int i = 0; i < cancellation_entry.second.size(); ++i) {
- backends_ss << TNetworkAddressToString(cancellation_entry.second[i]);
+ backends_ss << cancellation_entry.second[i];
if (i + 1 != cancellation_entry.second.size()) backends_ss << ", ";
}
VLOG_QUERY << "Backends failed for query " << PrintId(cancellation_entry.first)
@@ -1806,54 +1804,54 @@ void ImpalaServer::CancelQueriesOnFailedBackends(
}
}
-std::shared_ptr<const TBackendDescriptor> ImpalaServer::GetLocalBackendDescriptor() {
+std::shared_ptr<const BackendDescriptorPB> ImpalaServer::GetLocalBackendDescriptor() {
if (!services_started_.load()) return nullptr;
lock_guard<mutex> l(local_backend_descriptor_lock_);
// Check if the current backend descriptor needs to be initialized.
if (local_backend_descriptor_.get() == nullptr) {
- std::shared_ptr<TBackendDescriptor> new_be_desc =
- std::make_shared<TBackendDescriptor>();
+ std::shared_ptr<BackendDescriptorPB> new_be_desc =
+ std::make_shared<BackendDescriptorPB>();
BuildLocalBackendDescriptorInternal(new_be_desc.get());
local_backend_descriptor_ = new_be_desc;
}
// Check to see if it needs to be updated.
- if (IsShuttingDown() != local_backend_descriptor_->is_quiescing) {
- std::shared_ptr<TBackendDescriptor> new_be_desc =
- std::make_shared<TBackendDescriptor>(*local_backend_descriptor_);
- new_be_desc->is_quiescing = IsShuttingDown();
+ if (IsShuttingDown() != local_backend_descriptor_->is_quiescing()) {
+ std::shared_ptr<BackendDescriptorPB> new_be_desc =
+ std::make_shared<BackendDescriptorPB>(*local_backend_descriptor_);
+ new_be_desc->set_is_quiescing(IsShuttingDown());
local_backend_descriptor_ = new_be_desc;
}
return local_backend_descriptor_;
}
-void ImpalaServer::BuildLocalBackendDescriptorInternal(TBackendDescriptor* be_desc) {
+void ImpalaServer::BuildLocalBackendDescriptorInternal(BackendDescriptorPB* be_desc) {
DCHECK(services_started_.load());
bool is_quiescing = shutting_down_.Load() != 0;
- be_desc->__set_backend_id(exec_env_->backend_id());
- be_desc->__set_address(exec_env_->GetThriftBackendAddress());
- be_desc->__set_ip_address(exec_env_->ip_address());
- be_desc->__set_is_coordinator(FLAGS_is_coordinator);
- be_desc->__set_is_executor(FLAGS_is_executor);
+ *be_desc->mutable_backend_id() = exec_env_->backend_id();
+ *be_desc->mutable_address() = FromTNetworkAddress(exec_env_->GetThriftBackendAddress());
+ be_desc->set_ip_address(exec_env_->ip_address());
+ be_desc->set_is_coordinator(FLAGS_is_coordinator);
+ be_desc->set_is_executor(FLAGS_is_executor);
Webserver* webserver = ExecEnv::GetInstance()->webserver();
if (webserver != nullptr) {
- be_desc->__set_debug_http_address(
- MakeNetworkAddress(webserver->hostname(), webserver->port()));
- be_desc->__set_secure_webserver(webserver->IsSecure());
+ *be_desc->mutable_debug_http_address() =
+ MakeNetworkAddressPB(webserver->hostname(), webserver->port());
+ be_desc->set_secure_webserver(webserver->IsSecure());
}
const TNetworkAddress& krpc_address = exec_env_->krpc_address();
DCHECK(IsResolvedAddress(krpc_address));
- be_desc->__set_krpc_address(krpc_address);
+ *be_desc->mutable_krpc_address() = FromTNetworkAddress(krpc_address);
- be_desc->__set_admit_mem_limit(exec_env_->admit_mem_limit());
- be_desc->__set_admission_slots(exec_env_->admission_slots());
- be_desc->__set_is_quiescing(is_quiescing);
- be_desc->executor_groups = GetExecutorGroups(FLAGS_executor_groups);
+ be_desc->set_admit_mem_limit(exec_env_->admit_mem_limit());
+ be_desc->set_admission_slots(exec_env_->admission_slots());
+ be_desc->set_is_quiescing(is_quiescing);
+ SetExecutorGroups(FLAGS_executor_groups, be_desc);
}
ImpalaServer::QueryStateRecord::QueryStateRecord(
@@ -2289,14 +2287,14 @@ void ImpalaServer::UnregisterSessionTimeout(int32_t session_timeout) {
[&](const std::shared_ptr<ClientRequestState>& request_state) {
Coordinator* coord = request_state->GetCoordinator();
if (coord != nullptr) {
- TNetworkAddress address;
+ NetworkAddressPB address;
int64_t lag_time_ms = coord->GetMaxBackendStateLagMs(&address);
if (lag_time_ms > max_lag_ms) {
to_cancel.push_back(
CancellationWork::TerminatedByServer(request_state->query_id(),
Status(TErrorCode::UNRESPONSIVE_BACKEND,
PrintId(request_state->query_id()),
- TNetworkAddressToString(address), lag_time_ms, max_lag_ms),
+ NetworkAddressPBToString(address), lag_time_ms, max_lag_ms),
false /* unregister */));
}
}
diff --git a/be/src/service/impala-server.h b/be/src/service/impala-server.h
index 1396259..2449af7 100644
--- a/be/src/service/impala-server.h
+++ b/be/src/service/impala-server.h
@@ -456,7 +456,7 @@ class ImpalaServer : public ImpalaServiceIf,
int GetHS2Port();
/// Returns a current snapshot of the local backend descriptor.
- std::shared_ptr<const TBackendDescriptor> GetLocalBackendDescriptor();
+ std::shared_ptr<const BackendDescriptorPB> GetLocalBackendDescriptor();
/// Adds the query_id to the map from backend to the list of queries running or expected
/// to run there (query_locations_). After calling this function, the server will cancel
@@ -467,7 +467,7 @@ class ImpalaServer : public ImpalaServiceIf,
/// Takes a set of backend ids of active backends and cancels all the queries running on
/// failed ones (that is, ids not in the active set).
void CancelQueriesOnFailedBackends(
- const std::unordered_set<TBackendId>& current_membership);
+ const std::unordered_set<BackendIdPB>& current_membership);
/// Start the shutdown process. Return an error if it could not be started. Otherwise,
/// if it was successfully started by this or a previous call, return OK along with
@@ -816,7 +816,7 @@ class ImpalaServer : public ImpalaServiceIf,
WARN_UNUSED_RESULT;
/// Initializes the backend descriptor in 'be_desc' with the local backend information.
- void BuildLocalBackendDescriptorInternal(TBackendDescriptor* be_desc);
+ void BuildLocalBackendDescriptorInternal(BackendDescriptorPB* be_desc);
/// Snapshot of a query's state, archived in the query log. Not mutated after
/// construction.
@@ -1298,13 +1298,13 @@ class ImpalaServer : public ImpalaServiceIf,
/// Entries in the 'query_locations' map.
struct QueryLocationInfo {
- QueryLocationInfo(TNetworkAddress address, TUniqueId query_id) : address(address) {
+ QueryLocationInfo(NetworkAddressPB address, TUniqueId query_id) : address(address) {
query_ids.insert(query_id);
}
/// Used for logging and error messages so that users don't have to translate between
- /// the TBackendId and a hostname themselves.
- TNetworkAddress address;
+ /// the BackendIdPB and a hostname themselves.
+ NetworkAddressPB address;
/// Queries currently running or expected to run at this location.
std::unordered_set<TUniqueId> query_ids;
@@ -1312,12 +1312,12 @@ class ImpalaServer : public ImpalaServiceIf,
/// Contains info about what queries are running on each backend, so that they can be
/// cancelled if the backend goes down.
- typedef std::unordered_map<TBackendId, QueryLocationInfo> QueryLocations;
+ typedef std::unordered_map<BackendIdPB, QueryLocationInfo> QueryLocations;
QueryLocations query_locations_;
/// The local backend descriptor. Updated in GetLocalBackendDescriptor() and protected
/// by 'local_backend_descriptor_lock_';
- std::shared_ptr<const TBackendDescriptor> local_backend_descriptor_;
+ std::shared_ptr<const BackendDescriptorPB> local_backend_descriptor_;
std::mutex local_backend_descriptor_lock_;
/// UUID generator for session IDs and secrets. Uses system random device to get
diff --git a/be/src/statestore/CMakeLists.txt b/be/src/statestore/CMakeLists.txt
index a2bc4c0..65c26dc 100644
--- a/be/src/statestore/CMakeLists.txt
+++ b/be/src/statestore/CMakeLists.txt
@@ -22,8 +22,12 @@ set(LIBRARY_OUTPUT_PATH "${BUILD_OUTPUT_ROOT_DIRECTORY}/statestore")
# where to put generated binaries
set(EXECUTABLE_OUTPUT_PATH "${BUILD_OUTPUT_ROOT_DIRECTORY}/statestore")
+# Mark the protobuf file as generated
+set_source_files_properties(${STATESTORE_SERVICE_PROTO_SRCS} PROPERTIES GENERATED TRUE)
+
add_library(Statestore
failure-detector.cc
+ ${STATESTORE_SERVICE_PROTO_SRCS}
statestore.cc
statestore-subscriber.cc
statestored-main.cc
diff --git a/be/src/util/container-util.h b/be/src/util/container-util.h
index a4e2119..eeadf92 100644
--- a/be/src/util/container-util.h
+++ b/be/src/util/container-util.h
@@ -28,9 +28,9 @@
#include "gen-cpp/ErrorCodes_types.h"
#include "gen-cpp/Frontend_types.h"
-#include "gen-cpp/StatestoreService_types.h"
#include "gen-cpp/Status_types.h"
#include "gen-cpp/Types_types.h"
+#include "gen-cpp/common.pb.h"
/// Comparators for types that we commonly use in containers.
namespace impala {
@@ -62,6 +62,21 @@ inline bool operator<(const TUniqueId& lhs, const TUniqueId& rhs) {
return std::tie(lhs.hi, lhs.lo) < std::tie(rhs.hi, rhs.lo);
}
+// UniqueIdPB
+STATIC_ASSERT_SIZE(UniqueIdPB, 40);
+
+inline bool operator==(const UniqueIdPB& lhs, const UniqueIdPB& rhs) {
+ return lhs.hi() == rhs.hi() && lhs.lo() == rhs.lo();
+}
+
+inline bool operator!=(const UniqueIdPB& lhs, const UniqueIdPB& rhs) {
+ return !(lhs == rhs);
+}
+
+inline bool operator<(const UniqueIdPB& lhs, const UniqueIdPB& rhs) {
+ return lhs.hi() < rhs.hi() || (lhs.hi() == rhs.hi() && lhs.lo() < rhs.lo());
+}
+
// TNetworkAddress
STATIC_ASSERT_SIZE(TNetworkAddress, 24);
@@ -73,6 +88,17 @@ inline bool operator!=(const TNetworkAddress& lhs, const TNetworkAddress& rhs) {
return !(lhs == rhs);
}
+// NetworkAddressPB
+STATIC_ASSERT_SIZE(NetworkAddressPB, 40);
+
+inline bool operator==(const NetworkAddressPB& lhs, const NetworkAddressPB& rhs) {
+ return lhs.hostname() == rhs.hostname() && lhs.port() == rhs.port();
+}
+
+inline bool operator!=(const NetworkAddressPB& lhs, const NetworkAddressPB& rhs) {
+ return !(lhs == rhs);
+}
+
// TStatus
STATIC_ASSERT_SIZE(TStatus, 48);
@@ -98,6 +124,15 @@ inline std::size_t hash_value(const TNetworkAddress& host_port) {
return HashUtil::Hash(&host_port.port, sizeof(host_port.port), hash);
}
+/// Hash function for NetworkAddressPB. This function must be called hash_value to be
+/// picked up properly by boost.
+inline std::size_t hash_value(const NetworkAddressPB& host_port) {
+ uint32_t hash =
+ HashUtil::Hash(host_port.hostname().c_str(), host_port.hostname().length(), 0);
+ int32_t port = host_port.port();
+ return HashUtil::Hash(&port, sizeof(port), hash);
+}
+
} // end namespace impala
/// Hash function for std:: containers
@@ -109,6 +144,13 @@ template<> struct hash<impala::TNetworkAddress> {
}
};
+template <>
+struct hash<impala::NetworkAddressPB> {
+ std::size_t operator()(const impala::NetworkAddressPB& host_port) const {
+ return impala::hash_value(host_port);
+ }
+};
+
} // end namespace std
namespace impala {
diff --git a/be/src/util/debug-util.cc b/be/src/util/debug-util.cc
index 4e5f5a0..2e9f14a 100644
--- a/be/src/util/debug-util.cc
+++ b/be/src/util/debug-util.cc
@@ -108,6 +108,13 @@ string PrintId(const TUniqueId& id, const string& separator) {
return out.str();
}
+string PrintId(const UniqueIdPB& id, const string& separator) {
+ stringstream out;
+ // Outputting the separator string resets the stream width.
+ out << hex << setfill('0') << setw(16) << id.hi() << separator << setw(16) << id.lo();
+ return out.str();
+}
+
bool ParseId(const string& s, TUniqueId* id) {
// For backwards compatibility, this method parses two forms of query ID from text:
// - <hex-int64_t><colon><hex-int64_t> - this format is the standard going forward
diff --git a/be/src/util/debug-util.h b/be/src/util/debug-util.h
index eaaccf6..5126299 100644
--- a/be/src/util/debug-util.h
+++ b/be/src/util/debug-util.h
@@ -34,6 +34,7 @@
#include "gen-cpp/PlanNodes_types.h"
#include "gen-cpp/Types_types.h"
#include "gen-cpp/beeswax_types.h"
+#include "gen-cpp/common.pb.h"
#include "gen-cpp/parquet_types.h"
#include "gutil/macros.h"
@@ -83,6 +84,7 @@ std::string PrintBatch(RowBatch* batch);
/// Converts id to a string represantation. If necessary, the gdb equivalent is:
/// printf "%lx:%lx\n", id.hi, id.lo
std::string PrintId(const TUniqueId& id, const std::string& separator = ":");
+std::string PrintId(const UniqueIdPB& id, const std::string& separator = ":");
/// Returns the fully qualified path, e.g. "database.table.array_col.item.field"
std::string PrintPath(const TableDescriptor& tbl_desc, const SchemaPath& path);
diff --git a/be/src/util/network-util.cc b/be/src/util/network-util.cc
index d31076b..5a79182 100644
--- a/be/src/util/network-util.cc
+++ b/be/src/util/network-util.cc
@@ -120,6 +120,11 @@ bool IsResolvedAddress(const TNetworkAddress& addr) {
return sock.ParseString(addr.hostname, addr.port).ok();
}
+bool IsResolvedAddress(const NetworkAddressPB& addr) {
+ kudu::Sockaddr sock;
+ return sock.ParseString(addr.hostname(), addr.port()).ok();
+}
+
bool FindFirstNonLocalhost(const vector<string>& addresses, string* addr) {
for (const string& candidate: addresses) {
if (candidate != LOCALHOST_IP_STR) {
@@ -157,6 +162,13 @@ TNetworkAddress MakeNetworkAddress(const string& address) {
return ret;
}
+NetworkAddressPB MakeNetworkAddressPB(const string& hostname, int port) {
+ NetworkAddressPB ret;
+ ret.set_hostname(hostname);
+ ret.set_port(port);
+ return ret;
+}
+
bool IsWildcardAddress(const string& ipaddress) {
return ipaddress == "0.0.0.0";
}
@@ -180,6 +192,13 @@ TNetworkAddress FromNetworkAddressPB(const NetworkAddressPB& address) {
return t_address;
}
+NetworkAddressPB FromTNetworkAddress(const TNetworkAddress& address) {
+ NetworkAddressPB address_pb;
+ address_pb.set_hostname(address.hostname);
+ address_pb.set_port(address.port);
+ return address_pb;
+}
+
/// Pick a random port in the range of ephemeral ports
/// https://tools.ietf.org/html/rfc6335
int FindUnusedEphemeralPort() {
diff --git a/be/src/util/network-util.h b/be/src/util/network-util.h
index b6629af..cdd6975 100644
--- a/be/src/util/network-util.h
+++ b/be/src/util/network-util.h
@@ -15,11 +15,14 @@
// specific language governing permissions and limitations
// under the License.
+#pragma once
+
+#include <string>
+#include <vector>
+
#include "common/status.h"
#include "gen-cpp/common.pb.h"
-#include "gen-cpp/StatestoreService_types.h"
#include "gen-cpp/Types_types.h"
-#include <vector>
namespace kudu {
class Sockaddr;
@@ -35,6 +38,7 @@ typedef std::string IpAddr;
/// Returns true if 'addr' is a fully resolved IP address, rather than a fqdn + port.
bool IsResolvedAddress(const TNetworkAddress& addr);
+bool IsResolvedAddress(const NetworkAddressPB& addr);
/// Looks up all IP addresses associated with a given hostname and returns one of them via
/// 'address'. If the IP addresses of a host don't change, then subsequent calls will
@@ -50,8 +54,9 @@ bool FindFirstNonLocalhost(const std::vector<std::string>& addresses, std::strin
/// Returns OK if a hostname can be found, false otherwise.
Status GetHostname(std::string* hostname) WARN_UNUSED_RESULT;
-/// Utility method because Thrift does not supply useful constructors
+/// Utility methods because Thrift/protobuf do not supply useful constructors
TNetworkAddress MakeNetworkAddress(const std::string& hostname, int port);
+NetworkAddressPB MakeNetworkAddressPB(const std::string& hostname, int port);
/// Utility method to parse the given string into a network address.
/// Accepted format: "host:port" or "host". For the latter format the port is set to zero.
@@ -68,9 +73,16 @@ std::string TNetworkAddressToString(const TNetworkAddress& address);
/// Utility method to print a NetworkAddressPB as address:port.
std::string NetworkAddressPBToString(const NetworkAddressPB& address);
+inline ostream& operator<<(ostream& os, const NetworkAddressPB& address) {
+ return os << NetworkAddressPBToString(address);
+}
+
/// Utility method to convert a NetworkAddressPB to a TNetworkAddress.
TNetworkAddress FromNetworkAddressPB(const NetworkAddressPB& address);
+/// Utility method to convert a TNetworkAddress to a NetworkAddressPB.
+NetworkAddressPB FromTNetworkAddress(const TNetworkAddress& address);
+
/// Utility method to convert TNetworkAddress to Kudu sock addr.
/// Note that 'address' has to contain a resolved IP address.
Status TNetworkAddressToSockaddr(const TNetworkAddress& address,
diff --git a/be/src/util/uid-util.h b/be/src/util/uid-util.h
index 0a23f6a..8b0deed 100644
--- a/be/src/util/uid-util.h
+++ b/be/src/util/uid-util.h
@@ -34,6 +34,14 @@ inline void UUIDToTUniqueId(const boost::uuids::uuid& uuid, TUniqueId* unique_id
memcpy(&(unique_id->lo), &uuid.data[8], 8);
}
+inline void UUIDToUniqueIdPB(const boost::uuids::uuid& uuid, UniqueIdPB* unique_id) {
+ uint64_t hi, lo;
+ memcpy(&hi, &uuid.data[0], 8);
+ memcpy(&lo, &uuid.data[8], 8);
+ unique_id->set_hi(hi);
+ unique_id->set_lo(lo);
+}
+
inline void TUniqueIdToUniqueIdPB(
const TUniqueId& t_unique_id, UniqueIdPB* unique_id_pb) {
unique_id_pb->set_lo(t_unique_id.lo);
diff --git a/be/src/util/unique-id-hash.h b/be/src/util/unique-id-hash.h
index 1a73d8d..8481e4a 100644
--- a/be/src/util/unique-id-hash.h
+++ b/be/src/util/unique-id-hash.h
@@ -22,6 +22,7 @@
#include <boost/functional/hash.hpp>
#include "gen-cpp/Types_types.h" // for TUniqueId
+#include "gen-cpp/common.pb.h" // for UniqueIdPB
namespace impala {
@@ -32,6 +33,13 @@ inline std::size_t hash_value(const TUniqueId& id) {
return seed;
}
+inline std::size_t hash_value(const UniqueIdPB& id) {
+ std::size_t seed = 0;
+ boost::hash_combine(seed, id.lo());
+ boost::hash_combine(seed, id.hi());
+ return seed;
+}
+
} // namespace impala
/// Hash function for std:: containers
@@ -44,4 +52,11 @@ struct hash<impala::TUniqueId> {
}
};
+template <>
+struct hash<impala::UniqueIdPB> {
+ std::size_t operator()(const impala::UniqueIdPB& id) const {
+ return impala::hash_value(id);
+ }
+};
+
} // namespace std
diff --git a/common/protobuf/CMakeLists.txt b/common/protobuf/CMakeLists.txt
index 69a30a3..4856833 100644
--- a/common/protobuf/CMakeLists.txt
+++ b/common/protobuf/CMakeLists.txt
@@ -38,7 +38,7 @@ foreach(pb_src common row_batch)
set(${_PROTO_SRCS} ${${_PROTO_SRCS}} PARENT_SCOPE)
endforeach()
-foreach(pb_src data_stream_service control_service rpc_test)
+foreach(pb_src data_stream_service control_service rpc_test statestore_service)
string(TOUPPER ${pb_src} _PB_SRC_UPPER)
set(_PROTO_SRCS ${_PB_SRC_UPPER}_PROTO_SRCS)
set(_PROTO_HDRS ${_PB_SRC_UPPER}_PROTO_HDRS)
diff --git a/common/protobuf/statestore_service.proto b/common/protobuf/statestore_service.proto
new file mode 100644
index 0000000..ff9f7d9
--- /dev/null
+++ b/common/protobuf/statestore_service.proto
@@ -0,0 +1,79 @@
+// 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.
+
+syntax="proto2";
+
+package impala;
+
+import "common.proto";
+
+// Structure to describe an executor group. We use this to configure the executor group
+// for backends during startup and during cluster membership management.
+message ExecutorGroupDescPB {
+ // The name of the executor group.
+ optional string name = 1;
+
+ // The minimum size of the executor group to be considered healthy.
+ optional int64 min_size = 2;
+}
+
+// Structure serialised in the Impala backend topic. Each Impalad
+// constructs one BackendDescriptorPB, and registers it in the cluster-membership
+// topic. Impalads subscribe to this topic to learn of the location of
+// all other Impalads in the cluster. Impalads can act as coordinators, executors or
+// both.
+message BackendDescriptorPB {
+ // Unique identifier for this impalad. Generated on startup.
+ optional UniqueIdPB backend_id = 1;
+
+ // Network address of the thrift based ImpalaInternalService on this backend
+ optional NetworkAddressPB address = 2;
+
+ // IP address corresponding to address.hostname. Explicitly including this saves the
+ // cost of resolution at every Impalad (since IP addresses are needed for scheduling)
+ optional string ip_address = 3;
+
+ // True if this is a coordinator node
+ optional bool is_coordinator = 4;
+
+ // True if this is an executor node
+ optional bool is_executor = 5;
+
+ // The address of the debug HTTP server
+ optional NetworkAddressPB debug_http_address = 6;
+
+ // True if the debug webserver is secured (for correctly generating links)
+ optional bool secure_webserver = 7;
+
+ // IP address + port of KRPC based ImpalaInternalService on this backend
+ optional NetworkAddressPB krpc_address = 8;
+
+ // The amount of memory that can be admitted to this backend (in bytes).
+ optional int64 admit_mem_limit = 9;
+
+ // True if fragment instances should not be scheduled on this daemon because the
+ // daemon has been quiescing, e.g. if it shutting down.
+ optional bool is_quiescing = 10;
+
+ // The list of executor groups that this backend belongs to. Only valid if is_executor
+ // is set, and currently must contain exactly one entry.
+ repeated ExecutorGroupDescPB executor_groups = 11;
+
+ // The number of admission slots for this backend that can be occupied by running
+ // queries.
+ optional int64 admission_slots = 12;
+}
diff --git a/common/thrift/StatestoreService.thrift b/common/thrift/StatestoreService.thrift
index 3019b54..041b0df 100644
--- a/common/thrift/StatestoreService.thrift
+++ b/common/thrift/StatestoreService.thrift
@@ -44,62 +44,6 @@ struct TPoolStats {
3: required i64 backend_mem_reserved;
}
-// Structure to describe an executor group. We use this to configure the executor group
-// for backends during startup and during cluster membership management.
-struct TExecutorGroupDesc {
- // The name of the executor group.
- 1: required string name;
- // The minimum size of the executor group to be considered healthy.
- 2: required i64 min_size;
-}
-
-// Structure serialised in the Impala backend topic. Each Impalad
-// constructs one TBackendDescriptor, and registers it in the cluster-membership
-// topic. Impalads subscribe to this topic to learn of the location of
-// all other Impalads in the cluster. Impalads can act as coordinators, executors or
-// both.
-struct TBackendDescriptor {
- // Unique identifier for this impalad. Generated on startup.
- 1: required Types.TBackendId backend_id;
-
- // Network address of the thrift based ImpalaInternalService on this backend
- 2: required Types.TNetworkAddress address;
-
- // IP address corresponding to address.hostname. Explicitly including this saves the
- // cost of resolution at every Impalad (since IP addresses are needed for scheduling)
- 3: required string ip_address;
-
- // True if this is a coordinator node
- 4: required bool is_coordinator;
-
- // True if this is an executor node
- 5: required bool is_executor;
-
- // The address of the debug HTTP server
- 6: optional Types.TNetworkAddress debug_http_address;
-
- // True if the debug webserver is secured (for correctly generating links)
- 7: optional bool secure_webserver;
-
- // IP address + port of KRPC based ImpalaInternalService on this backend
- 8: optional Types.TNetworkAddress krpc_address;
-
- // The amount of memory that can be admitted to this backend (in bytes).
- 9: required i64 admit_mem_limit;
-
- // True if fragment instances should not be scheduled on this daemon because the
- // daemon has been quiescing, e.g. if it shutting down.
- 10: required bool is_quiescing;
-
- // The list of executor groups that this backend belongs to. Only valid if is_executor
- // is set, and currently must contain exactly one entry.
- 11: required list<TExecutorGroupDesc> executor_groups;
-
- // The number of admission slots for this backend that can be occupied by running
- // queries.
- 12: required i64 admission_slots;
-}
-
// Description of a single entry in a topic
struct TTopicItem {
// Human-readable topic entry identifier