You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by to...@apache.org on 2018/08/09 21:48:07 UTC
[1/9] impala git commit: IMPALA-7410: fix centos6 dependency
Repository: impala
Updated Branches:
refs/heads/master 3c9fef2ae -> 7f9a74ffc
IMPALA-7410: fix centos6 dependency
Update CDH_BUILD_NUM to address incorrect dependency for centos6.
Change-Id: I0fd2ddac11115d818a0118c74947532b773e0733
Reviewed-on: http://gerrit.cloudera.org:8080/11168
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/ab25d34b
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/ab25d34b
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/ab25d34b
Branch: refs/heads/master
Commit: ab25d34b1c10c0a9673e628ea6246852d535dae2
Parents: 3c9fef2
Author: Vuk Ercegovac <ve...@cloudera.com>
Authored: Wed Aug 8 11:56:23 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Thu Aug 9 00:34:44 2018 +0000
----------------------------------------------------------------------
bin/impala-config.sh | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/ab25d34b/bin/impala-config.sh
----------------------------------------------------------------------
diff --git a/bin/impala-config.sh b/bin/impala-config.sh
index 2e4782e..0940df6 100755
--- a/bin/impala-config.sh
+++ b/bin/impala-config.sh
@@ -160,7 +160,7 @@ unset IMPALA_KUDU_URL
: ${CDH_DOWNLOAD_HOST:=native-toolchain.s3.amazonaws.com}
export CDH_DOWNLOAD_HOST
export CDH_MAJOR_VERSION=6
-export CDH_BUILD_NUMBER=506967
+export CDH_BUILD_NUMBER=517354
export IMPALA_HADOOP_VERSION=3.0.0-cdh6.x-SNAPSHOT
export IMPALA_HBASE_VERSION=2.0.0-cdh6.x-SNAPSHOT
export IMPALA_HIVE_VERSION=2.1.1-cdh6.x-SNAPSHOT
[2/9] impala git commit: Exclude files with long mangled symbols from
whitespace checks
Posted by to...@apache.org.
Exclude files with long mangled symbols from whitespace checks
Change-Id: Ia99cfc9840cc0d80d2654089f7e776344831a6b2
Reviewed-on: http://gerrit.cloudera.org:8080/11169
Reviewed-by: Philip Zeyliger <ph...@cloudera.com>
Tested-by: Tim Armstrong <ta...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/b4d20ab7
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/b4d20ab7
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/b4d20ab7
Branch: refs/heads/master
Commit: b4d20ab78071059a981d7be06726a4042c1227de
Parents: ab25d34
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Wed Aug 8 12:11:34 2018 -0700
Committer: Tim Armstrong <ta...@cloudera.com>
Committed: Thu Aug 9 00:36:19 2018 +0000
----------------------------------------------------------------------
bin/jenkins/critique-gerrit-review.py | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/b4d20ab7/bin/jenkins/critique-gerrit-review.py
----------------------------------------------------------------------
diff --git a/bin/jenkins/critique-gerrit-review.py b/bin/jenkins/critique-gerrit-review.py
index e3873ab..548eac8 100755
--- a/bin/jenkins/critique-gerrit-review.py
+++ b/bin/jenkins/critique-gerrit-review.py
@@ -66,6 +66,8 @@ EXCLUDE_FILE_PATTERNS = [
re.compile(r".*be/src/kudu.*"), # Kudu source code may have different rules.
re.compile(r".*-benchmark.cc"), # Benchmark files tend to have long lines.
re.compile(r".*/function-registry/impala_functions.py"), # Many long strings.
+ re.compile(r".*/catalog/BuiltinsDb.java"), # Many long strings.
+ re.compile(r".*/codegen/gen_ir_descriptions.py"), # Many long strings.
re.compile(r".*/shell/ext-py/*") # Third-party code.
]
[6/9] impala git commit: IMPALA-7383: Configurable HMS and Sentry
policy DB
Posted by to...@apache.org.
IMPALA-7383: Configurable HMS and Sentry policy DB
Some developers keep multiple impala repos on their disk. Isolating
METASTORE_DB and SENTRY_POLICY_DB may help with switching between those
repos without reloading the data. This patch makes those DB names
configurable and default to an escaped IMPALA_HOME path.
Change-Id: I190d657cb95dfdf73ebd05e5dd24ef2a8e3156b8
Reviewed-on: http://gerrit.cloudera.org:8080/11104
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/2868bf56
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/2868bf56
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/2868bf56
Branch: refs/heads/master
Commit: 2868bf569afaf950b79a79a05893f314a9c51387
Parents: 4976ff3
Author: Tianyi Wang <ti...@apache.org>
Authored: Wed Aug 1 15:26:17 2018 -0700
Committer: Tianyi Wang <tw...@cloudera.com>
Committed: Thu Aug 9 18:07:40 2018 +0000
----------------------------------------------------------------------
bin/create-test-configuration.sh | 4 ++--
bin/impala-config.sh | 4 +++-
fe/src/test/resources/sentry-site.xml.template | 2 +-
3 files changed, 6 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/2868bf56/bin/create-test-configuration.sh
----------------------------------------------------------------------
diff --git a/bin/create-test-configuration.sh b/bin/create-test-configuration.sh
index 812154d..e68af9c 100755
--- a/bin/create-test-configuration.sh
+++ b/bin/create-test-configuration.sh
@@ -105,8 +105,8 @@ fi
if [ $CREATE_SENTRY_POLICY_DB -eq 1 ]; then
echo "Creating Sentry Policy Server DB"
- dropdb -U hiveuser sentry_policy 2> /dev/null || true
- createdb -U hiveuser sentry_policy
+ dropdb -U hiveuser $SENTRY_POLICY_DB 2> /dev/null || true
+ createdb -U hiveuser $SENTRY_POLICY_DB
fi
# Perform search-replace on $1, output to $2.
http://git-wip-us.apache.org/repos/asf/impala/blob/2868bf56/bin/impala-config.sh
----------------------------------------------------------------------
diff --git a/bin/impala-config.sh b/bin/impala-config.sh
index 0940df6..0fc3516 100755
--- a/bin/impala-config.sh
+++ b/bin/impala-config.sh
@@ -319,7 +319,9 @@ export ISILON_NAMENODE="${ISILON_NAMENODE-}"
export DEFAULT_FS="${DEFAULT_FS-hdfs://localhost:20500}"
export WAREHOUSE_LOCATION_PREFIX="${WAREHOUSE_LOCATION_PREFIX-}"
export LOCAL_FS="file:${WAREHOUSE_LOCATION_PREFIX}"
-export METASTORE_DB="hive_impala"
+ESCAPED_IMPALA_HOME=$(sed "s/[^0-9a-zA-Z]/_/g" <<< "$IMPALA_HOME")
+export METASTORE_DB=${METASTORE_DB-$(cut -c-63 <<< HMS$ESCAPED_IMPALA_HOME)}
+export SENTRY_POLICY_DB=${SENTRY_POLICY_DB-$(cut -c-63 <<< SP$ESCAPED_IMPALA_HOME)}
# Environment variables carrying AWS security credentials are prepared
# according to the following rules:
http://git-wip-us.apache.org/repos/asf/impala/blob/2868bf56/fe/src/test/resources/sentry-site.xml.template
----------------------------------------------------------------------
diff --git a/fe/src/test/resources/sentry-site.xml.template b/fe/src/test/resources/sentry-site.xml.template
index b569c5f..9d721ef 100644
--- a/fe/src/test/resources/sentry-site.xml.template
+++ b/fe/src/test/resources/sentry-site.xml.template
@@ -51,7 +51,7 @@
</property>
<property>
<name>sentry.store.jdbc.url</name>
- <value>jdbc:postgresql://localhost:5432/sentry_policy/;create=true</value>
+ <value>jdbc:postgresql://localhost:5432/${SENTRY_POLICY_DB}/;create=true</value>
</property>
<property>
<name>sentry.store.jdbc.user</name>
[5/9] impala git commit: IMPALA-6857: Add Jvm pause/GC Monitor
utility and expose JMX metrics
Posted by to...@apache.org.
IMPALA-6857: Add Jvm pause/GC Monitor utility and expose JMX metrics
Pause monitor:
=============
This commit adds a stripped down version of Hadoop's JvmPauseMonitor
class (https://bit.ly/2O6qSwm) . The core implementaion is borrowed
from hadoop-common project and the hadoop dependencies are removed.
- Removed dependency on AbstractService.
- Not relying on Hadoop's Configuration object for reading confs.
- Switched to Guava's implementation of Stopwatch.
This utility class can detect both GC/non-GC pauses. In case of GC
pauses, the GC metrics during the pause period are logged.
Sample Output:
=============
Detected pause in JVM or host machine (eg GC): pause of approximately
2356ms
GC pool 'PS MarkSweep' had collection(s): count=1 time=2241ms
GC pool 'PS Scavenge' had collection(s): count=3 time=352ms
Detected pause in JVM or host machine (eg GC): pause of approximately
1964ms
GC pool 'PS MarkSweep' had collection(s): count=1 time=2082ms
GC pool 'PS Scavenge' had collection(s): count=1 time=251ms
Detected pause in JVM or host machine (eg GC): pause of approximately
2120ms
GC pool 'PS MarkSweep' had collection(s): count=1 time=2454ms
Detected pause in JVM or host machine (eg GC): pause of approximately
2238ms
GC pool 'PS MarkSweep' had collection(s): count=5 time=13464ms
Detected pause in JVM or host machine (eg GC): pause of approximately
2233ms
GC pool 'PS MarkSweep' had collection(s): count=1 time=2733ms
JMX Metrics:
============
JMX metrics are now emmitted for Impala and Catalog JVMs at the web end
point /jmx.
- Impalad: http(s)://<impalad-host>:25000/jmx
- Catalogd: http(s)://<catalogd-host>:25020/jmx
Misc:
====
Renamed JvmMetric -> JvmMemoryMetric to make the intent more clear. It
doesn't relate to the functionality of the patch in anyway.
Testing:
=======
- Tested it manually with kill -SIGSTOP/-SIGCONT <pid>. Made sure that
the non-GC JVM pauses are logged.
- This class' functionality is tested manually by invoking it's main()
- Injected a memory leak into the Catalog server code and made sure the
GC is detected.
Change-Id: I30d897b7e063846ad6d8f88243e2c04264da0341
Reviewed-on: http://gerrit.cloudera.org:8080/10998
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/4976ff3c
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/4976ff3c
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/4976ff3c
Branch: refs/heads/master
Commit: 4976ff3c07f465915ac31312ca67519a600212e6
Parents: 5cb956e
Author: Bharath Vissapragada <bh...@cloudera.com>
Authored: Thu Jul 19 16:01:02 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Thu Aug 9 04:33:59 2018 +0000
----------------------------------------------------------------------
be/src/common/init.cc | 1 +
be/src/service/impala-http-handler.cc | 4 +-
be/src/util/default-path-handlers.cc | 40 ++-
be/src/util/jni-util.cc | 32 ++-
be/src/util/jni-util.h | 20 +-
be/src/util/memory-metrics.cc | 62 ++--
be/src/util/memory-metrics.h | 27 +-
be/src/util/metrics-test.cc | 4 +-
be/src/util/webserver-test.cc | 4 +-
be/src/util/webserver.cc | 13 +-
be/src/util/webserver.h | 10 +-
common/thrift/Frontend.thrift | 11 +-
.../java/org/apache/impala/common/JniUtil.java | 57 ++--
.../org/apache/impala/util/JMXJsonUtil.java | 281 +++++++++++++++++++
.../org/apache/impala/util/JvmPauseMonitor.java | 205 ++++++++++++++
.../org/apache/impala/util/JMXJsonUtilTest.java | 56 ++++
.../org/apache/impala/util/JniUtilTest.java | 54 ++++
tests/custom_cluster/test_pause_monitor.py | 38 +++
tests/webserver/test_web_pages.py | 21 +-
19 files changed, 850 insertions(+), 90 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/common/init.cc
----------------------------------------------------------------------
diff --git a/be/src/common/init.cc b/be/src/common/init.cc
index 2842c39..bc1065b 100644
--- a/be/src/common/init.cc
+++ b/be/src/common/init.cc
@@ -261,6 +261,7 @@ void impala::InitCommonRuntime(int argc, char** argv, bool init_jvm,
if (init_jvm) {
ABORT_IF_ERROR(JniUtil::Init());
InitJvmLoggingSupport();
+ ABORT_IF_ERROR(JniUtil::InitJvmPauseMonitor());
ZipUtil::InitJvm();
}
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/service/impala-http-handler.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-http-handler.cc b/be/src/service/impala-http-handler.cc
index f7ca5d6..ea41084 100644
--- a/be/src/service/impala-http-handler.cc
+++ b/be/src/service/impala-http-handler.cc
@@ -240,7 +240,7 @@ void ImpalaHttpHandler::QueryProfileEncodedHandler(const Webserver::ArgumentMap&
ss.str(Substitute("Could not obtain runtime profile: $0", status.GetDetail()));
}
}
- document->AddMember(Webserver::ENABLE_RAW_JSON_KEY, true, document->GetAllocator());
+ document->AddMember(Webserver::ENABLE_RAW_HTML_KEY, true, document->GetAllocator());
Value profile(ss.str().c_str(), document->GetAllocator());
document->AddMember("contents", profile, document->GetAllocator());
}
@@ -252,7 +252,7 @@ void ImpalaHttpHandler::InflightQueryIdsHandler(const Webserver::ArgumentMap& ar
[&](const std::shared_ptr<ClientRequestState>& request_state) {
ss << PrintId(request_state->query_id()) << "\n";
});
- document->AddMember(Webserver::ENABLE_RAW_JSON_KEY, true, document->GetAllocator());
+ document->AddMember(Webserver::ENABLE_RAW_HTML_KEY, true, document->GetAllocator());
Value query_ids(ss.str().c_str(), document->GetAllocator());
document->AddMember("contents", query_ids, document->GetAllocator());
}
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/util/default-path-handlers.cc
----------------------------------------------------------------------
diff --git a/be/src/util/default-path-handlers.cc b/be/src/util/default-path-handlers.cc
index 10966b4..9a50a79 100644
--- a/be/src/util/default-path-handlers.cc
+++ b/be/src/util/default-path-handlers.cc
@@ -26,6 +26,7 @@
#include <gutil/strings/substitute.h>
#include "common/logging.h"
+#include "rpc/jni-thrift-util.h"
#include "runtime/mem-tracker.h"
#include "runtime/exec-env.h"
#include "service/impala-server.h"
@@ -37,6 +38,7 @@
#include "util/cpu-info.h"
#include "util/disk-info.h"
#include "util/process-state-info.h"
+#include "util/jni-util.h"
#include "common/names.h"
@@ -192,6 +194,34 @@ void MemUsageHandler(MemTracker* mem_tracker, MetricGroup* metric_group,
}
}
+void JmxHandler(const Webserver::ArgumentMap& args, Document* document) {
+ document->AddMember(Webserver::ENABLE_PLAIN_JSON_KEY, true, document->GetAllocator());
+ TGetJMXJsonResponse result;
+ Status status = JniUtil::GetJMXJson(&result);
+ if (!status.ok()) {
+ Value error(status.GetDetail().c_str(), document->GetAllocator());
+ document->AddMember("error", error, document->GetAllocator());
+ VLOG(1) << "Error fetching JMX metrics: " << status.GetDetail();
+ return;
+ }
+ // Parse the JSON string returned from fe. We do an additional round of
+ // parsing to populate the JSON structure in the 'document' for our template
+ // rendering to work correctly. Otherwise the whole JSON content is considered
+ // as a single string mapped to another key.
+ Document doc(&document->GetAllocator());
+ doc.Parse<kParseDefaultFlags>(result.jmx_json.c_str());
+ if (doc.HasParseError()) {
+ Value error(doc.GetParseError(), document->GetAllocator());
+ document->AddMember("error", error, document->GetAllocator());
+ VLOG(1) << "Error fetching JMX metrics: " << doc.GetParseError();
+ return;
+ }
+ // Populate the members in the document.
+ for (Value::MemberIterator it = doc.MemberBegin(); it != doc.MemberEnd(); ++it) {
+ document->AddMember(it->name.GetString(), it->value, document->GetAllocator());
+ }
+}
+
namespace impala {
void RootHandler(const Webserver::ArgumentMap& args, Document* document) {
@@ -226,10 +256,16 @@ void RootHandler(const Webserver::ArgumentMap& args, Document* document) {
document->GetAllocator());
}
-void AddDefaultUrlCallbacks(
- Webserver* webserver, MemTracker* process_mem_tracker, MetricGroup* metric_group) {
+void AddDefaultUrlCallbacks(Webserver* webserver, MemTracker* process_mem_tracker,
+ MetricGroup* metric_group) {
webserver->RegisterUrlCallback("/logs", "logs.tmpl", LogsHandler);
webserver->RegisterUrlCallback("/varz", "flags.tmpl", FlagsHandler);
+ if (JniUtil::is_jvm_inited()) {
+ // JmxHandler outputs a plain JSON string and does not require a template to
+ // render. However RawUrlCallback only supports PLAIN content type.
+ // (TODO): Switch to RawUrlCallback when it supports JSON content-type.
+ webserver->RegisterUrlCallback("/jmx", "raw_text.tmpl", JmxHandler);
+ }
if (process_mem_tracker != NULL) {
auto callback = [process_mem_tracker, metric_group]
(const Webserver::ArgumentMap& args, Document* doc) {
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/util/jni-util.cc
----------------------------------------------------------------------
diff --git a/be/src/util/jni-util.cc b/be/src/util/jni-util.cc
index db7893b..f47e25d 100644
--- a/be/src/util/jni-util.cc
+++ b/be/src/util/jni-util.cc
@@ -64,10 +64,12 @@ bool JniScopedArrayCritical::Create(JNIEnv* env, jbyteArray jarr,
return true;
}
+bool JniUtil::jvm_inited_ = false;
jclass JniUtil::jni_util_cl_ = NULL;
jclass JniUtil::internal_exc_cl_ = NULL;
jmethodID JniUtil::get_jvm_metrics_id_ = NULL;
jmethodID JniUtil::get_jvm_threads_id_ = NULL;
+jmethodID JniUtil::get_jmx_json_ = NULL;
jmethodID JniUtil::throwable_to_string_id_ = NULL;
jmethodID JniUtil::throwable_to_stack_trace_id_ = NULL;
@@ -176,10 +178,10 @@ Status JniUtil::Init() {
}
get_jvm_metrics_id_ =
- env->GetStaticMethodID(jni_util_cl_, "getJvmMetrics", "([B)[B");
+ env->GetStaticMethodID(jni_util_cl_, "getJvmMemoryMetrics", "([B)[B");
if (get_jvm_metrics_id_ == NULL) {
if (env->ExceptionOccurred()) env->ExceptionDescribe();
- return Status("Failed to find JniUtil.getJvmMetrics method.");
+ return Status("Failed to find JniUtil.getJvmMemoryMetrics method.");
}
get_jvm_threads_id_ =
@@ -189,6 +191,13 @@ Status JniUtil::Init() {
return Status("Failed to find JniUtil.getJvmThreadsInfo method.");
}
+ get_jmx_json_ =
+ env->GetStaticMethodID(jni_util_cl_, "getJMXJson", "()[B");
+ if (get_jmx_json_ == NULL) {
+ if (env->ExceptionOccurred()) env->ExceptionDescribe();
+ return Status("Failed to find JniUtil.getJMXJson method.");
+ }
+ jvm_inited_ = true;
return Status::OK();
}
@@ -199,6 +208,17 @@ void JniUtil::InitLibhdfs() {
hdfsDisconnect(fs);
}
+Status JniUtil::InitJvmPauseMonitor() {
+ JNIEnv* env = getJNIEnv();
+ if (!env) return Status("Failed to get/create JVM.");
+ if (!jni_util_cl_) return Status("JniUtil::Init() not called.");
+ jmethodID init_jvm_pm_method;
+ JniMethodDescriptor init_jvm_pm_desc = {"initPauseMonitor", "()V", &init_jvm_pm_method};
+ RETURN_IF_ERROR(JniUtil::LoadStaticJniMethod(env, jni_util_cl_, &init_jvm_pm_desc));
+ RETURN_IF_ERROR(JniUtil::CallJniMethod(jni_util_cl_, init_jvm_pm_method));
+ return Status::OK();
+}
+
Status JniUtil::GetJniExceptionMsg(JNIEnv* env, bool log_stack, const string& prefix) {
jthrowable exc = env->ExceptionOccurred();
if (exc == nullptr) return Status::OK();
@@ -234,8 +254,8 @@ Status JniUtil::GetJniExceptionMsg(JNIEnv* env, bool log_stack, const string& pr
return Status(Substitute("$0$1", prefix, msg_str_guard.get()));
}
-Status JniUtil::GetJvmMetrics(const TGetJvmMetricsRequest& request,
- TGetJvmMetricsResponse* result) {
+Status JniUtil::GetJvmMemoryMetrics(const TGetJvmMemoryMetricsRequest& request,
+ TGetJvmMemoryMetricsResponse* result) {
return JniUtil::CallJniMethod(jni_util_class(), get_jvm_metrics_id_, request, result);
}
@@ -244,6 +264,10 @@ Status JniUtil::GetJvmThreadsInfo(const TGetJvmThreadsInfoRequest& request,
return JniUtil::CallJniMethod(jni_util_class(), get_jvm_threads_id_, request, result);
}
+Status JniUtil::GetJMXJson(TGetJMXJsonResponse* result) {
+ return JniUtil::CallJniMethod(jni_util_class(), get_jmx_json_, result);
+}
+
Status JniUtil::LoadJniMethod(JNIEnv* env, const jclass& jni_class,
JniMethodDescriptor* descriptor) {
(*descriptor->method_id) = env->GetMethodID(jni_class,
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/util/jni-util.h
----------------------------------------------------------------------
diff --git a/be/src/util/jni-util.h b/be/src/util/jni-util.h
index f0afb66..0176a13 100644
--- a/be/src/util/jni-util.h
+++ b/be/src/util/jni-util.h
@@ -219,6 +219,9 @@ class JniUtil {
/// Find JniUtil class, and get JniUtil.throwableToString method id
static Status Init() WARN_UNUSED_RESULT;
+ /// Initializes the JvmPauseMonitor.
+ static Status InitJvmPauseMonitor() WARN_UNUSED_RESULT;
+
/// Returns true if the given class could be found on the CLASSPATH in env.
/// Returns false otherwise, or if any other error occurred (e.g. a JNI exception).
/// This function does not log any errors or exceptions.
@@ -264,6 +267,9 @@ class JniUtil {
static jmethodID throwable_to_string_id() { return throwable_to_string_id_; }
static jmethodID throwable_to_stack_trace_id() { return throwable_to_stack_trace_id_; }
+ /// Returns true if an embedded JVM is initialized, false otherwise.
+ static bool is_jvm_inited() { return jvm_inited_; }
+
/// Global reference to java JniUtil class
static jclass jni_util_class() { return jni_util_cl_; }
@@ -278,14 +284,17 @@ class JniUtil {
/// Populates 'result' with a list of memory metrics from the Jvm. Returns Status::OK
/// unless there is an exception.
- static Status GetJvmMetrics(const TGetJvmMetricsRequest& request,
- TGetJvmMetricsResponse* result) WARN_UNUSED_RESULT;
+ static Status GetJvmMemoryMetrics(const TGetJvmMemoryMetricsRequest& request,
+ TGetJvmMemoryMetricsResponse* result) WARN_UNUSED_RESULT;
- // Populates 'result' with information about live JVM threads. Returns
- // Status::OK unless there is an exception.
+ /// Populates 'result' with information about live JVM threads. Returns
+ /// Status::OK unless there is an exception.
static Status GetJvmThreadsInfo(const TGetJvmThreadsInfoRequest& request,
TGetJvmThreadsInfoResponse* result) WARN_UNUSED_RESULT;
+ /// Gets JMX metrics of the JVM encoded as a JSON string.
+ static Status GetJMXJson(TGetJMXJsonResponse* result) WARN_UNUSED_RESULT;
+
/// Loads a method whose signature is in the supplied descriptor. Returns Status::OK
/// and sets descriptor->method_id to a JNI method handle if successful, otherwise an
/// error status is returned.
@@ -367,12 +376,15 @@ class JniUtil {
}
private:
+ // Set in Init() once the JVM is initialized.
+ static bool jvm_inited_;
static jclass jni_util_cl_;
static jclass internal_exc_cl_;
static jmethodID throwable_to_string_id_;
static jmethodID throwable_to_stack_trace_id_;
static jmethodID get_jvm_metrics_id_;
static jmethodID get_jvm_threads_id_;
+ static jmethodID get_jmx_json_;
};
}
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/util/memory-metrics.cc
----------------------------------------------------------------------
diff --git a/be/src/util/memory-metrics.cc b/be/src/util/memory-metrics.cc
index 390bd7c..04b94ef 100644
--- a/be/src/util/memory-metrics.cc
+++ b/be/src/util/memory-metrics.cc
@@ -118,7 +118,7 @@ Status impala::RegisterMemoryMetrics(MetricGroup* metrics, bool register_jvm_met
AggregateMemoryMetrics::TOTAL_USED = aggregate_metrics->RegisterMetric(
new SumGauge(MetricDefs::Get("memory.total-used"), used_metrics));
if (register_jvm_metrics) {
- RETURN_IF_ERROR(JvmMetric::InitMetrics(metrics->GetOrCreateChildGroup("jvm")));
+ RETURN_IF_ERROR(JvmMemoryMetric::InitMetrics(metrics->GetOrCreateChildGroup("jvm")));
}
if (FLAGS_enable_extended_memory_metrics && MemInfo::HaveSmaps()) {
@@ -158,53 +158,57 @@ void AggregateMemoryMetrics::Refresh() {
THP_KHUGEPAGED_DEFRAG->SetValue(thp_config.khugepaged_defrag);
}
-JvmMetric* JvmMetric::CreateAndRegister(MetricGroup* metrics, const string& key,
- const string& pool_name, JvmMetric::JvmMetricType type) {
+JvmMemoryMetric* JvmMemoryMetric::CreateAndRegister(
+ MetricGroup* metrics, const string& key, const string& pool_name,
+ JvmMemoryMetric::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(), ' ', '-');
- return metrics->RegisterMetric(new JvmMetric(MetricDefs::Get(key, pool_name_for_key),
- pool_name, type));
+ return metrics->RegisterMetric(
+ new JvmMemoryMetric(MetricDefs::Get(key, pool_name_for_key), pool_name, type));
}
-JvmMetric::JvmMetric(const TMetricDef& def, const string& mempool_name,
- JvmMetricType type) : IntGauge(def, 0) {
+JvmMemoryMetric::JvmMemoryMetric(const TMetricDef& def, const string& mempool_name,
+ JvmMemoryMetricType type) : IntGauge(def, 0) {
mempool_name_ = mempool_name;
metric_type_ = type;
}
-Status JvmMetric::InitMetrics(MetricGroup* metrics) {
+
+Status JvmMemoryMetric::InitMetrics(MetricGroup* metrics) {
DCHECK(metrics != nullptr);
- TGetJvmMetricsRequest request;
+ TGetJvmMemoryMetricsRequest request;
request.get_all = true;
- TGetJvmMetricsResponse response;
- RETURN_IF_ERROR(JniUtil::GetJvmMetrics(request, &response));
+ TGetJvmMemoryMetricsResponse response;
+ RETURN_IF_ERROR(JniUtil::GetJvmMemoryMetrics(request, &response));
for (const TJvmMemoryPool& usage: response.memory_pools) {
- JvmMetric::CreateAndRegister(metrics, "jvm.$0.max-usage-bytes", usage.name, MAX);
- JvmMetric::CreateAndRegister(metrics, "jvm.$0.current-usage-bytes", usage.name,
- CURRENT);
- JvmMetric::CreateAndRegister(metrics, "jvm.$0.committed-usage-bytes", usage.name,
- COMMITTED);
- JvmMetric::CreateAndRegister(metrics, "jvm.$0.init-usage-bytes", usage.name, INIT);
- JvmMetric::CreateAndRegister(metrics, "jvm.$0.peak-max-usage-bytes", usage.name,
+ JvmMemoryMetric::CreateAndRegister(
+ metrics, "jvm.$0.max-usage-bytes", usage.name, MAX);
+ JvmMemoryMetric::CreateAndRegister(
+ metrics, "jvm.$0.current-usage-bytes", usage.name, CURRENT);
+ JvmMemoryMetric::CreateAndRegister(
+ metrics, "jvm.$0.committed-usage-bytes", usage.name, COMMITTED);
+ JvmMemoryMetric::CreateAndRegister(
+ metrics, "jvm.$0.init-usage-bytes", usage.name, INIT);
+ JvmMemoryMetric::CreateAndRegister(metrics, "jvm.$0.peak-max-usage-bytes", usage.name,
PEAK_MAX);
- JvmMetric::CreateAndRegister(metrics, "jvm.$0.peak-current-usage-bytes", usage.name,
- PEAK_CURRENT);
- JvmMetric::CreateAndRegister(metrics, "jvm.$0.peak-committed-usage-bytes", usage.name,
- PEAK_COMMITTED);
- JvmMetric::CreateAndRegister(metrics, "jvm.$0.peak-init-usage-bytes", usage.name,
- PEAK_INIT);
+ JvmMemoryMetric::CreateAndRegister(
+ metrics, "jvm.$0.peak-current-usage-bytes", usage.name, PEAK_CURRENT);
+ JvmMemoryMetric::CreateAndRegister(
+ metrics, "jvm.$0.peak-committed-usage-bytes", usage.name, PEAK_COMMITTED);
+ JvmMemoryMetric::CreateAndRegister(
+ metrics, "jvm.$0.peak-init-usage-bytes", usage.name, PEAK_INIT);
}
return Status::OK();
}
-int64_t JvmMetric::GetValue() {
- TGetJvmMetricsRequest request;
+int64_t JvmMemoryMetric::GetValue() {
+ TGetJvmMemoryMetricsRequest request;
request.get_all = false;
request.__set_memory_pool(mempool_name_);
- TGetJvmMetricsResponse response;
- if (!JniUtil::GetJvmMetrics(request, &response).ok()) return 0;
+ 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_);
@@ -226,7 +230,7 @@ int64_t JvmMetric::GetValue() {
case PEAK_COMMITTED:
return pool.peak_committed;
default:
- DCHECK(false) << "Unknown JvmMetricType: " << metric_type_;
+ DCHECK(false) << "Unknown JvmMemoryMetricType: " << metric_type_;
}
return 0;
}
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/util/memory-metrics.h
----------------------------------------------------------------------
diff --git a/be/src/util/memory-metrics.h b/be/src/util/memory-metrics.h
index ed0e889..b2a2fdc 100644
--- a/be/src/util/memory-metrics.h
+++ b/be/src/util/memory-metrics.h
@@ -152,14 +152,14 @@ class SanitizerMallocMetric : public IntGauge {
}
};
-/// A JvmMetric 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 generation, survivor
-/// space, code cache and permanently tenured objects.
-class JvmMetric : public IntGauge {
+/// 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
+/// generation, survivor space, code cache and permanently tenured objects.
+class JvmMemoryMetric : public IntGauge {
public:
- /// Registers many Jvm memory metrics: one for every member of JvmMetricType for each
- /// pool (usually ~5 pools plus a synthetic 'total' pool).
+ /// Registers many Jvm memory metrics: one for every member of JvmMemoryMetricType for
+ /// each pool (usually ~5 pools plus a synthetic 'total' pool).
static Status InitMetrics(MetricGroup* metrics) WARN_UNUSED_RESULT;
/// Searches through jvm_metrics_response_ for a matching memory pool and pulls out the
@@ -168,7 +168,7 @@ class JvmMetric : public IntGauge {
private:
/// Each names one of the fields in TJvmMemoryPool.
- enum JvmMetricType {
+ enum JvmMemoryMetricType {
MAX,
INIT,
COMMITTED,
@@ -179,18 +179,19 @@ class JvmMetric : public IntGauge {
PEAK_CURRENT
};
- static JvmMetric* CreateAndRegister(MetricGroup* metrics, const std::string& key,
- const std::string& pool_name, JvmMetric::JvmMetricType type);
+ static JvmMemoryMetric* CreateAndRegister(MetricGroup* metrics, const std::string& key,
+ const std::string& pool_name, JvmMemoryMetric::JvmMemoryMetricType type);
- /// Private constructor to ensure only InitMetrics() can create JvmMetrics.
- JvmMetric(const TMetricDef& def, const std::string& mempool_name, JvmMetricType type);
+ /// Private constructor to ensure only InitMetrics() can create JvmMemoryMetrics.
+ JvmMemoryMetric(
+ const TMetricDef& def, const std::string& mempool_name, JvmMemoryMetricType type);
/// The name of the memory pool, defined by the Jvm.
std::string mempool_name_;
/// Each metric corresponds to one value; this tells us which value from the memory pool
/// that is.
- JvmMetricType metric_type_;
+ JvmMemoryMetricType metric_type_;
};
/// Metric that reports information about the buffer pool.
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/util/metrics-test.cc
----------------------------------------------------------------------
diff --git a/be/src/util/metrics-test.cc b/be/src/util/metrics-test.cc
index bfbfdfe..9aa50e9 100644
--- a/be/src/util/metrics-test.cc
+++ b/be/src/util/metrics-test.cc
@@ -247,8 +247,8 @@ TEST_F(MetricsTest, MemMetric) {
#endif
}
-TEST_F(MetricsTest, JvmMetrics) {
- MetricGroup metrics("JvmMetrics");
+TEST_F(MetricsTest, JvmMemoryMetrics) {
+ MetricGroup metrics("JvmMemoryMetrics");
ASSERT_OK(RegisterMemoryMetrics(&metrics, true, nullptr, nullptr));
IntGauge* jvm_total_used =
metrics.GetOrCreateChildGroup("jvm")->FindMetricForTesting<IntGauge>(
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/util/webserver-test.cc
----------------------------------------------------------------------
diff --git a/be/src/util/webserver-test.cc b/be/src/util/webserver-test.cc
index 78934c6..7132b74 100644
--- a/be/src/util/webserver-test.cc
+++ b/be/src/util/webserver-test.cc
@@ -135,7 +135,7 @@ void JsonCallback(bool always_text, const Webserver::ArgumentMap& args,
document->AddMember(TO_ESCAPE_KEY.c_str(), TO_ESCAPE_VALUE.c_str(),
document->GetAllocator());
if (always_text) {
- document->AddMember(Webserver::ENABLE_RAW_JSON_KEY, true, document->GetAllocator());
+ document->AddMember(Webserver::ENABLE_RAW_HTML_KEY, true, document->GetAllocator());
}
}
@@ -174,7 +174,7 @@ TEST(Webserver, JsonTest) {
Substitute("$0?raw", JSON_TEST_PATH), &raw_contents));
ASSERT_TRUE(raw_contents.str().find("text/plain") != string::npos);
- // Any callback that includes ENABLE_RAW_JSON_KEY should always return text.
+ // Any callback that includes ENABLE_RAW_HTML_KEY should always return text.
stringstream raw_cb_contents;
ASSERT_OK(HttpGet("localhost", FLAGS_webserver_port, RAW_TEXT_PATH,
&raw_cb_contents));
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/util/webserver.cc
----------------------------------------------------------------------
diff --git a/be/src/util/webserver.cc b/be/src/util/webserver.cc
index c8307b5..d5d9732 100644
--- a/be/src/util/webserver.cc
+++ b/be/src/util/webserver.cc
@@ -137,7 +137,9 @@ const char* GetDefaultDocumentRoot() {
namespace impala {
-const char* Webserver::ENABLE_RAW_JSON_KEY = "__raw__";
+const char* Webserver::ENABLE_RAW_HTML_KEY = "__raw__";
+
+const char* Webserver::ENABLE_PLAIN_JSON_KEY = "__json__";
// Supported HTTP response codes
enum ResponseCode {
@@ -435,9 +437,10 @@ void Webserver::RenderUrlWithTemplate(const ArgumentMap& arguments,
document.SetObject();
GetCommonJson(&document);
- bool raw_json = (arguments.find("json") != arguments.end());
url_handler.callback()(arguments, &document);
- if (raw_json) {
+ bool plain_json = (arguments.find("json") != arguments.end())
+ || document.HasMember(ENABLE_PLAIN_JSON_KEY);
+ if (plain_json) {
// Callbacks may optionally be rendered as a text-only, pretty-printed Json document
// (mostly for debugging or integration with third-party tools).
StringBuffer strbuf;
@@ -447,9 +450,9 @@ void Webserver::RenderUrlWithTemplate(const ArgumentMap& arguments,
*content_type = JSON;
} else {
if (arguments.find("raw") != arguments.end()) {
- document.AddMember(ENABLE_RAW_JSON_KEY, "true", document.GetAllocator());
+ document.AddMember(ENABLE_RAW_HTML_KEY, "true", document.GetAllocator());
}
- if (document.HasMember(ENABLE_RAW_JSON_KEY)) {
+ if (document.HasMember(ENABLE_RAW_HTML_KEY)) {
*content_type = PLAIN;
}
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/be/src/util/webserver.h
----------------------------------------------------------------------
diff --git a/be/src/util/webserver.h b/be/src/util/webserver.h
index c060d5e..1651d88 100644
--- a/be/src/util/webserver.h
+++ b/be/src/util/webserver.h
@@ -54,10 +54,16 @@ class Webserver {
typedef boost::function<void (const ArgumentMap& args, std::stringstream* output)>
RawUrlCallback;
- /// Any callback may add a member to their Json output with key ENABLE_RAW_JSON_KEY;
+ /// Any callback may add a member to their Json output with key ENABLE_RAW_HTML_KEY;
/// this causes the result of the template rendering process to be sent to the browser
/// as text, not HTML.
- static const char* ENABLE_RAW_JSON_KEY;
+ static const char* ENABLE_RAW_HTML_KEY;
+
+ /// Any callback may add a member to their Json output with key ENABLE_PLAIN_JSON_KEY;
+ /// this causes the result of the template rendering process to be sent to the browser
+ /// as pretty printed JSON plain text.
+ static const char* ENABLE_PLAIN_JSON_KEY;
+
/// Using this constructor, the webserver will bind to all available interfaces.
Webserver(const int port);
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/common/thrift/Frontend.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/Frontend.thrift b/common/thrift/Frontend.thrift
index 1245c94..abb2d77 100644
--- a/common/thrift/Frontend.thrift
+++ b/common/thrift/Frontend.thrift
@@ -758,7 +758,7 @@ struct TJvmMemoryPool {
}
// Request to get one or all sets of memory pool metrics.
-struct TGetJvmMetricsRequest {
+struct TGetJvmMemoryMetricsRequest {
// If set, return all pools
1: required bool get_all
@@ -766,8 +766,8 @@ struct TGetJvmMetricsRequest {
2: optional string memory_pool
}
-// Response from JniUtil::GetJvmMetrics()
-struct TGetJvmMetricsResponse {
+// 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
@@ -818,6 +818,11 @@ struct TGetJvmThreadsInfoResponse {
4: optional list<TJvmThreadInfo> threads
}
+struct TGetJMXJsonResponse {
+ // JMX of the JVM serialized to a json string.
+ 1: required string jmx_json
+}
+
struct TGetHadoopConfigRequest {
// The value of the <name> in the config <property>
1: required string name
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/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 6aec0d4..348b76d 100644
--- a/fe/src/main/java/org/apache/impala/common/JniUtil.java
+++ b/fe/src/main/java/org/apache/impala/common/JniUtil.java
@@ -17,10 +17,10 @@
package org.apache.impala.common;
-import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.io.Writer;
+import java.lang.management.GarbageCollectorMXBean;
import java.lang.management.ManagementFactory;
import java.lang.management.MemoryMXBean;
import java.lang.management.MemoryPoolMXBean;
@@ -31,6 +31,8 @@ import java.lang.management.ThreadInfo;
import java.util.ArrayList;
import java.util.Map;
+import org.apache.impala.thrift.TGetJMXJsonResponse;
+import org.apache.impala.util.JMXJsonUtil;
import org.apache.thrift.TBase;
import org.apache.thrift.TSerializer;
import org.apache.thrift.TDeserializer;
@@ -39,14 +41,17 @@ import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
-import org.apache.impala.thrift.TGetJvmMetricsRequest;
-import org.apache.impala.thrift.TGetJvmMetricsResponse;
+import org.apache.impala.thrift.TGetJvmMemoryMetricsRequest;
+import org.apache.impala.thrift.TGetJvmMemoryMetricsResponse;
import org.apache.impala.thrift.TGetJvmThreadsInfoRequest;
import org.apache.impala.thrift.TGetJvmThreadsInfoResponse;
import org.apache.impala.thrift.TJvmMemoryPool;
import org.apache.impala.thrift.TJvmThreadInfo;
+import org.apache.impala.util.JvmPauseMonitor;
+import org.apache.log4j.Logger;
/**
* Utility class with methods intended for JNI clients
*/
@@ -54,6 +59,15 @@ public class JniUtil {
private final static TBinaryProtocol.Factory protocolFactory_ =
new TBinaryProtocol.Factory();
+ private static final Logger LOG = Logger.getLogger(JniUtil.class);
+
+ /**
+ * Initializes the JvmPauseMonitor instance.
+ */
+ public static void initPauseMonitor() {
+ JvmPauseMonitor.INSTANCE.initPauseMonitor();
+ }
+
/**
* Returns a formatted string containing the simple exception name and the
* exception message without the full stack trace. Includes the
@@ -82,6 +96,18 @@ public class JniUtil {
}
/**
+ * Serializes input into a byte[] using a given protocol factory.
+ */
+ public static <T extends TBase<?, ?>, F extends TProtocolFactory>
+ byte[] serializeToThrift(T input, F protocolFactory) throws ImpalaException {
+ TSerializer serializer = new TSerializer(protocolFactory);
+ try {
+ return serializer.serialize(input);
+ } catch (TException e) {
+ throw new InternalException(e.getMessage());
+ }
+ }
+ /**
* Deserialize a serialized form of a Thrift data structure to its object form.
*/
public static <T extends TBase<?, ?>, F extends TProtocolFactory>
@@ -101,11 +127,11 @@ public class JniUtil {
* Impala metrics by the backend. A synthetic 'total' memory pool is included with
* aggregate statistics for all real pools.
*/
- public static byte[] getJvmMetrics(byte[] argument) throws ImpalaException {
- TGetJvmMetricsRequest request = new TGetJvmMetricsRequest();
+ public static byte[] getJvmMemoryMetrics(byte[] argument) throws ImpalaException {
+ TGetJvmMemoryMetricsRequest request = new TGetJvmMemoryMetricsRequest();
JniUtil.deserializeThrift(protocolFactory_, request, argument);
- TGetJvmMetricsResponse jvmMetrics = new TGetJvmMetricsResponse();
+ TGetJvmMemoryMetricsResponse jvmMetrics = new TGetJvmMemoryMetricsResponse();
jvmMetrics.setMemory_pools(new ArrayList<TJvmMemoryPool>());
TJvmMemoryPool totalUsage = new TJvmMemoryPool();
boolean is_total =
@@ -182,13 +208,7 @@ public class JniUtil {
nonHeap.setPeak_used(0);
jvmMetrics.getMemory_pools().add(nonHeap);
}
-
- TSerializer serializer = new TSerializer(protocolFactory_);
- try {
- return serializer.serialize(jvmMetrics);
- } catch (TException e) {
- throw new InternalException(e.getMessage());
- }
+ return serializeToThrift(jvmMetrics, protocolFactory_);
}
/**
@@ -215,13 +235,12 @@ public class JniUtil {
response.addToThreads(tThreadInfo);
}
}
+ return serializeToThrift(response, protocolFactory_);
+ }
- TSerializer serializer = new TSerializer(protocolFactory_);
- try {
- return serializer.serialize(response);
- } catch (TException e) {
- throw new InternalException(e.getMessage());
- }
+ public static byte[] getJMXJson() throws ImpalaException {
+ TGetJMXJsonResponse response = new TGetJMXJsonResponse(JMXJsonUtil.getJMXJson());
+ return serializeToThrift(response, protocolFactory_);
}
/**
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/fe/src/main/java/org/apache/impala/util/JMXJsonUtil.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/util/JMXJsonUtil.java b/fe/src/main/java/org/apache/impala/util/JMXJsonUtil.java
new file mode 100644
index 0000000..faaba23
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/util/JMXJsonUtil.java
@@ -0,0 +1,281 @@
+/*
+ * 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.
+ */
+
+package org.apache.impala.util;
+
+import org.apache.log4j.Logger;
+import org.codehaus.jackson.JsonFactory;
+import org.codehaus.jackson.JsonGenerator;
+
+import javax.management.AttributeNotFoundException;
+import javax.management.InstanceNotFoundException;
+import javax.management.IntrospectionException;
+import javax.management.MBeanAttributeInfo;
+import javax.management.MBeanException;
+import javax.management.MBeanInfo;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import javax.management.ReflectionException;
+import javax.management.RuntimeErrorException;
+import javax.management.RuntimeMBeanException;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeType;
+import javax.management.openmbean.TabularData;
+import java.io.IOException;
+import java.io.StringWriter;
+import java.lang.management.ManagementFactory;
+import java.lang.reflect.Array;
+import java.util.Iterator;
+import java.util.Set;
+
+/**
+ * Utility class that returns a JSON representation of the JMX beans.
+ * This is based on hadoop-common's implementation of JMXJsonServlet.
+ *
+ * Output format:
+ * {
+ * "beans" : [
+ * {
+ * "name":"bean-name"
+ * ...
+ * }
+ * ]
+ * }
+ * Each bean's attributes will be converted to a JSON object member.
+ * If the attribute is a boolean, a number, a string, or an array
+ * it will be converted to the JSON equivalent.
+ *
+ * If the value is a {@link CompositeData} then it will be converted
+ * to a JSON object with the keys as the name of the JSON member and
+ * the value is converted following these same rules.
+ * If the value is a {@link TabularData} then it will be converted
+ * to an array of the {@link CompositeData} elements that it contains.
+ * All other objects will be converted to a string and output as such.
+ * The bean's name and modelerType will be returned for all beans.
+ *
+ */
+public class JMXJsonUtil {
+ // MBean server instance
+ protected static transient MBeanServer mBeanServer =
+ ManagementFactory.getPlatformMBeanServer();
+
+ private static final Logger LOG = Logger.getLogger(JMXJsonUtil.class);
+
+ // Returns the JMX beans as a JSON string.
+ public static String getJMXJson() {
+ StringWriter writer = new StringWriter();
+ try {
+ JsonGenerator jg = null;
+ try {
+ JsonFactory jsonFactory = new JsonFactory();
+ jg = jsonFactory.createJsonGenerator(writer);
+ jg.disable(JsonGenerator.Feature.AUTO_CLOSE_TARGET);
+ jg.writeStartObject();
+ if (mBeanServer == null) {
+ jg.writeStringField("result", "ERROR");
+ jg.writeStringField("message", "No MBeanServer could be found");
+ jg.close();
+ LOG.error("No MBeanServer could be found.");
+ return writer.toString();
+ }
+ listBeans(jg);
+ } finally {
+ if (jg != null) {
+ jg.close();
+ }
+ if (writer != null) {
+ writer.close();
+ }
+ }
+ } catch ( IOException e ) {
+ LOG.error("Caught an exception while processing JMX request", e);
+ }
+ return writer.toString();
+ }
+
+ // Utility method that lists all the mbeans and write them using the supplied
+ // JsonGenerator.
+ private static void listBeans(JsonGenerator jg) throws IOException {
+ Set<ObjectName> names;
+ names = mBeanServer.queryNames(null, null);
+ jg.writeArrayFieldStart("beans");
+ Iterator<ObjectName> it = names.iterator();
+ while (it.hasNext()) {
+ ObjectName oname = it.next();
+ MBeanInfo minfo;
+ String code = "";
+ Object attributeinfo = null;
+ try {
+ minfo = mBeanServer.getMBeanInfo(oname);
+ code = minfo.getClassName();
+ String prs = "";
+ try {
+ if ("org.apache.commons.modeler.BaseModelMBean".equals(code)) {
+ prs = "modelerType";
+ code = (String) mBeanServer.getAttribute(oname, prs);
+ }
+ } catch (AttributeNotFoundException e) {
+ // If the modelerType attribute was not found, the class name is used
+ // instead.
+ LOG.error("getting attribute " + prs + " of " + oname
+ + " threw an exception", e);
+ } catch (MBeanException e) {
+ // The code inside the attribute getter threw an exception so log it,
+ // and fall back on the class name
+ LOG.error("getting attribute " + prs + " of " + oname
+ + " threw an exception", e);
+ } catch (RuntimeException e) {
+ // For some reason even with an MBeanException available to them
+ // Runtime exceptionscan still find their way through, so treat them
+ // the same as MBeanException
+ LOG.error("getting attribute " + prs + " of " + oname
+ + " threw an exception", e);
+ } catch ( ReflectionException e ) {
+ // This happens when the code inside the JMX bean (setter?? from the
+ // java docs) threw an exception, so log it and fall back on the
+ // class name
+ LOG.error("getting attribute " + prs + " of " + oname
+ + " threw an exception", e);
+ }
+ } catch (InstanceNotFoundException e) {
+ //Ignored for some reason the bean was not found so don't output it
+ continue;
+ } catch ( IntrospectionException | ReflectionException e ) {
+ // This is an internal error, something odd happened with reflection so
+ // log it and don't output the bean.
+ LOG.error("Problem while trying to process JMX query with MBean " + oname, e);
+ continue;
+ }
+ jg.writeStartObject();
+ jg.writeStringField("name", oname.toString());
+ jg.writeStringField("modelerType", code);
+ MBeanAttributeInfo attrs[] = minfo.getAttributes();
+ for (int i = 0; i < attrs.length; i++) {
+ writeAttribute(jg, oname, attrs[i]);
+ }
+ jg.writeEndObject();
+ }
+ jg.writeEndArray();
+ }
+
+ // Utility method to write mBean attributes.
+ private static void writeAttribute(JsonGenerator jg, ObjectName oname,
+ MBeanAttributeInfo attr) throws IOException {
+ if (!attr.isReadable()) {
+ return;
+ }
+ String attName = attr.getName();
+ if ("modelerType".equals(attName)) {
+ return;
+ }
+ if (attName.indexOf("=") >= 0 || attName.indexOf(":") >= 0
+ || attName.indexOf(" ") >= 0) {
+ return;
+ }
+ Object value = null;
+ try {
+ value = mBeanServer.getAttribute(oname, attName);
+ } catch (RuntimeMBeanException e) {
+ // UnsupportedOperationExceptions happen in the normal course of business,
+ // so no need to log them as errors all the time.
+ if (e.getCause() instanceof UnsupportedOperationException) {
+ LOG.trace("getting attribute "+attName+" of "+oname+" threw an exception", e);
+ } else {
+ LOG.error("getting attribute "+attName+" of "+oname+" threw an exception", e);
+ }
+ return;
+ } catch (RuntimeErrorException e) {
+ // RuntimeErrorException happens when an unexpected failure occurs in getAttribute
+ // for example https://issues.apache.org/jira/browse/DAEMON-120
+ LOG.debug("getting attribute "+attName+" of "+oname+" threw an exception", e);
+ return;
+ } catch (AttributeNotFoundException e) {
+ //Ignored the attribute was not found, which should never happen because the bean
+ //just told us that it has this attribute, but if this happens just don't output
+ //the attribute.
+ return;
+ } catch (MBeanException e) {
+ //The code inside the attribute getter threw an exception so log it, and
+ // skip outputting the attribute
+ LOG.error("getting attribute "+attName+" of "+oname+" threw an exception", e);
+ return;
+ } catch (RuntimeException e) {
+ //For some reason even with an MBeanException available to them Runtime exceptions
+ //can still find their way through, so treat them the same as MBeanException
+ LOG.error("getting attribute "+attName+" of "+oname+" threw an exception", e);
+ return;
+ } catch (ReflectionException e) {
+ //This happens when the code inside the JMX bean (setter?? from the java docs)
+ //threw an exception, so log it and skip outputting the attribute
+ LOG.error("getting attribute "+attName+" of "+oname+" threw an exception", e);
+ return;
+ } catch (InstanceNotFoundException e) {
+ //Ignored the mbean itself was not found, which should never happen because we
+ //just accessed it (perhaps something unregistered in-between) but if this
+ //happens just don't output the attribute.
+ return;
+ }
+ writeAttribute(jg, attName, value);
+ }
+
+ private static void writeAttribute(JsonGenerator jg, String attName, Object value)
+ throws IOException {
+ jg.writeFieldName(attName);
+ writeObject(jg, value);
+ }
+
+ private static void writeObject(JsonGenerator jg, Object value) throws IOException {
+ if(value == null) {
+ jg.writeNull();
+ } else {
+ Class<?> c = value.getClass();
+ if (c.isArray()) {
+ jg.writeStartArray();
+ int len = Array.getLength(value);
+ for (int j = 0; j < len; j++) {
+ Object item = Array.get(value, j);
+ writeObject(jg, item);
+ }
+ jg.writeEndArray();
+ } else if(value instanceof Number) {
+ Number n = (Number)value;
+ jg.writeNumber(n.toString());
+ } else if(value instanceof Boolean) {
+ Boolean b = (Boolean)value;
+ jg.writeBoolean(b);
+ } else if(value instanceof CompositeData) {
+ CompositeData cds = (CompositeData)value;
+ CompositeType comp = cds.getCompositeType();
+ Set<String> keys = comp.keySet();
+ jg.writeStartObject();
+ for(String key: keys) {
+ writeAttribute(jg, key, cds.get(key));
+ }
+ jg.writeEndObject();
+ } else if(value instanceof TabularData) {
+ TabularData tds = (TabularData)value;
+ jg.writeStartArray();
+ for(Object entry : tds.values()) {
+ writeObject(jg, entry);
+ }
+ jg.writeEndArray();
+ } else {
+ jg.writeString(value.toString());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/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
new file mode 100644
index 0000000..e4f28bd
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/util/JvmPauseMonitor.java
@@ -0,0 +1,205 @@
+/*
+ * 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.
+ */
+package org.apache.impala.util;
+
+import java.lang.management.GarbageCollectorMXBean;
+import java.lang.management.ManagementFactory;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.base.Stopwatch;
+
+/**
+ * Class which sets up a simple thread which runs in a loop sleeping
+ * for a short interval of time. If the sleep takes significantly longer
+ * than its target time, it implies that the JVM or host machine has
+ * paused processing, which may cause other problems. If such a pause is
+ * detected, the thread logs a message.
+ */
+public class JvmPauseMonitor {
+ private static final Logger LOG = Logger.getLogger(JvmPauseMonitor.class);
+
+ // The target sleep time.
+ private static final long SLEEP_INTERVAL_MS = 500;
+
+ // log WARN if we detect a pause longer than this threshold.
+ private long warnThresholdMs_;
+ private static final long WARN_THRESHOLD_MS = 10000;
+
+ // log INFO if we detect a pause longer than this threshold.
+ private long infoThresholdMs_;
+ private static final long INFO_THRESHOLD_MS = 1000;
+
+ // Daemon thread running the pause monitor loop.
+ private Thread monitorThread_;
+ private volatile boolean shouldRun = true;
+
+ // Singleton instance of this pause monitor.
+ public static JvmPauseMonitor INSTANCE = new JvmPauseMonitor();
+
+ // Initializes the pause monitor. No-op if called multiple times.
+ public static void initPauseMonitor() {
+ if (INSTANCE.isStarted()) return;
+ INSTANCE.init();
+ }
+
+ private JvmPauseMonitor() {
+ this(INFO_THRESHOLD_MS, WARN_THRESHOLD_MS);
+ }
+
+ private JvmPauseMonitor(long infoThresholdMs, long warnThresholdMs) {
+ this.infoThresholdMs_ = infoThresholdMs;
+ this.warnThresholdMs_ = warnThresholdMs;
+ }
+
+ protected void init() {
+ monitorThread_ = new Thread(new Monitor(), "JVM pause monitor");
+ monitorThread_.setDaemon(true);
+ monitorThread_.start();
+ }
+
+ public boolean isStarted() {
+ return monitorThread_ != null;
+ }
+
+ /**
+ * Helper method that formats the message to be logged, along with
+ * the GC metrics.
+ */
+ private String formatMessage(long extraSleepTime,
+ Map<String, GcTimes> gcTimesAfterSleep,
+ Map<String, GcTimes> gcTimesBeforeSleep) {
+
+ Set<String> gcBeanNames = Sets.intersection(
+ gcTimesAfterSleep.keySet(),
+ gcTimesBeforeSleep.keySet());
+ List<String> gcDiffs = Lists.newArrayList();
+ for (String name : gcBeanNames) {
+ GcTimes diff = gcTimesAfterSleep.get(name).subtract(
+ gcTimesBeforeSleep.get(name));
+ if (diff.gcCount != 0) {
+ gcDiffs.add("GC pool '" + name + "' had collection(s): " +
+ diff.toString());
+ }
+ }
+
+ String ret = "Detected pause in JVM or host machine (eg GC): " +
+ "pause of approximately " + extraSleepTime + "ms\n";
+ if (gcDiffs.isEmpty()) {
+ ret += "No GCs detected";
+ } else {
+ ret += Joiner.on("\n").join(gcDiffs);
+ }
+ return ret;
+ }
+
+ private Map<String, GcTimes> getGcTimes() {
+ Map<String, GcTimes> map = Maps.newHashMap();
+ List<GarbageCollectorMXBean> gcBeans =
+ ManagementFactory.getGarbageCollectorMXBeans();
+ for (GarbageCollectorMXBean gcBean : gcBeans) {
+ map.put(gcBean.getName(), new GcTimes(gcBean));
+ }
+ return map;
+ }
+
+ private static class GcTimes {
+ private GcTimes(GarbageCollectorMXBean gcBean) {
+ gcCount = gcBean.getCollectionCount();
+ gcTimeMillis = gcBean.getCollectionTime();
+ }
+
+ private GcTimes(long count, long time) {
+ this.gcCount = count;
+ this.gcTimeMillis = time;
+ }
+
+ private GcTimes subtract(GcTimes other) {
+ return new GcTimes(this.gcCount - other.gcCount,
+ this.gcTimeMillis - other.gcTimeMillis);
+ }
+
+ @Override
+ public String toString() {
+ return "count=" + gcCount + " time=" + gcTimeMillis + "ms";
+ }
+
+ private long gcCount;
+ private long gcTimeMillis;
+ }
+
+ /**
+ * Runnable instance of the pause monitor loop. Launched from serviceStart().
+ */
+ private class Monitor implements Runnable {
+ @Override
+ public void run() {
+ Stopwatch sw = new Stopwatch();
+ Map<String, GcTimes> gcTimesBeforeSleep = getGcTimes();
+ LOG.info("Starting JVM pause monitor");
+ while (shouldRun) {
+ sw.reset().start();
+ try {
+ Thread.sleep(SLEEP_INTERVAL_MS);
+ } catch (InterruptedException ie) {
+ return;
+ }
+ sw.stop();
+ long extraSleepTime = sw.elapsedTime(TimeUnit.MILLISECONDS) - SLEEP_INTERVAL_MS;
+ Map<String, GcTimes> gcTimesAfterSleep = getGcTimes();
+
+ if (extraSleepTime > warnThresholdMs_) {
+ LOG.warn(formatMessage(
+ extraSleepTime, gcTimesAfterSleep, gcTimesBeforeSleep));
+ } else if (extraSleepTime > infoThresholdMs_) {
+ LOG.info(formatMessage(
+ extraSleepTime, gcTimesAfterSleep, gcTimesBeforeSleep));
+ }
+ gcTimesBeforeSleep = gcTimesAfterSleep;
+ }
+ }
+ }
+
+ /**
+ * Simple 'main' to facilitate manual testing of the pause monitor.
+ *
+ * This main function just leaks memory into a list. Running this class
+ * with a 1GB heap will very quickly go into "GC hell" and result in
+ * log messages about the GC pauses.
+ */
+ @SuppressWarnings("resource")
+ public static void main(String []args) throws Exception {
+ JvmPauseMonitor monitor = new JvmPauseMonitor();
+ monitor.init();
+ List<String> list = Lists.newArrayList();
+ int i = 0;
+ while (true) {
+ list.add(String.valueOf(i++));
+ }
+ }
+}
+
+
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/fe/src/test/java/org/apache/impala/util/JMXJsonUtilTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/util/JMXJsonUtilTest.java b/fe/src/test/java/org/apache/impala/util/JMXJsonUtilTest.java
new file mode 100644
index 0000000..95e9c3d
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/util/JMXJsonUtilTest.java
@@ -0,0 +1,56 @@
+// 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.
+
+package org.apache.impala.util;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import org.apache.impala.common.ImpalaException;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Unit tests for JMXJsonUtil functionality.
+ */
+public class JMXJsonUtilTest {
+
+ // Validates the JSON string returned by JMXJsonUtil.getJMXJson()
+ @Test
+ public void testJMXMetrics() throws ImpalaException {
+ String jmxJson = JMXJsonUtil.getJMXJson();
+ JsonNode rootNode = null;
+ // Validate the JSON.
+ try {
+ rootNode = new ObjectMapper().readTree(jmxJson);
+ } catch (IOException e) {
+ fail("Invalid JSON returned by getMxJson(): " + jmxJson);
+ }
+ Preconditions.checkNotNull(rootNode);
+ assertTrue("Invalid JSON: " + jmxJson, rootNode.hasNonNull("beans"));
+ List<String> values = rootNode.get("beans").findValuesAsText("name");
+ assertTrue("Invalid JSON: " + jmxJson,
+ values.contains("java.lang:type=MemoryPool,name=Metaspace"));
+ assertTrue("Invalid JSON: " + jmxJson, values.contains("java.lang:type=Runtime"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/fe/src/test/java/org/apache/impala/util/JniUtilTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/util/JniUtilTest.java b/fe/src/test/java/org/apache/impala/util/JniUtilTest.java
new file mode 100644
index 0000000..6166179
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/util/JniUtilTest.java
@@ -0,0 +1,54 @@
+// 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.
+
+package org.apache.impala.util;
+
+import static org.junit.Assert.*;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import org.apache.impala.thrift.TGetJMXJsonResponse;
+import org.junit.Test;
+
+import org.apache.impala.common.ImpalaException;
+import org.apache.impala.common.JniUtil;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.protocol.TProtocolFactory;
+import org.apache.impala.thrift.TCacheJarParams;
+
+import java.io.IOException;
+
+/**
+ * Unit tests for JniUtil functions.
+ */
+public class JniUtilTest {
+
+ private static TBinaryProtocol.Factory protocolFactory_ = new TBinaryProtocol.Factory();
+
+ // Unit test for JniUtil.serializetoThrift().
+ @Test
+ public void testSerializeToThrift() throws ImpalaException {
+ // Serialize and deserialize an simple thrift object.
+ TCacheJarParams testObject = new TCacheJarParams("test string");
+ byte[] testObjBytes = JniUtil.serializeToThrift(testObject, protocolFactory_);
+
+ TCacheJarParams deserializedTestObj = new TCacheJarParams();
+ JniUtil.deserializeThrift(protocolFactory_, deserializedTestObj, testObjBytes);
+ assertEquals(deserializedTestObj.hdfs_location, "test string");
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/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
new file mode 100644
index 0000000..f4e616b
--- /dev/null
+++ b/tests/custom_cluster/test_pause_monitor.py
@@ -0,0 +1,38 @@
+# 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 signal
+import time
+
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+
+
+class TestPauseMonitor(CustomClusterTestSuite):
+ """Class for pause monitor tests."""
+
+ 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."""
+ impalad = self.cluster.get_first_impalad()
+ # Send a SIGSTOP for the process and block it for 5s.
+ 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)
+ # Check that the pause is detected.
+ self.assert_impalad_log_contains('INFO', "Detected pause in JVM or host machine")
http://git-wip-us.apache.org/repos/asf/impala/blob/4976ff3c/tests/webserver/test_web_pages.py
----------------------------------------------------------------------
diff --git a/tests/webserver/test_web_pages.py b/tests/webserver/test_web_pages.py
index 1ac1576..15d4f8f 100644
--- a/tests/webserver/test_web_pages.py
+++ b/tests/webserver/test_web_pages.py
@@ -36,6 +36,8 @@ class TestWebPage(ImpalaTestSuite):
RPCZ_URL = "http://localhost:{0}/rpcz"
THREAD_GROUP_URL = "http://localhost:{0}/thread-group"
METRICS_URL = "http://localhost:{0}/metrics"
+ JMX_URL = "http://localhost:{0}/jmx"
+
# 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
# log level endpoints, one without the statestore port and the
@@ -63,6 +65,20 @@ class TestWebPage(ImpalaTestSuite):
result = impalad.service.read_debug_webpage("query_profile_encoded?query_id=123")
assert result.startswith("Could not obtain runtime profile: Query id")
+ def test_jmx_endpoint(self):
+ """Tests that the /jmx endpoint on the Catalog and Impalads returns a valid json."""
+ for port in self.TEST_PORTS_WITHOUT_SS:
+ input_url = self.JMX_URL.format(port)
+ response = requests.get(input_url)
+ assert response.status_code == requests.codes.ok
+ assert "application/json" == response.headers['Content-Type']
+ jmx_json = ""
+ try:
+ jmx_json = json.loads(response.text)
+ assert "beans" in jmx_json.keys(), "Ill formatted JSON returned: %s" % jmx_json
+ except ValueError:
+ assert False, "Invalid JSON returned from /jmx endpoint: %s" % jmx_json
+
def get_and_check_status(self, url, string_to_search = "", ports_to_test = None):
"""Helper method that polls a given url and asserts the return code is ok and
the response contains the input string."""
@@ -73,7 +89,6 @@ class TestWebPage(ImpalaTestSuite):
response = requests.get(input_url)
assert response.status_code == requests.codes.ok\
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."""
@@ -83,8 +98,8 @@ class TestWebPage(ImpalaTestSuite):
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,
- ports_to_test=self.TEST_PORTS_WITHOUT_SS)
+ self.get_and_check_status(url, string_to_search,
+ ports_to_test=self.TEST_PORTS_WITHOUT_SS)
def test_content_type(self):
"""Checks that an appropriate content-type is set for various types of pages."""
[9/9] impala git commit: IMPALA-7320. Avoid calling getFileStatus()
for each partition when table is loaded
Posted by to...@apache.org.
IMPALA-7320. Avoid calling getFileStatus() for each partition when table is loaded
Prior to this patch, when a table is first loaded, the catalog iterated
over each of the partition directories and called getFileStatus() on
each, serially, to determine the overall access level of the table.
In some testing, each such call took 1-2ms, so this could add many
seconds to the overall table load time for a table with thousands of
partitions and also add to the NN load.
This patch adds some batch pre-fetching of file status information: for
any parent directory which contains more than one partition, we use the
listStatus() API to fetch the FileStatus objects in bulk.
A new unit test verifies the number of API calls made to the NameNode
during a table load.
Change-Id: I83e5ebc214d6620d165e13f8cc80f8fdda100734
Reviewed-on: http://gerrit.cloudera.org:8080/11027
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/7f9a74ff
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/7f9a74ff
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/7f9a74ff
Branch: refs/heads/master
Commit: 7f9a74ffcaf1818f1f3c9d427557acca21a627da
Parents: 8f9f91f
Author: Todd Lipcon <to...@cloudera.com>
Authored: Wed Jul 18 19:31:58 2018 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Thu Aug 9 21:39:30 2018 +0000
----------------------------------------------------------------------
.../org/apache/impala/catalog/HdfsTable.java | 149 ++++++++++++-------
.../apache/impala/util/FsPermissionCache.java | 62 ++++++++
.../apache/impala/util/FsPermissionChecker.java | 14 +-
.../org/apache/impala/catalog/CatalogTest.java | 52 +++++++
4 files changed, 218 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/7f9a74ff/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
index 38d1695..74f84e5 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -78,6 +78,7 @@ import org.apache.impala.thrift.TTableDescriptor;
import org.apache.impala.thrift.TTableType;
import org.apache.impala.util.AvroSchemaConverter;
import org.apache.impala.util.AvroSchemaUtils;
+import org.apache.impala.util.FsPermissionCache;
import org.apache.impala.util.FsPermissionChecker;
import org.apache.impala.util.HdfsCachingUtil;
import org.apache.impala.util.ListMap;
@@ -92,10 +93,12 @@ import com.codahale.metrics.Timer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
+import com.google.common.collect.HashMultiset;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import com.google.common.collect.Multiset;
import com.google.common.collect.Sets;
/**
@@ -765,10 +768,10 @@ public class HdfsTable extends Table implements FeFsTable {
// using createPartition() calls. A single partition path can correspond to multiple
// partitions.
HashMap<Path, List<HdfsPartition>> partsByPath = Maps.newHashMap();
+ FsPermissionCache permCache = preloadPermissionsCache(msPartitions);
Path tblLocation = FileSystemUtil.createFullyQualifiedPath(getHdfsBaseDirPath());
- FileSystem fs = tblLocation.getFileSystem(CONF);
- accessLevel_ = getAvailableAccessLevel(fs, tblLocation);
+ accessLevel_ = getAvailableAccessLevel(getFullName(), tblLocation, permCache);
if (msTbl.getPartitionKeysSize() == 0) {
Preconditions.checkArgument(msPartitions == null || msPartitions.isEmpty());
@@ -776,13 +779,14 @@ public class HdfsTable extends Table implements FeFsTable {
// We model partitions slightly differently to Hive - every file must exist in a
// partition, so add a single partition with no keys which will get all the
// files in the table's root directory.
- HdfsPartition part = createPartition(msTbl.getSd(), null);
+ HdfsPartition part = createPartition(msTbl.getSd(), null, permCache);
partsByPath.put(tblLocation, Lists.newArrayList(part));
if (isMarkedCached_) part.markCached();
addPartition(part);
} else {
for (org.apache.hadoop.hive.metastore.api.Partition msPartition: msPartitions) {
- HdfsPartition partition = createPartition(msPartition.getSd(), msPartition);
+ HdfsPartition partition = createPartition(msPartition.getSd(), msPartition,
+ permCache);
addPartition(partition);
// If the partition is null, its HDFS path does not exist, and it was not added
// to this table's partition list. Skip the partition.
@@ -802,26 +806,6 @@ public class HdfsTable extends Table implements FeFsTable {
}
/**
- * Helper method to load the partition file metadata from scratch. This method is
- * optimized for loading newly added partitions. For refreshing existing partitions
- * use refreshPartitionFileMetadata(HdfsPartition).
- */
- private void resetAndLoadPartitionFileMetadata(HdfsPartition partition) {
- Path partDir = partition.getLocationPath();
- try {
- FileMetadataLoadStats stats =
- resetAndLoadFileMetadata(partDir, Lists.newArrayList(partition));
- if (LOG.isDebugEnabled()) {
- LOG.debug(String.format("Loaded file metadata for %s %s", getFullName(),
- stats.debugString()));
- }
- } catch (Exception e) {
- LOG.error("Error loading file metadata for path: " + partDir.toString() +
- ". Partitions file metadata could be partially loaded.", e);
- }
- }
-
- /**
* Returns the thread pool size to load the file metadata of this table.
* 'numPaths' is the number of paths for which the file metadata should be loaded.
*
@@ -911,9 +895,10 @@ public class HdfsTable extends Table implements FeFsTable {
* from that.
* Always returns READ_WRITE for S3 and ADLS files.
*/
- private TAccessLevel getAvailableAccessLevel(FileSystem fs, Path location)
- throws IOException {
-
+ private static TAccessLevel getAvailableAccessLevel(String tableName,
+ Path location, FsPermissionCache permCache) throws IOException {
+ Preconditions.checkNotNull(location);
+ FileSystem fs = location.getFileSystem(CONF);
// Avoid calling getPermissions() on file path for S3 files, as that makes a round
// trip to S3. Also, the S3A connector is currently unable to manage S3 permissions,
// so for now it is safe to assume that all files(objects) have READ_WRITE
@@ -928,11 +913,9 @@ public class HdfsTable extends Table implements FeFsTable {
// permissions to hadoop users/groups (HADOOP-14437).
if (FileSystemUtil.isADLFileSystem(fs)) return TAccessLevel.READ_WRITE;
- FsPermissionChecker permissionChecker = FsPermissionChecker.getInstance();
while (location != null) {
try {
- FsPermissionChecker.Permissions perms =
- permissionChecker.getPermissions(fs, location);
+ FsPermissionChecker.Permissions perms = permCache.getPermissions(location);
if (perms.canReadAndWrite()) {
return TAccessLevel.READ_WRITE;
} else if (perms.canRead()) {
@@ -946,39 +929,29 @@ public class HdfsTable extends Table implements FeFsTable {
}
}
// Should never get here.
- Preconditions.checkNotNull(location, "Error: no path ancestor exists");
- return TAccessLevel.NONE;
+ throw new NullPointerException("Error determining access level for table " +
+ tableName + ": no path ancestor exists for path: " + location);
}
/**
- * Creates a new HdfsPartition object to be added to HdfsTable's partition list.
+ * Creates new HdfsPartition objects to be added to HdfsTable's partition list.
* Partitions may be empty, or may not even exist in the filesystem (a partition's
* location may have been changed to a new path that is about to be created by an
* INSERT). Also loads the file metadata for this partition. Returns new partition
* if successful or null if none was created.
*
- * Throws CatalogException if the supplied storage descriptor contains metadata that
- * Impala can't understand.
- */
- public HdfsPartition createAndLoadPartition(
- org.apache.hadoop.hive.metastore.api.Partition msPartition)
- throws CatalogException {
- HdfsPartition hdfsPartition = createPartition(msPartition.getSd(), msPartition);
- resetAndLoadPartitionFileMetadata(hdfsPartition);
- return hdfsPartition;
- }
-
- /**
- * Same as createAndLoadPartition() but is optimized for loading file metadata of
- * newly created HdfsPartitions in parallel.
+ * Throws CatalogException if one of the supplied storage descriptors contains metadata
+ * that Impala can't understand.
*/
public List<HdfsPartition> createAndLoadPartitions(
List<org.apache.hadoop.hive.metastore.api.Partition> msPartitions)
throws CatalogException {
HashMap<Path, List<HdfsPartition>> partsByPath = Maps.newHashMap();
List<HdfsPartition> addedParts = Lists.newArrayList();
+ FsPermissionCache permCache = preloadPermissionsCache(msPartitions);
for (org.apache.hadoop.hive.metastore.api.Partition partition: msPartitions) {
- HdfsPartition hdfsPartition = createPartition(partition.getSd(), partition);
+ HdfsPartition hdfsPartition = createPartition(partition.getSd(), partition,
+ permCache);
Preconditions.checkNotNull(hdfsPartition);
addedParts.add(hdfsPartition);
Path partitionPath = hdfsPartition.getLocationPath();
@@ -998,8 +971,8 @@ public class HdfsTable extends Table implements FeFsTable {
* object.
*/
private HdfsPartition createPartition(StorageDescriptor storageDescriptor,
- org.apache.hadoop.hive.metastore.api.Partition msPartition)
- throws CatalogException {
+ org.apache.hadoop.hive.metastore.api.Partition msPartition,
+ FsPermissionCache permCache) throws CatalogException {
HdfsStorageDescriptor fileFormatDescriptor =
HdfsStorageDescriptor.fromStorageDescriptor(this.name_, storageDescriptor);
List<LiteralExpr> keyValues;
@@ -1010,13 +983,14 @@ public class HdfsTable extends Table implements FeFsTable {
}
Path partDirPath = new Path(storageDescriptor.getLocation());
try {
- FileSystem fs = partDirPath.getFileSystem(CONF);
if (msPartition != null) {
HdfsCachingUtil.validateCacheParams(msPartition.getParameters());
}
+ TAccessLevel accessLevel = getAvailableAccessLevel(getFullName(), partDirPath,
+ permCache);
HdfsPartition partition =
new HdfsPartition(this, msPartition, keyValues, fileFormatDescriptor,
- new ArrayList<FileDescriptor>(), getAvailableAccessLevel(fs, partDirPath));
+ new ArrayList<FileDescriptor>(), accessLevel);
partition.checkWellFormed();
// Set the partition's #rows.
if (msPartition != null && msPartition.getParameters() != null) {
@@ -1273,8 +1247,8 @@ public class HdfsTable extends Table implements FeFsTable {
hdfsBaseDir_ = msTbl.getSd().getLocation();
isMarkedCached_ = HdfsCachingUtil.validateCacheParams(msTbl.getParameters());
Path location = new Path(hdfsBaseDir_);
- FileSystem fs = location.getFileSystem(CONF);
- accessLevel_ = getAvailableAccessLevel(fs, location);
+ accessLevel_ = getAvailableAccessLevel(getFullName(), location,
+ new FsPermissionCache());
setMetaStoreTable(msTbl);
}
@@ -1290,7 +1264,7 @@ public class HdfsTable extends Table implements FeFsTable {
org.apache.hadoop.hive.metastore.api.Table msTbl = getMetaStoreTable();
Preconditions.checkNotNull(msTbl);
setPrototypePartition(msTbl.getSd());
- HdfsPartition part = createPartition(msTbl.getSd(), null);
+ HdfsPartition part = createPartition(msTbl.getSd(), null, new FsPermissionCache());
addPartition(part);
if (isMarkedCached_) part.markCached();
refreshPartitionFileMetadata(part);
@@ -1577,8 +1551,11 @@ public class HdfsTable extends Table implements FeFsTable {
msPartitions.addAll(MetaStoreUtil.fetchPartitionsByName(client,
Lists.newArrayList(partitionNames), db_.getName(), name_));
+ FsPermissionCache permCache = preloadPermissionsCache(msPartitions);
+
for (org.apache.hadoop.hive.metastore.api.Partition msPartition: msPartitions) {
- HdfsPartition partition = createPartition(msPartition.getSd(), msPartition);
+ HdfsPartition partition = createPartition(msPartition.getSd(), msPartition,
+ permCache);
addPartition(partition);
// If the partition is null, its HDFS path does not exist, and it was not added to
// this table's partition list. Skip the partition.
@@ -1587,6 +1564,64 @@ public class HdfsTable extends Table implements FeFsTable {
}
}
+ /**
+ * For each of the partitions in 'msPartitions' with a location inside the table's
+ * base directory, attempt to pre-cache the associated file permissions into the
+ * returned cache. This takes advantage of the fact that many partition directories will
+ * be in the same parent directories, and we can bulk fetch the permissions with a
+ * single round trip to the filesystem instead of individually looking up each.
+ */
+ private FsPermissionCache preloadPermissionsCache(List<Partition> msPartitions) {
+ FsPermissionCache permCache = new FsPermissionCache();
+ // Only preload permissions if the number of partitions to be added is
+ // large (3x) relative to the number of existing partitions. This covers
+ // two common cases:
+ //
+ // 1) initial load of a table (no existing partition metadata)
+ // 2) ALTER TABLE RECOVER PARTITIONS after creating a table pointing to
+ // an already-existing partition directory tree
+ //
+ // Without this heuristic, we would end up using a "listStatus" call to
+ // potentially fetch a bunch of irrelevant information about existing
+ // partitions when we only want to know about a small number of newly-added
+ // partitions.
+ if (msPartitions.size() < partitionMap_.size() * 3) return permCache;
+
+ // TODO(todd): when HDFS-13616 (batch listing of multiple directories)
+ // is implemented, we could likely implement this with a single round
+ // trip.
+ Multiset<Path> parentPaths = HashMultiset.create();
+ for (Partition p : msPartitions) {
+ // We only do this optimization for partitions which are within the table's base
+ // directory. Otherwise we risk a case where a user has specified an external
+ // partition location that is in a directory containing a high number of irrelevant
+ // files, and we'll potentially regress performance compared to just looking up
+ // the partition file directly.
+ String loc = p.getSd().getLocation();
+ if (!loc.startsWith(hdfsBaseDir_)) continue;
+ Path parent = new Path(loc).getParent();
+ if (parent == null) continue;
+ parentPaths.add(parent);
+ }
+
+ // For any paths that contain more than one partition, issue a listStatus
+ // and pre-cache the resulting permissions.
+ for (Multiset.Entry<Path> entry : parentPaths.entrySet()) {
+ if (entry.getCount() == 1) continue;
+ Path p = entry.getElement();
+ try {
+ FileSystem fs = p.getFileSystem(CONF);
+ permCache.precacheChildrenOf(fs, p);
+ } catch (IOException ioe) {
+ // If we fail to pre-warm the cache we'll just wait for later when we
+ // try to actually load the individual permissions, at which point
+ // we can handle the issue accordingly.
+ LOG.debug("Unable to bulk-load permissions for parent path: " + p, ioe);
+ }
+ }
+ return permCache;
+ }
+
@Override
protected List<String> getColumnNamesWithHmsStats() {
List<String> ret = Lists.newArrayList();
@@ -2024,7 +2059,7 @@ public class HdfsTable extends Table implements FeFsTable {
public void reloadPartition(HdfsPartition oldPartition, Partition hmsPartition)
throws CatalogException {
HdfsPartition refreshedPartition = createPartition(
- hmsPartition.getSd(), hmsPartition);
+ hmsPartition.getSd(), hmsPartition, new FsPermissionCache());
refreshPartitionFileMetadata(refreshedPartition);
Preconditions.checkArgument(oldPartition == null
|| HdfsPartition.KV_COMPARATOR.compare(oldPartition, refreshedPartition) == 0);
http://git-wip-us.apache.org/repos/asf/impala/blob/7f9a74ff/fe/src/main/java/org/apache/impala/util/FsPermissionCache.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/util/FsPermissionCache.java b/fe/src/main/java/org/apache/impala/util/FsPermissionCache.java
new file mode 100644
index 0000000..48e8ecf
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/util/FsPermissionCache.java
@@ -0,0 +1,62 @@
+// 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.
+
+package org.apache.impala.util;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.impala.util.FsPermissionChecker.Permissions;
+
+/**
+ * Simple non-thread-safe cache for resolved file permissions. This allows
+ * pre-caching permissions by listing the status of all files within a directory,
+ * and then using that cache to avoid round trips to the FileSystem for later
+ * queries of those paths.
+ */
+public class FsPermissionCache {
+ private static Configuration CONF = new Configuration();
+ private Map<Path, Permissions> cache_ = new HashMap<>();
+
+ public Permissions getPermissions(Path location) throws IOException {
+ Permissions perms = cache_.get(location);
+ if (perms != null) return perms;
+ FsPermissionChecker checker = FsPermissionChecker.getInstance();
+ FileSystem fs = location.getFileSystem(CONF);
+ perms = checker.getPermissions(fs, location);
+ cache_.put(location, perms);
+ return perms;
+ }
+
+ public void precacheChildrenOf(FileSystem fs, Path p)
+ throws FileNotFoundException, IOException {
+ FsPermissionChecker checker = FsPermissionChecker.getInstance();
+ RemoteIterator<FileStatus> iter = fs.listStatusIterator(p);
+ while (iter.hasNext()) {
+ FileStatus status = iter.next();
+ Permissions perms = checker.getPermissions(fs, status);
+ cache_.put(status.getPath(), perms);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/7f9a74ff/fe/src/main/java/org/apache/impala/util/FsPermissionChecker.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/util/FsPermissionChecker.java b/fe/src/main/java/org/apache/impala/util/FsPermissionChecker.java
index 136c525..db5c555 100644
--- a/fe/src/main/java/org/apache/impala/util/FsPermissionChecker.java
+++ b/fe/src/main/java/org/apache/impala/util/FsPermissionChecker.java
@@ -283,11 +283,21 @@ public class FsPermissionChecker {
public Permissions getPermissions(FileSystem fs, Path path) throws IOException {
Preconditions.checkNotNull(fs);
Preconditions.checkNotNull(path);
+ return getPermissions(fs, fs.getFileStatus(path));
+ }
+
+ /**
+ * Returns a Permissions object for the given FileStatus object. In the common
+ * case that ACLs are not in use, this does not require any additional round-trip
+ * to the FileSystem. This allows batch construction using APIs like
+ * FileSystem.listStatus(...).
+ */
+ public Permissions getPermissions(FileSystem fs, FileStatus fileStatus)
+ throws IOException {
AclStatus aclStatus = null;
- FileStatus fileStatus = fs.getFileStatus(path);
if (fileStatus.getPermission().getAclBit()) {
try {
- aclStatus = fs.getAclStatus(path);
+ aclStatus = fs.getAclStatus(fileStatus.getPath());
} catch (AclException ex) {
if (LOG.isTraceEnabled()) {
LOG.trace(
http://git-wip-us.apache.org/repos/asf/impala/blob/7f9a74ff/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java b/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
index 856fc95..7ff5054 100644
--- a/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -30,14 +31,20 @@ import java.util.List;
import java.util.Set;
import org.apache.commons.codec.binary.Base64;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.GlobalStorageStatistics;
+import org.apache.hadoop.fs.StorageStatistics;
+import org.apache.hadoop.hdfs.DFSOpsCountStatistics;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.impala.analysis.FunctionName;
import org.apache.impala.analysis.LiteralExpr;
import org.apache.impala.analysis.NumericLiteral;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
+import org.apache.impala.common.Reference;
import org.apache.impala.testutil.CatalogServiceTestCatalog;
import org.apache.impala.thrift.TFunctionBinaryType;
+import org.apache.impala.thrift.TTableName;
import org.junit.Test;
import com.google.common.base.Strings;
@@ -283,6 +290,51 @@ public class CatalogTest {
catalog_.getOrLoadTable("functional", "AllTypes"));
}
+ /**
+ * Regression test for IMPALA-7320: we should use batch APIs to fetch
+ * file permissions for partitions.
+ */
+ @Test
+ public void testNumberOfGetFileStatusCalls() throws CatalogException, IOException {
+ // Reset the filesystem statistics and load the table, ensuring that it's
+ // loaded fresh by invalidating it first.
+ GlobalStorageStatistics stats = FileSystem.getGlobalStorageStatistics();
+ stats.reset();
+ catalog_.invalidateTable(new TTableName("functional", "alltypes"),
+ /*tblWasRemoved=*/new Reference<Boolean>(),
+ /*dbWasAdded=*/new Reference<Boolean>());
+
+ HdfsTable table = (HdfsTable)catalog_.getOrLoadTable("functional", "AllTypes");
+ StorageStatistics opsCounts = stats.get(DFSOpsCountStatistics.NAME);
+
+ // We expect:
+ // - one listLocatedStatus() per partition, to get the file info
+ // - one listStatus() for the month=2010/ dir
+ // - one listStatus() for the month=2009/ dir
+ long expectedCalls = table.getPartitionIds().size() + 2;
+ // Due to HDFS-13747, the listStatus calls are incorrectly accounted as
+ // op_list_located_status. So, we'll just add up the two to make our
+ // assertion resilient to this bug.
+ long seenCalls = opsCounts.getLong("op_list_located_status") +
+ opsCounts.getLong("op_list_status");
+ assertEquals(expectedCalls, seenCalls);
+
+ // We expect only one getFileStatus call, for the top-level directory.
+ assertEquals(1L, (long)opsCounts.getLong("op_get_file_status"));
+
+ // Now test REFRESH on the table...
+ stats.reset();
+ catalog_.reloadTable(table);
+
+ // Again, we expect only one getFileStatus call, for the top-level directory.
+ assertEquals(1L, (long)opsCounts.getLong("op_get_file_status"));
+ // REFRESH calls listStatus on each of the partitions, but doesn't re-check
+ // the permissions of the partition directories themselves.
+ assertEquals(table.getPartitionIds().size(),
+ (long)opsCounts.getLong("op_list_status"));
+ }
+
+
@Test
public void TestPartitions() throws CatalogException {
HdfsTable table =
[3/9] impala git commit: IMPALA-6153: Execute UpdateFilter() only for
executing queries
Posted by to...@apache.org.
IMPALA-6153: Execute UpdateFilter() only for executing queries
This change ensures that coordinator::UpdateFilter is executed
only for queries which are in the EXECUTING state. Additionally,
it also guarantees that coordinator::ReleaseExecResources is
executed only when no other thread is executing UpdateFilter.
Testing: Ran all back-end tests.
Additionally, ran the exhaustive tests with DCHECKs which would
be triggered if the ReleaseExecResources() is executed while any
other thread is executing UpdateFilter().
Change-Id: I669db217f86db5ff2802335e7b1ae8027ea7161c
Reviewed-on: http://gerrit.cloudera.org:8080/11005
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/2ced56c0
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/2ced56c0
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/2ced56c0
Branch: refs/heads/master
Commit: 2ced56c00b73ecdc92f1021f5d0129a41553e4f2
Parents: b4d20ab
Author: poojanilangekar <po...@cloudera.com>
Authored: Mon Jul 16 11:02:39 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Thu Aug 9 01:04:41 2018 +0000
----------------------------------------------------------------------
be/src/runtime/coordinator.cc | 43 +++++++++++++++++++++++++-------------
be/src/runtime/coordinator.h | 18 ++++++++++++++--
2 files changed, 45 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/2ced56c0/be/src/runtime/coordinator.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index 4b87b69..8a3213e 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -76,7 +76,7 @@ Coordinator::Coordinator(
Coordinator::~Coordinator() {
// Must have entered a terminal exec state guaranteeing resources were released.
- DCHECK_NE(exec_state_.Load(), ExecState::EXECUTING);
+ DCHECK(!IsExecuting());
DCHECK_LE(backend_exec_complete_barrier_->pending(), 0);
// Release the coordinator's reference to the query control structures.
if (query_state_ != nullptr) {
@@ -261,6 +261,7 @@ void Coordinator::InitFilterRoutingTable() {
DCHECK(!filter_routing_table_complete_)
<< "InitFilterRoutingTable() called after setting filter_routing_table_complete_";
+ lock_guard<shared_mutex> lock(filter_lock_);
for (const FragmentExecParams& fragment_params: schedule_.fragment_exec_params()) {
int num_instances = fragment_params.instance_exec_params.size();
DCHECK_GT(num_instances, 0);
@@ -384,7 +385,6 @@ string Coordinator::FilterDebugString() {
table_printer.AddColumn("Tgt. Node(s)", false);
table_printer.AddColumn("Target type", false);
table_printer.AddColumn("Partition filter", false);
-
// Distribution metrics are only meaningful if the coordinator is routing the filter.
if (filter_mode_ == TRuntimeFilterMode::GLOBAL) {
table_printer.AddColumn("Pending (Expected)", false);
@@ -392,7 +392,6 @@ string Coordinator::FilterDebugString() {
table_printer.AddColumn("Completed", false);
}
table_printer.AddColumn("Enabled", false);
- lock_guard<SpinLock> l(filter_lock_);
for (FilterRoutingTable::value_type& v: filter_routing_table_) {
vector<string> row;
const FilterState& state = v.second;
@@ -449,7 +448,7 @@ Status Coordinator::SetNonErrorTerminalState(const ExecState state) {
{
lock_guard<SpinLock> l(exec_state_lock_);
// May have already entered a terminal state, in which case nothing to do.
- if (exec_state_.Load() != ExecState::EXECUTING) return exec_status_;
+ if (!IsExecuting()) return exec_status_;
DCHECK(exec_status_.ok()) << exec_status_;
exec_state_.Store(state);
if (state == ExecState::CANCELLED) exec_status_ = Status::CANCELLED;
@@ -709,7 +708,7 @@ Status Coordinator::UpdateBackendExecStatus(const TReportExecStatusParams& param
}
// If query execution has terminated, return a cancelled status to force the fragment
// instance to stop executing.
- return exec_state_.Load() == ExecState::EXECUTING ? Status::OK() : Status::CANCELLED;
+ return IsExecuting() ? Status::OK() : Status::CANCELLED;
}
// TODO: add histogram/percentile
@@ -750,16 +749,14 @@ string Coordinator::GetErrorLog() {
}
void Coordinator::ReleaseExecResources() {
+ lock_guard<shared_mutex> lock(filter_lock_);
if (filter_routing_table_.size() > 0) {
query_profile_->AddInfoString("Final filter table", FilterDebugString());
}
- {
- lock_guard<SpinLock> l(filter_lock_);
- for (auto& filter : filter_routing_table_) {
- FilterState* state = &filter.second;
- state->Disable(filter_mem_tracker_);
- }
+ for (auto& filter : filter_routing_table_) {
+ FilterState* state = &filter.second;
+ state->Disable(filter_mem_tracker_);
}
// This may be NULL while executing UDFs.
if (filter_mem_tracker_ != nullptr) filter_mem_tracker_->Close();
@@ -779,6 +776,7 @@ void Coordinator::ReleaseAdmissionControlResources() {
}
void Coordinator::UpdateFilter(const TUpdateFilterParams& params) {
+ shared_lock<shared_mutex> lock(filter_lock_);
DCHECK_NE(filter_mode_, TRuntimeFilterMode::OFF)
<< "UpdateFilter() called although runtime filters are disabled";
DCHECK(exec_rpcs_complete_barrier_.get() != nullptr)
@@ -791,7 +789,12 @@ void Coordinator::UpdateFilter(const TUpdateFilterParams& params) {
TPublishFilterParams rpc_params;
unordered_set<int> target_fragment_idxs;
{
- lock_guard<SpinLock> l(filter_lock_);
+ lock_guard<SpinLock> l(filter_update_lock_);
+ if (!IsExecuting()) {
+ LOG(INFO) << "Filter update received for non-executing query with id: "
+ << query_id();
+ return;
+ }
FilterRoutingTable::iterator it = filter_routing_table_.find(params.filter_id);
if (it == filter_routing_table_.end()) {
LOG(INFO) << "Could not find filter with id: " << params.filter_id;
@@ -834,9 +837,7 @@ void Coordinator::UpdateFilter(const TUpdateFilterParams& params) {
if (state->is_bloom_filter()) {
// Assign outgoing bloom filter.
TBloomFilter& aggregated_filter = state->bloom_filter();
- filter_mem_tracker_->Release(aggregated_filter.directory.size());
- // TODO: Track memory used by 'rpc_params'.
swap(rpc_params.bloom_filter, aggregated_filter);
DCHECK(rpc_params.bloom_filter.always_false || rpc_params.bloom_filter.always_true
|| !rpc_params.bloom_filter.directory.empty());
@@ -858,10 +859,19 @@ void Coordinator::UpdateFilter(const TUpdateFilterParams& params) {
// Waited for exec_rpcs_complete_barrier_ so backend_states_ is valid.
for (BackendState* bs: backend_states_) {
for (int fragment_idx: target_fragment_idxs) {
+ if (!IsExecuting()) goto cleanup;
rpc_params.__set_dst_fragment_idx(fragment_idx);
bs->PublishFilter(rpc_params);
}
}
+
+cleanup:
+ // For bloom filters, the memory used in the filter_routing_table_ is transfered to
+ // rpc_params. Hence the Release() function on the filter_mem_tracker_ is called
+ // here to ensure that the MemTracker is updated after the memory is actually freed.
+ if (rpc_params.__isset.bloom_filter) {
+ filter_mem_tracker_->Release(rpc_params.bloom_filter.directory.size());
+ }
}
void Coordinator::FilterState::ApplyUpdate(const TUpdateFilterParams& params,
@@ -976,3 +986,8 @@ const TFinalizeParams* Coordinator::finalize_params() const {
return schedule_.request().__isset.finalize_params
? &schedule_.request().finalize_params : nullptr;
}
+
+bool Coordinator::IsExecuting() {
+ ExecState current_state = exec_state_.Load();
+ return current_state == ExecState::EXECUTING;
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/2ced56c0/be/src/runtime/coordinator.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h
index a0dce35..0b0312c 100644
--- a/be/src/runtime/coordinator.h
+++ b/be/src/runtime/coordinator.h
@@ -21,6 +21,7 @@
#include <string>
#include <vector>
#include <boost/scoped_ptr.hpp>
+#include <boost/thread/shared_mutex.hpp>
#include <boost/unordered_map.hpp>
#include <rapidjson/document.h>
@@ -89,7 +90,8 @@ class QueryState;
///
/// Lock ordering: (lower-numbered acquired before higher-numbered)
/// 1. wait_lock_
-/// 2. exec_state_lock_, backend_states_init_lock_, filter_lock_,
+/// 2. filter_lock_
+/// 3. exec_state_lock_, backend_states_init_lock_, filter_update_lock_,
/// ExecSummary::lock (leafs)
///
/// TODO: move into separate subdirectory and move nested classes into separate files
@@ -296,8 +298,16 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// - ERROR: error status
Status exec_status_;
+ /// Synchronizes updates to the filter_routing_table_.
+ SpinLock filter_update_lock_;
+
/// Protects filter_routing_table_.
- SpinLock filter_lock_;
+ /// Usage pattern:
+ /// 1. To update filter_routing_table_: Acquire shared access on filter_lock_ and
+ /// upgrade to exclusive access by subsequently acquiring filter_update_lock_.
+ /// 2. To read, initialize/destroy filter_routing_table: Directly acquire exclusive
+ /// access on filter_lock_.
+ boost::shared_mutex filter_lock_;
/// Map from filter ID to filter.
typedef boost::unordered_map<int32_t, FilterState> FilterRoutingTable;
@@ -319,6 +329,7 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
const TUniqueId& query_id() const;
/// Returns a pretty-printed table of the current filter state.
+ /// Caller must have exclusive access to filter_lock_.
std::string FilterDebugString();
/// Called when the query is done executing due to reaching EOS or client
@@ -445,6 +456,9 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
///
/// The ExecState state-machine ensures this is called exactly once.
void ReleaseAdmissionControlResources();
+
+ /// Checks the exec_state_ of the query and returns true if the query is executing.
+ bool IsExecuting();
};
}
[7/9] impala git commit: IMPALA-7411,
IMPALA-7414. Fix failing tests on local filesystem
Posted by to...@apache.org.
IMPALA-7411, IMPALA-7414. Fix failing tests on local filesystem
The new test added by IMPALA-7308 failed on local-filesystem builds
because the warehouse path is not directly at /test-warehouse. This
fix prefixes the path appropriately with $FILESYSTEM_PREFIX
Additionally, the fix for IMPALA-5542 made a similar mistake
constructing a path on the Python side of a test case. Fixed by using
the get_fs_path function.
Change-Id: I6922e24a74576d0d000e8e2645a235868583c1e1
Reviewed-on: http://gerrit.cloudera.org:8080/11164
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/d958b477
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/d958b477
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/d958b477
Branch: refs/heads/master
Commit: d958b4779ca3f6403780bc700e4f88a2c18589d7
Parents: 2868bf5
Author: Todd Lipcon <to...@apache.org>
Authored: Wed Aug 8 10:25:12 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Thu Aug 9 19:18:08 2018 +0000
----------------------------------------------------------------------
.../queries/QueryTest/incompatible_avro_partition.test | 3 ++-
tests/query_test/test_scanners.py | 3 ++-
2 files changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/d958b477/testdata/workloads/functional-query/queries/QueryTest/incompatible_avro_partition.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/incompatible_avro_partition.test b/testdata/workloads/functional-query/queries/QueryTest/incompatible_avro_partition.test
index 9ca0df7..23f7abc 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/incompatible_avro_partition.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/incompatible_avro_partition.test
@@ -47,7 +47,8 @@ int, boolean, tinyint, smallint, int, bigint, float, double, string, char, strin
====
---- QUERY
# Add incompatible data in the avro partition.
-alter table mixed partition (part = 2) set location '/test-warehouse/alltypes_avro/year=2009/month=1';
+alter table mixed partition (part = 2)
+ set location '$FILESYSTEM_PREFIX/test-warehouse/alltypes_avro/year=2009/month=1';
refresh mixed;
====
---- QUERY
http://git-wip-us.apache.org/repos/asf/impala/blob/d958b477/tests/query_test/test_scanners.py
----------------------------------------------------------------------
diff --git a/tests/query_test/test_scanners.py b/tests/query_test/test_scanners.py
index afbac0f..2e61d20 100644
--- a/tests/query_test/test_scanners.py
+++ b/tests/query_test/test_scanners.py
@@ -295,7 +295,8 @@ class TestParquet(ImpalaTestSuite):
local_file = os.path.join(os.environ['IMPALA_HOME'],
'testdata/data/%s' % filename)
assert os.path.isfile(local_file)
- hdfs_file = '/test-warehouse/{0}.db/{1}'.format(unique_database, filename)
+ hdfs_file = get_fs_path('/test-warehouse/{0}.db/{1}'.format(
+ unique_database, filename))
check_call(['hdfs', 'dfs', '-copyFromLocal', '-f', local_file, hdfs_file])
qualified_table_name = '%s.%s' % (unique_database, table_name)
[8/9] impala git commit: IMPALA-7399: Add script in lib/python to
generate junit XML.
Posted by to...@apache.org.
IMPALA-7399: Add script in lib/python to generate junit XML.
This patch adds a script to generate junit XML reports for arbitrary
build steps. It's also being used to seed the creation of an internal
python library for Impala development that can be pip installed into
a development environment.
Change-Id: If6024d74075ea69b8ee20d1fc3cc9c1ff821ba5b
Reviewed-on: http://gerrit.cloudera.org:8080/11128
Reviewed-by: David Knupp <dk...@cloudera.com>
Tested-by: David Knupp <dk...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/8f9f91f3
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/8f9f91f3
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/8f9f91f3
Branch: refs/heads/master
Commit: 8f9f91f38bebba3c69fc15cc522840941b871658
Parents: d958b47
Author: David Knupp <dk...@cloudera.com>
Authored: Fri Aug 3 14:44:06 2018 -0700
Committer: David Knupp <dk...@cloudera.com>
Committed: Thu Aug 9 20:53:48 2018 +0000
----------------------------------------------------------------------
bin/rat_exclude_files.txt | 2 +
lib/python/impala_py_lib/__init__.py | 0
lib/python/impala_py_lib/jenkins/__init__.py | 0
.../impala_py_lib/jenkins/generate_junitxml.py | 161 +++++++++++++++++++
lib/python/requirements.txt | 20 +++
lib/python/setup.py | 69 ++++++++
6 files changed, 252 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/8f9f91f3/bin/rat_exclude_files.txt
----------------------------------------------------------------------
diff --git a/bin/rat_exclude_files.txt b/bin/rat_exclude_files.txt
index 170dd24..980cfba 100644
--- a/bin/rat_exclude_files.txt
+++ b/bin/rat_exclude_files.txt
@@ -19,6 +19,8 @@ testdata/__init__.py
tests/__init__.py
bin/diagnostics/__init__.py
www/index.html
+lib/python/impala_py_lib/__init__.py
+lib/python/impala_py_lib/jenkins/__init__.py
# See $IMPALA_HOME/LICENSE.txt
be/src/gutil/*
http://git-wip-us.apache.org/repos/asf/impala/blob/8f9f91f3/lib/python/impala_py_lib/__init__.py
----------------------------------------------------------------------
diff --git a/lib/python/impala_py_lib/__init__.py b/lib/python/impala_py_lib/__init__.py
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/impala/blob/8f9f91f3/lib/python/impala_py_lib/jenkins/__init__.py
----------------------------------------------------------------------
diff --git a/lib/python/impala_py_lib/jenkins/__init__.py b/lib/python/impala_py_lib/jenkins/__init__.py
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/impala/blob/8f9f91f3/lib/python/impala_py_lib/jenkins/generate_junitxml.py
----------------------------------------------------------------------
diff --git a/lib/python/impala_py_lib/jenkins/generate_junitxml.py b/lib/python/impala_py_lib/jenkins/generate_junitxml.py
new file mode 100644
index 0000000..ff93aa4
--- /dev/null
+++ b/lib/python/impala_py_lib/jenkins/generate_junitxml.py
@@ -0,0 +1,161 @@
+#!/usr/bin/env python
+# 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.
+
+"""
+A script for generating arbitrary junit XML reports while building Impala.
+These files will be consumed by jenkins.impala.io to generate reports for
+easier triaging of build and setup errors.
+"""
+import argparse
+import errno
+import os
+import pytz
+import textwrap
+
+from datetime import datetime as dt
+from junit_xml import TestSuite, TestCase
+
+IMPALA_HOME = os.getenv('IMPALA_HOME', '.')
+
+
+def get_xml_content(file_or_string=None):
+ """
+ Derive content for the XML report.
+
+ Args:
+ file_or_string: a path to a file, or a plain string
+
+ If the supplied parameter is the path to a file, the contents will be inserted
+ into the XML report. If the parameter is just plain string, use that as the
+ content for the report.
+ """
+ if file_or_string is None:
+ content = ''
+ elif os.path.exists(file_or_string):
+ with open(file_or_string, 'r') as f:
+ content = f.read()
+ else:
+ content = file_or_string
+ return content
+
+
+def generate_xml_file(testsuite, junitxml_logdir='.'):
+ """
+ Create a timestamped XML report file.
+
+ Args:
+ testsuite: junit_xml.TestSuite object
+ junitxml_logdir: path to directory where the file will be created
+
+ Return:
+ junit_log_file: path to the generated file
+ """
+ ts_string = testsuite.timestamp.strftime('%Y%m%d_%H_%M_%S')
+ junit_log_file = os.path.join(junitxml_logdir,
+ '{}.{}.xml'.format(testsuite.name, ts_string))
+
+ with open(junit_log_file, 'w') as f:
+ TestSuite.to_file(f, [testsuite], prettyprint=True)
+
+ return junit_log_file
+
+
+def get_options():
+ """Parse and return command line options."""
+ parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+
+ # Required options
+ parser.add_argument("--phase",
+ default="buildall",
+ help="General build phase or script.")
+ parser.add_argument("--step",
+ required=True,
+ help=textwrap.dedent(
+ """Specific build step or child script being run.
+ Each step must be unique for the given build phase.""")
+ )
+ parser.add_argument("-t", "--time",
+ type=float,
+ default=0,
+ help="If known, the elapsed time in seconds for this step.")
+ parser.add_argument("--stdout",
+ help=textwrap.dedent(
+ """Standard output to include in the XML report. Can be
+ either a string or the path to a file..""")
+ )
+ parser.add_argument("--stderr",
+ help=textwrap.dedent(
+ """Standard error to include in the XML report. Can be
+ either a string or the path to a file.""")
+ )
+ parser.add_argument("--error",
+ help=textwrap.dedent(
+ """If specified, the XML report will mark this as an error.
+ This should be a brief explanation for the error.""")
+ )
+
+ return parser.parse_args()
+
+
+def main():
+ """
+ Create a "testcase" for each invocation of the script, and output the results
+ of the test case to an XML file within $IMPALA_HOME/logs/extra_junit_xml_logs.
+ The log file name will use "phase" and "step" values provided on the command
+ line to structure the report. The XML report filename will follow the form:
+
+ junitxml_logger.<phase>.<step>.<time_stamp>.xml
+
+ Phase can be repeated in a given test run, but the step leaf node, which is
+ equivalent to a "test case", must be unique within each phase.
+ """
+ junitxml_logdir = os.path.join(IMPALA_HOME, 'logs', 'extra_junit_xml_logs')
+
+ # The equivalent of mkdir -p
+ try:
+ os.makedirs(junitxml_logdir)
+ except OSError as e:
+ if e.errno == errno.EEXIST and os.path.isdir(junitxml_logdir):
+ pass
+ else:
+ raise
+
+ options = get_options()
+ root_name, _ = os.path.splitext(os.path.basename(__file__))
+
+ tc = TestCase(classname='{}.{}'.format(root_name, options.phase),
+ name=options.step,
+ elapsed_sec=options.time,
+ stdout=get_xml_content(options.stdout),
+ stderr=get_xml_content(options.stderr))
+
+ # Specifying an error message for any step causes the buid to be marked as invalid.
+ if options.error:
+ tc.add_error_info(get_xml_content(options.error))
+ assert tc.is_error()
+
+ testsuite = TestSuite(name='{}.{}.{}'.format(root_name, options.phase, options.step),
+ timestamp=dt.utcnow().replace(tzinfo=pytz.UTC),
+ test_cases=[tc])
+
+ xml_report = generate_xml_file(testsuite, junitxml_logdir)
+ print("Generated: {}".format(xml_report))
+
+
+if "__main__" == __name__:
+ main()
http://git-wip-us.apache.org/repos/asf/impala/blob/8f9f91f3/lib/python/requirements.txt
----------------------------------------------------------------------
diff --git a/lib/python/requirements.txt b/lib/python/requirements.txt
new file mode 100644
index 0000000..1995eb0
--- /dev/null
+++ b/lib/python/requirements.txt
@@ -0,0 +1,20 @@
+# 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.
+
+junit-xml==1.8
+pytz==2018.5
+six==1.11.0
http://git-wip-us.apache.org/repos/asf/impala/blob/8f9f91f3/lib/python/setup.py
----------------------------------------------------------------------
diff --git a/lib/python/setup.py b/lib/python/setup.py
new file mode 100644
index 0000000..c2b3cec
--- /dev/null
+++ b/lib/python/setup.py
@@ -0,0 +1,69 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+"""This library package contains Impala jenkins-related utilities.
+The tooling here is intended for Impala testing and is not installed
+as part of production Impala clusters.
+"""
+
+from __future__ import absolute_import
+
+from setuptools import find_packages
+
+try:
+ from setuptools import setup
+except ImportError:
+ from distutils.core import setup
+
+
+def parse_requirements(requirements_file='requirements.txt'):
+ """
+ Parse requirements from the requirements file, stripping comments.
+
+ Args:
+ requirements_file: path to a requirements file
+
+ Returns:
+ a list of python packages
+ """
+ lines = []
+ with open(requirements_file) as reqs:
+ for _ in reqs:
+ line = _.split('#')[0]
+ if line.strip():
+ lines.append(line)
+ return lines
+
+
+setup(
+ name='impala_py_lib',
+ version='0.0.1',
+ author_email='dev@impala.apache.org',
+ description='Internal python libraries and utilities for Impala development',
+ packages=find_packages(),
+ include_package_data=True,
+ install_requires=parse_requirements(),
+ entry_points={
+ 'console_scripts': [
+ 'generate-junitxml = impala_py_lib.jenkins.generate_junitxml:main'
+ ]
+ }
+)
[4/9] impala git commit: IMPALA-7361: Fix flakiness in
test_heterogeneous_proc_mem_limit
Posted by to...@apache.org.
IMPALA-7361: Fix flakiness in test_heterogeneous_proc_mem_limit
This patch fixes flakiness in test_heterogeneous_proc_mem_limit wherein
one of the fragments on a host from a previous query lingers and holds
on to resources which causes the next query to fail the test since it
expects the cluster to be idle at that point.
Change-Id: I3e5c9b0c6a99d7157640c02e6b3c808b4ae9e73c
Reviewed-on: http://gerrit.cloudera.org:8080/11166
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/5cb956e3
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/5cb956e3
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/5cb956e3
Branch: refs/heads/master
Commit: 5cb956e3fd86ba6f9b49a273737f46467b4a1936
Parents: 2ced56c
Author: Bikramjeet Vig <bi...@cloudera.com>
Authored: Wed Aug 8 11:19:15 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Thu Aug 9 03:04:33 2018 +0000
----------------------------------------------------------------------
tests/custom_cluster/test_admission_controller.py | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/5cb956e3/tests/custom_cluster/test_admission_controller.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py
index 32fd971..4b8974a 100644
--- a/tests/custom_cluster/test_admission_controller.py
+++ b/tests/custom_cluster/test_admission_controller.py
@@ -500,6 +500,9 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
"node is greater than process mem limit 2.00 GB of \S+", str(e)), str(e)
# Exercise queuing checks in admission controller.
try:
+ # Wait for previous queries to finish to avoid flakiness.
+ for impalad in self.cluster.impalads:
+ impalad.service.wait_for_metric_value("impala-server.num-fragments-in-flight", 0)
impalad_with_2g_mem = self.cluster.impalads[2].service.create_beeswax_client()
impalad_with_2g_mem.set_configuration_option('mem_limit', '1G')
impalad_with_2g_mem.execute_async("select sleep(1000)")