You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2021/01/15 07:28:47 UTC
[impala] 01/03: IMPALA-9975 (part 2): Introduce new admission
control daemon
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 91adb33b22eb132b80091742c631e627058c9984
Author: Thomas Tauber-Marshall <tm...@cloudera.com>
AuthorDate: Wed Aug 12 10:53:10 2020 -0700
IMPALA-9975 (part 2): Introduce new admission control daemon
A recent patch (IMPALA-9930) introduces a new admission control rpc
service, which can be configured to perform admission control for
coordinators. In that patch, the admission service runs in an impalad.
This patch separates the service out to run in a new daemon, called
the admissiond. It also integrates this new daemon with the build
infrastructure around Docker.
Some notable changes:
- Adds a new class, AdmissiondEnv, which performs the same function
for the admissiond as ExecEnv does for impalads.
- The '/admission' http endpoint is exposed on the admissiond's webui
if the admission control service is in use, otherwise it is exposed
on coordinator impalad's webuis.
- start-impala-cluster.py takes a new flag --enable_admission_service
which configures the minicluster to have an admissiond with all
coordinators using it for admission control.
- Coordinators are now configured to use the admission service by
specifying the startup flag --admission_service_host. This is
intended to mirror the configuration of the statestored/catalogd
location.
Testing:
- Existing tests for the admission control serivce are modified to run
with an admissiond.
- Manually ran start-impala-cluster.py with --enable_admission_service
and --docker_network to verify Docker integration.
Change-Id: Id677814b31e9193035e8cf0d08aba0ce388a0ad9
Reviewed-on: http://gerrit.cloudera.org:8080/16891
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
CMakeLists.txt | 2 +-
be/src/runtime/exec-env.cc | 99 ++++-------------
be/src/runtime/exec-env.h | 20 ++--
be/src/scheduling/CMakeLists.txt | 2 +
be/src/scheduling/admission-control-client.cc | 10 +-
be/src/scheduling/admission-control-service.cc | 45 ++++----
be/src/scheduling/admission-control-service.h | 8 +-
be/src/scheduling/admission-controller.h | 13 ++-
be/src/scheduling/admissiond-env.cc | 123 +++++++++++++++++++++
be/src/scheduling/admissiond-env.h | 82 ++++++++++++++
be/src/scheduling/admissiond-main.cc | 52 +++++++++
.../scheduling/remote-admission-control-client.cc | 15 +--
.../scheduling/remote-admission-control-client.h | 3 -
be/src/service/CMakeLists.txt | 8 +-
be/src/service/daemon-main.cc | 5 +
be/src/service/impala-http-handler.cc | 64 +++++++----
be/src/service/impala-http-handler.h | 26 ++++-
be/src/service/impala-server.cc | 3 +-
be/src/util/mem-info.cc | 71 ++++++++++++
be/src/util/mem-info.h | 7 ++
bin/start-impala-cluster.py | 65 +++++++++--
common/thrift/metrics.json | 10 ++
docker/CMakeLists.txt | 1 +
docker/admissiond/Dockerfile | 32 ++++++
docker/impala_base/Dockerfile | 1 +
tests/common/custom_cluster_test_suite.py | 5 +-
tests/common/impala_cluster.py | 48 ++++++--
tests/common/impala_service.py | 7 ++
tests/custom_cluster/test_admission_controller.py | 37 ++-----
29 files changed, 658 insertions(+), 206 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 875cfb1..4bd9fb5 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -452,7 +452,7 @@ add_custom_target(notests_independent_targets DEPENDS
java cscope tarballs
)
add_custom_target(notests_regular_targets DEPENDS
- impalad statestored catalogd fesupport loggingsupport ImpalaUdf udasample udfsample
+ impalad statestored catalogd admissiond fesupport loggingsupport ImpalaUdf udasample udfsample
)
add_custom_target(notests_all_targets DEPENDS
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index 1cfd6ec..fe06948 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -43,7 +43,6 @@
#include "runtime/query-exec-mgr.h"
#include "runtime/thread-resource-mgr.h"
#include "runtime/tmp-file-mgr.h"
-#include "scheduling/admission-control-service.h"
#include "scheduling/admission-controller.h"
#include "scheduling/cluster-membership-mgr.h"
#include "scheduling/request-pool-service.h"
@@ -53,7 +52,6 @@
#include "service/frontend.h"
#include "service/impala-server.h"
#include "statestore/statestore-subscriber.h"
-#include "util/cgroup-util.h"
#include "util/cpu-info.h"
#include "util/debug-util.h"
#include "util/default-path-handlers.h"
@@ -136,6 +134,8 @@ DECLARE_bool(is_executor);
DECLARE_string(webserver_interface);
DECLARE_int32(webserver_port);
DECLARE_int64(tcmalloc_max_total_thread_cache_bytes);
+DECLARE_string(admission_service_host);
+DECLARE_int32(admission_service_port);
DECLARE_string(ssl_client_ca_certificate);
@@ -163,12 +163,6 @@ DEFINE_int32(metrics_webserver_port, 0,
DEFINE_string(metrics_webserver_interface, "",
"Interface to start metrics webserver on. If blank, webserver binds to 0.0.0.0");
-DEFINE_bool(is_admission_controller, false,
- "(Experimental) If true, this impalad will export the AdmissionControlService "
- "interface, allowing it to perform admission control for coordinators that have "
- "--admission_control_service_addr set to point to this impalad. This flag will be "
- "removed in a future version, see IMPALA-9155.");
-
const static string DEFAULT_FS = "fs.defaultFS";
// The multiplier for how many queries a dedicated coordinator can run compared to an
@@ -289,7 +283,9 @@ ExecEnv::ExecEnv(int krpc_port, int subscriber_port, int webserver_port,
hdfs_op_thread_pool_.reset(
CreateHdfsOpThreadPool("hdfs-worker-pool", FLAGS_num_hdfs_worker_threads, 1024));
}
- if (FLAGS_is_coordinator || FLAGS_is_admission_controller) {
+ if (FLAGS_is_coordinator && !AdmissionServiceEnabled()) {
+ // We only need a Scheduler if we're performing admission control locally, i.e. if
+ // this is a coordinator and there isn't an admissiond.
scheduler_.reset(new Scheduler(metrics_.get(), request_pool_service_.get()));
}
@@ -306,6 +302,16 @@ ExecEnv::ExecEnv(int krpc_port, int subscriber_port, int webserver_port,
FLAGS_metrics_webserver_port, metrics_.get(), Webserver::AuthMode::NONE));
}
+ if (AdmissionServiceEnabled()) {
+ admission_service_address_ =
+ MakeNetworkAddress(FLAGS_admission_service_host, FLAGS_admission_service_port);
+ if (!IsResolvedAddress(admission_service_address_)) {
+ IpAddr ip;
+ ABORT_IF_ERROR(HostnameToIpAddr(FLAGS_admission_service_host, &ip));
+ admission_service_address_ = MakeNetworkAddress(ip, FLAGS_admission_service_port);
+ }
+ }
+
exec_env_ = this;
}
@@ -419,11 +425,6 @@ Status ExecEnv::Init() {
data_svc_.reset(new DataStreamService(rpc_metrics_));
RETURN_IF_ERROR(data_svc_->Init());
RETURN_IF_ERROR(stream_mgr_->Init(data_svc_->mem_tracker()));
- // Initialize the AdmissionControlService, if enabled.
- if (FLAGS_is_admission_controller) {
- admission_svc_.reset(new AdmissionControlService(rpc_metrics_));
- RETURN_IF_ERROR(admission_svc_->Init());
- }
// Bump thread cache to 1GB to reduce contention for TCMalloc central
// list's spinlock.
@@ -494,72 +495,6 @@ Status ExecEnv::InitHadoopConfig() {
return Status::OK();
}
-Status ExecEnv::ChooseProcessMemLimit(int64_t* bytes_limit) {
- // Depending on the system configuration, we detect the total amount of memory
- // available to the system - either the available physical memory, or if overcommitting
- // is turned off, we use the memory commit limit from /proc/meminfo (see IMPALA-1690).
- // The 'memory' CGroup can also impose a lower limit on memory consumption,
- // so we take the minimum of the system memory and the CGroup memory limit.
- int64_t avail_mem = MemInfo::physical_mem();
- bool use_commit_limit =
- MemInfo::vm_overcommit() == 2 && MemInfo::commit_limit() < MemInfo::physical_mem();
- if (use_commit_limit) {
- avail_mem = MemInfo::commit_limit();
- // There might be the case of misconfiguration, when on a system swap is disabled
- // and overcommitting is turned off the actual usable memory is less than the
- // available physical memory.
- LOG(WARNING) << "This system shows a discrepancy between the available "
- << "memory and the memory commit limit allowed by the "
- << "operating system. ( Mem: " << MemInfo::physical_mem()
- << "<=> CommitLimit: " << MemInfo::commit_limit() << "). "
- << "Impala will adhere to the smaller value when setting the process "
- << "memory limit. Please verify the system configuration. Specifically, "
- << "/proc/sys/vm/overcommit_memory and "
- << "/proc/sys/vm/overcommit_ratio.";
- }
- LOG(INFO) << "System memory available: " << PrettyPrinter::PrintBytes(avail_mem)
- << " (from " << (use_commit_limit ? "commit limit" : "physical mem") << ")";
- int64_t cgroup_mem_limit;
- Status cgroup_mem_status = CGroupUtil::FindCGroupMemLimit(&cgroup_mem_limit);
- if (cgroup_mem_status.ok()) {
- if (cgroup_mem_limit < avail_mem) {
- avail_mem = cgroup_mem_limit;
- LOG(INFO) << "CGroup memory limit for this process reduces physical memory "
- << "available to: " << PrettyPrinter::PrintBytes(avail_mem);
- }
- } else {
- LOG(WARNING) << "Could not detect CGroup memory limit, assuming unlimited: $0"
- << cgroup_mem_status.GetDetail();
- }
- bool is_percent;
- *bytes_limit = ParseUtil::ParseMemSpec(FLAGS_mem_limit, &is_percent, avail_mem);
- // ParseMemSpec() returns -1 for invalid input and 0 to mean unlimited. From Impala
- // 2.11 onwards we do not support unlimited process memory limits.
- if (*bytes_limit <= 0) {
- return Status(Substitute("The process memory limit (--mem_limit) must be a positive "
- "bytes value or percentage: $0", FLAGS_mem_limit));
- }
- if (is_percent) {
- LOG(INFO) << "Using process memory limit: " << PrettyPrinter::PrintBytes(*bytes_limit)
- << " (--mem_limit=" << FLAGS_mem_limit << " of "
- << PrettyPrinter::PrintBytes(avail_mem) << ")";
- } else {
- LOG(INFO) << "Using process memory limit: " << PrettyPrinter::PrintBytes(*bytes_limit)
- << " (--mem_limit=" << FLAGS_mem_limit << ")";
- }
- if (*bytes_limit > MemInfo::physical_mem()) {
- LOG(WARNING) << "Process memory limit " << PrettyPrinter::PrintBytes(*bytes_limit)
- << " exceeds physical memory of "
- << PrettyPrinter::PrintBytes(MemInfo::physical_mem());
- }
- if (cgroup_mem_status.ok() && *bytes_limit > cgroup_mem_limit) {
- LOG(WARNING) << "Process Memory limit " << PrettyPrinter::PrintBytes(*bytes_limit)
- << " exceeds CGroup memory limit of "
- << PrettyPrinter::PrintBytes(cgroup_mem_limit);
- }
- return Status::OK();
-}
-
Status ExecEnv::StartStatestoreSubscriberService() {
LOG(INFO) << "Starting statestore subscriber service";
@@ -693,4 +628,8 @@ Status ExecEnv::GetKuduClient(
return Status::OK();
}
+bool ExecEnv::AdmissionServiceEnabled() const {
+ return !FLAGS_admission_service_host.empty();
+}
+
} // namespace impala
diff --git a/be/src/runtime/exec-env.h b/be/src/runtime/exec-env.h
index 0fed647..e378d5b 100644
--- a/be/src/runtime/exec-env.h
+++ b/be/src/runtime/exec-env.h
@@ -41,7 +41,6 @@ class KuduClient;
namespace impala {
class AdmissionController;
-class AdmissionControlService;
class BufferPool;
class CallableThreadPool;
class ClusterMembershipMgr;
@@ -144,7 +143,6 @@ class ExecEnv {
ClusterMembershipMgr* cluster_membership_mgr() { return cluster_membership_mgr_.get(); }
Scheduler* scheduler() { return scheduler_.get(); }
AdmissionController* admission_controller() { return admission_controller_.get(); }
- AdmissionControlService* admission_control_service() { return admission_svc_.get(); }
StatestoreSubscriber* subscriber() { return statestore_subscriber_.get(); }
const TNetworkAddress& configured_backend_address() const {
@@ -176,6 +174,13 @@ class ExecEnv {
int64_t admit_mem_limit() const { return admit_mem_limit_; }
int64_t admission_slots() const { return admission_slots_; }
+ const TNetworkAddress& admission_service_address() const {
+ return admission_service_address_;
+ }
+
+ /// Returns true if the admission control service is enabled.
+ bool AdmissionServiceEnabled() const;
+
private:
// Used to uniquely identify this impalad.
BackendIdPB backend_id_;
@@ -209,7 +214,6 @@ class ExecEnv {
boost::scoped_ptr<RpcMgr> rpc_mgr_;
boost::scoped_ptr<ControlService> control_svc_;
boost::scoped_ptr<DataStreamService> data_svc_;
- boost::scoped_ptr<AdmissionControlService> admission_svc_;
/// Query-wide buffer pool and the root reservation tracker for the pool. The
/// reservation limit is equal to the maximum capacity of the pool. Created in
@@ -279,15 +283,13 @@ class ExecEnv {
/// this backend. Queries take up multiple slots only when mt_dop > 1.
int64_t admission_slots_;
+ /// If the admission control service is enabled, the resolved IP address and port where
+ /// the service is running.
+ TNetworkAddress admission_service_address_;
+
/// Initialize ExecEnv based on Hadoop config from frontend.
Status InitHadoopConfig();
- /// Choose a memory limit (returned in *bytes_limit) based on the --mem_limit flag and
- /// the memory available to the daemon process. Returns an error if the memory limit is
- /// invalid or another error is encountered that should prevent starting up the daemon.
- /// Logs the memory limit chosen and any relevant diagnostics related to that choice.
- Status ChooseProcessMemLimit(int64_t* bytes_limit);
-
/// Initialise 'buffer_pool_' and 'buffer_reservation_' with given capacity.
void InitBufferPool(int64_t min_page_len, int64_t capacity, int64_t clean_pages_limit);
diff --git a/be/src/scheduling/CMakeLists.txt b/be/src/scheduling/CMakeLists.txt
index 05a80de..61bd37c 100644
--- a/be/src/scheduling/CMakeLists.txt
+++ b/be/src/scheduling/CMakeLists.txt
@@ -27,10 +27,12 @@ set_source_files_properties(${ADMISSION_CONTROL_SERVICE_PROTO_SRCS} PROPERTIES G
# TODO: Move other scheduling-related classes here
add_library(Scheduling STATIC
+ admissiond-env.cc
${ADMISSION_CONTROL_SERVICE_PROTO_SRCS}
admission-controller.cc
admission-control-client.cc
admission-control-service.cc
+ admissiond-main.cc
executor-blacklist.cc
cluster-membership-mgr.cc
cluster-membership-test-util.cc
diff --git a/be/src/scheduling/admission-control-client.cc b/be/src/scheduling/admission-control-client.cc
index bf1cc7a..98f788a 100644
--- a/be/src/scheduling/admission-control-client.cc
+++ b/be/src/scheduling/admission-control-client.cc
@@ -17,14 +17,12 @@
#include "scheduling/admission-control-client.h"
+#include "runtime/exec-env.h"
#include "scheduling/local-admission-control-client.h"
#include "scheduling/remote-admission-control-client.h"
#include "common/names.h"
-DECLARE_string(admission_control_service_addr);
-DECLARE_bool(is_admission_controller);
-
namespace impala {
// Profile query events
@@ -36,10 +34,10 @@ const string AdmissionControlClient::QUERY_EVENT_COMPLETED_ADMISSION =
void AdmissionControlClient::Create(
const TQueryCtx& query_ctx, unique_ptr<AdmissionControlClient>* client) {
- if (FLAGS_is_admission_controller || FLAGS_admission_control_service_addr.empty()) {
- client->reset(new LocalAdmissionControlClient(query_ctx.query_id));
- } else {
+ if (ExecEnv::GetInstance()->AdmissionServiceEnabled()) {
client->reset(new RemoteAdmissionControlClient(query_ctx));
+ } else {
+ client->reset(new LocalAdmissionControlClient(query_ctx.query_id));
}
}
diff --git a/be/src/scheduling/admission-control-service.cc b/be/src/scheduling/admission-control-service.cc
index ccfe9ee..6235765 100644
--- a/be/src/scheduling/admission-control-service.cc
+++ b/be/src/scheduling/admission-control-service.cc
@@ -28,6 +28,7 @@
#include "runtime/exec-env.h"
#include "runtime/mem-tracker.h"
#include "scheduling/admission-controller.h"
+#include "scheduling/admissiond-env.h"
#include "util/cpu-info.h"
#include "util/kudu-status-util.h"
#include "util/memory-metrics.h"
@@ -51,10 +52,10 @@ DEFINE_int32(admission_thread_pool_size, 5,
DEFINE_int32(max_admission_queue_size, 50,
"(Advanced) Max size of the queue for the AdmitQuery thread pool.");
-DEFINE_string(admission_control_service_addr, "",
- "(Experimental) If provided, queries submitted to this impalad will be scheduled and "
- "admitted by contacting the admission control service at the specified address. This "
- "flag will be removed in a future version, see IMPALA-9155.");
+DEFINE_string(admission_service_host, "",
+ "If provided, queries submitted to this impalad will be scheduled and admitted by "
+ "contacting the admission control service at the specified address and "
+ "--admission_service_port.");
DEFINE_int32(admission_status_wait_time_ms, 100,
"(Advanced) The number of milliseconds the GetQueryStatus() rpc in the admission "
"control service will wait for admission to complete before returning.");
@@ -71,9 +72,9 @@ namespace impala {
} while (false)
AdmissionControlService::AdmissionControlService(MetricGroup* metric_group)
- : AdmissionControlServiceIf(ExecEnv::GetInstance()->rpc_mgr()->metric_entity(),
- ExecEnv::GetInstance()->rpc_mgr()->result_tracker()) {
- MemTracker* process_mem_tracker = ExecEnv::GetInstance()->process_mem_tracker();
+ : AdmissionControlServiceIf(AdmissiondEnv::GetInstance()->rpc_mgr()->metric_entity(),
+ AdmissiondEnv::GetInstance()->rpc_mgr()->result_tracker()) {
+ MemTracker* process_mem_tracker = AdmissiondEnv::GetInstance()->process_mem_tracker();
bool is_percent; // not used
int64_t bytes_limit =
ParseUtil::ParseMemSpec(FLAGS_admission_control_service_queue_mem_limit,
@@ -96,9 +97,9 @@ Status AdmissionControlService::Init() {
CpuInfo::num_cores();
// The maximum queue length is set to maximum 32-bit value. Its actual capacity is
// bound by memory consumption against 'mem_tracker_'.
- RETURN_IF_ERROR(ExecEnv::GetInstance()->rpc_mgr()->RegisterService(num_svc_threads,
- std::numeric_limits<int32_t>::max(), this, mem_tracker_.get(),
- ExecEnv::GetInstance()->rpc_metrics()));
+ RETURN_IF_ERROR(AdmissiondEnv::GetInstance()->rpc_mgr()->RegisterService(
+ num_svc_threads, std::numeric_limits<int32_t>::max(), this, mem_tracker_.get(),
+ AdmissiondEnv::GetInstance()->rpc_metrics()));
admission_thread_pool_.reset(
new ThreadPool<UniqueIdPB>("admission-control-service", "admission-worker",
@@ -109,10 +110,16 @@ Status AdmissionControlService::Init() {
return Status::OK();
}
-Status AdmissionControlService::GetProxy(const TNetworkAddress& address,
- const string& hostname, unique_ptr<AdmissionControlServiceProxy>* proxy) {
+void AdmissionControlService::Join() {
+ admission_thread_pool_->Join();
+}
+
+Status AdmissionControlService::GetProxy(
+ unique_ptr<AdmissionControlServiceProxy>* proxy) {
// Create a AdmissionControlService proxy to the destination.
- RETURN_IF_ERROR(ExecEnv::GetInstance()->rpc_mgr()->GetProxy(address, hostname, proxy));
+ RETURN_IF_ERROR(ExecEnv::GetInstance()->rpc_mgr()->GetProxy(
+ ExecEnv::GetInstance()->admission_service_address(), FLAGS_admission_service_host,
+ proxy));
return Status::OK();
}
@@ -153,9 +160,9 @@ void AdmissionControlService::GetQueryStatus(const GetQueryStatusRequestPB* req,
if (!admission_state->admission_done) {
bool timed_out;
admission_state->admit_status =
- ExecEnv::GetInstance()->admission_controller()->WaitOnQueued(req->query_id(),
- &admission_state->schedule, FLAGS_admission_status_wait_time_ms,
- &timed_out);
+ AdmissiondEnv::GetInstance()->admission_controller()->WaitOnQueued(
+ req->query_id(), &admission_state->schedule,
+ FLAGS_admission_status_wait_time_ms, &timed_out);
if (!timed_out) {
admission_state->admission_done = true;
if (admission_state->admit_status.ok()) {
@@ -204,7 +211,7 @@ void AdmissionControlService::ReleaseQuery(const ReleaseQueryRequestPB* req,
{
lock_guard<mutex> l(admission_state->lock);
if (!admission_state->released) {
- ExecEnv::GetInstance()->admission_controller()->ReleaseQuery(
+ AdmissiondEnv::GetInstance()->admission_controller()->ReleaseQuery(
req->query_id(), req->peak_mem_consumption());
admission_state->released = true;
} else {
@@ -239,7 +246,7 @@ void AdmissionControlService::ReleaseQueryBackends(
admission_state->unreleased_backends.erase(it);
}
- ExecEnv::GetInstance()->admission_controller()->ReleaseQueryBackends(
+ AdmissiondEnv::GetInstance()->admission_controller()->ReleaseQueryBackends(
req->query_id(), host_addrs);
}
@@ -272,7 +279,7 @@ void AdmissionControlService::AdmitFromThreadPool(UniqueIdPB query_id) {
admission_state->summary_profile,
admission_state->blacklisted_executor_addresses};
admission_state->admit_status =
- ExecEnv::GetInstance()->admission_controller()->SubmitForAdmission(request,
+ AdmissiondEnv::GetInstance()->admission_controller()->SubmitForAdmission(request,
&admission_state->admit_outcome, &admission_state->schedule, &queued,
&admission_state->request_pool);
admission_state->submitted = true;
diff --git a/be/src/scheduling/admission-control-service.h b/be/src/scheduling/admission-control-service.h
index 96ca8fa..23ebce6 100644
--- a/be/src/scheduling/admission-control-service.h
+++ b/be/src/scheduling/admission-control-service.h
@@ -50,6 +50,9 @@ class AdmissionControlService : public AdmissionControlServiceIf,
/// This mustn't be called until RPC manager has been initialized.
Status Init();
+ /// Blocks until the service shuts down.
+ void Join();
+
virtual void AdmitQuery(const AdmitQueryRequestPB* req, AdmitQueryResponsePB* resp,
kudu::rpc::RpcContext* context) override;
virtual void GetQueryStatus(const GetQueryStatusRequestPB* req,
@@ -61,10 +64,9 @@ class AdmissionControlService : public AdmissionControlServiceIf,
virtual void CancelAdmission(const CancelAdmissionRequestPB* req,
CancelAdmissionResponsePB* resp, kudu::rpc::RpcContext* context) override;
- /// Gets a AdmissionControlService proxy to a server with 'address' and 'hostname'.
+ /// Gets a AdmissionControlService proxy to the configured admission control service.
/// The newly created proxy is returned in 'proxy'. Returns error status on failure.
- static Status GetProxy(const TNetworkAddress& address, const std::string& hostname,
- std::unique_ptr<AdmissionControlServiceProxy>* proxy);
+ static Status GetProxy(std::unique_ptr<AdmissionControlServiceProxy>* proxy);
private:
friend class ImpalaHttpHandler;
diff --git a/be/src/scheduling/admission-controller.h b/be/src/scheduling/admission-controller.h
index a81e54a..b281730 100644
--- a/be/src/scheduling/admission-controller.h
+++ b/be/src/scheduling/admission-controller.h
@@ -65,8 +65,17 @@ enum class AdmissionOutcome {
/// queue size, incoming queries will be rejected. Requests in the queue will time out
/// after a configurable timeout.
///
-/// Depending on the -is_coordinator startup flag, multiple impalads can act as a
-/// coordinator and thus also an admission controller, so some cluster state must be
+/// Admission control can be run in two possible modes:
+/// - Distributed mode: each coordinator performs admission control independently, based
+/// on eventually consistent info about the admission decisions made by other
+/// coordinators, which is distributed by the statestore. This is the traditional Impala
+/// set up.
+/// - Admission service mode: coordinators are configured with --admission_service_host to
+/// send their queries to the admissiond which performs all admission control based on
+/// its complete view of cluster load. This mode is currently considered experimental.
+///
+/// In distributed mode, depending on the -is_coordinator flag, multiple impalads can act
+/// as a coordinator and thus also an admission controller, so some cluster state must be
/// shared between impalads in order to make admission decisions on any of them. Every
/// coordinator maintains some per-pool and per-host statistics related to the requests it
/// itself is servicing as the admission controller. Some of these local admission
diff --git a/be/src/scheduling/admissiond-env.cc b/be/src/scheduling/admissiond-env.cc
new file mode 100644
index 0000000..9f15f38
--- /dev/null
+++ b/be/src/scheduling/admissiond-env.cc
@@ -0,0 +1,123 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "scheduling/admissiond-env.h"
+
+#include "common/daemon-env.h"
+#include "rpc/rpc-mgr.h"
+#include "runtime/mem-tracker.h"
+#include "scheduling/admission-control-service.h"
+#include "scheduling/admission-controller.h"
+#include "scheduling/cluster-membership-mgr.h"
+#include "scheduling/scheduler.h"
+#include "service/impala-http-handler.h"
+#include "util/default-path-handlers.h"
+#include "util/mem-info.h"
+#include "util/memory-metrics.h"
+#include "util/metrics.h"
+#include "util/uid-util.h"
+
+#include "common/names.h"
+
+DEFINE_int32(
+ admission_service_port, 29500, "The port where the admission control service runs");
+
+DECLARE_string(state_store_host);
+DECLARE_int32(state_store_port);
+DECLARE_int32(state_store_subscriber_port);
+DECLARE_string(hostname);
+
+namespace impala {
+
+AdmissiondEnv* AdmissiondEnv::admissiond_env_ = nullptr;
+
+AdmissiondEnv::AdmissiondEnv()
+ : pool_mem_trackers_(new PoolMemTrackerRegistry),
+ request_pool_service_(new RequestPoolService(DaemonEnv::GetInstance()->metrics())),
+ rpc_mgr_(new RpcMgr(IsInternalTlsConfigured())),
+ rpc_metrics_(DaemonEnv::GetInstance()->metrics()->GetOrCreateChildGroup("rpc")) {
+ MetricGroup* metrics = DaemonEnv::GetInstance()->metrics();
+
+ TNetworkAddress admission_service_addr =
+ MakeNetworkAddress(FLAGS_hostname, FLAGS_admission_service_port);
+ TNetworkAddress subscriber_address =
+ MakeNetworkAddress(FLAGS_hostname, FLAGS_state_store_subscriber_port);
+ TNetworkAddress statestore_address =
+ MakeNetworkAddress(FLAGS_state_store_host, FLAGS_state_store_port);
+ statestore_subscriber_.reset(new StatestoreSubscriber(
+ Substitute("admissiond@$0", TNetworkAddressToString(admission_service_addr)),
+ subscriber_address, statestore_address, metrics));
+
+ scheduler_.reset(new Scheduler(metrics, request_pool_service()));
+ cluster_membership_mgr_.reset(new ClusterMembershipMgr(
+ PrintId(DaemonEnv::GetInstance()->backend_id()), subscriber(), metrics));
+ admission_controller_.reset(new AdmissionController(cluster_membership_mgr(),
+ subscriber(), request_pool_service(), metrics, scheduler(), pool_mem_trackers(),
+ admission_service_addr));
+ http_handler_.reset(ImpalaHttpHandler::CreateAdmissiondHandler(
+ admission_controller_.get(), cluster_membership_mgr_.get()));
+
+ admissiond_env_ = this;
+}
+
+AdmissiondEnv::~AdmissiondEnv() {
+ if (rpc_mgr_ != nullptr) rpc_mgr_->Shutdown();
+}
+
+Status AdmissiondEnv::Init() {
+ int64_t bytes_limit;
+ RETURN_IF_ERROR(ChooseProcessMemLimit(&bytes_limit));
+ mem_tracker_.reset(
+ new MemTracker(AggregateMemoryMetrics::TOTAL_USED, bytes_limit, "Process"));
+ mem_tracker_->RegisterMetrics(
+ DaemonEnv::GetInstance()->metrics(), "mem-tracker.process");
+
+ http_handler_->RegisterHandlers(DaemonEnv::GetInstance()->webserver());
+
+ string ip_address;
+ RETURN_IF_ERROR(HostnameToIpAddr(FLAGS_hostname, &ip_address));
+ TNetworkAddress ip_addr_port;
+ ip_addr_port.__set_hostname(ip_address);
+ ip_addr_port.__set_port(FLAGS_admission_service_port);
+ RETURN_IF_ERROR(rpc_mgr_->Init(ip_addr_port));
+ admission_control_svc_.reset(
+ new AdmissionControlService(DaemonEnv::GetInstance()->metrics()));
+ RETURN_IF_ERROR(admission_control_svc_->Init());
+
+ RETURN_IF_ERROR(cluster_membership_mgr_->Init());
+ RETURN_IF_ERROR(admission_controller_->Init());
+ return Status::OK();
+}
+
+Status AdmissiondEnv::StartServices() {
+ LOG(INFO) << "Starting statestore subscriber service";
+ // Must happen after all topic registrations / callbacks are done
+ if (statestore_subscriber_.get() != nullptr) {
+ Status status = statestore_subscriber_->Start();
+ if (!status.ok()) {
+ status.AddDetail("Statestore subscriber did not start up.");
+ return status;
+ }
+ }
+
+ LOG(INFO) << "Starting KRPC service.";
+ RETURN_IF_ERROR(rpc_mgr_->StartServices());
+
+ return Status::OK();
+}
+
+} // namespace impala
diff --git a/be/src/scheduling/admissiond-env.h b/be/src/scheduling/admissiond-env.h
new file mode 100644
index 0000000..e6efff4
--- /dev/null
+++ b/be/src/scheduling/admissiond-env.h
@@ -0,0 +1,82 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include "common/global-types.h"
+#include "common/status.h"
+#include "gen-cpp/common.pb.h"
+#include "util/metrics.h"
+#include "util/network-util.h"
+
+namespace impala {
+
+class AdmissionController;
+class AdmissionControlService;
+class ClusterMembershipMgr;
+class ImpalaHttpHandler;
+class MemTracker;
+class MetricGroup;
+class PoolMemTrackerRegistry;
+class RequestPoolService;
+class RpcMgr;
+class Scheduler;
+class StatestoreSubscriber;
+
+/// Contains and initializes singleton objects needed for the admisison control service,
+/// analogous to ExecEnv for impalads.
+class AdmissiondEnv {
+ public:
+ static AdmissiondEnv* GetInstance() { return admissiond_env_; }
+
+ AdmissiondEnv();
+ ~AdmissiondEnv();
+
+ Status Init() WARN_UNUSED_RESULT;
+ Status StartServices() WARN_UNUSED_RESULT;
+
+ AdmissionController* admission_controller() { return admission_controller_.get(); }
+ AdmissionControlService* admission_control_service() {
+ return admission_control_svc_.get();
+ }
+ ClusterMembershipMgr* cluster_membership_mgr() { return cluster_membership_mgr_.get(); }
+ MemTracker* process_mem_tracker() { return mem_tracker_.get(); }
+ MetricGroup* rpc_metrics() { return rpc_metrics_; }
+ PoolMemTrackerRegistry* pool_mem_trackers() { return pool_mem_trackers_.get(); }
+ RequestPoolService* request_pool_service() { return request_pool_service_.get(); }
+ RpcMgr* rpc_mgr() { return rpc_mgr_.get(); }
+ Scheduler* scheduler() { return scheduler_.get(); }
+ StatestoreSubscriber* subscriber() { return statestore_subscriber_.get(); }
+
+ private:
+ static AdmissiondEnv* admissiond_env_;
+
+ std::unique_ptr<AdmissionController> admission_controller_;
+ std::unique_ptr<AdmissionControlService> admission_control_svc_;
+ std::unique_ptr<ClusterMembershipMgr> cluster_membership_mgr_;
+ std::unique_ptr<ImpalaHttpHandler> http_handler_;
+ std::unique_ptr<MemTracker> mem_tracker_;
+ std::unique_ptr<PoolMemTrackerRegistry> pool_mem_trackers_;
+ std::unique_ptr<RequestPoolService> request_pool_service_;
+ std::unique_ptr<RpcMgr> rpc_mgr_;
+ std::unique_ptr<Scheduler> scheduler_;
+ std::unique_ptr<StatestoreSubscriber> statestore_subscriber_;
+
+ MetricGroup* rpc_metrics_ = nullptr;
+};
+
+} // namespace impala
diff --git a/be/src/scheduling/admissiond-main.cc b/be/src/scheduling/admissiond-main.cc
new file mode 100644
index 0000000..6d81e95
--- /dev/null
+++ b/be/src/scheduling/admissiond-main.cc
@@ -0,0 +1,52 @@
+// 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.
+
+//
+// This file contains the main() function for the admission controller process,
+// which exports the KRPC service AdmissionControlService.
+
+#include <gutil/strings/substitute.h>
+
+#include "common/daemon-env.h"
+#include "common/init.h"
+#include "common/logging.h"
+#include "scheduling/admission-control-service.h"
+#include "scheduling/admissiond-env.h"
+#include "util/network-util.h"
+
+DECLARE_int32(webserver_port);
+DECLARE_int32(state_store_subscriber_port);
+
+using namespace impala;
+using namespace strings;
+
+int AdmissiondMain(int argc, char** argv) {
+ // Override default for webserver port
+ FLAGS_webserver_port = 25030;
+ FLAGS_state_store_subscriber_port = 23030;
+ InitCommonRuntime(argc, argv, /* init_jvm */ true);
+
+ DaemonEnv daemon_env("admissiond");
+ ABORT_IF_ERROR(daemon_env.Init(/* init_jvm */ true));
+
+ AdmissiondEnv admissiond_env;
+ ABORT_IF_ERROR(admissiond_env.Init());
+ ABORT_IF_ERROR(admissiond_env.StartServices());
+ admissiond_env.admission_control_service()->Join();
+
+ return 0;
+}
diff --git a/be/src/scheduling/remote-admission-control-client.cc b/be/src/scheduling/remote-admission-control-client.cc
index c24d854..6183daf 100644
--- a/be/src/scheduling/remote-admission-control-client.cc
+++ b/be/src/scheduling/remote-admission-control-client.cc
@@ -33,7 +33,6 @@
#include "common/names.h"
-DECLARE_string(admission_control_service_addr);
DEFINE_int32(admission_status_retry_time_ms, 10,
"(Advanced) The number of milliseconds coordinators will wait before retrying the "
@@ -45,8 +44,7 @@ using namespace kudu::rpc;
namespace impala {
RemoteAdmissionControlClient::RemoteAdmissionControlClient(const TQueryCtx& query_ctx)
- : query_ctx_(query_ctx),
- address_(MakeNetworkAddress(FLAGS_admission_control_service_addr)) {
+ : query_ctx_(query_ctx) {
TUniqueIdToUniqueIdPB(query_ctx.query_id, &query_id_);
}
@@ -58,7 +56,7 @@ Status RemoteAdmissionControlClient::SubmitForAdmission(
query_events, AdmissionControlClient::QUERY_EVENT_COMPLETED_ADMISSION);
std::unique_ptr<AdmissionControlServiceProxy> proxy;
- RETURN_IF_ERROR(AdmissionControlService::GetProxy(address_, address_.hostname, &proxy));
+ RETURN_IF_ERROR(AdmissionControlService::GetProxy(&proxy));
AdmitQueryRequestPB req;
AdmitQueryResponsePB resp;
RpcController rpc_controller;
@@ -137,8 +135,7 @@ Status RemoteAdmissionControlClient::SubmitForAdmission(
void RemoteAdmissionControlClient::ReleaseQuery(int64_t peak_mem_consumption) {
std::unique_ptr<AdmissionControlServiceProxy> proxy;
- Status get_proxy_status =
- AdmissionControlService::GetProxy(address_, address_.hostname, &proxy);
+ Status get_proxy_status = AdmissionControlService::GetProxy(&proxy);
if (!get_proxy_status.ok()) {
LOG(ERROR) << "ReleaseQuery for " << query_id_
<< " failed to get proxy: " << get_proxy_status;
@@ -169,8 +166,7 @@ void RemoteAdmissionControlClient::ReleaseQuery(int64_t peak_mem_consumption) {
void RemoteAdmissionControlClient::ReleaseQueryBackends(
const vector<NetworkAddressPB>& host_addrs) {
std::unique_ptr<AdmissionControlServiceProxy> proxy;
- Status get_proxy_status =
- AdmissionControlService::GetProxy(address_, address_.hostname, &proxy);
+ Status get_proxy_status = AdmissionControlService::GetProxy(&proxy);
if (!get_proxy_status.ok()) {
LOG(ERROR) << "ReleaseQueryBackends for " << query_id_
<< " failed to get proxy: " << get_proxy_status;
@@ -213,8 +209,7 @@ void RemoteAdmissionControlClient::CancelAdmission() {
}
std::unique_ptr<AdmissionControlServiceProxy> proxy;
- Status get_proxy_status =
- AdmissionControlService::GetProxy(address_, address_.hostname, &proxy);
+ Status get_proxy_status = AdmissionControlService::GetProxy(&proxy);
if (!get_proxy_status.ok()) {
LOG(WARNING) << "CancelAdmission for " << query_id_
<< " failed to get proxy: " << get_proxy_status;
diff --git a/be/src/scheduling/remote-admission-control-client.h b/be/src/scheduling/remote-admission-control-client.h
index 59fed78..d06929f 100644
--- a/be/src/scheduling/remote-admission-control-client.h
+++ b/be/src/scheduling/remote-admission-control-client.h
@@ -50,9 +50,6 @@ class RemoteAdmissionControlClient : public AdmissionControlClient {
// The id of the query being considered for admission.
UniqueIdPB query_id_;
- /// The address of the remote admission controller to use.
- TNetworkAddress address_;
-
/// Protects 'pending_admit_' and 'cancelled_'.
std::mutex lock_;
diff --git a/be/src/service/CMakeLists.txt b/be/src/service/CMakeLists.txt
index 7b7dfef..747e83e 100644
--- a/be/src/service/CMakeLists.txt
+++ b/be/src/service/CMakeLists.txt
@@ -82,6 +82,7 @@ add_executable(unifiedbetests
# the impalad binary.
set (STATESTORED_SYMLINK ${EXECUTABLE_OUTPUT_PATH}/statestored)
set (CATALOGD_SYMLINK ${EXECUTABLE_OUTPUT_PATH}/catalogd)
+set (ADMISSIOND_SYMLINK ${EXECUTABLE_OUTPUT_PATH}/admissiond)
ADD_CUSTOM_COMMAND(
OUTPUT ${STATESTORED_SYMLINK}
@@ -91,10 +92,15 @@ ADD_CUSTOM_COMMAND(
OUTPUT ${CATALOGD_SYMLINK}
COMMAND ${CMAKE_COMMAND} -E create_symlink impalad ${CATALOGD_SYMLINK}
DEPENDS impalad)
+ADD_CUSTOM_COMMAND(
+ OUTPUT ${ADMISSIOND_SYMLINK}
+ COMMAND ${CMAKE_COMMAND} -E create_symlink impalad ${ADMISSIOND_SYMLINK}
+ DEPENDS impalad)
ADD_CUSTOM_TARGET(statestored ALL DEPENDS ${STATESTORED_SYMLINK})
ADD_CUSTOM_TARGET(catalogd ALL DEPENDS ${CATALOGD_SYMLINK})
+ADD_CUSTOM_TARGET(admissiond ALL DEPENDS ${ADMISSIOND_SYMLINK})
-ADD_CUSTOM_TARGET(daemons DEPENDS impalad statestored catalogd fesupport)
+ADD_CUSTOM_TARGET(daemons DEPENDS impalad statestored catalogd admissiond fesupport)
# This permits libraries loaded by dlopen to link to the symbols in the program.
set_target_properties(impalad PROPERTIES LINK_FLAGS -rdynamic)
diff --git a/be/src/service/daemon-main.cc b/be/src/service/daemon-main.cc
index f44cb45..e23491f 100644
--- a/be/src/service/daemon-main.cc
+++ b/be/src/service/daemon-main.cc
@@ -25,6 +25,7 @@ using namespace std;
int StatestoredMain(int, char**);
int CatalogdMain(int, char**);
int ImpaladMain(int, char**);
+int AdmissiondMain(int, char**);
int main(int argc, char** argv) {
path cmd_line_path(argv[0]);
@@ -41,6 +42,10 @@ int main(int argc, char** argv) {
return CatalogdMain(argc, argv);
}
+ if (daemon == "admissiond") {
+ return AdmissiondMain(argc, argv);
+ }
+
cerr << "Unknown daemon name: " << daemon
<< " (valid options: impalad, catalogd, statestored)" << endl;
exit(1);
diff --git a/be/src/service/impala-http-handler.cc b/be/src/service/impala-http-handler.cc
index 65307d5..677f50c 100644
--- a/be/src/service/impala-http-handler.cc
+++ b/be/src/service/impala-http-handler.cc
@@ -38,6 +38,7 @@
#include "runtime/timestamp-value.inline.h"
#include "scheduling/admission-control-service.h"
#include "scheduling/admission-controller.h"
+#include "scheduling/admissiond-env.h"
#include "scheduling/cluster-membership-mgr.h"
#include "service/client-request-state.h"
#include "service/frontend.h"
@@ -65,8 +66,7 @@ using namespace strings;
DECLARE_int32(query_log_size);
DECLARE_int32(query_stmt_size);
DECLARE_bool(use_local_catalog);
-DECLARE_bool(is_admission_controller);
-DECLARE_string(admission_control_service_addr);
+DECLARE_string(admission_service_host);
namespace {
@@ -94,9 +94,31 @@ static Status ParseIdFromRequest(const Webserver::WebRequest& req, TUniqueId* id
}
+ImpalaHttpHandler::ImpalaHttpHandler(ImpalaServer* server,
+ AdmissionController* admission_controller,
+ ClusterMembershipMgr* cluster_membership_mgr, bool is_admissiond)
+ : server_(server),
+ admission_controller_(admission_controller),
+ cluster_membership_mgr_(cluster_membership_mgr),
+ is_admissiond_(is_admissiond) {}
+
void ImpalaHttpHandler::RegisterHandlers(Webserver* webserver, bool metrics_only) {
DCHECK(webserver != NULL);
+ if (is_admissiond_) {
+ // The admissiond only exposes a subset of endpoints that have info relevant to
+ // admission control.
+ webserver->RegisterUrlCallback("/backends", "backends.tmpl",
+ MakeCallback(this, &ImpalaHttpHandler::BackendsHandler), true);
+
+ webserver->RegisterUrlCallback("/admission", "admission_controller.tmpl",
+ MakeCallback(this, &ImpalaHttpHandler::AdmissionStateHandler), true);
+
+ webserver->RegisterUrlCallback("/resource_pool_reset", "",
+ MakeCallback(this, &ImpalaHttpHandler::ResetResourcePoolStatsHandler), false);
+ return;
+ }
+
Webserver::RawUrlCallback healthz_callback =
[this](const auto& req, auto* data, auto* response) {
return this->HealthzHandler(req, data, response);
@@ -173,15 +195,14 @@ void ImpalaHttpHandler::RegisterHandlers(Webserver* webserver, bool metrics_only
[this](const auto& req, auto* doc) {
this->QuerySummaryHandler(false, false, req, doc); }, false);
- // Only enable the admission control endpoints for impalads that are running admission
- // control, which is all impalads if the admission control service is not enabled,
- // indicated by 'admission_control_service_addr' not being specified.
- if (FLAGS_admission_control_service_addr.empty() || FLAGS_is_admission_controller) {
+ // Only enable the admission control endpoints for impalads if the admission service is
+ // not enabled, otherwise these will be exposed on the admissiond.
+ if (!ExecEnv::GetInstance()->AdmissionServiceEnabled()) {
webserver->RegisterUrlCallback("/admission", "admission_controller.tmpl",
- MakeCallback(this, &ImpalaHttpHandler::AdmissionStateHandler), true);
+ MakeCallback(this, &ImpalaHttpHandler::AdmissionStateHandler), true);
webserver->RegisterUrlCallback("/resource_pool_reset", "",
- MakeCallback(this, &ImpalaHttpHandler::ResetResourcePoolStatsHandler), false);
+ MakeCallback(this, &ImpalaHttpHandler::ResetResourcePoolStatsHandler), false);
}
RegisterLogLevelCallbacks(webserver, true);
@@ -955,17 +976,15 @@ void ImpalaHttpHandler::QuerySummaryHandler(bool include_json_plan, bool include
void ImpalaHttpHandler::BackendsHandler(const Webserver::WebRequest& req,
Document* document) {
AdmissionController::PerHostStats host_stats;
- ExecEnv::GetInstance()->admission_controller()->PopulatePerHostMemReservedAndAdmitted(
- &host_stats);
+ DCHECK(admission_controller_ != nullptr);
+ admission_controller_->PopulatePerHostMemReservedAndAdmitted(&host_stats);
Value backends_list(kArrayType);
int num_active_backends = 0;
int num_quiescing_backends = 0;
int num_blacklisted_backends = 0;
- ClusterMembershipMgr* cluster_membership_mgr =
- ExecEnv::GetInstance()->cluster_membership_mgr();
- DCHECK(cluster_membership_mgr != nullptr);
+ DCHECK(cluster_membership_mgr_ != nullptr);
ClusterMembershipMgr::SnapshotPtr membership_snapshot =
- cluster_membership_mgr->GetSnapshot();
+ cluster_membership_mgr_->GetSnapshot();
DCHECK(membership_snapshot.get() != nullptr);
for (const auto& entry : membership_snapshot->current_backends) {
BackendDescriptorPB backend = entry.second;
@@ -1057,14 +1076,13 @@ void ImpalaHttpHandler::BackendsHandler(const Webserver::WebRequest& req,
void ImpalaHttpHandler::AdmissionStateHandler(
const Webserver::WebRequest& req, Document* document) {
const auto& args = req.parsed_args;
- AdmissionController* ac = ExecEnv::GetInstance()->admission_controller();
Webserver::ArgumentMap::const_iterator pool_name_arg = args.find("pool_name");
bool get_all_pools = (pool_name_arg == args.end());
Value resource_pools(kArrayType);
if (get_all_pools) {
- ac->AllPoolsToJson( &resource_pools, document);
+ admission_controller_->AllPoolsToJson(&resource_pools, document);
} else {
- ac->PoolToJson(pool_name_arg->second, &resource_pools, document);
+ admission_controller_->PoolToJson(pool_name_arg->second, &resource_pools, document);
}
// Now get running queries from CRS map.
@@ -1077,8 +1095,8 @@ void ImpalaHttpHandler::AdmissionStateHandler(
unsigned long num_backends;
};
unordered_map<string, vector<QueryInfo>> running_queries;
- if (FLAGS_is_admission_controller) {
- ExecEnv::GetInstance()
+ if (is_admissiond_) {
+ AdmissiondEnv::GetInstance()
->admission_control_service()
->admission_state_map_.DoFuncForAllEntries(
[&running_queries](
@@ -1148,7 +1166,8 @@ void ImpalaHttpHandler::AdmissionStateHandler(
"running_queries", queries_in_pool, document->GetAllocator());
}
int64_t ms_since_last_statestore_update;
- string staleness_detail = ac->GetStalenessDetail("", &ms_since_last_statestore_update);
+ string staleness_detail =
+ admission_controller_->GetStalenessDetail("", &ms_since_last_statestore_update);
// In order to embed a plain json inside the webpage generated by mustache, we need
// to stringify it and write it out as a json element. We do not need to pretty-print
@@ -1178,9 +1197,8 @@ void ImpalaHttpHandler::ResetResourcePoolStatsHandler(
Webserver::ArgumentMap::const_iterator pool_name_arg = args.find("pool_name");
bool reset_all_pools = (pool_name_arg == args.end());
if (reset_all_pools) {
- ExecEnv::GetInstance()->admission_controller()->ResetAllPoolInformationalStats();
+ admission_controller_->ResetAllPoolInformationalStats();
} else {
- ExecEnv::GetInstance()->admission_controller()->ResetPoolInformationalStats(
- pool_name_arg->second);
+ admission_controller_->ResetPoolInformationalStats(pool_name_arg->second);
}
}
diff --git a/be/src/service/impala-http-handler.h b/be/src/service/impala-http-handler.h
index a491f11..a0eb747 100644
--- a/be/src/service/impala-http-handler.h
+++ b/be/src/service/impala-http-handler.h
@@ -28,19 +28,43 @@ using kudu::HttpStatusCode;
namespace impala {
+class AdmissionController;
+class ClusterMembershipMgr;
+
/// Handles all webserver callbacks for an ImpalaServer. This class is a friend of
/// ImpalaServer in order to access the internal state needed to generate the debug
/// webpages.
class ImpalaHttpHandler {
public:
- ImpalaHttpHandler(ImpalaServer* server) : server_(server) { }
+ static ImpalaHttpHandler* CreateImpaladHandler(ImpalaServer* server,
+ AdmissionController* admission_controller,
+ ClusterMembershipMgr* cluster_membership_mgr) {
+ return new ImpalaHttpHandler(
+ server, admission_controller, cluster_membership_mgr, false);
+ }
+
+ static ImpalaHttpHandler* CreateAdmissiondHandler(
+ AdmissionController* admission_controller,
+ ClusterMembershipMgr* cluster_membership_mgr) {
+ return new ImpalaHttpHandler(
+ nullptr, admission_controller, cluster_membership_mgr, true);
+ }
/// Registers per-Impalad webserver callbacks. If 'metrics_only' is true, only registers
/// the callbacks needed by the metrics server, i.e. /healthz.
void RegisterHandlers(Webserver* webserver, bool metrics_only = false);
private:
+ ImpalaHttpHandler(ImpalaServer* server, AdmissionController* admission_controller,
+ ClusterMembershipMgr* cluster_membership_mgr, bool is_admissiond);
+
ImpalaServer* server_;
+ AdmissionController* admission_controller_;
+ ClusterMembershipMgr* cluster_membership_mgr_;
+
+ /// If true, this is an admissiond and we'll only expose admission related endpoints,
+ /// otherwise its an impalad.
+ bool is_admissiond_;
/// Raw callback to indicate whether the server is ready to accept queries.
void HealthzHandler(const Webserver::WebRequest& req, std::stringstream* data,
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 347c9ba..1812d5d 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -2710,7 +2710,8 @@ Status ImpalaServer::Start(
// We must register the HTTP handlers after registering the ImpalaServer with the
// ExecEnv. Otherwise the HTTP handlers will try to resolve the ImpalaServer through the
// ExecEnv singleton and will receive a nullptr.
- http_handler_.reset(new ImpalaHttpHandler(this));
+ http_handler_.reset(ImpalaHttpHandler::CreateImpaladHandler(
+ this, exec_env_->admission_controller(), exec_env_->cluster_membership_mgr()));
http_handler_->RegisterHandlers(exec_env_->webserver());
if (exec_env_->metrics_webserver() != nullptr) {
http_handler_->RegisterHandlers(
diff --git a/be/src/util/mem-info.cc b/be/src/util/mem-info.cc
index a3f9d9d..df932cb 100644
--- a/be/src/util/mem-info.cc
+++ b/be/src/util/mem-info.cc
@@ -28,6 +28,8 @@
#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>
+#include "util/cgroup-util.h"
+#include "util/parse-util.h"
#include "util/pretty-printer.h"
#include "common/names.h"
@@ -36,6 +38,8 @@ using boost::algorithm::is_any_of;
using boost::algorithm::split;
using boost::algorithm::token_compress_on;
+DECLARE_string(mem_limit);
+
namespace impala {
bool MemInfo::initialized_ = false;
@@ -193,4 +197,71 @@ string ThpConfig::DebugString() const {
stream << " khugepaged defrag: " << khugepaged_defrag << endl;
return stream.str();
}
+
+Status ChooseProcessMemLimit(int64_t* bytes_limit) {
+ // Depending on the system configuration, we detect the total amount of memory
+ // available to the system - either the available physical memory, or if overcommitting
+ // is turned off, we use the memory commit limit from /proc/meminfo (see IMPALA-1690).
+ // The 'memory' CGroup can also impose a lower limit on memory consumption,
+ // so we take the minimum of the system memory and the CGroup memory limit.
+ int64_t avail_mem = MemInfo::physical_mem();
+ bool use_commit_limit =
+ MemInfo::vm_overcommit() == 2 && MemInfo::commit_limit() < MemInfo::physical_mem();
+ if (use_commit_limit) {
+ avail_mem = MemInfo::commit_limit();
+ // There might be the case of misconfiguration, when on a system swap is disabled
+ // and overcommitting is turned off the actual usable memory is less than the
+ // available physical memory.
+ LOG(WARNING) << "This system shows a discrepancy between the available "
+ << "memory and the memory commit limit allowed by the "
+ << "operating system. ( Mem: " << MemInfo::physical_mem()
+ << "<=> CommitLimit: " << MemInfo::commit_limit() << "). "
+ << "Impala will adhere to the smaller value when setting the process "
+ << "memory limit. Please verify the system configuration. Specifically, "
+ << "/proc/sys/vm/overcommit_memory and "
+ << "/proc/sys/vm/overcommit_ratio.";
+ }
+ LOG(INFO) << "System memory available: " << PrettyPrinter::PrintBytes(avail_mem)
+ << " (from " << (use_commit_limit ? "commit limit" : "physical mem") << ")";
+ int64_t cgroup_mem_limit;
+ Status cgroup_mem_status = CGroupUtil::FindCGroupMemLimit(&cgroup_mem_limit);
+ if (cgroup_mem_status.ok()) {
+ if (cgroup_mem_limit < avail_mem) {
+ avail_mem = cgroup_mem_limit;
+ LOG(INFO) << "CGroup memory limit for this process reduces physical memory "
+ << "available to: " << PrettyPrinter::PrintBytes(avail_mem);
+ }
+ } else {
+ LOG(WARNING) << "Could not detect CGroup memory limit, assuming unlimited: $0"
+ << cgroup_mem_status.GetDetail();
+ }
+ bool is_percent;
+ *bytes_limit = ParseUtil::ParseMemSpec(FLAGS_mem_limit, &is_percent, avail_mem);
+ // ParseMemSpec() returns -1 for invalid input and 0 to mean unlimited. From Impala
+ // 2.11 onwards we do not support unlimited process memory limits.
+ if (*bytes_limit <= 0) {
+ return Status(Substitute("The process memory limit (--mem_limit) must be a positive "
+ "bytes value or percentage: $0",
+ FLAGS_mem_limit));
+ }
+ if (is_percent) {
+ LOG(INFO) << "Using process memory limit: " << PrettyPrinter::PrintBytes(*bytes_limit)
+ << " (--mem_limit=" << FLAGS_mem_limit << " of "
+ << PrettyPrinter::PrintBytes(avail_mem) << ")";
+ } else {
+ LOG(INFO) << "Using process memory limit: " << PrettyPrinter::PrintBytes(*bytes_limit)
+ << " (--mem_limit=" << FLAGS_mem_limit << ")";
+ }
+ if (*bytes_limit > MemInfo::physical_mem()) {
+ LOG(WARNING) << "Process memory limit " << PrettyPrinter::PrintBytes(*bytes_limit)
+ << " exceeds physical memory of "
+ << PrettyPrinter::PrintBytes(MemInfo::physical_mem());
+ }
+ if (cgroup_mem_status.ok() && *bytes_limit > cgroup_mem_limit) {
+ LOG(WARNING) << "Process Memory limit " << PrettyPrinter::PrintBytes(*bytes_limit)
+ << " exceeds CGroup memory limit of "
+ << PrettyPrinter::PrintBytes(cgroup_mem_limit);
+ }
+ return Status::OK();
+}
}
diff --git a/be/src/util/mem-info.h b/be/src/util/mem-info.h
index cf4d23e..f36e8f7 100644
--- a/be/src/util/mem-info.h
+++ b/be/src/util/mem-info.h
@@ -21,6 +21,7 @@
#include <string>
#include "common/logging.h"
+#include "common/status.h"
namespace impala {
@@ -120,4 +121,10 @@ class MemInfo {
/// If overcommit is turned off the maximum allocatable memory
static int64_t commit_limit_;
};
+
+/// Choose a memory limit (returned in *bytes_limit) based on the --mem_limit flag and
+/// the memory available to the daemon process. Returns an error if the memory limit is
+/// invalid or another error is encountered that should prevent starting up the daemon.
+/// Logs the memory limit chosen and any relevant diagnostics related to that choice.
+Status ChooseProcessMemLimit(int64_t* bytes_limit);
}
diff --git a/bin/start-impala-cluster.py b/bin/start-impala-cluster.py
index 15abdf1..7767730 100755
--- a/bin/start-impala-cluster.py
+++ b/bin/start-impala-cluster.py
@@ -39,14 +39,15 @@ from tests.common.impala_cluster import (ImpalaCluster, DEFAULT_BEESWAX_PORT,
DEFAULT_HS2_PORT, DEFAULT_KRPC_PORT, DEFAULT_HS2_HTTP_PORT,
DEFAULT_STATE_STORE_SUBSCRIBER_PORT, DEFAULT_IMPALAD_WEBSERVER_PORT,
DEFAULT_STATESTORED_WEBSERVER_PORT, DEFAULT_CATALOGD_WEBSERVER_PORT,
- DEFAULT_CATALOGD_JVM_DEBUG_PORT, DEFAULT_IMPALAD_JVM_DEBUG_PORT,
- find_user_processes, run_daemon)
+ DEFAULT_ADMISSIOND_WEBSERVER_PORT, DEFAULT_CATALOGD_JVM_DEBUG_PORT,
+ DEFAULT_IMPALAD_JVM_DEBUG_PORT, find_user_processes, run_daemon)
LOG = logging.getLogger(os.path.splitext(os.path.basename(__file__))[0])
LOG.setLevel(level=logging.DEBUG)
KUDU_MASTER_HOSTS = os.getenv("KUDU_MASTER_HOSTS", "127.0.0.1")
DEFAULT_IMPALA_MAX_LOG_FILES = os.environ.get("IMPALA_MAX_LOG_FILES", 10)
+INTERNAL_LISTEN_HOST = os.getenv("INTERNAL_LISTEN_HOST", "localhost")
# Options
parser = OptionParser()
@@ -69,6 +70,9 @@ parser.add_option("--state_store_args", dest="state_store_args", action="append"
parser.add_option("--catalogd_args", dest="catalogd_args", action="append",
type="string", default=[],
help="Additional arguments to pass to the Catalog Service at startup")
+parser.add_option("--admissiond_args", dest="admissiond_args",
+ action="append", type="string", default=[], help="Additional arguments "
+ "to pass to the Admission Control Service at startup")
parser.add_option("--kill", "--kill_only", dest="kill_only", action="store_true",
default=False, help="Instead of starting the cluster, just kill all"
" the running impalads and the statestored.")
@@ -128,6 +132,11 @@ parser.add_option("--data_cache_eviction_policy", dest="data_cache_eviction_poli
parser.add_option("--data_cache_enable_tracing", dest="data_cache_enable_tracing",
action="store_true", default=False,
help="If the data cache is enabled, this enables tracing accesses.")
+parser.add_option("--enable_admission_service", dest="enable_admission_service",
+ action="store_true", default=False,
+ help="If true, enables the Admissison Control Service - the cluster "
+ "will be launched with an admissiond and all coordinators configured "
+ "to use it for admission control.")
# For testing: list of comma-separated delays, in milliseconds, that delay impalad catalog
# replica initialization. The ith delay is applied to the ith impalad.
@@ -283,13 +292,20 @@ def build_catalogd_arg_list():
combine_arg_list_opts(options.catalogd_args))
+def build_admissiond_arg_list():
+ """Build a list of command line arguments to pass to the admissiond."""
+ return (build_logging_args("admissiond") + build_kerberos_args("admissiond") +
+ combine_arg_list_opts(options.admissiond_args))
+
+
def build_impalad_arg_lists(cluster_size, num_coordinators, use_exclusive_coordinators,
- remap_ports, start_idx=0):
+ remap_ports, start_idx=0, admissiond_host=INTERNAL_LISTEN_HOST):
"""Build the argument lists for impala daemons in the cluster. Returns a list of
argument lists, one for each impala daemon in the cluster. Each argument list is
a list of strings. 'num_coordinators' and 'use_exclusive_coordinators' allow setting
up the cluster with dedicated coordinators. If 'remap_ports' is true, the impalad
- ports are changed from their default values to avoid port conflicts."""
+ ports are changed from their default values to avoid port conflicts. If the admission
+ service is enabled, 'admissiond_host' is the hostname for the admissiond."""
# TODO: currently we build a big string blob then split it. It would be better to
# build up the lists directly.
@@ -396,6 +412,10 @@ def build_impalad_arg_lists(cluster_size, num_coordinators, use_exclusive_coordi
tracing_args=tracing_args)
args = "{tracing_args} {args}".format(tracing_args=tracing_args, args=args)
+ if options.enable_admission_service:
+ args = "{args} -admission_service_host={host}".format(
+ args=args, host=admissiond_host)
+
# Appended at the end so they can override previous args.
if i < len(per_impalad_args):
args = "{args} {per_impalad_args}".format(
@@ -406,10 +426,10 @@ def build_impalad_arg_lists(cluster_size, num_coordinators, use_exclusive_coordi
def build_kerberos_args(daemon):
"""If the cluster is kerberized, returns arguments to pass to daemon process.
- daemon should either be "impalad", "catalogd" or "statestored"."""
+ daemon should either be "impalad", "catalogd", "statestored", or "admissiond"."""
# Note: this code has probably bit-rotted but is preserved in case someone needs to
# revive the kerberized minicluster.
- assert daemon in ("impalad", "catalogd", "statestored")
+ assert daemon in ("impalad", "catalogd", "statestored", "admissiond")
if call([os.path.join(IMPALA_HOME, "testdata/cluster/admin"), "is_kerberized"]) != 0:
return []
args = ["-keytab_file={0}".format(os.getenv("KRB5_KTNAME")),
@@ -450,10 +470,10 @@ class MiniClusterOperations(object):
"""
def get_cluster(self):
"""Return an ImpalaCluster instance."""
- return ImpalaCluster()
+ return ImpalaCluster(use_admission_service=options.enable_admission_service)
def kill_all_daemons(self, force=False):
- kill_matching_processes(["catalogd", "impalad", "statestored"], force)
+ kill_matching_processes(["catalogd", "impalad", "statestored", "admissiond"], force)
def kill_all_impalads(self, force=False):
kill_matching_processes(["impalad"], force=force)
@@ -464,6 +484,9 @@ class MiniClusterOperations(object):
def kill_statestored(self, force=False):
kill_matching_processes(["statestored"], force=force)
+ def kill_admissiond(self, force=False):
+ kill_matching_processes(["admissiond"], force=force)
+
def start_statestore(self):
LOG.info("Starting State Store logging to {log_dir}/statestored.INFO".format(
log_dir=options.log_dir))
@@ -483,6 +506,15 @@ class MiniClusterOperations(object):
raise RuntimeError("Unable to start catalogd. Check log or file permissions"
" for more details.")
+ def start_admissiond(self):
+ LOG.info("Starting Admission Control Service logging to {log_dir}/admissiond.INFO"
+ .format(log_dir=options.log_dir))
+ output_file = os.path.join(options.log_dir, "admissiond-out.log")
+ run_daemon_with_options("admissiond", build_admissiond_arg_list(), output_file)
+ if not check_process_exists("admissiond", 10):
+ raise RuntimeError("Unable to start admissiond. Check log or file permissions"
+ " for more details.")
+
def start_impalads(self, cluster_size, num_coordinators, use_exclusive_coordinators,
start_idx=0):
"""Start 'cluster_size' impalad instances. The first 'num_coordinator' instances will
@@ -531,11 +563,13 @@ class DockerMiniClusterOperations(object):
def get_cluster(self):
"""Return an ImpalaCluster instance."""
- return ImpalaCluster(docker_network=self.network_name)
+ return ImpalaCluster(docker_network=self.network_name,
+ use_admission_service=options.enable_admission_service)
def kill_all_daemons(self, force=False):
self.kill_statestored(force=force)
self.kill_catalogd(force=force)
+ self.kill_admissiond(force=force)
self.kill_all_impalads(force=force)
def kill_all_impalads(self, force=False):
@@ -554,6 +588,9 @@ class DockerMiniClusterOperations(object):
def kill_statestored(self, force=False):
self.__stop_container__("statestored")
+ def kill_admissiond(self, force=False):
+ self.__stop_container__("admissiond")
+
def start_statestore(self):
self.__run_container__("statestored", build_statestored_arg_list(),
{DEFAULT_STATESTORED_WEBSERVER_PORT: DEFAULT_STATESTORED_WEBSERVER_PORT})
@@ -562,9 +599,13 @@ class DockerMiniClusterOperations(object):
self.__run_container__("catalogd", build_catalogd_arg_list(),
{DEFAULT_CATALOGD_WEBSERVER_PORT: DEFAULT_CATALOGD_WEBSERVER_PORT})
+ def start_admissiond(self):
+ self.__run_container__("admissiond", build_admissiond_arg_list(),
+ {DEFAULT_ADMISSIOND_WEBSERVER_PORT: DEFAULT_ADMISSIOND_WEBSERVER_PORT})
+
def start_impalads(self, cluster_size, num_coordinators, use_exclusive_coordinators):
- impalad_arg_lists = build_impalad_arg_lists(
- cluster_size, num_coordinators, use_exclusive_coordinators, remap_ports=False)
+ impalad_arg_lists = build_impalad_arg_lists(cluster_size, num_coordinators,
+ use_exclusive_coordinators, remap_ports=False, admissiond_host="admissiond")
assert cluster_size == len(impalad_arg_lists)
mem_limit = compute_impalad_mem_limit(cluster_size)
for i in xrange(cluster_size):
@@ -774,6 +815,8 @@ if __name__ == "__main__":
else:
cluster_ops.start_statestore()
cluster_ops.start_catalogd()
+ if options.enable_admission_service:
+ cluster_ops.start_admissiond()
cluster_ops.start_impalads(options.cluster_size, options.num_coordinators,
options.use_exclusive_coordinators)
# Sleep briefly to reduce log spam: the cluster takes some time to start up.
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index 560a494..b77f4ba 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -2754,5 +2754,15 @@
"units": "NONE",
"kind": "COUNTER",
"key": "admission-controller.total-dequeue-failed-coordinator-limited"
+ },
+ {
+ "description": "The full version string of the Admission Control Server.",
+ "contexts": [
+ "ADMISSIOND"
+ ],
+ "label": "Admission Control Service Version",
+ "units": "NONE",
+ "kind": "PROPERTY",
+ "key": "admissiond.version"
}
]
diff --git a/docker/CMakeLists.txt b/docker/CMakeLists.txt
index dfb6a33..60fd8c2 100644
--- a/docker/CMakeLists.txt
+++ b/docker/CMakeLists.txt
@@ -119,6 +119,7 @@ if (NOT ${DISTRO_BASE_IMAGE} STREQUAL "UNSUPPORTED")
add_daemon_docker_images(impalad_executor)
add_daemon_docker_images(catalogd)
add_daemon_docker_images(statestored)
+ add_daemon_docker_images(admissiond)
# Generate a text file with all of the release daemon images.
file(WRITE ${CMAKE_SOURCE_DIR}/docker/docker-images.txt "${daemon_image_names}")
diff --git a/docker/admissiond/Dockerfile b/docker/admissiond/Dockerfile
new file mode 100644
index 0000000..c2135a0
--- /dev/null
+++ b/docker/admissiond/Dockerfile
@@ -0,0 +1,32 @@
+# 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.
+
+ARG BASE_IMAGE=impala_base
+FROM ${BASE_IMAGE}
+
+# IMPALA-10068: Split out jars for admissiond Docker images
+# For now, we copy in all Impala dependencies. One way to decrease the admissiond image
+# size would be to only copy in jars required by admissiond.
+COPY --chown=impala lib /opt/impala/lib
+
+# Externally-facing ports
+# Debug webserver
+EXPOSE 25030
+
+ENTRYPOINT ["/opt/impala/bin/daemon_entrypoint.sh", "/opt/impala/bin/admissiond",\
+ "-log_dir=/opt/impala/logs", "-abort_on_config_error=false",\
+ "-state_store_host=statestored", "-use_resolved_hostname=true"]
diff --git a/docker/impala_base/Dockerfile b/docker/impala_base/Dockerfile
index 74db54b..955a46f 100644
--- a/docker/impala_base/Dockerfile
+++ b/docker/impala_base/Dockerfile
@@ -41,6 +41,7 @@ COPY --chown=impala www /opt/impala/www
COPY --chown=impala bin /opt/impala/bin
# Symlink here instead of in setup_build_context to avoid duplicate binaries.
RUN cd /opt/impala/bin && ln -s impalad statestored && ln -s impalad catalogd && \
+ ln -s impalad admissiond && \
# Create conf directory for later config injection.
mkdir /opt/impala/conf && \
# Create logs directory to collect container logs.
diff --git a/tests/common/custom_cluster_test_suite.py b/tests/common/custom_cluster_test_suite.py
index 6e83046..448a591 100644
--- a/tests/common/custom_cluster_test_suite.py
+++ b/tests/common/custom_cluster_test_suite.py
@@ -38,6 +38,7 @@ NUM_COORDINATORS = DEFAULT_CLUSTER_SIZE
IMPALAD_ARGS = 'impalad_args'
STATESTORED_ARGS = 'state_store_args'
CATALOGD_ARGS = 'catalogd_args'
+ADMISSIOND_ARGS = 'admissiond_args'
KUDU_ARGS = 'kudu_args'
# Additional args passed to the start-impala-cluster script.
START_ARGS = 'start_args'
@@ -135,7 +136,7 @@ class CustomClusterTestSuite(ImpalaTestSuite):
def setup_method(self, method):
cluster_args = list()
- for arg in [IMPALAD_ARGS, STATESTORED_ARGS, CATALOGD_ARGS]:
+ for arg in [IMPALAD_ARGS, STATESTORED_ARGS, CATALOGD_ARGS, ADMISSIOND_ARGS]:
if arg in method.func_dict:
cluster_args.append("--%s=%s " % (arg, method.func_dict[arg]))
if START_ARGS in method.func_dict:
@@ -285,6 +286,8 @@ class CustomClusterTestSuite(ImpalaTestSuite):
# cluster_size (# of impalad) + 1 (for catalogd).
if expected_subscribers == 0:
expected_subscribers = expected_num_executors + 1
+ if "--enable_admission_service" in options:
+ expected_subscribers += 1
statestored.service.wait_for_live_subscribers(expected_subscribers,
timeout=statestored_timeout_s)
diff --git a/tests/common/impala_cluster.py b/tests/common/impala_cluster.py
index 35face8..305b461 100644
--- a/tests/common/impala_cluster.py
+++ b/tests/common/impala_cluster.py
@@ -33,6 +33,7 @@ from time import sleep
import tests.common.environ
from tests.common.impala_service import (
+ AdmissiondService,
CatalogdService,
ImpaladService,
StateStoredService)
@@ -57,6 +58,7 @@ DEFAULT_STATE_STORE_SUBSCRIBER_PORT = 23000
DEFAULT_IMPALAD_WEBSERVER_PORT = 25000
DEFAULT_STATESTORED_WEBSERVER_PORT = 25010
DEFAULT_CATALOGD_WEBSERVER_PORT = 25020
+DEFAULT_ADMISSIOND_WEBSERVER_PORT = 25030
DEFAULT_IMPALAD_JVM_DEBUG_PORT = 30000
DEFAULT_CATALOGD_JVM_DEBUG_PORT = 30030
@@ -74,8 +76,9 @@ CLUSTER_WAIT_TIMEOUT_IN_SECONDS = 240
# * The docker minicluster with one container per process connected to a user-defined
# bridge network.
class ImpalaCluster(object):
- def __init__(self, docker_network=None):
+ def __init__(self, docker_network=None, use_admission_service=False):
self.docker_network = docker_network
+ self.use_admission_service = use_admission_service
self.refresh()
@classmethod
@@ -90,13 +93,18 @@ class ImpalaCluster(object):
Helpful to confirm that processes have been killed.
"""
if self.docker_network is None:
- self.__impalads, self.__statestoreds, self.__catalogd =\
+ self.__impalads, self.__statestoreds, self.__catalogd, self.__admissiond =\
self.__build_impala_process_lists()
else:
- self.__impalads, self.__statestoreds, self.__catalogd =\
+ self.__impalads, self.__statestoreds, self.__catalogd, self.__admissiond =\
self.__find_docker_containers()
- LOG.debug("Found %d impalad/%d statestored/%d catalogd process(es)" %
- (len(self.__impalads), len(self.__statestoreds), 1 if self.__catalogd else 0))
+ admissiond_str = ""
+ if self.use_admission_service:
+ admissiond_str = "/%d admissiond" % (1 if self.__admissiond else 0)
+
+ LOG.debug("Found %d impalad/%d statestored/%d catalogd%s process(es)" %
+ (len(self.__impalads), len(self.__statestoreds), 1 if self.__catalogd else 0,
+ admissiond_str))
@property
def statestored(self):
@@ -119,6 +127,11 @@ class ImpalaCluster(object):
"""Returns the catalogd process, or None if no catalogd process was found"""
return self.__catalogd
+ @property
+ def admissiond(self):
+ """Returns the admisisond process, or None if no admissiond process was found"""
+ return self.__admissiond
+
def get_first_impalad(self):
return self.impalads[0]
@@ -220,7 +233,9 @@ class ImpalaCluster(object):
impalads = list()
statestored = list()
catalogd = None
- for binary, process in find_user_processes(['impalad', 'catalogd', 'statestored']):
+ admissiond = None
+ daemons = ['impalad', 'catalogd', 'statestored', 'admissiond']
+ for binary, process in find_user_processes(daemons):
# IMPALA-6889: When a process shuts down and becomes a zombie its cmdline becomes
# empty for a brief moment, before it gets reaped by its parent (see man proc). We
# copy the cmdline to prevent it from changing between the following checks and
@@ -240,9 +255,11 @@ class ImpalaCluster(object):
statestored.append(StateStoreProcess(cmdline))
elif binary == 'catalogd':
catalogd = CatalogdProcess(cmdline)
+ elif binary == 'admissiond':
+ admissiond = AdmissiondProcess(cmdline)
self.__sort_impalads(impalads)
- return impalads, statestored, catalogd
+ return impalads, statestored, catalogd, admissiond
def __find_docker_containers(self):
"""
@@ -251,6 +268,7 @@ class ImpalaCluster(object):
impalads = []
statestoreds = []
catalogd = None
+ admissiond = None
output = check_output(["docker", "network", "inspect", self.docker_network])
# Only one network should be present in the top level array.
for container_id in json.loads(output)[0]["Containers"]:
@@ -277,8 +295,12 @@ class ImpalaCluster(object):
assert catalogd is None
catalogd = CatalogdProcess(args, container_id=container_id,
port_map=port_map)
+ elif executable == 'admissiond':
+ assert admissiond is None
+ admissiond = AdmissiondProcess(args, container_id=container_id,
+ port_map=port_map)
self.__sort_impalads(impalads)
- return impalads, statestoreds, catalogd
+ return impalads, statestoreds, catalogd, admissiond
def __sort_impalads(self, impalads):
"""Does an in-place sort of a list of ImpaladProcess objects into a canonical order.
@@ -548,6 +570,16 @@ class CatalogdProcess(BaseImpalaProcess):
expected_value=1, timeout=30)
+# Represents an admission control process.
+class AdmissiondProcess(BaseImpalaProcess):
+ def __init__(self, cmd, container_id=None, port_map=None):
+ super(AdmissiondProcess, self).__init__(cmd, container_id, port_map)
+ self.service = AdmissiondService(self.hostname, self.webserver_interface,
+ self.get_webserver_port(), self._get_webserver_certificate_file())
+
+ def _get_default_webserver_port(self):
+ return DEFAULT_ADMISSIOND_WEBSERVER_PORT
+
def find_user_processes(binaries):
"""Returns an iterator over all processes owned by the current user with a matching
binary name from the provided list. Return a iterable of tuples, with each tuple
diff --git a/tests/common/impala_service.py b/tests/common/impala_service.py
index 97e4d74..ef868ed 100644
--- a/tests/common/impala_service.py
+++ b/tests/common/impala_service.py
@@ -494,3 +494,10 @@ class CatalogdService(BaseImpalaService):
LOG.info('Catalogd version not yet available.')
sleep(interval)
assert False, 'Catalog version not ready in expected time.'
+
+
+class AdmissiondService(BaseImpalaService):
+ def __init__(self, hostname, webserver_interface, webserver_port,
+ webserver_certificate_file):
+ super(AdmissiondService, self).__init__(
+ hostname, webserver_interface, webserver_port, webserver_certificate_file)
diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py
index f2bfa0c..d67795a 100644
--- a/tests/custom_cluster/test_admission_controller.py
+++ b/tests/custom_cluster/test_admission_controller.py
@@ -1318,28 +1318,20 @@ class TestAdmissionControllerWithACService(TestAdmissionController):
minicluster configured to perform all admission control."""
def get_ac_process(self):
- return self.cluster.impalads[1]
+ return self.cluster.admissiond
def get_ac_log_name(self):
- return "impalad_node1"
+ return "admissiond"
def setup_method(self, method):
if self.exploration_strategy() != 'exhaustive':
pytest.skip('runs only in exhaustive')
- PER_IMPALAD_ACS_ARGS = [
- '--is_admission_controller=false',
- '--is_admission_controller=true',
- '--is_admission_controller=false',
- ]
if 'start_args' not in method.func_dict:
method.func_dict['start_args'] = list()
- method.func_dict['start_args'].append(
- "--per_impalad_args=" + ";".join(PER_IMPALAD_ACS_ARGS))
- if 'impalad_args' not in method.func_dict:
- method.func_dict["impalad_args"] = ""
- method.func_dict["impalad_args"] +=\
- " --admission_control_service_addr=127.0.0.1:27001 "
+ method.func_dict["start_args"].append("--enable_admission_service")
+ if "impalad_args" in method.func_dict:
+ method.func_dict["admissiond_args"] = method.func_dict["impalad_args"]
super(TestAdmissionController, self).setup_method(method)
@@ -1930,26 +1922,17 @@ class TestAdmissionControllerStressWithACService(TestAdmissionControllerStress):
in the minicluster configured to perform all admission control."""
def get_ac_processes(self):
- return [self.cluster.impalads[1]]
+ return [self.cluster.admissiond]
def get_ac_log_name(self):
- return "impalad_node1"
+ return "admissiond"
def setup_method(self, method):
if self.exploration_strategy() != 'exhaustive':
pytest.skip('runs only in exhaustive')
-
- PER_IMPALAD_ACS_ARGS = [
- '--is_admission_controller=false',
- '--is_admission_controller=true',
- '--is_admission_controller=false',
- ]
if 'start_args' not in method.func_dict:
method.func_dict['start_args'] = list()
- method.func_dict['start_args'].append(
- "--per_impalad_args=" + ";".join(PER_IMPALAD_ACS_ARGS))
- if 'impalad_args' not in method.func_dict:
- method.func_dict["impalad_args"] = ""
- method.func_dict["impalad_args"] +=\
- " --admission_control_service_addr=127.0.0.1:27001 "
+ method.func_dict["start_args"].append("--enable_admission_service")
+ if "impalad_args" in method.func_dict:
+ method.func_dict["admissiond_args"] = method.func_dict["impalad_args"]
super(TestAdmissionControllerStress, self).setup_method(method)