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/09/27 20:43:12 UTC
[2/7] impala git commit: IMPALA-7596. Adding JvmPauseMonitor (and
other GC) metrics to Impala metrics.
IMPALA-7596. Adding JvmPauseMonitor (and other GC) metrics to Impala metrics.
Following up to IMPALA-6857, it's useful for monitoring tools to see if
the pause monitor is getting triggered, and to see other GC metrics.
The Java side here, and the Thrift side, were easy enough.
However, the Impala metric implementation here caused us to call into
the frontend to read through the JMX memory beans 72 times, because each
call to GetValue() was getting all the data for the pool. This structure
made it hard to add additional, non-pool, metrics, and it felt wasteful.
To combat this, I added a cache of 10 seconds for getting the metrics
from the Frontend. The counters will typically re-use the same data.
There are five metrics here, and to avoid yet another enum class, I used
C++ lambdas to capture which field of the Thrift object I care about. If
folks like the approach, I think it can simplify way the enums for the
pool metrics as well.
I measured the cost of calling into the metrics code by
looping the metrics-gathering 100 times and looking at CPU
time for the process using this script:
START_CPU=$(cat /proc/$(fuser 25000/tcp 2> /dev/null | tr -d ' ')/stat | awk '{ print $14 + $15 }')
for i in $(seq 100); do
curl http://localhost:25000/jsonmetrics?json > /dev/null 2> /dev/null
done
END_CPU=$( cat /proc/$(fuser 25000/tcp 2> /dev/null | tr -d ' ')/stat | awk '{ print $14 + $15 }')
echo $START_CPU $END_CPU $(($END_CPU - $START_CPU))
On a release build on my development machine, gathering metrics 100
times took 0.16 cpu seconds without this change and 0.07 cpu seconds
with this change. The measurement accuracy here is 0.01 (I spot-checked
this with using the cpuacct cgroup infrastructure which gives you nanos,
but it was more painful to script), but this convinces me that this is a
net improvement.
Change-Id: Ia707393962ad94ef715ec015b3fe3bb1769104a2
Reviewed-on: http://gerrit.cloudera.org:8080/11468
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/abd23064
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/abd23064
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/abd23064
Branch: refs/heads/master
Commit: abd230647fa92db29ac3719096eb4ebc7c151069
Parents: 009b252
Author: Philip Zeyliger <ph...@cloudera.com>
Authored: Tue Sep 18 15:52:31 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Thu Sep 27 01:55:09 2018 +0000
----------------------------------------------------------------------
be/src/util/jni-util.cc | 8 +-
be/src/util/jni-util.h | 2 +-
be/src/util/memory-metrics.cc | 155 +++++++++++++------
be/src/util/memory-metrics.h | 77 +++++++--
common/thrift/Frontend.thrift | 21 +--
common/thrift/metrics.json | 55 +++++++
.../java/org/apache/impala/common/JniUtil.java | 142 ++++++++---------
.../org/apache/impala/util/JvmPauseMonitor.java | 23 +++
tests/custom_cluster/test_pause_monitor.py | 8 +-
tests/observability/__init__.py | 0
tests/observability/test_jvm_metrics.py | 41 +++++
tests/run-tests.py | 2 +-
12 files changed, 389 insertions(+), 145 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/be/src/util/jni-util.cc
----------------------------------------------------------------------
diff --git a/be/src/util/jni-util.cc b/be/src/util/jni-util.cc
index 3e5beac..714086b 100644
--- a/be/src/util/jni-util.cc
+++ b/be/src/util/jni-util.cc
@@ -182,7 +182,7 @@ Status JniUtil::Init() {
}
get_jvm_metrics_id_ =
- env->GetStaticMethodID(jni_util_cl_, "getJvmMemoryMetrics", "([B)[B");
+ env->GetStaticMethodID(jni_util_cl_, "getJvmMemoryMetrics", "()[B");
if (get_jvm_metrics_id_ == NULL) {
if (env->ExceptionOccurred()) env->ExceptionDescribe();
return Status("Failed to find JniUtil.getJvmMemoryMetrics method.");
@@ -260,10 +260,8 @@ Status JniUtil::GetJniExceptionMsg(JNIEnv* env, bool log_stack, const string& pr
return Status(Substitute("$0$1", prefix, msg_str_guard.get()));
}
-Status JniUtil::GetJvmMemoryMetrics(const TGetJvmMemoryMetricsRequest& request,
- TGetJvmMemoryMetricsResponse* result) {
- return JniCall::static_method(jni_util_class(), get_jvm_metrics_id_)
- .with_thrift_arg(request).Call(result);
+Status JniUtil::GetJvmMemoryMetrics(TGetJvmMemoryMetricsResponse* result) {
+ return JniCall::static_method(jni_util_class(), get_jvm_metrics_id_).Call(result);
}
Status JniUtil::GetJvmThreadsInfo(const TGetJvmThreadsInfoRequest& request,
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/be/src/util/jni-util.h
----------------------------------------------------------------------
diff --git a/be/src/util/jni-util.h b/be/src/util/jni-util.h
index c4030a4..576f943 100644
--- a/be/src/util/jni-util.h
+++ b/be/src/util/jni-util.h
@@ -372,7 +372,7 @@ class JniUtil {
/// Populates 'result' with a list of memory metrics from the Jvm. Returns Status::OK
/// unless there is an exception.
- static Status GetJvmMemoryMetrics(const TGetJvmMemoryMetricsRequest& request,
+ static Status GetJvmMemoryMetrics(
TGetJvmMemoryMetricsResponse* result) WARN_UNUSED_RESULT;
/// Populates 'result' with information about live JVM threads. Returns
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/be/src/util/memory-metrics.cc
----------------------------------------------------------------------
diff --git a/be/src/util/memory-metrics.cc b/be/src/util/memory-metrics.cc
index 04b94ef..3476e9e 100644
--- a/be/src/util/memory-metrics.cc
+++ b/be/src/util/memory-metrics.cc
@@ -29,6 +29,7 @@
#include "util/time.h"
using boost::algorithm::to_lower;
+using boost::lock_guard;
using namespace impala;
using namespace strings;
@@ -160,7 +161,7 @@ void AggregateMemoryMetrics::Refresh() {
JvmMemoryMetric* JvmMemoryMetric::CreateAndRegister(
MetricGroup* metrics, const string& key, const string& pool_name,
- JvmMemoryMetric::JvmMemoryMetricType type) {
+ JvmMemoryMetricType type) {
string pool_name_for_key = pool_name;
to_lower(pool_name_for_key);
replace(pool_name_for_key.begin(), pool_name_for_key.end(), ' ', '-');
@@ -174,67 +175,133 @@ JvmMemoryMetric::JvmMemoryMetric(const TMetricDef& def, const string& mempool_na
metric_type_ = type;
}
+int64_t JvmMemoryMetric::GetValue() {
+ return JvmMetricCache::GetInstance()->GetPoolMetric(mempool_name_, metric_type_);
+}
Status JvmMemoryMetric::InitMetrics(MetricGroup* metrics) {
- DCHECK(metrics != nullptr);
- TGetJvmMemoryMetricsRequest request;
- request.get_all = true;
- TGetJvmMemoryMetricsResponse response;
- RETURN_IF_ERROR(JniUtil::GetJvmMemoryMetrics(request, &response));
- for (const TJvmMemoryPool& usage: response.memory_pools) {
+ vector<string> names = JvmMetricCache::GetInstance()->GetPoolNames();
+ for (const string& name: names) {
JvmMemoryMetric::CreateAndRegister(
- metrics, "jvm.$0.max-usage-bytes", usage.name, MAX);
+ metrics, "jvm.$0.max-usage-bytes", name, MAX);
JvmMemoryMetric::CreateAndRegister(
- metrics, "jvm.$0.current-usage-bytes", usage.name, CURRENT);
+ metrics, "jvm.$0.current-usage-bytes", name, CURRENT);
JvmMemoryMetric::CreateAndRegister(
- metrics, "jvm.$0.committed-usage-bytes", usage.name, COMMITTED);
+ metrics, "jvm.$0.committed-usage-bytes", name, COMMITTED);
JvmMemoryMetric::CreateAndRegister(
- metrics, "jvm.$0.init-usage-bytes", usage.name, INIT);
- JvmMemoryMetric::CreateAndRegister(metrics, "jvm.$0.peak-max-usage-bytes", usage.name,
+ metrics, "jvm.$0.init-usage-bytes", name, INIT);
+ JvmMemoryMetric::CreateAndRegister(metrics, "jvm.$0.peak-max-usage-bytes", name,
PEAK_MAX);
JvmMemoryMetric::CreateAndRegister(
- metrics, "jvm.$0.peak-current-usage-bytes", usage.name, PEAK_CURRENT);
+ metrics, "jvm.$0.peak-current-usage-bytes", name, PEAK_CURRENT);
JvmMemoryMetric::CreateAndRegister(
- metrics, "jvm.$0.peak-committed-usage-bytes", usage.name, PEAK_COMMITTED);
+ metrics, "jvm.$0.peak-committed-usage-bytes", name, PEAK_COMMITTED);
JvmMemoryMetric::CreateAndRegister(
- metrics, "jvm.$0.peak-init-usage-bytes", usage.name, PEAK_INIT);
+ metrics, "jvm.$0.peak-init-usage-bytes", name, PEAK_INIT);
}
-
+ JvmMemoryCounterMetric::CreateAndRegister(metrics, "jvm.gc_time_millis",
+ [](const TGetJvmMemoryMetricsResponse& r) { return r.gc_time_millis; });
+ JvmMemoryCounterMetric::CreateAndRegister(metrics, "jvm.gc_num_info_threshold_exceeded",
+ [](const TGetJvmMemoryMetricsResponse& r) {
+ return r.gc_num_info_threshold_exceeded;
+ });
+ JvmMemoryCounterMetric::CreateAndRegister(metrics, "jvm.gc_num_warn_threshold_exceeded",
+ [](const TGetJvmMemoryMetricsResponse& r) {
+ return r.gc_num_warn_threshold_exceeded;
+ });
+ JvmMemoryCounterMetric::CreateAndRegister(metrics, "jvm.gc_count",
+ [](const TGetJvmMemoryMetricsResponse& r) { return r.gc_count; });
+ JvmMemoryCounterMetric::CreateAndRegister(metrics,
+ "jvm.gc_total_extra_sleep_time_millis", [](const TGetJvmMemoryMetricsResponse& r) {
+ return r.gc_total_extra_sleep_time_millis;
+ });
return Status::OK();
}
-int64_t JvmMemoryMetric::GetValue() {
- TGetJvmMemoryMetricsRequest request;
- request.get_all = false;
- request.__set_memory_pool(mempool_name_);
- TGetJvmMemoryMetricsResponse response;
- if (!JniUtil::GetJvmMemoryMetrics(request, &response).ok()) return 0;
- if (response.memory_pools.size() != 1) return 0;
- TJvmMemoryPool& pool = response.memory_pools[0];
- DCHECK(pool.name == mempool_name_);
- switch (metric_type_) {
- case MAX:
- return pool.max;
- case INIT:
- return pool.init;
- case CURRENT:
- return pool.used;
- case COMMITTED:
- return pool.committed;
- case PEAK_MAX:
- return pool.peak_max;
- case PEAK_INIT:
- return pool.peak_init;
- case PEAK_CURRENT:
- return pool.peak_used;
- case PEAK_COMMITTED:
- return pool.peak_committed;
- default:
- DCHECK(false) << "Unknown JvmMemoryMetricType: " << metric_type_;
+JvmMemoryCounterMetric* JvmMemoryCounterMetric::CreateAndRegister(
+ MetricGroup* metrics, const string& key,
+ int64_t(*accessor)(const TGetJvmMemoryMetricsResponse&)) {
+ return metrics->RegisterMetric(
+ new JvmMemoryCounterMetric(MetricDefs::Get(key), accessor));
+}
+
+JvmMemoryCounterMetric::JvmMemoryCounterMetric(const TMetricDef& def,
+ int64_t(*accessor)(const TGetJvmMemoryMetricsResponse&))
+ : IntCounter(def, 0) {
+ accessor_ = accessor;
+}
+
+int64_t JvmMemoryCounterMetric::GetValue() {
+ return JvmMetricCache::GetInstance()->GetCounterMetric(accessor_);
+}
+
+JvmMetricCache* JvmMetricCache::GetInstance() {
+ static JvmMetricCache instance;
+ return &instance;
+}
+
+void JvmMetricCache::GrabMetricsIfNecessary() {
+ int64_t now = MonotonicMillis();
+ if (now - last_fetch_ < CACHE_PERIOD_MILLIS) return;
+ Status status = JniUtil::GetJvmMemoryMetrics(&last_response_);
+ if (!status.ok()) {
+ VLOG_QUERY << "Couldn't retrieve JVM Memory Metrics: " << status.GetDetail();
+ return;
}
+ last_fetch_ = now;
+ return;
+}
+
+int64_t JvmMetricCache::GetCounterMetric(
+ int64_t(*accessor)(const TGetJvmMemoryMetricsResponse&)) {
+ lock_guard<boost::mutex> lock_guard(lock_);
+ GrabMetricsIfNecessary();
+ return accessor(last_response_);
+}
+
+int64_t JvmMetricCache::GetPoolMetric(const std::string& mempool_name,
+ JvmMemoryMetricType type) {
+ lock_guard<boost::mutex> lock_guard(lock_);
+ GrabMetricsIfNecessary();
+
+ for (const TJvmMemoryPool& pool : last_response_.memory_pools) {
+ if (pool.name == mempool_name) {
+ switch (type) {
+ case MAX:
+ return pool.max;
+ case INIT:
+ return pool.init;
+ case CURRENT:
+ return pool.used;
+ case COMMITTED:
+ return pool.committed;
+ case PEAK_MAX:
+ return pool.peak_max;
+ case PEAK_INIT:
+ return pool.peak_init;
+ case PEAK_CURRENT:
+ return pool.peak_used;
+ case PEAK_COMMITTED:
+ return pool.peak_committed;
+ default:
+ DCHECK(false) << "Unknown JvmMemoryMetricType: " << type;
+ }
+ }
+ }
+ DCHECK(false) << "Could not find pool: " << mempool_name;
return 0;
}
+vector<string> JvmMetricCache::GetPoolNames() {
+ lock_guard<boost::mutex> lock_guard(lock_);
+ GrabMetricsIfNecessary();
+ vector<string> names;
+ for (const TJvmMemoryPool& usage: last_response_.memory_pools) {
+ names.push_back(usage.name);
+ }
+ return names;
+}
+
Status BufferPoolMetric::InitMetrics(MetricGroup* metrics,
ReservationTracker* global_reservations, BufferPool* buffer_pool) {
LIMIT = metrics->RegisterMetric(
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/be/src/util/memory-metrics.h
----------------------------------------------------------------------
diff --git a/be/src/util/memory-metrics.h b/be/src/util/memory-metrics.h
index b2a2fdc..70ec497 100644
--- a/be/src/util/memory-metrics.h
+++ b/be/src/util/memory-metrics.h
@@ -37,6 +37,18 @@ class MemTracker;
class ReservationTracker;
class Thread;
+/// Each names one of the fields in TJvmMemoryPool.
+enum JvmMemoryMetricType {
+ MAX,
+ INIT,
+ COMMITTED,
+ CURRENT,
+ PEAK_MAX,
+ PEAK_INIT,
+ PEAK_COMMITTED,
+ PEAK_CURRENT
+};
+
/// Memory metrics including TCMalloc and BufferPool memory.
class AggregateMemoryMetrics {
public:
@@ -152,6 +164,40 @@ class SanitizerMallocMetric : public IntGauge {
}
};
+// A singleton for caching the gathering of JVM Metrics for 1 second, to amortize
+// the cost of getting all the JVM metrics.
+//
+// Clients should get the singleton via GetInstance() and call the
+// Get* methods. Internally, the Get* methods are synchronized with
+// lock_.
+class JvmMetricCache {
+ public:
+ /// Retrieves a metric for a given pool.
+ long GetPoolMetric(const std::string& mempool_name, JvmMemoryMetricType type);
+ /// Retrieves a single counter metric from the response.
+ long GetCounterMetric(int64_t(*accessor)(const TGetJvmMemoryMetricsResponse&));
+ /// Returns all pool names.
+ vector<string> GetPoolNames();
+ /// Returns singleton instance.
+ static JvmMetricCache* GetInstance();
+
+ private:
+ /// Updates metrics if over CACHE_PERIOD_MILLIS has elapsed.
+ void GrabMetricsIfNecessary();
+
+ boost::mutex lock_;
+ /// Time when metrics were last fetched, using MonotonicMillis().
+ /// Protected by lock_.
+ int64_t last_fetch_ = 0;
+ /// Last available metrics.
+ /// Protected by lock_.
+ TGetJvmMemoryMetricsResponse last_response_;
+
+ static const int64_t CACHE_PERIOD_MILLIS = 1000;
+ JvmMetricCache() { }
+ DISALLOW_COPY_AND_ASSIGN(JvmMetricCache);
+};
+
/// A JvmMemoryMetric corresponds to one value drawn from one 'memory pool' in the JVM. A
/// memory pool is an area of memory assigned for one particular aspect of memory
/// management. For example Hotspot has pools for the permanent generation, the old
@@ -167,20 +213,8 @@ class JvmMemoryMetric : public IntGauge {
virtual int64_t GetValue() override;
private:
- /// Each names one of the fields in TJvmMemoryPool.
- enum JvmMemoryMetricType {
- MAX,
- INIT,
- COMMITTED,
- CURRENT,
- PEAK_MAX,
- PEAK_INIT,
- PEAK_COMMITTED,
- PEAK_CURRENT
- };
-
static JvmMemoryMetric* CreateAndRegister(MetricGroup* metrics, const std::string& key,
- const std::string& pool_name, JvmMemoryMetric::JvmMemoryMetricType type);
+ const std::string& pool_name, JvmMemoryMetricType type);
/// Private constructor to ensure only InitMetrics() can create JvmMemoryMetrics.
JvmMemoryMetric(
@@ -194,6 +228,23 @@ class JvmMemoryMetric : public IntGauge {
JvmMemoryMetricType metric_type_;
};
+// A counter that represents metrics about JVM Memory. It acesses the underlying
+// data via JniUtil::GetJvmMemoryMetrics() via JvmMetricCache.
+class JvmMemoryCounterMetric : public IntCounter {
+ virtual int64_t GetValue() override;
+
+ private:
+ friend class JvmMemoryMetric;
+ static JvmMemoryCounterMetric* CreateAndRegister(MetricGroup* metrics,
+ const string& key,
+ int64_t(*accessor)(const TGetJvmMemoryMetricsResponse&));
+ /// Private constructor; used via CreateAndRegister
+ JvmMemoryCounterMetric(const TMetricDef& def,
+ int64_t(*accessor)(const TGetJvmMemoryMetricsResponse&));
+
+ int64_t(*accessor_)(const TGetJvmMemoryMetricsResponse&);
+};
+
/// Metric that reports information about the buffer pool.
class BufferPoolMetric : public IntGauge {
public:
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/common/thrift/Frontend.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/Frontend.thrift b/common/thrift/Frontend.thrift
index 249636b..86588d3 100644
--- a/common/thrift/Frontend.thrift
+++ b/common/thrift/Frontend.thrift
@@ -792,20 +792,23 @@ struct TJvmMemoryPool {
9: required string name
}
-// Request to get one or all sets of memory pool metrics.
-struct TGetJvmMemoryMetricsRequest {
- // If set, return all pools
- 1: required bool get_all
-
- // If get_all is false, this must be set to the name of the memory pool to return.
- 2: optional string memory_pool
-}
-
// Response from JniUtil::GetJvmMemoryMetrics()
struct TGetJvmMemoryMetricsResponse {
// One entry for every pool tracked by the Jvm, plus a synthetic aggregate pool called
// 'total'
1: required list<TJvmMemoryPool> memory_pools
+
+ // Metrics from JvmPauseMonitor, measuring how much time is spend
+ // pausing, presumably because of Garbage Collection. These
+ // names are consistent with Hadoop's metric names.
+ 2: required i64 gc_num_warn_threshold_exceeded
+ 3: required i64 gc_num_info_threshold_exceeded
+ 4: required i64 gc_total_extra_sleep_time_millis
+
+ // Metrics for JVM Garbage Collection, from the management beans;
+ // these are cumulative across all types of GCs.
+ 5: required i64 gc_count
+ 6: required i64 gc_time_millis
}
// Contains information about a JVM thread
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/common/thrift/metrics.json
----------------------------------------------------------------------
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index 97f4275..f0cdcb6 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -1505,6 +1505,61 @@
"key": "jvm.$0.peak-max-usage-bytes"
},
{
+ "description": "Jvm Garbage Collection Time",
+ "contexts": [
+ "CATALOGSERVER",
+ "IMPALAD"
+ ],
+ "label": "Jvm Garbage Collection Time",
+ "units": "TIME_MS",
+ "kind": "COUNTER",
+ "key": "jvm.gc_time_millis"
+ },
+ {
+ "description": "Jvm Garbage Collection Count",
+ "contexts": [
+ "CATALOGSERVER",
+ "IMPALAD"
+ ],
+ "label": "Jvm Garbage Collection Count",
+ "units": "UNIT",
+ "kind": "COUNTER",
+ "key": "jvm.gc_count"
+ },
+ {
+ "description": "Jvm Pause Detection Warning Threshold Exceeded",
+ "contexts": [
+ "CATALOGSERVER",
+ "IMPALAD"
+ ],
+ "label": "Jvm Pause Detector Warn Threshold Exceeded",
+ "units": "UNIT",
+ "kind": "COUNTER",
+ "key": "jvm.gc_num_warn_threshold_exceeded"
+ },
+ {
+ "description": "Jvm Pause Detection Info Threshold Exceeded",
+ "contexts": [
+ "CATALOGSERVER",
+ "IMPALAD"
+ ],
+ "label": "Jvm Pause Detector Info Threshold Exceeded",
+ "units": "UNIT",
+ "kind": "COUNTER",
+ "key": "jvm.gc_num_info_threshold_exceeded"
+ },
+ {
+ "description": "Jvm Pause Detection Extra Sleep Time",
+ "contexts": [
+ "CATALOGSERVER",
+ "IMPALAD"
+ ],
+ "label": "Jvm Pause Detector Extra Sleep Time",
+ "units": "TIME_MS",
+ "kind": "COUNTER",
+ "key": "jvm.gc_total_extra_sleep_time_millis"
+ },
+ {
"description": "Distribution of query latencies",
"contexts": [
"IMPALAD"
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/fe/src/main/java/org/apache/impala/common/JniUtil.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/common/JniUtil.java b/fe/src/main/java/org/apache/impala/common/JniUtil.java
index f9368fa..37e2769 100644
--- a/fe/src/main/java/org/apache/impala/common/JniUtil.java
+++ b/fe/src/main/java/org/apache/impala/common/JniUtil.java
@@ -43,7 +43,6 @@ import org.apache.thrift.protocol.TProtocolFactory;
import com.google.common.base.Joiner;
import com.google.common.base.Strings;
-import org.apache.impala.thrift.TGetJvmMemoryMetricsRequest;
import org.apache.impala.thrift.TGetJvmMemoryMetricsResponse;
import org.apache.impala.thrift.TGetJvmThreadsInfoRequest;
import org.apache.impala.thrift.TGetJvmThreadsInfoResponse;
@@ -125,89 +124,92 @@ public class JniUtil {
/**
* Collect the JVM's memory statistics into a thrift structure for translation into
* Impala metrics by the backend. A synthetic 'total' memory pool is included with
- * aggregate statistics for all real pools.
+ * aggregate statistics for all real pools. Metrics for the JvmPauseMonitor
+ * and Garbage Collection are also included.
*/
- public static byte[] getJvmMemoryMetrics(byte[] argument) throws ImpalaException {
- TGetJvmMemoryMetricsRequest request = new TGetJvmMemoryMetricsRequest();
- JniUtil.deserializeThrift(protocolFactory_, request, argument);
-
+ public static byte[] getJvmMemoryMetrics() throws ImpalaException {
TGetJvmMemoryMetricsResponse jvmMetrics = new TGetJvmMemoryMetricsResponse();
jvmMetrics.setMemory_pools(new ArrayList<TJvmMemoryPool>());
TJvmMemoryPool totalUsage = new TJvmMemoryPool();
- boolean is_total =
- request.getMemory_pool() != null && request.getMemory_pool().equals("total");
- if (request.get_all || is_total) {
- totalUsage.setName("total");
- jvmMetrics.getMemory_pools().add(totalUsage);
- }
+ totalUsage.setName("total");
+ jvmMetrics.getMemory_pools().add(totalUsage);
+
for (MemoryPoolMXBean memBean: ManagementFactory.getMemoryPoolMXBeans()) {
- if (request.get_all || is_total ||
- memBean.getName().equals(request.getMemory_pool())) {
- TJvmMemoryPool usage = new TJvmMemoryPool();
- MemoryUsage beanUsage = memBean.getUsage();
- usage.setCommitted(beanUsage.getCommitted());
- usage.setInit(beanUsage.getInit());
- usage.setMax(beanUsage.getMax());
- usage.setUsed(beanUsage.getUsed());
- usage.setName(memBean.getName());
+ TJvmMemoryPool usage = new TJvmMemoryPool();
+ MemoryUsage beanUsage = memBean.getUsage();
+ usage.setCommitted(beanUsage.getCommitted());
+ usage.setInit(beanUsage.getInit());
+ usage.setMax(beanUsage.getMax());
+ usage.setUsed(beanUsage.getUsed());
+ usage.setName(memBean.getName());
- totalUsage.committed += beanUsage.getCommitted();
- totalUsage.init += beanUsage.getInit();
- totalUsage.max += beanUsage.getMax();
- totalUsage.used += beanUsage.getUsed();
+ totalUsage.committed += beanUsage.getCommitted();
+ totalUsage.init += beanUsage.getInit();
+ totalUsage.max += beanUsage.getMax();
+ totalUsage.used += beanUsage.getUsed();
- MemoryUsage peakUsage = memBean.getPeakUsage();
- usage.setPeak_committed(peakUsage.getCommitted());
- usage.setPeak_init(peakUsage.getInit());
- usage.setPeak_max(peakUsage.getMax());
- usage.setPeak_used(peakUsage.getUsed());
+ MemoryUsage peakUsage = memBean.getPeakUsage();
+ usage.setPeak_committed(peakUsage.getCommitted());
+ usage.setPeak_init(peakUsage.getInit());
+ usage.setPeak_max(peakUsage.getMax());
+ usage.setPeak_used(peakUsage.getUsed());
- totalUsage.peak_committed += peakUsage.getCommitted();
- totalUsage.peak_init += peakUsage.getInit();
- totalUsage.peak_max += peakUsage.getMax();
- totalUsage.peak_used += peakUsage.getUsed();
+ totalUsage.peak_committed += peakUsage.getCommitted();
+ totalUsage.peak_init += peakUsage.getInit();
+ totalUsage.peak_max += peakUsage.getMax();
+ totalUsage.peak_used += peakUsage.getUsed();
- if (!is_total) {
- jvmMetrics.getMemory_pools().add(usage);
- if (!request.get_all) break;
- }
- }
+ jvmMetrics.getMemory_pools().add(usage);
}
- if (request.get_all || request.getMemory_pool().equals("heap")) {
- // Populate heap usage
- MemoryMXBean mBean = ManagementFactory.getMemoryMXBean();
- TJvmMemoryPool heap = new TJvmMemoryPool();
- MemoryUsage heapUsage = mBean.getHeapMemoryUsage();
- heap.setCommitted(heapUsage.getCommitted());
- heap.setInit(heapUsage.getInit());
- heap.setMax(heapUsage.getMax());
- heap.setUsed(heapUsage.getUsed());
- heap.setName("heap");
- heap.setPeak_committed(0);
- heap.setPeak_init(0);
- heap.setPeak_max(0);
- heap.setPeak_used(0);
- jvmMetrics.getMemory_pools().add(heap);
- }
+ // Populate heap usage
+ MemoryMXBean mBean = ManagementFactory.getMemoryMXBean();
+ TJvmMemoryPool heap = new TJvmMemoryPool();
+ MemoryUsage heapUsage = mBean.getHeapMemoryUsage();
+ heap.setCommitted(heapUsage.getCommitted());
+ heap.setInit(heapUsage.getInit());
+ heap.setMax(heapUsage.getMax());
+ heap.setUsed(heapUsage.getUsed());
+ heap.setName("heap");
+ heap.setPeak_committed(0);
+ heap.setPeak_init(0);
+ heap.setPeak_max(0);
+ heap.setPeak_used(0);
+ jvmMetrics.getMemory_pools().add(heap);
+
+ // Populate non-heap usage
+ TJvmMemoryPool nonHeap = new TJvmMemoryPool();
+ MemoryUsage nonHeapUsage = mBean.getNonHeapMemoryUsage();
+ nonHeap.setCommitted(nonHeapUsage.getCommitted());
+ nonHeap.setInit(nonHeapUsage.getInit());
+ nonHeap.setMax(nonHeapUsage.getMax());
+ nonHeap.setUsed(nonHeapUsage.getUsed());
+ nonHeap.setName("non-heap");
+ nonHeap.setPeak_committed(0);
+ nonHeap.setPeak_init(0);
+ nonHeap.setPeak_max(0);
+ nonHeap.setPeak_used(0);
+ jvmMetrics.getMemory_pools().add(nonHeap);
- if (request.get_all || request.getMemory_pool().equals("non-heap")) {
- // Populate non-heap usage
- MemoryMXBean mBean = ManagementFactory.getMemoryMXBean();
- TJvmMemoryPool nonHeap = new TJvmMemoryPool();
- MemoryUsage nonHeapUsage = mBean.getNonHeapMemoryUsage();
- nonHeap.setCommitted(nonHeapUsage.getCommitted());
- nonHeap.setInit(nonHeapUsage.getInit());
- nonHeap.setMax(nonHeapUsage.getMax());
- nonHeap.setUsed(nonHeapUsage.getUsed());
- nonHeap.setName("non-heap");
- nonHeap.setPeak_committed(0);
- nonHeap.setPeak_init(0);
- nonHeap.setPeak_max(0);
- nonHeap.setPeak_used(0);
- jvmMetrics.getMemory_pools().add(nonHeap);
+ // Populate JvmPauseMonitor metrics
+ jvmMetrics.setGc_num_warn_threshold_exceeded(
+ JvmPauseMonitor.INSTANCE.getNumGcWarnThresholdExceeded());
+ jvmMetrics.setGc_num_info_threshold_exceeded(
+ JvmPauseMonitor.INSTANCE.getNumGcInfoThresholdExceeded());
+ jvmMetrics.setGc_total_extra_sleep_time_millis(
+ JvmPauseMonitor.INSTANCE.getTotalGcExtraSleepTime());
+
+ // And Garbage Collector metrics
+ long gcCount = 0;
+ long gcTimeMillis = 0;
+ for (GarbageCollectorMXBean bean : ManagementFactory.getGarbageCollectorMXBeans()) {
+ gcCount += bean.getCollectionCount();
+ gcTimeMillis += bean.getCollectionTime();
}
+ jvmMetrics.setGc_count(gcCount);
+ jvmMetrics.setGc_time_millis(gcTimeMillis);
+
return serializeToThrift(jvmMetrics, protocolFactory_);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/fe/src/main/java/org/apache/impala/util/JvmPauseMonitor.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/util/JvmPauseMonitor.java b/fe/src/main/java/org/apache/impala/util/JvmPauseMonitor.java
index 8dbca6c..2c09b45 100644
--- a/fe/src/main/java/org/apache/impala/util/JvmPauseMonitor.java
+++ b/fe/src/main/java/org/apache/impala/util/JvmPauseMonitor.java
@@ -59,6 +59,14 @@ public class JvmPauseMonitor {
private long infoThresholdMs_;
private static final long INFO_THRESHOLD_MS = 1000;
+ // Overall metrics
+ // Volatile to allow populating metrics concurrently with the values
+ // being updated without staleness (but with no other synchronization
+ // guarantees).
+ private volatile long numGcWarnThresholdExceeded = 0;
+ private volatile long numGcInfoThresholdExceeded = 0;
+ private volatile long totalGcExtraSleepTime = 0;
+
// Daemon thread running the pause monitor loop.
private Thread monitorThread_;
private volatile boolean shouldRun = true;
@@ -92,6 +100,18 @@ public class JvmPauseMonitor {
return monitorThread_ != null;
}
+ public long getNumGcWarnThresholdExceeded() {
+ return numGcWarnThresholdExceeded;
+ }
+
+ public long getNumGcInfoThresholdExceeded() {
+ return numGcInfoThresholdExceeded;
+ }
+
+ public long getTotalGcExtraSleepTime() {
+ return totalGcExtraSleepTime;
+ }
+
/**
* Helper method that formats the message to be logged, along with
* the GC metrics.
@@ -181,12 +201,15 @@ public class JvmPauseMonitor {
Map<String, GcTimes> gcTimesAfterSleep = getGcTimes();
if (extraSleepTime > warnThresholdMs_) {
+ ++numGcWarnThresholdExceeded;
LOG.warn(formatMessage(
extraSleepTime, gcTimesAfterSleep, gcTimesBeforeSleep));
} else if (extraSleepTime > infoThresholdMs_) {
+ ++numGcInfoThresholdExceeded;
LOG.info(formatMessage(
extraSleepTime, gcTimesAfterSleep, gcTimesBeforeSleep));
}
+ totalGcExtraSleepTime += extraSleepTime;
gcTimesBeforeSleep = gcTimesAfterSleep;
if (deadlockCheckIntervalS_ > 0 &&
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/tests/custom_cluster/test_pause_monitor.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_pause_monitor.py b/tests/custom_cluster/test_pause_monitor.py
index f4e616b..166316f 100644
--- a/tests/custom_cluster/test_pause_monitor.py
+++ b/tests/custom_cluster/test_pause_monitor.py
@@ -24,6 +24,7 @@ from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
class TestPauseMonitor(CustomClusterTestSuite):
"""Class for pause monitor tests."""
+ @CustomClusterTestSuite.with_args("--logbuflevel=-1")
def test_jvm_pause_monitor_logs_entries(self):
"""This test injects a non-GC pause and confirms that that the JVM pause
monitor detects and logs it."""
@@ -32,7 +33,10 @@ class TestPauseMonitor(CustomClusterTestSuite):
impalad.kill(signal.SIGSTOP)
time.sleep(5)
impalad.kill(signal.SIGCONT)
- # Wait for a few seconds for the logs to get flushed.
- time.sleep(5)
+ # Wait for over a second for the cache metrics to expire.
+ time.sleep(1.2)
# Check that the pause is detected.
self.assert_impalad_log_contains('INFO', "Detected pause in JVM or host machine")
+ # Check that the metrics we have for this updated as well
+ assert impalad.service.get_metric_value("jvm.gc_num_info_threshold_exceeded") > 0
+ assert impalad.service.get_metric_value("jvm.gc_total_extra_sleep_time_millis") > 0
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/tests/observability/__init__.py
----------------------------------------------------------------------
diff --git a/tests/observability/__init__.py b/tests/observability/__init__.py
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/tests/observability/test_jvm_metrics.py
----------------------------------------------------------------------
diff --git a/tests/observability/test_jvm_metrics.py b/tests/observability/test_jvm_metrics.py
new file mode 100644
index 0000000..80deb81
--- /dev/null
+++ b/tests/observability/test_jvm_metrics.py
@@ -0,0 +1,41 @@
+# 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.
+
+from tests.common.impala_test_suite import ImpalaTestSuite
+
+
+class TestJvmMetrics(ImpalaTestSuite):
+ """Class for checking JVM Metrics."""
+
+ def test_jvm_metrics(self):
+ """The following metrics should exist and come from the JVM."""
+ impalad = self.impalad_test_service
+ # These metrics can't possibly be zero, so we assert that
+ # they're positive.
+ NON_ZERO_METRICS = [
+ "jvm.heap.current-usage-bytes",
+ "jvm.heap.max-usage-bytes",
+ "jvm.non-heap.committed-usage-bytes",
+ "jvm.total.current-usage-bytes",
+ ]
+ for m in NON_ZERO_METRICS:
+ assert impalad.get_metric_value(m) > 0
+
+ # One can imagine somehow there being no GCs, so do
+ # a weaker assertion that they're non-negative and non-None.
+ for m in ["jvm.gc_count", "jvm.gc_time_millis"]:
+ assert impalad.get_metric_value(m) >= 0
http://git-wip-us.apache.org/repos/asf/impala/blob/abd23064/tests/run-tests.py
----------------------------------------------------------------------
diff --git a/tests/run-tests.py b/tests/run-tests.py
index 78c152b..c2fd420 100755
--- a/tests/run-tests.py
+++ b/tests/run-tests.py
@@ -37,7 +37,7 @@ from _pytest.config import FILE_OR_DIR
# We whitelist valid test directories. If a new test directory is added, update this.
VALID_TEST_DIRS = ['failure', 'query_test', 'stress', 'unittests', 'aux_query_tests',
'shell', 'hs2', 'catalog_service', 'metadata', 'data_errors',
- 'statestore', 'infra']
+ 'statestore', 'infra', 'observability']
TEST_DIR = os.path.join(os.environ['IMPALA_HOME'], 'tests')
RESULT_DIR = os.path.join(os.environ['IMPALA_EE_TEST_LOGS_DIR'], 'results')