You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2018/02/21 22:23:04 UTC
[3/4] impala git commit: IMPALA-6269: Expose KRPC metrics on debug
webpage
IMPALA-6269: Expose KRPC metrics on debug webpage
This change exposes KRPC metrics on the /rpcz debug web page.
This change also exposes metrics for rejected RPCs on the /metrics debug
web page. See here for an example: https://git.io/vAczm
This change also fixes a bug in PrettyPrinter::GetByteUnit(), which
previously did not work for unsigned values due to an implicit cast.
This change contains tests to check that the metrics show up in /rpcz
and /metrics and that they update as expected when executing queries.
This change is based on a change by Sailesh Mukil.
Change-Id: I7af7c1a84a5be82c979ca4ef1edf35167493be3f
Reviewed-on: http://gerrit.cloudera.org:8080/9292
Reviewed-by: Lars Volker <lv...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/74dd9e9b
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/74dd9e9b
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/74dd9e9b
Branch: refs/heads/master
Commit: 74dd9e9bfe32541e9ab6bddb8b839f3e491361f1
Parents: 678bf28
Author: Lars Volker <lv...@cloudera.com>
Authored: Tue Jan 23 11:01:29 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Wed Feb 21 10:19:06 2018 +0000
----------------------------------------------------------------------
be/src/rpc/impala-service-pool.cc | 103 +++++++++++++--
be/src/rpc/impala-service-pool.h | 26 ++--
be/src/rpc/rpc-mgr-kerberized-test.cc | 8 +-
be/src/rpc/rpc-mgr-test-base.h | 28 +++--
be/src/rpc/rpc-mgr-test.cc | 14 ++-
be/src/rpc/rpc-mgr.cc | 52 +++++++-
be/src/rpc/rpc-mgr.h | 15 ++-
be/src/rpc/rpc-trace.cc | 19 ++-
be/src/rpc/rpc-trace.h | 3 +-
be/src/runtime/exec-env.cc | 3 +-
be/src/runtime/exec-env.h | 4 +-
be/src/service/impalad-main.cc | 2 +-
be/src/util/histogram-metric.h | 37 ++++--
be/src/util/pretty-printer.h | 51 +++++---
common/thrift/Metrics.thrift | 3 +-
common/thrift/metrics.json | 10 ++
tests/custom_cluster/test_krpc_metrics.py | 97 +++++++++++++++
tests/webserver/test_web_pages.py | 30 +++++
www/rpcz.tmpl | 166 ++++++++++++++++++++-----
19 files changed, 547 insertions(+), 124 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/rpc/impala-service-pool.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/impala-service-pool.cc b/be/src/rpc/impala-service-pool.cc
index 35a5d6d..fb50656 100644
--- a/be/src/rpc/impala-service-pool.cc
+++ b/be/src/rpc/impala-service-pool.cc
@@ -31,30 +31,49 @@
#include "kudu/rpc/messenger.h"
#include "kudu/rpc/service_if.h"
#include "kudu/rpc/service_queue.h"
+#include "kudu/util/hdr_histogram.h"
#include "kudu/util/metrics.h"
#include "kudu/util/status.h"
#include "kudu/util/trace.h"
+#include "runtime/exec-env.h"
#include "runtime/mem-tracker.h"
#include "common/names.h"
#include "common/status.h"
-METRIC_DEFINE_histogram(server, impala_unused,
+METRIC_DEFINE_histogram(server, impala_incoming_queue_time,
"RPC Queue Time",
kudu::MetricUnit::kMicroseconds,
"Number of microseconds incoming RPC requests spend in the worker queue",
60000000LU, 3);
+using namespace rapidjson;
+
namespace impala {
+// Metric key format for rpc call duration metrics.
+const string RPC_QUEUE_OVERFLOW_METRIC_KEY = "rpc.$0.rpcs_queue_overflow";
ImpalaServicePool::ImpalaServicePool(const scoped_refptr<kudu::MetricEntity>& entity,
- size_t service_queue_length, kudu::rpc::ServiceIf* service,
+ size_t service_queue_length, kudu::rpc::GeneratedServiceIf* service,
MemTracker* service_mem_tracker)
: service_mem_tracker_(service_mem_tracker),
service_(service),
service_queue_(service_queue_length),
- unused_histogram_(METRIC_impala_unused.Instantiate(entity)) {
+ incoming_queue_time_(METRIC_impala_incoming_queue_time.Instantiate(entity)) {
DCHECK(service_mem_tracker_ != nullptr);
+ const TMetricDef& overflow_metric_def =
+ MetricDefs::Get(RPC_QUEUE_OVERFLOW_METRIC_KEY, service_->service_name());
+ rpcs_queue_overflow_ = ExecEnv::GetInstance()->rpc_metrics()->RegisterMetric(
+ new IntCounter(overflow_metric_def, 0L));
+ // Initialize additional histograms for each method of the service.
+ // TODO: Retrieve these from KRPC once KUDU-2313 has been implemented.
+ for (const auto& method : service_->methods_by_name()) {
+ const string& method_name = method.first;
+ string payload_size_name = Substitute("$0-payload-size", method_name);
+ payload_size_histograms_[method_name].reset(new HistogramMetric(
+ MakeTMetricDef(method_name, TMetricKind::HISTOGRAM, TUnit::BYTES),
+ 1024 * 1024 * 1024, 3));
+ }
}
ImpalaServicePool::~ImpalaServicePool() {
@@ -101,7 +120,7 @@ void ImpalaServicePool::RejectTooBusy(kudu::rpc::InboundCall* c) {
service_->service_name(),
c->remote_address().ToString(),
service_queue_.max_size());
- rpcs_queue_overflow_.Add(1);
+ rpcs_queue_overflow_->Increment(1);
FailAndReleaseRpc(kudu::rpc::ErrorStatusPB::ERROR_SERVER_TOO_BUSY,
kudu::Status::ServiceUnavailable(err_msg), c);
VLOG(1) << err_msg << " Contents of service queue:\n"
@@ -202,12 +221,11 @@ void ImpalaServicePool::RunThread() {
}
// We need to call RecordHandlingStarted() to update the InboundCall timing.
- incoming->RecordHandlingStarted(unused_histogram_);
+ incoming->RecordHandlingStarted(incoming_queue_time_);
ADOPT_TRACE(incoming->trace());
if (UNLIKELY(incoming->ClientTimedOut())) {
TRACE_TO(incoming->trace(), "Skipping call since client already timed out"); // NOLINT(*)
- rpcs_timed_out_in_queue_.Add(1);
// Respond as a failure, even though the client will probably ignore
// the response anyway.
@@ -217,10 +235,13 @@ void ImpalaServicePool::RunThread() {
continue;
}
- TRACE_TO(incoming->trace(), "Handling call"); // NOLINT(*)
+ const string& method_name = incoming->remote_method().method_name();
+ int64_t transfer_size = incoming->GetTransferSize();
+ payload_size_histograms_[method_name]->Update(transfer_size);
- // Release the InboundCall pointer -- when the call is responded to,
- // it will get deleted at that point.
+ TRACE_TO(incoming->trace(), "Handling call"); // NOLINT(*)
+ // Release the InboundCall pointer -- when the call is responded to, it will get
+ // deleted at that point.
service_->Handle(incoming.release());
}
}
@@ -229,4 +250,68 @@ const string ImpalaServicePool::service_name() const {
return service_->service_name();
}
+// Render a kudu::Histogram into a human readable string representation.
+// TODO: Switch to structured JSON (IMPALA-6545).
+const string KrpcHistogramToString(const kudu::Histogram* histogram) {
+ DCHECK(histogram != nullptr);
+ DCHECK_EQ(histogram->prototype()->unit(), kudu::MetricUnit::kMicroseconds);
+ kudu::HdrHistogram snapshot(*histogram->histogram());
+ return HistogramMetric::HistogramToHumanReadable(&snapshot, TUnit::TIME_US);
+}
+
+// Expose the service pool metrics by storing them as JSON in 'value'.
+void ImpalaServicePool::ToJson(rapidjson::Value* value, rapidjson::Document* document) {
+ // Add pool metrics.
+ Value service_name_val(service_name().c_str(), document->GetAllocator());
+ value->AddMember("service_name", service_name_val, document->GetAllocator());
+ value->AddMember("queue_size", service_queue_.estimated_queue_length(),
+ document->GetAllocator());
+ value->AddMember("idle_threads", service_queue_.estimated_idle_worker_count(),
+ document->GetAllocator());
+ value->AddMember("rpcs_queue_overflow", rpcs_queue_overflow_->GetValue(),
+ document->GetAllocator());
+
+ Value mem_usage(PrettyPrinter::Print(service_mem_tracker_->consumption(),
+ TUnit::BYTES).c_str(), document->GetAllocator());
+ value->AddMember("mem_usage", mem_usage, document->GetAllocator());
+
+ Value mem_peak(PrettyPrinter::Print(service_mem_tracker_->peak_consumption(),
+ TUnit::BYTES).c_str(), document->GetAllocator());
+ value->AddMember("mem_peak", mem_peak, document->GetAllocator());
+
+ Value incoming_queue_time(KrpcHistogramToString(incoming_queue_time_.get()).c_str(),
+ document->GetAllocator());
+ value->AddMember("incoming_queue_time", incoming_queue_time,
+ document->GetAllocator());
+
+ // Add method specific metrics.
+ const kudu::rpc::GeneratedServiceIf::MethodInfoMap& method_infos =
+ service_->methods_by_name();
+ Value rpc_method_metrics(kArrayType);
+ for (const auto& method : method_infos) {
+ Value method_entry(kObjectType);
+
+ const string& method_name = method.first;
+ Value method_name_val(method_name.c_str(), document->GetAllocator());
+ method_entry.AddMember("method_name", method_name_val, document->GetAllocator());
+
+ kudu::rpc::RpcMethodInfo* method_info = method.second.get();
+ kudu::Histogram* handler_latency = method_info->handler_latency_histogram.get();
+ Value handler_latency_val(KrpcHistogramToString(handler_latency).c_str(),
+ document->GetAllocator());
+ method_entry.AddMember("handler_latency", handler_latency_val,
+ document->GetAllocator());
+
+ HistogramMetric* payload_size = payload_size_histograms_[method_name].get();
+ DCHECK(payload_size != nullptr);
+ Value payload_size_val(payload_size->ToHumanReadable().c_str(),
+ document->GetAllocator());
+ method_entry.AddMember("payload_size", payload_size_val, document->GetAllocator());
+
+ rpc_method_metrics.PushBack(method_entry, document->GetAllocator());
+ }
+ value->AddMember("rpc_method_metrics", rpc_method_metrics, document->GetAllocator());
+}
+
+
} // namespace impala
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/rpc/impala-service-pool.h
----------------------------------------------------------------------
diff --git a/be/src/rpc/impala-service-pool.h b/be/src/rpc/impala-service-pool.h
index 624e937..9d34366 100644
--- a/be/src/rpc/impala-service-pool.h
+++ b/be/src/rpc/impala-service-pool.h
@@ -46,7 +46,7 @@ class ImpalaServicePool : public kudu::rpc::RpcService {
/// 'service_mem_tracker' is the MemTracker for tracking the memory usage of RPC
/// payloads in the service queue.
ImpalaServicePool(const scoped_refptr<kudu::MetricEntity>& entity,
- size_t service_queue_length, kudu::rpc::ServiceIf* service,
+ size_t service_queue_length, kudu::rpc::GeneratedServiceIf* service,
MemTracker* service_mem_tracker);
virtual ~ImpalaServicePool();
@@ -57,13 +57,17 @@ class ImpalaServicePool : public kudu::rpc::RpcService {
/// Shut down the queue and the thread pool.
virtual void Shutdown();
- kudu::rpc::RpcMethodInfo* LookupMethod(const kudu::rpc::RemoteMethod& method) override;
+ virtual kudu::rpc::RpcMethodInfo* LookupMethod(const kudu::rpc::RemoteMethod& method)
+ override;
virtual kudu::Status
QueueInboundCall(gscoped_ptr<kudu::rpc::InboundCall> call) OVERRIDE;
const std::string service_name() const;
+ /// Expose the service pool metrics by storing them as JSON in 'value'.
+ void ToJson(rapidjson::Value* value, rapidjson::Document* document);
+
private:
void RunThread();
void RejectTooBusy(kudu::rpc::InboundCall* c);
@@ -80,7 +84,7 @@ class ImpalaServicePool : public kudu::rpc::RpcService {
MemTracker* const service_mem_tracker_;
/// Reference to the implementation of the RPC handlers. Not owned.
- kudu::rpc::ServiceIf* const service_;
+ kudu::rpc::GeneratedServiceIf* const service_;
/// The set of service threads started to process incoming RPC calls.
std::vector<std::unique_ptr<Thread>> threads_;
@@ -88,17 +92,15 @@ class ImpalaServicePool : public kudu::rpc::RpcService {
/// The pending RPCs to be dequeued by the service threads.
kudu::rpc::LifoServiceQueue service_queue_;
- /// TODO: Display these metrics in the debug webpage. IMPALA-6269
- /// Number of RPCs that timed out while waiting in the service queue.
- AtomicInt32 rpcs_timed_out_in_queue_;
+ /// Histogram to track time spent by requests in the krpc incoming requests queue.
+ scoped_refptr<kudu::Histogram> incoming_queue_time_;
- /// Number of RPCs that were rejected due to the queue being full.
- AtomicInt32 rpcs_queue_overflow_;
+ /// Histogram for incoming request payload size for each method of this service.
+ std::unordered_map<std::string, std::unique_ptr<HistogramMetric>>
+ payload_size_histograms_;
- /// Dummy histogram needed to call InboundCall::RecordHandlingStarted() to set
- /// appropriate internal KRPC state. Unused otherwise.
- /// TODO: Consider displaying this histogram in the debug webpage. IMPALA-6269
- scoped_refptr<kudu::Histogram> unused_histogram_;
+ /// Number of RPCs that were rejected due to the queue being full. Not owned.
+ IntCounter* rpcs_queue_overflow_= nullptr;
/// Protects against concurrent Shutdown() operations.
/// TODO: This seems implausible given our current usage pattern.
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/rpc/rpc-mgr-kerberized-test.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/rpc-mgr-kerberized-test.cc b/be/src/rpc/rpc-mgr-kerberized-test.cc
index 08c1971..bb4b9db 100644
--- a/be/src/rpc/rpc-mgr-kerberized-test.cc
+++ b/be/src/rpc/rpc-mgr-kerberized-test.cc
@@ -16,6 +16,7 @@
// under the License.
#include "rpc/rpc-mgr-test-base.h"
+#include "service/fe-support.h"
DECLARE_string(ssl_client_ca_certificate);
DECLARE_string(ssl_server_certificate);
@@ -27,7 +28,7 @@ static int kdc_port = GetServerPort();
class RpcMgrKerberizedTest :
public RpcMgrTestBase<testing::TestWithParam<KerberosSwitch> > {
- virtual void SetUp() {
+ virtual void SetUp() override {
IpAddr ip;
ASSERT_OK(HostnameToIpAddr(FLAGS_hostname, &ip));
string spn = Substitute("impala-test/$0", ip);
@@ -42,7 +43,7 @@ class RpcMgrKerberizedTest :
RpcMgrTestBase::SetUp();
}
- virtual void TearDown() {
+ virtual void TearDown() override {
ASSERT_OK(kdc_wrapper_->TearDownMiniKDC(GetParam()));
RpcMgrTestBase::TearDown();
}
@@ -82,7 +83,8 @@ TEST_P(RpcMgrKerberizedTest, MultipleServicesTls) {
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
- impala::InitCommonRuntime(argc, argv, false, impala::TestInfo::BE_TEST);
+ impala::InitCommonRuntime(argc, argv, true, impala::TestInfo::BE_TEST);
+ impala::InitFeSupport();
// Fill in the path of the current binary for use by the tests.
CURRENT_EXECUTABLE_PATH = argv[0];
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/rpc/rpc-mgr-test-base.h
----------------------------------------------------------------------
diff --git a/be/src/rpc/rpc-mgr-test-base.h b/be/src/rpc/rpc-mgr-test-base.h
index 4a79040..ce063f8 100644
--- a/be/src/rpc/rpc-mgr-test-base.h
+++ b/be/src/rpc/rpc-mgr-test-base.h
@@ -26,6 +26,7 @@
#include "kudu/util/monotime.h"
#include "kudu/util/status.h"
#include "rpc/auth-provider.h"
+#include "runtime/exec-env.h"
#include "runtime/mem-tracker.h"
#include "testutil/gtest-util.h"
#include "testutil/mini-kdc-wrapper.h"
@@ -40,7 +41,7 @@
#include "common/names.h"
-using kudu::rpc::ServiceIf;
+using kudu::rpc::GeneratedServiceIf;
using kudu::rpc::RpcController;
using kudu::rpc::RpcContext;
using kudu::rpc::RpcSidecar;
@@ -132,7 +133,7 @@ template <class T> class RpcMgrTestBase : public T {
// Takes over ownership of the newly created 'service' which needs to have a lifetime
// as long as 'rpc_mgr_' as RpcMgr::Shutdown() will call Shutdown() of 'service'.
- ServiceIf* TakeOverService(std::unique_ptr<ServiceIf> service) {
+ GeneratedServiceIf* TakeOverService(std::unique_ptr<GeneratedServiceIf> service) {
services_.emplace_back(move(service));
return services_.back().get();
}
@@ -141,14 +142,15 @@ template <class T> class RpcMgrTestBase : public T {
TNetworkAddress krpc_address_;
RpcMgr rpc_mgr_;
- virtual void SetUp() {
+ virtual void SetUp() override {
IpAddr ip;
ASSERT_OK(HostnameToIpAddr(FLAGS_hostname, &ip));
krpc_address_ = MakeNetworkAddress(ip, SERVICE_PORT);
+ exec_env_.reset(new ExecEnv());
ASSERT_OK(rpc_mgr_.Init());
}
- virtual void TearDown() {
+ virtual void TearDown() override {
rpc_mgr_.Shutdown();
}
@@ -156,7 +158,10 @@ template <class T> class RpcMgrTestBase : public T {
int32_t payload_[PAYLOAD_SIZE];
// Own all the services used by the test.
- std::vector<std::unique_ptr<ServiceIf>> services_;
+ std::vector<std::unique_ptr<GeneratedServiceIf>> services_;
+
+ // Required to set up the RPC metric groups used by the service pool.
+ std::unique_ptr<ExecEnv> exec_env_;
};
typedef std::function<void(RpcContext*)> ServiceCB;
@@ -169,8 +174,8 @@ class PingServiceImpl : public PingServiceIf {
ServiceCB cb = [](RpcContext* ctx) { ctx->RespondSuccess(); })
: PingServiceIf(entity, tracker), mem_tracker_(-1, "Ping Service"), cb_(cb) {}
- virtual void Ping(
- const PingRequestPB* request, PingResponsePB* response, RpcContext* context) {
+ virtual void Ping(const PingRequestPB* request, PingResponsePB* response, RpcContext*
+ context) override {
response->set_int_response(42);
// Incoming requests will already be tracked and we need to release the memory.
mem_tracker_.Release(context->GetTransferSize());
@@ -194,7 +199,7 @@ class ScanMemServiceImpl : public ScanMemServiceIf {
// The request comes with an int 'pattern' and a payload of int array sent with
// sidecar. Scan the array to make sure every element matches 'pattern'.
virtual void ScanMem(const ScanMemRequestPB* request, ScanMemResponsePB* response,
- RpcContext* context) {
+ RpcContext* context) override {
int32_t pattern = request->pattern();
Slice payload;
ASSERT_OK(
@@ -228,15 +233,16 @@ template <class T>
Status RunMultipleServicesTestTemplate(RpcMgrTestBase<T>* test_base,
RpcMgr* rpc_mgr, const TNetworkAddress& krpc_address) {
// Test that a service can be started, and will respond to requests.
- ServiceIf* ping_impl = test_base->TakeOverService(make_unique<PingServiceImpl>(
+ GeneratedServiceIf* ping_impl = test_base->TakeOverService(make_unique<PingServiceImpl>(
rpc_mgr->metric_entity(), rpc_mgr->result_tracker()));
RETURN_IF_ERROR(rpc_mgr->RegisterService(10, 10, ping_impl,
static_cast<PingServiceImpl*>(ping_impl)->mem_tracker()));
// Test that a second service, that verifies the RPC payload is not corrupted,
// can be started.
- ServiceIf* scan_mem_impl = test_base->TakeOverService(make_unique<ScanMemServiceImpl>(
- rpc_mgr->metric_entity(), rpc_mgr->result_tracker()));
+ GeneratedServiceIf* scan_mem_impl = test_base->TakeOverService(
+ make_unique<ScanMemServiceImpl>(rpc_mgr->metric_entity(),
+ rpc_mgr->result_tracker()));
RETURN_IF_ERROR(rpc_mgr->RegisterService(10, 10, scan_mem_impl,
static_cast<ScanMemServiceImpl*>(scan_mem_impl)->mem_tracker()));
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/rpc/rpc-mgr-test.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/rpc-mgr-test.cc b/be/src/rpc/rpc-mgr-test.cc
index 8d5312f..e90838e 100644
--- a/be/src/rpc/rpc-mgr-test.cc
+++ b/be/src/rpc/rpc-mgr-test.cc
@@ -16,8 +16,9 @@
// under the License.
#include "rpc/rpc-mgr-test-base.h"
+#include "service/fe-support.h"
-using kudu::rpc::ServiceIf;
+using kudu::rpc::GeneratedServiceIf;
using kudu::rpc::RpcController;
using kudu::rpc::RpcContext;
using kudu::MonoDelta;
@@ -31,11 +32,11 @@ namespace impala {
// For tests that do not require kerberized testing, we use RpcTest.
class RpcMgrTest : public RpcMgrTestBase<testing::Test> {
- virtual void SetUp() {
+ virtual void SetUp() override {
RpcMgrTestBase::SetUp();
}
- virtual void TearDown() {
+ virtual void TearDown() override {
RpcMgrTestBase::TearDown();
}
};
@@ -178,7 +179,7 @@ TEST_F(RpcMgrTest, SlowCallback) {
// Test a service which is slow to respond and has a short queue.
// Set a timeout on the client side. Expect either a client timeout
// or the service queue filling up.
- ServiceIf* ping_impl = TakeOverService(make_unique<PingServiceImpl>(
+ GeneratedServiceIf* ping_impl = TakeOverService(make_unique<PingServiceImpl>(
rpc_mgr_.metric_entity(), rpc_mgr_.result_tracker(), slow_cb));
const int num_service_threads = 1;
const int queue_size = 3;
@@ -204,7 +205,7 @@ TEST_F(RpcMgrTest, SlowCallback) {
}
TEST_F(RpcMgrTest, AsyncCall) {
- ServiceIf* scan_mem_impl = TakeOverService(make_unique<ScanMemServiceImpl>(
+ GeneratedServiceIf* scan_mem_impl = TakeOverService(make_unique<ScanMemServiceImpl>(
rpc_mgr_.metric_entity(), rpc_mgr_.result_tracker()));
ASSERT_OK(rpc_mgr_.RegisterService(10, 10, scan_mem_impl,
static_cast<ScanMemServiceImpl*>(scan_mem_impl)->mem_tracker()));
@@ -257,7 +258,8 @@ TEST_F(RpcMgrTest, NegotiationTimeout) {
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
- impala::InitCommonRuntime(argc, argv, false, impala::TestInfo::BE_TEST);
+ impala::InitCommonRuntime(argc, argv, true, impala::TestInfo::BE_TEST);
+ impala::InitFeSupport();
// Fill in the path of the current binary for use by the tests.
CURRENT_EXECUTABLE_PATH = argv[0];
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/rpc/rpc-mgr.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/rpc-mgr.cc b/be/src/rpc/rpc-mgr.cc
index 7e05f38..ed9614e 100644
--- a/be/src/rpc/rpc-mgr.cc
+++ b/be/src/rpc/rpc-mgr.cc
@@ -31,14 +31,19 @@
#include "common/names.h"
+using namespace rapidjson;
+
using kudu::HostPort;
using kudu::MetricEntity;
using kudu::MonoDelta;
using kudu::rpc::AcceptorPool;
+using kudu::rpc::DumpRunningRpcsRequestPB;
+using kudu::rpc::DumpRunningRpcsResponsePB;
using kudu::rpc::MessengerBuilder;
using kudu::rpc::Messenger;
+using kudu::rpc::RpcConnectionPB;
using kudu::rpc::RpcController;
-using kudu::rpc::ServiceIf;
+using kudu::rpc::GeneratedServiceIf;
using kudu::Sockaddr;
DECLARE_string(hostname);
@@ -127,7 +132,7 @@ Status RpcMgr::Init() {
}
Status RpcMgr::RegisterService(int32_t num_service_threads, int32_t service_queue_depth,
- ServiceIf* service_ptr, MemTracker* service_mem_tracker) {
+ GeneratedServiceIf* service_ptr, MemTracker* service_mem_tracker) {
DCHECK(is_inited()) << "Must call Init() before RegisterService()";
DCHECK(!services_started_) << "Cannot call RegisterService() after StartServices()";
scoped_refptr<ImpalaServicePool> service_pool = new ImpalaServicePool(
@@ -154,10 +159,9 @@ Status RpcMgr::StartServices(const TNetworkAddress& address) {
RETURN_IF_ERROR(TNetworkAddressToSockaddr(address, &sockaddr));
// Call the messenger to create an AcceptorPool for us.
- shared_ptr<AcceptorPool> acceptor_pool;
- KUDU_RETURN_IF_ERROR(messenger_->AddAcceptorPool(sockaddr, &acceptor_pool),
+ KUDU_RETURN_IF_ERROR(messenger_->AddAcceptorPool(sockaddr, &acceptor_pool_),
"Failed to add acceptor pool");
- KUDU_RETURN_IF_ERROR(acceptor_pool->Start(FLAGS_num_acceptor_threads),
+ KUDU_RETURN_IF_ERROR(acceptor_pool_->Start(FLAGS_num_acceptor_threads),
"Acceptor pool failed to start");
VLOG_QUERY << "Started " << FLAGS_num_acceptor_threads << " acceptor threads";
services_started_ = true;
@@ -167,6 +171,7 @@ Status RpcMgr::StartServices(const TNetworkAddress& address) {
void RpcMgr::Shutdown() {
if (messenger_.get() == nullptr) return;
for (auto service_pool : service_pools_) service_pool->Shutdown();
+ acceptor_pool_.reset();
messenger_->UnregisterAllServices();
messenger_->Shutdown();
@@ -180,4 +185,41 @@ bool RpcMgr::IsServerTooBusy(const RpcController& rpc_controller) {
err->code() == kudu::rpc::ErrorStatusPB::ERROR_SERVER_TOO_BUSY;
}
+void RpcMgr::ToJson(Document* document) {
+ if (messenger_.get() == nullptr) return;
+ // Add acceptor metrics.
+ int64_t num_accepted = 0;
+ if (acceptor_pool_.get() != nullptr) {
+ num_accepted = acceptor_pool_->num_rpc_connections_accepted();
+ }
+ document->AddMember("rpc_connections_accepted", num_accepted, document->GetAllocator());
+
+ // Add messenger metrics.
+ DumpRunningRpcsResponsePB response;
+ messenger_->DumpRunningRpcs(DumpRunningRpcsRequestPB(), &response);
+
+ int64_t num_inbound_calls_in_flight = 0;
+ for (const RpcConnectionPB& conn : response.inbound_connections()) {
+ num_inbound_calls_in_flight += conn.calls_in_flight().size();
+ }
+ document->AddMember("num_inbound_calls_in_flight", num_inbound_calls_in_flight,
+ document->GetAllocator());
+
+ int64_t num_outbound_calls_in_flight = 0;
+ for (const RpcConnectionPB& conn : response.outbound_connections()) {
+ num_outbound_calls_in_flight += conn.calls_in_flight().size();
+ }
+ document->AddMember("num_outbound_calls_in_flight", num_outbound_calls_in_flight,
+ document->GetAllocator());
+
+ // Add service pool metrics
+ Value services(kArrayType);
+ for (auto service_pool : service_pools_) {
+ Value service_entry(kObjectType);
+ service_pool->ToJson(&service_entry, document);
+ services.PushBack(service_entry, document->GetAllocator());
+ }
+ document->AddMember("services", services, document->GetAllocator());
+}
+
} // namespace impala
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/rpc/rpc-mgr.h
----------------------------------------------------------------------
diff --git a/be/src/rpc/rpc-mgr.h b/be/src/rpc/rpc-mgr.h
index e87b559..c7107d2 100644
--- a/be/src/rpc/rpc-mgr.h
+++ b/be/src/rpc/rpc-mgr.h
@@ -29,7 +29,7 @@
namespace kudu {
namespace rpc {
class RpcController;
-class ServiceIf;
+class GeneratedServiceIf;
} // rpc
} // kudu
@@ -95,6 +95,9 @@ namespace impala {
/// Inbound connection set-up is handled by a small fixed-size pool of 'acceptor'
/// threads. The number of threads that accept new TCP connection requests to the service
/// port is configurable via FLAGS_acceptor_threads.
+///
+/// If 'use_tls' is true, then the underlying messenger is configured with the required
+/// certificates, and encryption is enabled and marked as required.
class RpcMgr {
public:
RpcMgr(bool use_tls = false) : use_tls_(use_tls) {}
@@ -127,7 +130,7 @@ class RpcMgr {
///
/// It is an error to call this after StartServices() has been called.
Status RegisterService(int32_t num_service_threads, int32_t service_queue_depth,
- kudu::rpc::ServiceIf* service_ptr, MemTracker* service_mem_tracker)
+ kudu::rpc::GeneratedServiceIf* service_ptr, MemTracker* service_mem_tracker)
WARN_UNUSED_RESULT;
/// Creates a new proxy for a remote service of type P at location 'address', and places
@@ -157,6 +160,11 @@ class RpcMgr {
std::shared_ptr<kudu::rpc::Messenger> messenger() { return messenger_; }
+ /// Writes a JSON representation of the RpcMgr's metrics to a value named 'services' in
+ /// 'document'. It will include the number of RPCs accepted so far, the number of calls
+ /// in flight, and metrics and histograms for each service and their methods.
+ void ToJson(rapidjson::Document* document);
+
~RpcMgr() {
DCHECK_EQ(service_pools_.size(), 0)
<< "Must call Shutdown() before destroying RpcMgr";
@@ -175,6 +183,9 @@ class RpcMgr {
/// track results for idempotent RPC calls.
const scoped_refptr<kudu::rpc::ResultTracker> tracker_;
+ /// Holds a reference to the acceptor pool. Shared ownership with messenger_.
+ std::shared_ptr<kudu::rpc::AcceptorPool> acceptor_pool_;
+
/// Container for reactor threads which run event loops for RPC services, plus acceptor
/// threads which manage connection setup. Has to be a shared_ptr as required by
/// MessangerBuilder::Build().
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/rpc/rpc-trace.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/rpc-trace.cc b/be/src/rpc/rpc-trace.cc
index f4da19b..028f397 100644
--- a/be/src/rpc/rpc-trace.cc
+++ b/be/src/rpc/rpc-trace.cc
@@ -22,6 +22,7 @@
#include <gutil/strings/substitute.h>
#include "common/logging.h"
+#include "rpc/rpc-mgr.h"
#include "util/debug-util.h"
#include "util/time.h"
#include "util/webserver.h"
@@ -39,6 +40,8 @@ const string RPC_PROCESSING_TIME_DISTRIBUTION_METRIC_KEY = "rpc-method.$0.call_d
// web-based summary page.
class RpcEventHandlerManager {
public:
+ RpcEventHandlerManager(RpcMgr* rpc_mgr) : rpc_mgr_(rpc_mgr) {}
+
// Adds an event handler to the list of those tracked
void RegisterEventHandler(RpcEventHandler* event_handler);
@@ -64,14 +67,19 @@ class RpcEventHandlerManager {
// after they are started, event handlers have a lifetime equivalent to the length of
// the process.
vector<RpcEventHandler*> event_handlers_;
+
+ // Points to an RpcMgr. If this is not null, then its metrics will be included in the
+ // output of JsonCallback. Not owned, but the object must be guaranteed to live as long
+ // as the process lives.
+ RpcMgr* rpc_mgr_ = nullptr;
};
// Only instance of RpcEventHandlerManager
scoped_ptr<RpcEventHandlerManager> handler_manager;
-void impala::InitRpcEventTracing(Webserver* webserver) {
- handler_manager.reset(new RpcEventHandlerManager());
- if (webserver != NULL) {
+void impala::InitRpcEventTracing(Webserver* webserver, RpcMgr* rpc_mgr) {
+ handler_manager.reset(new RpcEventHandlerManager(rpc_mgr));
+ if (webserver != nullptr) {
Webserver::UrlCallback json = bind<void>(
mem_fn(&RpcEventHandlerManager::JsonCallback), handler_manager.get(), _1, _2);
webserver->RegisterUrlCallback("/rpcz", "rpcz.tmpl", json);
@@ -83,7 +91,7 @@ void impala::InitRpcEventTracing(Webserver* webserver) {
}
void RpcEventHandlerManager::RegisterEventHandler(RpcEventHandler* event_handler) {
- DCHECK(event_handler != NULL);
+ DCHECK(event_handler != nullptr);
lock_guard<mutex> l(lock_);
event_handlers_.push_back(event_handler);
}
@@ -98,6 +106,7 @@ void RpcEventHandlerManager::JsonCallback(const Webserver::ArgumentMap& args,
servers.PushBack(server, document->GetAllocator());
}
document->AddMember("servers", servers, document->GetAllocator());
+ if (rpc_mgr_ != nullptr) rpc_mgr_->ToJson(document);
}
void RpcEventHandlerManager::ResetCallback(const Webserver::ArgumentMap& args,
@@ -137,7 +146,7 @@ void RpcEventHandler::ResetAll() {
RpcEventHandler::RpcEventHandler(const string& server_name, MetricGroup* metrics) :
server_name_(server_name), metrics_(metrics) {
- if (handler_manager.get() != NULL) handler_manager->RegisterEventHandler(this);
+ if (handler_manager.get() != nullptr) handler_manager->RegisterEventHandler(this);
}
void RpcEventHandler::ToJson(Value* server, Document* document) {
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/rpc/rpc-trace.h
----------------------------------------------------------------------
diff --git a/be/src/rpc/rpc-trace.h b/be/src/rpc/rpc-trace.h
index 1bd7823..12c7395 100644
--- a/be/src/rpc/rpc-trace.h
+++ b/be/src/rpc/rpc-trace.h
@@ -31,6 +31,7 @@
namespace impala {
class MetricGroup;
+class RpcMgr;
class Webserver;
/// An RpcEventHandler is called every time an Rpc is started and completed. There is at
@@ -126,7 +127,7 @@ class RpcEventHandler : public apache::thrift::TProcessorEventHandler {
};
/// Initialises rpc event tracing, must be called before any RpcEventHandlers are created.
-void InitRpcEventTracing(Webserver* webserver);
+void InitRpcEventTracing(Webserver* webserver, RpcMgr* = nullptr);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/runtime/exec-env.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index ce8dae0..8a51d88 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -67,7 +67,7 @@
#include "common/names.h"
using boost::algorithm::join;
-using kudu::rpc::ServiceIf;
+using kudu::rpc::GeneratedServiceIf;
using namespace strings;
DEFINE_string(catalog_service_host, "localhost",
@@ -152,6 +152,7 @@ ExecEnv::ExecEnv(const string& hostname, int backend_port, int krpc_port,
frontend_(new Frontend()),
async_rpc_pool_(new CallableThreadPool("rpc-pool", "async-rpc-sender", 8, 10000)),
query_exec_mgr_(new QueryExecMgr()),
+ rpc_metrics_(metrics_->GetOrCreateChildGroup("rpc")),
enable_webserver_(FLAGS_enable_webserver && webserver_port > 0),
backend_address_(MakeNetworkAddress(hostname, backend_port)) {
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/runtime/exec-env.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/exec-env.h b/be/src/runtime/exec-env.h
index 3f050c9..7741276 100644
--- a/be/src/runtime/exec-env.h
+++ b/be/src/runtime/exec-env.h
@@ -83,7 +83,7 @@ class ExecEnv {
int subscriber_port, int webserver_port, const std::string& statestore_host,
int statestore_port);
- /// Returns the first created exec env instance. In a normal impalad, this is
+ /// Returns the most recently created exec env instance. In a normal impalad, this is
/// the only instance. In test setups with multiple ExecEnv's per process,
/// we return the most recently created instance.
static ExecEnv* GetInstance() { return exec_env_; }
@@ -123,6 +123,7 @@ class ExecEnv {
io::DiskIoMgr* disk_io_mgr() { return disk_io_mgr_.get(); }
Webserver* webserver() { return webserver_.get(); }
MetricGroup* metrics() { return metrics_.get(); }
+ MetricGroup* rpc_metrics() { return rpc_metrics_; }
MemTracker* process_mem_tracker() { return mem_tracker_.get(); }
ThreadResourceMgr* thread_mgr() { return thread_mgr_.get(); }
HdfsOpThreadPool* hdfs_op_thread_pool() { return hdfs_op_thread_pool_.get(); }
@@ -210,6 +211,7 @@ class ExecEnv {
/// Not owned by this class
ImpalaServer* impala_server_ = nullptr;
+ MetricGroup* rpc_metrics_ = nullptr;
bool enable_webserver_;
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/service/impalad-main.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impalad-main.cc b/be/src/service/impalad-main.cc
index 8463467..b4ec3f0 100644
--- a/be/src/service/impalad-main.cc
+++ b/be/src/service/impalad-main.cc
@@ -75,7 +75,7 @@ int ImpaladMain(int argc, char** argv) {
ABORT_IF_ERROR(StartMemoryMaintenanceThread()); // Memory metrics are created in Init().
ABORT_IF_ERROR(
StartThreadInstrumentation(exec_env.metrics(), exec_env.webserver(), true));
- InitRpcEventTracing(exec_env.webserver());
+ InitRpcEventTracing(exec_env.webserver(), exec_env.rpc_mgr());
boost::shared_ptr<ImpalaServer> impala_server(new ImpalaServer(&exec_env));
Status status =
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/util/histogram-metric.h
----------------------------------------------------------------------
diff --git a/be/src/util/histogram-metric.h b/be/src/util/histogram-metric.h
index a520947..d4e09e4 100644
--- a/be/src/util/histogram-metric.h
+++ b/be/src/util/histogram-metric.h
@@ -39,7 +39,7 @@ class HistogramMetric : public Metric {
DCHECK_EQ(TMetricKind::HISTOGRAM, def.kind);
}
- virtual void ToJson(rapidjson::Document* document, rapidjson::Value* value) {
+ virtual void ToJson(rapidjson::Document* document, rapidjson::Value* value) override {
rapidjson::Value container(rapidjson::kObjectType);
AddStandardFields(document, &container);
@@ -84,36 +84,47 @@ class HistogramMetric : public Metric {
histogram_.reset(new HdrHistogram(highest, digits));
}
- virtual void ToLegacyJson(rapidjson::Document*) { }
+ virtual void ToLegacyJson(rapidjson::Document*) override {}
const TUnit::type& unit() const { return unit_; }
- virtual std::string ToHumanReadable() {
+ virtual std::string ToHumanReadable() override {
boost::lock_guard<SpinLock> l(lock_);
+ return HistogramToHumanReadable(histogram_.get(), unit_);
+ }
+
+ /// Render a HdrHistogram into a human readable string representation. The histogram
+ /// type is a template parameter so that it accepts both Impala's and Kudu's
+ /// HdrHistogram classes.
+ template <class T>
+ static std::string HistogramToHumanReadable(T* histogram, TUnit::type unit) {
+ DCHECK(histogram != nullptr);
std::stringstream out;
- out << "Count: " << histogram_->TotalCount() << ", "
- << "min / max: " << PrettyPrinter::Print(histogram_->MinValue(), unit_)
- << " / " << PrettyPrinter::Print(histogram_->MaxValue(), unit_) << ", "
+ out << "Count: " << histogram->TotalCount() << ", "
+ << "min / max: " << PrettyPrinter::Print(histogram->MinValue(), unit)
+ << " / " << PrettyPrinter::Print(histogram->MaxValue(), unit) << ", "
<< "25th %-ile: "
- << PrettyPrinter::Print(histogram_->ValueAtPercentile(25), unit_) << ", "
+ << PrettyPrinter::Print(histogram->ValueAtPercentile(25), unit) << ", "
<< "50th %-ile: "
- << PrettyPrinter::Print(histogram_->ValueAtPercentile(50), unit_) << ", "
+ << PrettyPrinter::Print(histogram->ValueAtPercentile(50), unit) << ", "
<< "75th %-ile: "
- << PrettyPrinter::Print(histogram_->ValueAtPercentile(75), unit_) << ", "
+ << PrettyPrinter::Print(histogram->ValueAtPercentile(75), unit) << ", "
<< "90th %-ile: "
- << PrettyPrinter::Print(histogram_->ValueAtPercentile(90), unit_) << ", "
+ << PrettyPrinter::Print(histogram->ValueAtPercentile(90), unit) << ", "
<< "95th %-ile: "
- << PrettyPrinter::Print(histogram_->ValueAtPercentile(95), unit_) << ", "
+ << PrettyPrinter::Print(histogram->ValueAtPercentile(95), unit) << ", "
<< "99.9th %-ile: "
- << PrettyPrinter::Print(histogram_->ValueAtPercentile(99.9), unit_);
+ << PrettyPrinter::Print(histogram->ValueAtPercentile(99.9), unit);
return out.str();
}
private:
- // Protects histogram_ pointer itself.
+ /// Protects histogram_ pointer itself.
SpinLock lock_;
boost::scoped_ptr<HdrHistogram> histogram_;
const TUnit::type unit_;
+
+ DISALLOW_COPY_AND_ASSIGN(HistogramMetric);
};
}
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/be/src/util/pretty-printer.h
----------------------------------------------------------------------
diff --git a/be/src/util/pretty-printer.h b/be/src/util/pretty-printer.h
index 5e76db1..03ba99e 100644
--- a/be/src/util/pretty-printer.h
+++ b/be/src/util/pretty-printer.h
@@ -91,22 +91,12 @@ class PrettyPrinter {
}
case TUnit::TIME_NS: {
- ss << std::setprecision(TIME_NS_PRECISION);
- if (value >= BILLION) {
- /// If the time is over a second, print it up to ms.
- value /= MILLION;
- PrintTimeMs(value, &ss);
- } else if (value >= MILLION) {
- /// if the time is over a ms, print it up to microsecond in the unit of ms.
- ss << DOUBLE_TRUNCATE(static_cast<double>(value) / MILLION, TIME_NS_PRECISION)
- << "ms";
- } else if (value > 1000) {
- /// if the time is over a microsecond, print it using unit microsecond
- ss << DOUBLE_TRUNCATE(static_cast<double>(value) / 1000, TIME_NS_PRECISION)
- << "us";
- } else {
- ss << DOUBLE_TRUNCATE(value, TIME_NS_PRECISION) << "ns";
- }
+ PrintTimeNs(value, &ss);
+ break;
+ }
+
+ case TUnit::TIME_US: {
+ PrintTimeNs(value * THOUSAND, &ss);
break;
}
@@ -204,13 +194,13 @@ class PrettyPrinter {
if (value == 0) {
*unit = "";
return value;
- } else if (value >= GIGABYTE || value <= -GIGABYTE) {
+ } else if (value >= GIGABYTE || (value < 0 && value <= -GIGABYTE)) {
*unit = "GB";
return value / (double) GIGABYTE;
- } else if (value >= MEGABYTE || value <= -MEGABYTE ) {
+ } else if (value >= MEGABYTE || (value < 0 && value <= -MEGABYTE)) {
*unit = "MB";
return value / (double) MEGABYTE;
- } else if (value >= KILOBYTE || value <= -KILOBYTE) {
+ } else if (value >= KILOBYTE || (value < 0 && value <= -KILOBYTE)) {
*unit = "KB";
return value / (double) KILOBYTE;
} else {
@@ -247,7 +237,28 @@ class PrettyPrinter {
return fmod(value, 1. * modulus);
}
- /// Print the value (time in ms) to ss
+ /// Pretty print the value (time in ns) to ss.
+ template <typename T>
+ static void PrintTimeNs(T value, std::stringstream* ss) {
+ *ss << std::setprecision(TIME_NS_PRECISION);
+ if (value >= BILLION) {
+ /// If the time is over a second, print it up to ms.
+ value /= MILLION;
+ PrintTimeMs(value, ss);
+ } else if (value >= MILLION) {
+ /// if the time is over a ms, print it up to microsecond in the unit of ms.
+ *ss << DOUBLE_TRUNCATE(static_cast<double>(value) / MILLION, TIME_NS_PRECISION)
+ << "ms";
+ } else if (value > THOUSAND) {
+ /// if the time is over a microsecond, print it using unit microsecond.
+ *ss << DOUBLE_TRUNCATE(static_cast<double>(value) / THOUSAND, TIME_NS_PRECISION)
+ << "us";
+ } else {
+ *ss << DOUBLE_TRUNCATE(value, TIME_NS_PRECISION) << "ns";
+ }
+ }
+
+ /// Print the value (time in ms) to ss.
template <typename T>
static void PrintTimeMs(T value, std::stringstream* ss) {
DCHECK_GE(value, static_cast<T>(0));
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/common/thrift/Metrics.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/Metrics.thrift b/common/thrift/Metrics.thrift
index 04f5946..4f2c7f2 100644
--- a/common/thrift/Metrics.thrift
+++ b/common/thrift/Metrics.thrift
@@ -32,7 +32,8 @@ enum TUnit {
// No units at all, may not be a numerical quantity
NONE,
TIME_MS,
- TIME_S
+ TIME_S,
+ TIME_US
}
// The kind of value that a metric represents.
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/common/thrift/metrics.json
----------------------------------------------------------------------
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index b457741..6328cd4 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -1635,5 +1635,15 @@
"units": "NONE",
"kind": "PROPERTY",
"key": "kudu-client.version"
+ },
+ {
+ "description": "Service $0: Total number of incoming RPCs that were rejected due to overflow of the service queue.",
+ "contexts": [
+ "IMPALAD"
+ ],
+ "label": "Service $0 Incoming RPC Queue Overflows",
+ "units": "UNIT",
+ "kind": "COUNTER",
+ "key": "rpc.$0.rpcs_queue_overflow"
}
]
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/tests/custom_cluster/test_krpc_metrics.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_krpc_metrics.py b/tests/custom_cluster/test_krpc_metrics.py
new file mode 100644
index 0000000..ec56f41
--- /dev/null
+++ b/tests/custom_cluster/test_krpc_metrics.py
@@ -0,0 +1,97 @@
+# 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.
+
+import json
+import pytest
+import requests
+import time
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+from tests.common.impala_cluster import ImpalaCluster
+from tests.common.skip import SkipIf, SkipIfBuildType
+from tests.verifiers.mem_usage_verifier import MemUsageVerifier
+
+@SkipIf.not_krpc
+class TestKrpcMetrics(CustomClusterTestSuite):
+ """Test for KRPC metrics that require special arguments during cluster startup."""
+ RPCZ_URL = 'http://localhost:25000/rpcz?json'
+ METRICS_URL = 'http://localhost:25000/metrics?json'
+ TEST_QUERY = 'select count(*) from tpch_parquet.lineitem l1 \
+ join tpch_parquet.lineitem l2 where l1.l_orderkey = l2.l_orderkey;'
+
+ @classmethod
+ def get_workload(self):
+ return 'functional-query'
+
+ @classmethod
+ def setup_class(cls):
+ if cls.exploration_strategy() != 'exhaustive':
+ pytest.skip('runs only in exhaustive')
+ super(TestKrpcMetrics, cls).setup_class()
+
+ def get_debug_page(self, page_url):
+ """Returns the content of the debug page 'page_url' as json."""
+ response = requests.get(page_url)
+ assert response.status_code == requests.codes.ok
+ return json.loads(response.text)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args('-datastream_service_queue_mem_limit=1B \
+ -datastream_service_num_svc_threads=1')
+ def test_krpc_queue_overflow_rpcz(self, vector):
+ """Test that rejected RPCs show up on the /rpcz debug web page.
+ """
+ def get_rpc_overflows():
+ rpcz = self.get_debug_page(self.RPCZ_URL)
+ assert len(rpcz['services']) > 0
+ for s in rpcz['services']:
+ if s['service_name'] == 'impala.DataStreamService':
+ return int(s['rpcs_queue_overflow'])
+ assert False, "Could not find DataStreamService metrics"
+
+ before = get_rpc_overflows()
+ assert before == 0
+ self.client.execute(self.TEST_QUERY)
+ after = get_rpc_overflows()
+
+ assert before < after
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args('-datastream_service_queue_mem_limit=1B \
+ -datastream_service_num_svc_threads=1')
+ def test_krpc_queue_overflow_metrics(self, vector):
+ """Test that rejected RPCs show up on the /metrics debug web page.
+ """
+ def iter_metrics(group):
+ for m in group['metrics']:
+ yield m
+ for c in group['child_groups']:
+ for m in iter_metrics(c):
+ yield m
+
+ def get_metric(name):
+ metrics = self.get_debug_page(self.METRICS_URL)['metric_group']
+ for m in iter_metrics(metrics):
+ if m['name'] == name:
+ return int(m['value'])
+
+ metric_name = 'rpc.impala.DataStreamService.rpcs_queue_overflow'
+ before = get_metric(metric_name)
+ assert before == 0
+
+ self.client.execute(self.TEST_QUERY)
+ after = get_metric(metric_name)
+ assert before < after
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/tests/webserver/test_web_pages.py
----------------------------------------------------------------------
diff --git a/tests/webserver/test_web_pages.py b/tests/webserver/test_web_pages.py
index 8dd17a4..c736cb3 100644
--- a/tests/webserver/test_web_pages.py
+++ b/tests/webserver/test_web_pages.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+from tests.common.skip import SkipIf
from tests.common.impala_cluster import ImpalaCluster
from tests.common.impala_test_suite import ImpalaTestSuite
import json
@@ -32,6 +33,7 @@ class TestWebPage(ImpalaTestSuite):
TABLE_METRICS_URL = "http://localhost:{0}/table_metrics"
QUERY_BACKENDS_URL = "http://localhost:{0}/query_backends"
QUERY_FINSTANCES_URL = "http://localhost:{0}/query_finstances"
+ RPCZ_URL = "http://localhost:{0}/rpcz"
THREAD_GROUP_URL = "http://localhost:{0}/thread-group"
# log4j changes do not apply to the statestore since it doesn't
# have an embedded JVM. So we make two sets of ports to test the
@@ -72,6 +74,11 @@ class TestWebPage(ImpalaTestSuite):
and string_to_search in response.text, "Offending url: " + input_url
return response.text
+ def get_debug_page(self, page_url):
+ """Returns the content of the debug page 'page_url' as json."""
+ response = self.get_and_check_status(page_url + "?json", ports_to_test=[25000])
+ return json.loads(response)
+
def get_and_check_status_jvm(self, url, string_to_search = ""):
"""Calls get_and_check_status() for impalad and catalogd only"""
return self.get_and_check_status(url, string_to_search,
@@ -229,3 +236,26 @@ class TestWebPage(ImpalaTestSuite):
for pattern in expected_name_patterns:
assert any(pattern in t for t in thread_names), \
"Could not find thread matching '%s'" % pattern
+
+ @SkipIf.not_krpc
+ def test_krpc_rpcz(self):
+ """Test that KRPC metrics are exposed in /rpcz and that they are updated when
+ executing a query."""
+ TEST_QUERY = "select count(c2.string_col) from \
+ functional.alltypestiny join functional.alltypessmall c2"
+ SVC_NAME = 'impala.DataStreamService'
+
+ def get_svc_metrics(svc_name):
+ rpcz = self.get_debug_page(self.RPCZ_URL)
+ assert len(rpcz['services']) > 0
+ for s in rpcz['services']:
+ if s['service_name'] == svc_name:
+ assert len(s['rpc_method_metrics']) > 0, '%s metrics are empty' % svc_name
+ return sorted(s['rpc_method_metrics'], key=lambda m: m['method_name'])
+ assert False, 'Could not find metrics for %s' % svc_name
+
+ before = get_svc_metrics(SVC_NAME)
+ self.client.execute(TEST_QUERY)
+ after = get_svc_metrics(SVC_NAME)
+
+ assert before != after
http://git-wip-us.apache.org/repos/asf/impala/blob/74dd9e9b/www/rpcz.tmpl
----------------------------------------------------------------------
diff --git a/www/rpcz.tmpl b/www/rpcz.tmpl
index e62e662..8ade9ef 100644
--- a/www/rpcz.tmpl
+++ b/www/rpcz.tmpl
@@ -18,11 +18,7 @@ under the License.
-->
{{> www/common-header.tmpl }}
-<h2>RPC durations
- <button class="btn btn-warning btn-xs" onClick="reset_all();">
- Reset all
- </button>
-</h2>
+<h2>RPC durations</h2>
<p class="lead">This page shows the durations of all RPCs served by this
<samp>{{__common__.process-name}}</samp> process.
@@ -33,6 +29,70 @@ under the License.
<span id="refresh_on">Auto-refresh on</span>
</label> Last updated: <span id="last-updated"></span>
+{{?services}}
+<h2>KRPC Services</h2>
+{{/services}}
+{{#services}}
+<h3><samp>{{service_name}}</samp></h3>
+<table class="table table-hover table-bordered" id="{{service_name}}_metrics">
+ <tbody>
+ <tr>
+ <td>
+ <table class="table table-hover">
+ <tr>
+ <th>Queue Size</th>
+ <th>Idle Threads</th>
+ <th>Current Memory Usage</th>
+ <th>Peak Memory Usage</th>
+ <th>RPCs Rejected due to Queue Overflow</th>
+ </tr>
+ <tr>
+ <td id="{{service_name}}_queue_size">{{queue_size}}</td>
+ <td id="{{service_name}}_idle_threads">{{idle_threads}}</td>
+ <td id="{{service_name}}_mem_usage">{{mem_usage}}</td>
+ <td id="{{service_name}}_mem_peak">{{mem_peak}}</td>
+ <td id="{{service_name}}_rpcs_queue_overflow">{{rpcs_queue_overflow}}</td>
+ </tr>
+ </table>
+ <table class="table table-hover">
+ <tr>
+ <th>Incoming Queueing Time</th>
+ <td id="{{service_name}}_incoming_queue_time" colspan=2>{{incoming_queue_time}}
+ </td>
+ </tr>
+ </table>
+ </td>
+ </tr>
+ <tr>
+ <td>
+ {{#rpc_method_metrics}}
+ <table class="table table-bordered table-hover">
+ <tr><td colspan=2>
+ <strong>Method: <i>{{method_name}}</i></strong>
+ </td></tr>
+ <tr>
+ <td>Handler Latency</td>
+ <td class="{{method_name}}_handler_latency">{{handler_latency}}</td>
+ </tr>
+ <tr>
+ <td>Payload Size</td>
+ <td class="{{method_name}}_payload_size">{{payload_size}}</td>
+ </tr>
+ </table>
+ {{/rpc_method_metrics}}
+ </td>
+ </tr>
+ </tbody>
+</table>
+{{/services}}
+
+{{?servers}}
+<h2>Impala RPC Services
+ <button class="btn btn-warning btn-xs" onClick="reset_all();">
+ Reset all
+ </button>
+</h2>
+{{/servers}}
{{#servers}}
<h3><samp>{{name}} </samp>
@@ -83,44 +143,84 @@ function reset_server(server) {
xhr.send();
}
+// Update all metrics for services in "servers", which use Impala's old RPC layer.
+function update_impala_services(json) {
+ for (var i = 0; i < json["servers"].length; ++i) {
+ var tbl_json = json["servers"][i];
+ var table = document.getElementById(tbl_json["name"]);
+ if (!table) continue;
+ // Delete all existing rows, stopping at 1 to save the header
+ for (var j = table.rows.length - 1; j >= 1; --j) table.deleteRow(j);
+ tbl_json["methods"].forEach(function(method) {
+ var row = table.insertRow();
+ row.insertCell().innerHTML = "<samp>" + method.name + "</samp>";
+ row.insertCell().innerHTML = method.summary;
+ row.insertCell().innerHTML = method.in_flight;
+ var reset_cell = row.insertCell();
+ reset_cell.align = "center";
+ var button = document.createElement("button");
+ button.className = "btn btn-warning btn-xs";
+ button.appendChild(document.createTextNode("Reset"));
+ button.onclick = function() { reset_method(method.server_name, method.name); }
+ reset_cell.appendChild(button);
+ });
+ }
+}
+
+// Update all krpc metrics from the "services" member of "json".
+function update_krpc_services(json) {
+ // Update each service
+ for (var i = 0; i < json["services"].length; ++i) {
+ var svc_json = json["services"][i];
+ var svc_name = svc_json["service_name"];
+ var table = document.getElementById(svc_name + "_metrics");
+
+ // Skip updates for unknown services.
+ if (!table) continue;
+
+ // Update service metrics
+ var keys = ["queue_size", "idle_threads", "mem_usage", "mem_peak",
+ "rpcs_queue_overflow", "incoming_queue_time"];
+ for (var j = 0; j < keys.length; ++j) {
+ var key = keys[j];
+ var cell = document.getElementById(svc_name + "_" + key);
+ // Skip update for unknown values.
+ if (!cell) continue;
+ cell.innerHTML = svc_json[key];
+ }
+
+ // Update metrics for individual methods.
+ var num_methods = svc_json["rpc_method_metrics"].length;
+ for (var j = 0; j < num_methods; ++j) {
+ var method_json = svc_json["rpc_method_metrics"][j];
+ var method_name = method_json["method_name"];
+ // Update all metrics for this method.
+ var keys = ["handler_latency", "payload_size"];
+ for (var l = 0; l < keys.length; ++l) {
+ var key = keys[l];
+ var cell = $(table).find("." + method_name + "_" + key)[0];
+ // Skip update for unknown values.
+ if (!cell) continue;
+ cell.innerHTML = method_json[key];
+ }
+ }
+ }
+}
+
function refresh() {
var xhr = new XMLHttpRequest();
xhr.responseType = 'text';
xhr.timeout = 60000;
xhr.onload = function(ignored_arg) {
- if (xhr.status != 200) {
- return;
- }
+ if (xhr.status != 200) return;
var blob = xhr.response;
json = JSON.parse(blob);
- for (var i = 0; i < json["servers"].length; ++i) {
- var tbl_json = json["servers"][i];
- var table = document.getElementById(tbl_json["name"]);
- if (!table) continue;
- // Delete all existing rows, stopping at 1 to save the header
- for (var j = table.rows.length - 1; j >= 1; --j) {
- table.deleteRow(j);
- }
- tbl_json["methods"].forEach(function(method) {
- var row = table.insertRow();
- row.insertCell().innerHTML = "<samp>" + method.name + "</samp>";
- row.insertCell().innerHTML = method.summary;
- row.insertCell().innerHTML = method.in_flight;
- var reset_cell = row.insertCell();
- reset_cell.align = "center";
- var button = document.createElement("button");
- button.className = "btn btn-warning btn-xs";
- button.appendChild(document.createTextNode("Reset"));
- button.onclick = function() { reset_method(method.server_name, method.name); }
-
- reset_cell.appendChild(button);
- });
- }
-
+ update_impala_services(json);
+ update_krpc_services(json);
document.getElementById("last-updated").textContent = new Date();
}
- xhr.ontimeout = function(){ }
+ xhr.ontimeout = function() {}
xhr.open('GET', "/rpcz?json", true);
xhr.send();
}