| // 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 "service/impala-http-handler.h" |
| |
| #include <algorithm> |
| #include <sstream> |
| #include <boost/lexical_cast.hpp> |
| #include <boost/thread/mutex.hpp> |
| #include <rapidjson/prettywriter.h> |
| #include <rapidjson/stringbuffer.h> |
| |
| #include "catalog/catalog-util.h" |
| #include "gen-cpp/beeswax_types.h" |
| #include "gutil/strings/join.h" |
| #include "gutil/strings/substitute.h" |
| #include "runtime/coordinator.h" |
| #include "runtime/exec-env.h" |
| #include "runtime/mem-tracker.h" |
| #include "runtime/query-state.h" |
| #include "runtime/timestamp-value.h" |
| #include "runtime/timestamp-value.inline.h" |
| #include "scheduling/admission-controller.h" |
| #include "scheduling/cluster-membership-mgr.h" |
| #include "service/client-request-state.h" |
| #include "service/frontend.h" |
| #include "service/impala-server.h" |
| #include "thrift/protocol/TDebugProtocol.h" |
| #include "util/coding-util.h" |
| #include "util/logging-support.h" |
| #include "util/pretty-printer.h" |
| #include "util/redactor.h" |
| #include "util/summary-util.h" |
| #include "util/time.h" |
| #include "util/uid-util.h" |
| #include "util/webserver.h" |
| |
| #include "common/names.h" |
| |
| using namespace apache::hive::service::cli::thrift; |
| using namespace apache::thrift; |
| using namespace beeswax; |
| using namespace impala; |
| using namespace rapidjson; |
| using namespace strings; |
| |
| DECLARE_int32(query_log_size); |
| DECLARE_bool(use_local_catalog); |
| |
| namespace { |
| |
| // Helper method to turn a class + a method to invoke into a UrlCallback |
| template<typename T, class F> |
| Webserver::UrlCallback MakeCallback(T* caller, const F& fnc) { |
| return [caller, fnc](const auto& req, auto* doc) { |
| (caller->*fnc)(req, doc); |
| }; |
| } |
| |
| // We expect the id to be passed as one parameter. Eg: 'query_id' or 'session_id'. |
| // Returns true if the id was present and valid; false otherwise. |
| static Status ParseIdFromRequest(const Webserver::WebRequest& req, TUniqueId* id, |
| const std::string &to_find) { |
| const auto& args = req.parsed_args; |
| Webserver::ArgumentMap::const_iterator it = args.find(to_find); |
| if (it == args.end()) { |
| return Status(Substitute("No '$0' argument found.", to_find)); |
| } else { |
| if (ParseId(it->second, id)) return Status::OK(); |
| return Status(Substitute("Could not parse '$0' argument: $1", to_find, it->second)); |
| } |
| } |
| |
| } |
| |
| void ImpalaHttpHandler::RegisterHandlers(Webserver* webserver) { |
| DCHECK(webserver != NULL); |
| |
| Webserver::RawUrlCallback healthz_callback = |
| [this](const auto& req, auto* data, auto* response) { |
| return this->HealthzHandler(req, data, response); |
| }; |
| webserver->RegisterUrlCallback("/healthz", healthz_callback); |
| |
| webserver->RegisterUrlCallback("/backends", "backends.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::BackendsHandler), true); |
| |
| webserver->RegisterUrlCallback("/hadoop-varz", "hadoop-varz.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::HadoopVarzHandler), true); |
| |
| webserver->RegisterUrlCallback("/queries", "queries.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::QueryStateHandler), true); |
| |
| webserver->RegisterUrlCallback("/sessions", "sessions.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::SessionsHandler), true); |
| |
| webserver->RegisterUrlCallback("/catalog", "catalog.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::CatalogHandler), true); |
| |
| if(!FLAGS_use_local_catalog) { |
| // The /catalog_object endpoint is disabled if local_catalog_mode is used |
| // since metadata is partially fetched on demand |
| webserver->RegisterUrlCallback("/catalog_object", "catalog_object.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::CatalogObjectsHandler), false); |
| } |
| |
| webserver->RegisterUrlCallback("/query_profile", "query_profile.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::QueryProfileHandler), false); |
| |
| webserver->RegisterUrlCallback("/query_memory", "query_memory.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::QueryMemoryHandler), false); |
| |
| webserver->RegisterUrlCallback("/query_backends", "query_backends.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::QueryBackendsHandler), false); |
| |
| webserver->RegisterUrlCallback("/query_finstances", "query_finstances.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::QueryFInstancesHandler), false); |
| |
| webserver->RegisterUrlCallback("/cancel_query", "common-pre.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::CancelQueryHandler), false); |
| |
| webserver->RegisterUrlCallback("/close_session", "common-pre.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::CloseSessionHandler), false); |
| |
| webserver->RegisterUrlCallback("/query_profile_encoded", "raw_text.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::QueryProfileEncodedHandler), false); |
| |
| webserver->RegisterUrlCallback("/query_profile_plain_text", "raw_text.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::QueryProfileTextHandler), false); |
| |
| webserver->RegisterUrlCallback("/query_profile_json", "raw_text.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::QueryProfileJsonHandler), false); |
| |
| webserver->RegisterUrlCallback("/inflight_query_ids", "raw_text.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::InflightQueryIdsHandler), false); |
| |
| webserver->RegisterUrlCallback("/query_summary", "query_summary.tmpl", |
| [this](const auto& req, auto* doc) { |
| this->QuerySummaryHandler(false, true, req, doc); }, false); |
| |
| webserver->RegisterUrlCallback("/query_plan", "query_plan.tmpl", |
| [this](const auto& req, auto* doc) { |
| this->QuerySummaryHandler(true, true, req, doc); }, false); |
| |
| webserver->RegisterUrlCallback("/query_plan_text", "query_plan_text.tmpl", |
| [this](const auto& req, auto* doc) { |
| this->QuerySummaryHandler(false, false, req, doc); }, false); |
| |
| webserver->RegisterUrlCallback("/query_stmt", "query_stmt.tmpl", |
| [this](const auto& req, auto* doc) { |
| this->QuerySummaryHandler(false, false, req, doc); }, false); |
| |
| webserver->RegisterUrlCallback("/admission", "admission_controller.tmpl", |
| MakeCallback(this, &ImpalaHttpHandler::AdmissionStateHandler), true); |
| |
| webserver->RegisterUrlCallback("/resource_pool_reset", "", |
| MakeCallback(this, &ImpalaHttpHandler::ResetResourcePoolStatsHandler), false); |
| |
| RegisterLogLevelCallbacks(webserver, true); |
| } |
| |
| void ImpalaHttpHandler::HealthzHandler(const Webserver::WebRequest& req, |
| std::stringstream* data, HttpStatusCode* response) { |
| if (server_->IsHealthy()) { |
| (*data) << "OK"; |
| *response = HttpStatusCode::Ok; |
| return; |
| } |
| *(data) << "Not Available"; |
| *response = HttpStatusCode::ServiceUnavailable; |
| } |
| |
| void ImpalaHttpHandler::HadoopVarzHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| TGetAllHadoopConfigsResponse response; |
| Status status = server_->exec_env_->frontend()->GetAllHadoopConfigs(&response); |
| if (!status.ok()) return; |
| |
| Value configs(kArrayType); |
| typedef map<string, string> ConfigMap; |
| for (const ConfigMap::value_type& config: response.configs) { |
| Value key(config.first.c_str(), document->GetAllocator()); |
| Value value(config.second.c_str(), document->GetAllocator()); |
| Value config_json(kObjectType); |
| config_json.AddMember("key", key, document->GetAllocator()); |
| config_json.AddMember("value", value, document->GetAllocator()); |
| configs.PushBack(config_json, document->GetAllocator()); |
| } |
| document->AddMember("configs", configs, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::CancelQueryHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| TUniqueId unique_id; |
| Status status = ParseIdFromRequest(req, &unique_id, "query_id"); |
| if (!status.ok()) { |
| Value error(status.GetDetail().c_str(), document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| return; |
| } |
| Status cause("Cancelled from Impala's debug web interface"); |
| // Web UI doesn't have access to secret so we can't validate it. We assume that |
| // web UI is allowed to close queries. |
| status = server_->UnregisterQuery(unique_id, true, &cause); |
| if (!status.ok()) { |
| Value error(status.GetDetail().c_str(), document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| return; |
| } |
| Value message("Query cancellation successful", document->GetAllocator()); |
| document->AddMember("contents", message, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::CloseSessionHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| TUniqueId unique_id; |
| Status status = ParseIdFromRequest(req, &unique_id, "session_id"); |
| if (!status.ok()) { |
| Value error(status.GetDetail().c_str(), document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| return; |
| } |
| Status cause("Session closed from Impala's debug web interface"); |
| // Web UI doesn't have access to secret so we can't validate it. We assume that |
| // web UI is allowed to close sessions. |
| status = server_->CloseSessionInternal(unique_id, |
| ImpalaServer::SecretArg::SkipSecretCheck(), /* ignore_if_absent= */ false); |
| if (!status.ok()) { |
| Value error(status.GetDetail().c_str(), document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| return; |
| } |
| stringstream ss; |
| ss << "Session " << PrintId(unique_id) << " closed successfully"; |
| Value message(ss.str().c_str(), document->GetAllocator()); |
| document->AddMember("contents", message, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::QueryProfileHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| TUniqueId unique_id; |
| Status parse_status = ParseIdFromRequest(req, &unique_id, "query_id"); |
| if (!parse_status.ok()) { |
| Value error(parse_status.GetDetail().c_str(), document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| return; |
| } |
| |
| stringstream ss; |
| Status status = server_->GetRuntimeProfileOutput( |
| unique_id, "", TRuntimeProfileFormat::STRING, &ss, nullptr, nullptr); |
| if (!status.ok()) { |
| Value error(status.GetDetail().c_str(), document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| return; |
| } |
| |
| Value profile(ss.str().c_str(), document->GetAllocator()); |
| document->AddMember("profile", profile, document->GetAllocator()); |
| const auto& args = req.parsed_args; |
| Value query_id(args.find("query_id")->second.c_str(), document->GetAllocator()); |
| document->AddMember("query_id", query_id, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::QueryProfileHelper(const Webserver::WebRequest& req, |
| Document* document, TRuntimeProfileFormat::type format) { |
| TUniqueId unique_id; |
| stringstream ss; |
| Status status = ParseIdFromRequest(req, &unique_id, "query_id"); |
| if (!status.ok()) { |
| ss << status.GetDetail(); |
| } else { |
| Status status = server_->GetRuntimeProfileOutput( |
| unique_id, "", format, &ss, nullptr, document); |
| if (!status.ok()) { |
| ss.str(Substitute("Could not obtain runtime profile: $0", status.GetDetail())); |
| } |
| } |
| // JSON format contents already been added inside document in GetRuntimeProfileOutput() |
| if (format != TRuntimeProfileFormat::JSON){ |
| Value profile(ss.str().c_str(), document->GetAllocator()); |
| document->AddMember("contents", profile, document->GetAllocator()); |
| } |
| } |
| |
| void ImpalaHttpHandler::QueryProfileEncodedHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| QueryProfileHelper(req, document, TRuntimeProfileFormat::BASE64); |
| document->AddMember(rapidjson::StringRef(Webserver::ENABLE_RAW_HTML_KEY), true, |
| document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::QueryProfileTextHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| QueryProfileHelper(req, document, TRuntimeProfileFormat::STRING); |
| document->AddMember(rapidjson::StringRef(Webserver::ENABLE_RAW_HTML_KEY), true, |
| document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::QueryProfileJsonHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| QueryProfileHelper(req, document, TRuntimeProfileFormat::JSON); |
| document->AddMember(rapidjson::StringRef(Webserver::ENABLE_PLAIN_JSON_KEY), true, |
| document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::InflightQueryIdsHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| stringstream ss; |
| server_->client_request_state_map_.DoFuncForAllEntries( |
| [&](const std::shared_ptr<ClientRequestState>& request_state) { |
| ss << PrintId(request_state->query_id()) << "\n"; |
| }); |
| document->AddMember(rapidjson::StringRef(Webserver::ENABLE_RAW_HTML_KEY), true, |
| document->GetAllocator()); |
| Value query_ids(ss.str().c_str(), document->GetAllocator()); |
| document->AddMember("contents", query_ids, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::QueryMemoryHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| TUniqueId unique_id; |
| Status parse_status = ParseIdFromRequest(req, &unique_id, "query_id"); |
| if (!parse_status.ok()) { |
| Value error(parse_status.GetDetail().c_str(), document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| return; |
| } |
| QueryState::ScopedRef qs(unique_id); |
| string mem_usage_text; |
| // Only queries that have started execution have a MemTracker to get usage from. |
| if (qs.get() != nullptr) { |
| mem_usage_text = qs->query_mem_tracker()->LogUsage(MemTracker::UNLIMITED_DEPTH); |
| } else { |
| mem_usage_text = "The query has either finished or has not started execution yet, " |
| "current memory consumption is not available."; |
| } |
| |
| Value mem_usage(mem_usage_text.c_str(), document->GetAllocator()); |
| document->AddMember("mem_usage", mem_usage, document->GetAllocator()); |
| const auto& args = req.parsed_args; |
| Value query_id(args.find("query_id")->second.c_str(), document->GetAllocator()); |
| document->AddMember("query_id", query_id, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::QueryStateToJson(const ImpalaServer::QueryStateRecord& record, |
| Value* value, Document* document) { |
| Value user(record.effective_user.c_str(), document->GetAllocator()); |
| value->AddMember("effective_user", user, document->GetAllocator()); |
| |
| Value default_db(record.default_db.c_str(), document->GetAllocator()); |
| value->AddMember("default_db", default_db, document->GetAllocator()); |
| |
| // Redact the query string |
| Value stmt(RedactCopy(record.stmt).c_str(), document->GetAllocator()); |
| value->AddMember("stmt", stmt, document->GetAllocator()); |
| |
| Value stmt_type(_TStmtType_VALUES_TO_NAMES.find(record.stmt_type)->second, |
| document->GetAllocator()); |
| value->AddMember("stmt_type", stmt_type, document->GetAllocator()); |
| |
| Value start_time(ToStringFromUnixMicros(record.start_time_us, |
| TimePrecision::Nanosecond).c_str(), document->GetAllocator()); |
| value->AddMember("start_time", start_time, document->GetAllocator()); |
| |
| Value end_time(ToStringFromUnixMicros(record.end_time_us, |
| TimePrecision::Nanosecond).c_str(), document->GetAllocator()); |
| value->AddMember("end_time", end_time, document->GetAllocator()); |
| |
| // record.end_time_us might still be zero if the query is not yet done |
| // Use the current Unix time in that case. Note that the duration can be |
| // negative if a system clock reset happened after the query was initiated. |
| int64_t end_time_us = record.end_time_us > 0LL ? record.end_time_us : UnixMicros(); |
| int64_t duration_us = end_time_us - record.start_time_us; |
| const string& printed_duration = PrettyPrinter::Print(duration_us * NANOS_PER_MICRO, |
| TUnit::TIME_NS); |
| Value val_duration(printed_duration.c_str(), document->GetAllocator()); |
| value->AddMember("duration", val_duration, document->GetAllocator()); |
| |
| string progress = "N/A"; |
| if (record.has_coord) { |
| stringstream ss; |
| ss << record.num_complete_fragments << " / " << record.total_fragments |
| << " (" << setw(4); |
| if (record.total_fragments == 0) { |
| ss << "0%)"; |
| } else { |
| ss << (100.0 * record.num_complete_fragments / (1.f * record.total_fragments)) |
| << "%)"; |
| } |
| progress = ss.str(); |
| } |
| Value progress_json(progress.c_str(), document->GetAllocator()); |
| value->AddMember("progress", progress_json, document->GetAllocator()); |
| |
| Value state(_QueryState_VALUES_TO_NAMES.find(record.query_state)->second, |
| document->GetAllocator()); |
| value->AddMember("state", state, document->GetAllocator()); |
| |
| value->AddMember("rows_fetched", record.num_rows_fetched, document->GetAllocator()); |
| |
| Value query_id(PrintId(record.id).c_str(), document->GetAllocator()); |
| value->AddMember("query_id", query_id, document->GetAllocator()); |
| |
| if (record.event_sequence.labels.size() > 0) { |
| Value last_event(record.event_sequence.labels.back().c_str(), |
| document->GetAllocator()); |
| value->AddMember("last_event", last_event, document->GetAllocator()); |
| } |
| |
| // Waiting to be closed. |
| bool waiting = record.query_state == beeswax::QueryState::EXCEPTION || |
| record.all_rows_returned; |
| value->AddMember("waiting", waiting, document->GetAllocator()); |
| value->AddMember("executing", !waiting, document->GetAllocator()); |
| |
| int64_t waiting_time = impala::UnixMillis() - record.last_active_time_ms; |
| string waiting_time_str = ""; |
| if (waiting_time > 0) { |
| waiting_time_str = PrettyPrinter::Print(waiting_time, TUnit::TIME_MS); |
| } |
| Value val_waiting_time(waiting_time_str.c_str(), document->GetAllocator()); |
| value->AddMember("waiting_time", val_waiting_time, document->GetAllocator()); |
| |
| Value resource_pool(record.resource_pool.c_str(), document->GetAllocator()); |
| value->AddMember("resource_pool", resource_pool, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::QueryStateHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| set<ImpalaServer::QueryStateRecord, ImpalaServer::QueryStateRecordLessThan> |
| sorted_query_records; |
| |
| server_->client_request_state_map_.DoFuncForAllEntries( |
| [&](const std::shared_ptr<ClientRequestState>& request_state) { |
| sorted_query_records.insert(ImpalaServer::QueryStateRecord(*request_state)); |
| }); |
| |
| Value in_flight_queries(kArrayType); |
| int64_t num_waiting_queries = 0; |
| for (const ImpalaServer::QueryStateRecord& record: sorted_query_records) { |
| Value record_json(kObjectType); |
| QueryStateToJson(record, &record_json, document); |
| |
| if (record_json["waiting"].GetBool()) ++num_waiting_queries; |
| |
| in_flight_queries.PushBack(record_json, document->GetAllocator()); |
| } |
| document->AddMember("in_flight_queries", in_flight_queries, document->GetAllocator()); |
| document->AddMember("num_in_flight_queries", |
| static_cast<uint64_t>(sorted_query_records.size()), |
| document->GetAllocator()); |
| document->AddMember("num_executing_queries", |
| sorted_query_records.size() - num_waiting_queries, |
| document->GetAllocator()); |
| document->AddMember("num_waiting_queries", num_waiting_queries, |
| document->GetAllocator()); |
| document->AddMember("waiting-tooltip", "These queries are no longer executing, either " |
| "because they encountered an error or because they have returned all of their " |
| "results, but they are still active so that their results can be inspected. To " |
| "free the resources they are using, they must be closed.", |
| document->GetAllocator()); |
| |
| Value completed_queries(kArrayType); |
| { |
| lock_guard<mutex> l(server_->query_log_lock_); |
| for (const ImpalaServer::QueryStateRecord& log_entry: server_->query_log_) { |
| Value record_json(kObjectType); |
| QueryStateToJson(log_entry, &record_json, document); |
| completed_queries.PushBack(record_json, document->GetAllocator()); |
| } |
| } |
| document->AddMember("completed_queries", completed_queries, document->GetAllocator()); |
| document->AddMember("completed_log_size", FLAGS_query_log_size, |
| document->GetAllocator()); |
| |
| Value query_locations(kArrayType); |
| { |
| lock_guard<mutex> l(server_->query_locations_lock_); |
| for (const ImpalaServer::QueryLocations::value_type& location: |
| server_->query_locations_) { |
| Value location_json(kObjectType); |
| Value location_name(TNetworkAddressToString(location.first).c_str(), |
| document->GetAllocator()); |
| location_json.AddMember("location", location_name, document->GetAllocator()); |
| location_json.AddMember("count", static_cast<uint64_t>(location.second.size()), |
| document->GetAllocator()); |
| query_locations.PushBack(location_json, document->GetAllocator()); |
| } |
| } |
| document->AddMember("query_locations", query_locations, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::SessionsHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| lock_guard<mutex> l(server_->session_state_map_lock_); |
| Value sessions(kArrayType); |
| int num_active = 0; |
| for (const ImpalaServer::SessionStateMap::value_type& session: |
| server_->session_state_map_) { |
| shared_ptr<ImpalaServer::SessionState> state = session.second; |
| Value session_json(kObjectType); |
| Value type(PrintThriftEnum(state->session_type).c_str(), document->GetAllocator()); |
| session_json.AddMember("type", type, document->GetAllocator()); |
| |
| session_json.AddMember("inflight_queries", |
| static_cast<uint64_t>(state->inflight_queries.size()), |
| document->GetAllocator()); |
| session_json.AddMember("total_queries", state->total_queries, |
| document->GetAllocator()); |
| |
| Value user(state->connected_user.c_str(), document->GetAllocator()); |
| session_json.AddMember("user", user, document->GetAllocator()); |
| |
| Value delegated_user(state->do_as_user.c_str(), document->GetAllocator()); |
| session_json.AddMember("delegated_user", delegated_user, document->GetAllocator()); |
| |
| Value session_id(PrintId(session.first).c_str(), document->GetAllocator()); |
| session_json.AddMember("session_id", session_id, document->GetAllocator()); |
| |
| Value network_address(TNetworkAddressToString(state->network_address).c_str(), |
| document->GetAllocator()); |
| session_json.AddMember("network_address", network_address, document->GetAllocator()); |
| |
| Value default_db(state->database.c_str(), document->GetAllocator()); |
| session_json.AddMember("default_database", default_db, document->GetAllocator()); |
| |
| Value start_time(ToStringFromUnixMillis(session.second->start_time_ms, |
| TimePrecision::Second).c_str(), document->GetAllocator()); |
| session_json.AddMember("start_time", start_time, document->GetAllocator()); |
| session_json.AddMember( |
| "start_time_sort", session.second->start_time_ms, document->GetAllocator()); |
| |
| Value last_accessed(ToStringFromUnixMillis(session.second->last_accessed_ms, |
| TimePrecision::Second).c_str(), document->GetAllocator()); |
| session_json.AddMember("last_accessed", last_accessed, document->GetAllocator()); |
| session_json.AddMember( |
| "last_accessed_sort", session.second->last_accessed_ms, document->GetAllocator()); |
| |
| session_json.AddMember("session_timeout", state->session_timeout, |
| document->GetAllocator()); |
| session_json.AddMember("expired", state->expired, document->GetAllocator()); |
| session_json.AddMember("closed", state->closed, document->GetAllocator()); |
| if (!state->expired && !state->closed) ++num_active; |
| session_json.AddMember("ref_count", state->ref_count, document->GetAllocator()); |
| sessions.PushBack(session_json, document->GetAllocator()); |
| } |
| |
| document->AddMember("sessions", sessions, document->GetAllocator()); |
| document->AddMember("num_sessions", |
| static_cast<uint64_t>(server_->session_state_map_.size()), |
| document->GetAllocator()); |
| document->AddMember("num_active", num_active, document->GetAllocator()); |
| document->AddMember("num_inactive", server_->session_state_map_.size() - num_active, |
| document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::CatalogHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| TGetDbsResult get_dbs_result; |
| Status status = server_->exec_env_->frontend()->GetDbs(NULL, NULL, &get_dbs_result); |
| if (!status.ok()) { |
| Value error(status.GetDetail().c_str(), document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| return; |
| } |
| |
| Value databases(kArrayType); |
| for (const TDatabase& db: get_dbs_result.dbs) { |
| Value database(kObjectType); |
| Value str(db.db_name.c_str(), document->GetAllocator()); |
| database.AddMember("name", str, document->GetAllocator()); |
| |
| TGetTablesResult get_table_results; |
| Status status = server_->exec_env_->frontend()->GetTableNames( |
| db.db_name, NULL, NULL, &get_table_results); |
| if (!status.ok()) { |
| Value error(status.GetDetail().c_str(), document->GetAllocator()); |
| database.AddMember("error", error, document->GetAllocator()); |
| continue; |
| } |
| |
| Value table_array(kArrayType); |
| for (const string& table: get_table_results.tables) { |
| Value table_obj(kObjectType); |
| if(!FLAGS_use_local_catalog){ |
| // Creates hyperlinks for /catalog_object. This is disabled in local catalog mode |
| Value fq_name(Substitute("$0.$1", db.db_name, table).c_str(), |
| document->GetAllocator()); |
| table_obj.AddMember("fqtn", fq_name, document->GetAllocator()); |
| } |
| Value table_name(table.c_str(), document->GetAllocator()); |
| table_obj.AddMember("name", table_name, document->GetAllocator()); |
| table_array.PushBack(table_obj, document->GetAllocator()); |
| } |
| database.AddMember("num_tables", table_array.Size(), document->GetAllocator()); |
| database.AddMember("tables", table_array, document->GetAllocator()); |
| Value use_local_catalog(FLAGS_use_local_catalog); |
| database.AddMember("use_local_catalog", use_local_catalog, |
| document->GetAllocator()); |
| databases.PushBack(database, document->GetAllocator()); |
| } |
| document->AddMember("databases", databases, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::CatalogObjectsHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| DCHECK(!FLAGS_use_local_catalog); |
| const auto& args = req.parsed_args; |
| Webserver::ArgumentMap::const_iterator object_type_arg = args.find("object_type"); |
| Webserver::ArgumentMap::const_iterator object_name_arg = args.find("object_name"); |
| if (object_type_arg != args.end() && object_name_arg != args.end()) { |
| TCatalogObjectType::type object_type = |
| TCatalogObjectTypeFromName(object_type_arg->second); |
| |
| // Get the object type and name from the topic entry key |
| TCatalogObject request; |
| TCatalogObject result; |
| Status status = TCatalogObjectFromObjectName(object_type, object_name_arg->second, &request); |
| if (status.ok()) { |
| // Get the object and dump its contents. |
| status = server_->exec_env_->frontend()->GetCatalogObject(request, &result); |
| } |
| if (status.ok()) { |
| Value debug_string(ThriftDebugString(result).c_str(), document->GetAllocator()); |
| document->AddMember("thrift_string", debug_string, document->GetAllocator()); |
| } else { |
| Value error(status.GetDetail().c_str(), document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| } |
| } else { |
| Value error("Please specify values for the object_type and object_name parameters.", |
| document->GetAllocator()); |
| document->AddMember("error", error, document->GetAllocator()); |
| } |
| } |
| |
| namespace { |
| |
| // Helper for PlanToJson(), processes a single list of plan nodes which are the |
| // DFS-flattened representation of a single plan fragment. Called recursively, the |
| // iterator parameter is updated in place so that when a recursive call returns, the |
| // caller is pointing at the next of its children. |
| void PlanToJsonHelper(const map<TPlanNodeId, TPlanNodeExecSummary>& summaries, |
| const vector<TPlanNode>& nodes, |
| vector<TPlanNode>::const_iterator* it, rapidjson::Document* document, Value* value) { |
| Value children(kArrayType); |
| Value label((*it)->label.c_str(), document->GetAllocator()); |
| value->AddMember("label", label, document->GetAllocator()); |
| // Node "details" may contain exprs which should be redacted. |
| Value label_detail(RedactCopy((*it)->label_detail).c_str(), document->GetAllocator()); |
| value->AddMember("label_detail", label_detail, document->GetAllocator()); |
| |
| TPlanNodeId id = (*it)->node_id; |
| map<TPlanNodeId, TPlanNodeExecSummary>::const_iterator summary = summaries.find(id); |
| if (summary != summaries.end()) { |
| int64_t cardinality = 0; |
| int64_t max_time = 0L; |
| int64_t total_time = 0; |
| for (const TExecStats& stat: summary->second.exec_stats) { |
| if (summary->second.is_broadcast) { |
| // Avoid multiple-counting for recipients of broadcasts. |
| cardinality = ::max(cardinality, stat.cardinality); |
| } else { |
| cardinality += stat.cardinality; |
| } |
| total_time += stat.latency_ns; |
| max_time = ::max(max_time, stat.latency_ns); |
| } |
| value->AddMember("output_card", cardinality, document->GetAllocator()); |
| value->AddMember("num_instances", |
| static_cast<uint64_t>(summary->second.exec_stats.size()), |
| document->GetAllocator()); |
| if (summary->second.is_broadcast) { |
| value->AddMember("is_broadcast", true, document->GetAllocator()); |
| } |
| |
| const string& max_time_str = PrettyPrinter::Print(max_time, TUnit::TIME_NS); |
| Value max_time_str_json(max_time_str.c_str(), document->GetAllocator()); |
| value->AddMember("max_time", max_time_str_json, document->GetAllocator()); |
| value->AddMember("max_time_val", max_time, document->GetAllocator()); |
| |
| // Round to the nearest ns, to workaround a bug in pretty-printing a fraction of a |
| // ns. See IMPALA-1800. |
| const string& avg_time_str = PrettyPrinter::Print( |
| // A bug may occasionally cause 1-instance nodes to appear to have 0 instances. |
| total_time / ::max(static_cast<int>(summary->second.exec_stats.size()), 1), |
| TUnit::TIME_NS); |
| Value avg_time_str_json(avg_time_str.c_str(), document->GetAllocator()); |
| value->AddMember("avg_time", avg_time_str_json, document->GetAllocator()); |
| } |
| |
| int num_children = (*it)->num_children; |
| for (int i = 0; i < num_children; ++i) { |
| ++(*it); |
| Value container(kObjectType); |
| PlanToJsonHelper(summaries, nodes, it, document, &container); |
| children.PushBack(container, document->GetAllocator()); |
| } |
| value->AddMember("children", children, document->GetAllocator()); |
| } |
| |
| // Helper method which converts a list of plan fragments into a single JSON document, with |
| // the following schema: |
| // "plan_nodes": [ |
| // { |
| // "label": "12:AGGREGATE", |
| // "label_detail": "FINALIZE", |
| // "output_card": 23456, |
| // "num_instances": 34, |
| // "max_time": "1m23s", |
| // "avg_time": "1.3ms", |
| // "children": [ |
| // { |
| // "label": "11:EXCHANGE", |
| // "label_detail": "UNPARTITIONED", |
| // "children": [] |
| // } |
| // ] |
| // }, |
| // { |
| // "label": "07:AGGREGATE", |
| // "label_detail": "", |
| // "children": [], |
| // "data_stream_target": "11:EXCHANGE" |
| // } |
| // ] |
| void PlanToJson(const vector<TPlanFragment>& fragments, const TExecSummary& summary, |
| rapidjson::Document* document, Value* value) { |
| // Build a map from id to label so that we can resolve the targets of data-stream sinks |
| // and connect plan fragments. |
| map<TPlanNodeId, string> label_map; |
| for (const TPlanFragment& fragment: fragments) { |
| for (const TPlanNode& node: fragment.plan.nodes) { |
| label_map[node.node_id] = node.label; |
| } |
| } |
| |
| map<TPlanNodeId, TPlanNodeExecSummary> exec_summaries; |
| for (const TPlanNodeExecSummary& s: summary.nodes) { |
| exec_summaries[s.node_id] = s; |
| } |
| |
| Value nodes(kArrayType); |
| for (const TPlanFragment& fragment: fragments) { |
| Value plan_fragment(kObjectType); |
| vector<TPlanNode>::const_iterator it = fragment.plan.nodes.begin(); |
| PlanToJsonHelper(exec_summaries, fragment.plan.nodes, &it, document, &plan_fragment); |
| if (fragment.__isset.output_sink) { |
| const TDataSink& sink = fragment.output_sink; |
| if (sink.__isset.stream_sink) { |
| Value target(label_map[sink.stream_sink.dest_node_id].c_str(), |
| document->GetAllocator()); |
| plan_fragment.AddMember("data_stream_target", target, document->GetAllocator()); |
| } |
| } |
| nodes.PushBack(plan_fragment, document->GetAllocator()); |
| } |
| value->AddMember("plan_nodes", nodes, document->GetAllocator()); |
| } |
| |
| } |
| |
| void ImpalaHttpHandler::QueryBackendsHandler( |
| const Webserver::WebRequest& req, Document* document) { |
| TUniqueId query_id; |
| Status status = ParseIdFromRequest(req, &query_id, "query_id"); |
| Value query_id_val(PrintId(query_id).c_str(), document->GetAllocator()); |
| document->AddMember("query_id", query_id_val, document->GetAllocator()); |
| if (!status.ok()) { |
| // Redact the error message, it may contain part or all of the query. |
| Value json_error(RedactCopy(status.GetDetail()).c_str(), document->GetAllocator()); |
| document->AddMember("error", json_error, document->GetAllocator()); |
| return; |
| } |
| |
| shared_ptr<ClientRequestState> request_state = server_->GetClientRequestState(query_id); |
| if (request_state.get() == nullptr || request_state->GetCoordinator() == nullptr) { |
| return; |
| } |
| |
| request_state->GetCoordinator()->BackendsToJson(document); |
| } |
| |
| void ImpalaHttpHandler::QueryFInstancesHandler( |
| const Webserver::WebRequest& req, Document* document) { |
| TUniqueId query_id; |
| Status status = ParseIdFromRequest(req, &query_id, "query_id"); |
| Value query_id_val(PrintId(query_id).c_str(), document->GetAllocator()); |
| document->AddMember("query_id", query_id_val, document->GetAllocator()); |
| if (!status.ok()) { |
| // Redact the error message, it may contain part or all of the query. |
| Value json_error(RedactCopy(status.GetDetail()).c_str(), document->GetAllocator()); |
| document->AddMember("error", json_error, document->GetAllocator()); |
| return; |
| } |
| |
| shared_ptr<ClientRequestState> request_state = server_->GetClientRequestState(query_id); |
| if (request_state.get() == nullptr || request_state->GetCoordinator() == nullptr) { |
| return; |
| } |
| |
| request_state->GetCoordinator()->FInstanceStatsToJson(document); |
| } |
| |
| void ImpalaHttpHandler::QuerySummaryHandler(bool include_json_plan, bool include_summary, |
| const Webserver::WebRequest& req, Document* document) { |
| TUniqueId query_id; |
| Status status = ParseIdFromRequest(req, &query_id, "query_id"); |
| Value query_id_val(PrintId(query_id).c_str(), document->GetAllocator()); |
| document->AddMember("query_id", query_id_val, document->GetAllocator()); |
| if (!status.ok()) { |
| // Redact the error message, it may contain part or all of the query. |
| Value json_error(RedactCopy(status.GetDetail()).c_str(), document->GetAllocator()); |
| document->AddMember("error", json_error, document->GetAllocator()); |
| return; |
| } |
| |
| TExecSummary summary; |
| string stmt; |
| string plan; |
| Status query_status; |
| bool found = false; |
| vector<TPlanFragment> fragments; |
| |
| // Search the in-flight queries first, followed by the archived ones. |
| { |
| shared_ptr<ClientRequestState> request_state = |
| server_->GetClientRequestState(query_id); |
| if (request_state != nullptr) { |
| found = true; |
| // If the query plan isn't generated, avoid waiting for the request |
| // state lock to be acquired, since it could potentially be an expensive |
| // call, if the table Catalog metadata loading is in progress. Instead |
| // update the caller that the plan information is unavailable. |
| if (request_state->operation_state() == TOperationState::INITIALIZED_STATE) { |
| document->AddMember( |
| "plan_metadata_unavailable", "true", document->GetAllocator()); |
| return; |
| } |
| lock_guard<mutex> l(*request_state->lock()); |
| query_status = request_state->query_status(); |
| stmt = request_state->sql_stmt(); |
| plan = request_state->exec_request().query_exec_request.query_plan; |
| if ((include_json_plan || include_summary) |
| && request_state->GetCoordinator() != nullptr) { |
| request_state->GetCoordinator()->GetTExecSummary(&summary); |
| } |
| if (include_json_plan) { |
| for (const TPlanExecInfo& plan_exec_info: |
| request_state->exec_request().query_exec_request.plan_exec_info) { |
| for (const TPlanFragment& fragment: plan_exec_info.fragments) { |
| fragments.push_back(fragment); |
| } |
| } |
| } |
| } |
| } |
| |
| if (!found) { |
| lock_guard<mutex> l(server_->query_log_lock_); |
| ImpalaServer::QueryLogIndex::const_iterator query_record = |
| server_->query_log_index_.find(query_id); |
| if (query_record == server_->query_log_index_.end()) { |
| const string& err = Substitute("Unknown query id: $0", PrintId(query_id)); |
| Value json_error(err.c_str(), document->GetAllocator()); |
| document->AddMember("error", json_error, document->GetAllocator()); |
| return; |
| } |
| if (include_json_plan || include_summary) { |
| summary = query_record->second->exec_summary; |
| } |
| stmt = query_record->second->stmt; |
| plan = query_record->second->plan; |
| query_status = query_record->second->query_status; |
| if (include_json_plan) { |
| fragments = query_record->second->fragments; |
| } |
| } |
| |
| if (include_json_plan) { |
| Value v(kObjectType); |
| PlanToJson(fragments, summary, document, &v); |
| document->AddMember("plan_json", v, document->GetAllocator()); |
| } |
| if (include_summary) { |
| const string& printed_summary = PrintExecSummary(summary); |
| Value json_summary(printed_summary.c_str(), document->GetAllocator()); |
| document->AddMember("summary", json_summary, document->GetAllocator()); |
| } |
| Value json_stmt(RedactCopy(stmt).c_str(), document->GetAllocator()); |
| document->AddMember("stmt", json_stmt, document->GetAllocator()); |
| Value json_plan_text(RedactCopy(plan).c_str(), document->GetAllocator()); |
| document->AddMember("plan", json_plan_text, document->GetAllocator()); |
| |
| // Redact the error in case the query is contained in the error message. |
| Value json_status(query_status.ok() ? "OK" : |
| RedactCopy(query_status.GetDetail()).c_str(), document->GetAllocator()); |
| document->AddMember("status", json_status, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::BackendsHandler(const Webserver::WebRequest& req, |
| Document* document) { |
| AdmissionController::PerHostStats host_stats; |
| ExecEnv::GetInstance()->admission_controller()->PopulatePerHostMemReservedAndAdmitted( |
| &host_stats); |
| Value backends_list(kArrayType); |
| ClusterMembershipMgr* cluster_membership_mgr = |
| ExecEnv::GetInstance()->cluster_membership_mgr(); |
| DCHECK(cluster_membership_mgr != nullptr); |
| ClusterMembershipMgr::SnapshotPtr membership_snapshot = |
| cluster_membership_mgr->GetSnapshot(); |
| DCHECK(membership_snapshot.get() != nullptr); |
| for (const auto& entry : membership_snapshot->current_backends) { |
| TBackendDescriptor backend = entry.second; |
| Value backend_obj(kObjectType); |
| string address = TNetworkAddressToString(backend.address); |
| Value str(address.c_str(), document->GetAllocator()); |
| Value krpc_address( |
| TNetworkAddressToString(backend.krpc_address).c_str(), document->GetAllocator()); |
| backend_obj.AddMember("address", str, document->GetAllocator()); |
| backend_obj.AddMember("krpc_address", krpc_address, document->GetAllocator()); |
| string webserver_url = |
| Substitute("$0://$1", backend.secure_webserver ? "https" : "http", |
| TNetworkAddressToString(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()); |
| 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 |
| // value of zero is used for those addresses. |
| Value mem_reserved(PrettyPrinter::PrintBytes( |
| host_stats[address].mem_reserved).c_str(), document->GetAllocator()); |
| backend_obj.AddMember("mem_reserved", mem_reserved, document->GetAllocator()); |
| 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("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); |
| Value executor_groups(JoinStrings(group_names, ", ").c_str(), |
| document->GetAllocator()); |
| backend_obj.AddMember("executor_groups", executor_groups, document->GetAllocator()); |
| backends_list.PushBack(backend_obj, document->GetAllocator()); |
| } |
| document->AddMember("backends", backends_list, document->GetAllocator()); |
| } |
| |
| 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); |
| } else { |
| ac->PoolToJson(pool_name_arg->second, &resource_pools, document); |
| } |
| |
| // Now get running queries from CRS map. |
| struct QueryInfo { |
| TUniqueId query_id; |
| int64_t executor_mem_limit; |
| int64_t executor_mem_to_admit; |
| int64_t coord_mem_limit; |
| int64_t coord_mem_to_admit; |
| unsigned long num_backends; |
| }; |
| unordered_map<string, vector<QueryInfo>> running_queries; |
| server_->client_request_state_map_.DoFuncForAllEntries([&running_queries]( |
| const std::shared_ptr<ClientRequestState>& request_state) { |
| // Make sure only queries past admission control are added. |
| auto query_state = request_state->operation_state(); |
| if (query_state != TOperationState::INITIALIZED_STATE |
| && query_state != TOperationState::PENDING_STATE |
| && request_state->schedule() != nullptr) |
| running_queries[request_state->request_pool()].push_back( |
| {request_state->query_id(), request_state->schedule()->per_backend_mem_limit(), |
| request_state->schedule()->per_backend_mem_to_admit(), |
| request_state->schedule()->coord_backend_mem_limit(), |
| request_state->schedule()->coord_backend_mem_to_admit(), |
| static_cast<unsigned long>( |
| request_state->schedule()->per_backend_exec_params().size())}); |
| }); |
| |
| // Add the running queries to the resource_pools json. |
| for (int i = 0; i < resource_pools.Size(); i++) { |
| DCHECK(resource_pools[i].IsObject()); |
| Value::MemberIterator it = resource_pools[i].GetObject().FindMember("pool_name"); |
| DCHECK(it != resource_pools[i].GetObject().MemberEnd()); |
| DCHECK(it->value.IsString()); |
| string pool_name(it->value.GetString()); |
| // Now add running queries to the json. |
| auto query_list = running_queries.find(pool_name); |
| if (query_list == running_queries.end()) continue; |
| vector<QueryInfo>& info_array = query_list->second; |
| Value queries_in_pool(rapidjson::kArrayType); |
| for (QueryInfo info : info_array) { |
| Value query_info(rapidjson::kObjectType); |
| Value query_id(PrintId(info.query_id).c_str(), document->GetAllocator()); |
| query_info.AddMember("query_id", query_id, document->GetAllocator()); |
| query_info.AddMember( |
| "mem_limit", info.executor_mem_limit, document->GetAllocator()); |
| query_info.AddMember( |
| "mem_limit_to_admit", info.executor_mem_to_admit, document->GetAllocator()); |
| query_info.AddMember( |
| "coord_mem_limit", info.coord_mem_limit, document->GetAllocator()); |
| query_info.AddMember( |
| "coord_mem_to_admit", info.coord_mem_to_admit, document->GetAllocator()); |
| query_info.AddMember("num_backends", info.num_backends, document->GetAllocator()); |
| queries_in_pool.PushBack(query_info, document->GetAllocator()); |
| } |
| resource_pools[i].GetObject().AddMember( |
| "running_queries", queries_in_pool, document->GetAllocator()); |
| } |
| int64_t ms_since_last_statestore_update; |
| string staleness_detail = ac->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. |
| rapidjson::StringBuffer strbuf; |
| PrettyWriter<rapidjson::StringBuffer> writer(strbuf); |
| resource_pools.Accept(writer); |
| Value raw_json(strbuf.GetString(), document->GetAllocator()); |
| document->AddMember("resource_pools_plain_json", raw_json, document->GetAllocator()); |
| document->AddMember("resource_pools", resource_pools, document->GetAllocator()); |
| document->AddMember("statestore_admission_control_time_since_last_update_ms", |
| ms_since_last_statestore_update, document->GetAllocator()); |
| if (!staleness_detail.empty()) { |
| Value staleness_detail_json(staleness_detail.c_str(), document->GetAllocator()); |
| document->AddMember("statestore_update_staleness_detail", staleness_detail_json, |
| document->GetAllocator()); |
| } |
| |
| // Indicator that helps render UI elements based on this condition. |
| document->AddMember("get_all_pools", get_all_pools, document->GetAllocator()); |
| } |
| |
| void ImpalaHttpHandler::ResetResourcePoolStatsHandler( |
| const Webserver::WebRequest& req, Document* document) { |
| const auto& args = req.parsed_args; |
| 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(); |
| } else { |
| ExecEnv::GetInstance()->admission_controller()->ResetPoolInformationalStats( |
| pool_name_arg->second); |
| } |
| } |