You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by bo...@apache.org on 2019/02/27 10:28:17 UTC
[impala] branch master updated (b74d632 -> adced30)
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from b74d632 IMPALA-8250: Fix a handful of JNI usage errors.
new 46e0339 IMPALA-7645: Add a query option to set the default table file format
new 8b143f5 IMPALA-8252: Only write node_metadata if it's set
new 360b23b IMPALA-7975 : Improve supportability of the automatic invalidate feature
new adced30 IMPALA-8223: remove {mem-pool,hash-table}.total-bytes
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/catalog/catalog-server.cc | 30 ++++-
be/src/catalog/catalog-server.h | 4 +
be/src/catalog/catalog.cc | 6 +
be/src/catalog/catalog.h | 5 +
be/src/catalog/catalogd-main.cc | 4 +-
be/src/exec/hash-table.cc | 4 -
be/src/exec/hdfs-orc-scanner.cc | 3 -
be/src/runtime/mem-pool.cc | 9 --
be/src/service/query-options-test.cc | 2 +
be/src/service/query-options.cc | 22 ++++
be/src/service/query-options.h | 5 +-
be/src/util/CMakeLists.txt | 6 +-
be/src/util/event-metrics.cc | 108 ++++++++++++++++
be/src/util/event-metrics.h | 86 +++++++++++++
be/src/util/impalad-metrics.cc | 12 --
be/src/util/impalad-metrics.h | 8 --
be/src/util/runtime-profile-test.cc | 17 +++
be/src/util/runtime-profile.cc | 6 +-
common/thrift/ImpalaInternalService.thrift | 4 +
common/thrift/ImpalaService.thrift | 3 +
common/thrift/JniCatalog.thrift | 38 ++++++
common/thrift/metrics.json | 102 ++++++++++++---
fe/src/main/cup/sql-parser.cup | 5 +-
.../java/org/apache/impala/analysis/TableDef.java | 18 ++-
.../impala/catalog/CatalogServiceCatalog.java | 26 +++-
.../catalog/events/ExternalEventsProcessor.java | 14 ++
.../impala/catalog/events/MetastoreEvents.java | 114 ++++++++--------
.../catalog/events/MetastoreEventsProcessor.java | 143 ++++++++++++++++++---
.../impala/catalog/events/NoOpEventProcessor.java | 22 ++++
.../java/org/apache/impala/common/Metrics.java | 62 +++++++--
.../java/org/apache/impala/service/JniCatalog.java | 6 +
.../events/MetastoreEventsProcessorTest.java | 94 +++++++++++++-
.../functional-query/queries/QueryTest/set.test | 8 ++
tests/metadata/test_ddl.py | 34 +++++
tests/verifiers/metric_verifier.py | 2 -
www/{table_metrics.tmpl => events.tmpl} | 4 +-
36 files changed, 869 insertions(+), 167 deletions(-)
create mode 100644 be/src/util/event-metrics.cc
create mode 100644 be/src/util/event-metrics.h
copy www/{table_metrics.tmpl => events.tmpl} (90%)
[impala] 04/04: IMPALA-8223: remove {mem-pool,
hash-table}.total-bytes
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit adced30896ade0f71e64cc2d5bad1cd9f43119b7
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Tue Feb 19 14:55:40 2019 -0800
IMPALA-8223: remove {mem-pool,hash-table}.total-bytes
These metrics have some issues (see the JIRA) and are not very relevant
in a world where most of the memory comes from the BufferPool
Change-Id: I61c8ca1ddf20676252eae958bc2d293e0ec67ecf
Reviewed-on: http://gerrit.cloudera.org:8080/12528
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/hash-table.cc | 4 ----
be/src/exec/hdfs-orc-scanner.cc | 3 ---
be/src/runtime/mem-pool.cc | 9 ---------
be/src/util/impalad-metrics.cc | 12 ------------
be/src/util/impalad-metrics.h | 8 --------
common/thrift/metrics.json | 20 --------------------
tests/verifiers/metric_verifier.py | 2 --
7 files changed, 58 deletions(-)
diff --git a/be/src/exec/hash-table.cc b/be/src/exec/hash-table.cc
index 576deca..58a9212 100644
--- a/be/src/exec/hash-table.cc
+++ b/be/src/exec/hash-table.cc
@@ -425,9 +425,6 @@ void HashTable::Close() {
if ((num_buckets_ > LARGE_HT) || (num_probes_ > HEAVILY_USED)) VLOG(2) << PrintStats();
for (auto& data_page : data_pages_) allocator_->Free(move(data_page));
data_pages_.clear();
- if (ImpaladMetrics::HASH_TABLE_TOTAL_BYTES != NULL) {
- ImpaladMetrics::HASH_TABLE_TOTAL_BYTES->Increment(-total_data_page_size_);
- }
if (bucket_allocation_ != nullptr) allocator_->Free(move(bucket_allocation_));
}
@@ -505,7 +502,6 @@ bool HashTable::GrowNodeArray(Status* status) {
if (!status->ok() || allocation == nullptr) return false;
next_node_ = reinterpret_cast<DuplicateNode*>(allocation->data());
data_pages_.push_back(move(allocation));
- ImpaladMetrics::HASH_TABLE_TOTAL_BYTES->Increment(DATA_PAGE_SIZE);
node_remaining_current_page_ = DATA_PAGE_SIZE / sizeof(DuplicateNode);
total_data_page_size_ += DATA_PAGE_SIZE;
return true;
diff --git a/be/src/exec/hdfs-orc-scanner.cc b/be/src/exec/hdfs-orc-scanner.cc
index a0083bf..a9ef594 100644
--- a/be/src/exec/hdfs-orc-scanner.cc
+++ b/be/src/exec/hdfs-orc-scanner.cc
@@ -67,7 +67,6 @@ void HdfsOrcScanner::OrcMemPool::FreeAll() {
}
mem_tracker_->Release(total_bytes_released);
chunk_sizes_.clear();
- ImpaladMetrics::MEM_POOL_TOTAL_BYTES->Increment(-total_bytes_released);
}
// orc-reader will not check the malloc result. We throw an exception if we can't
@@ -83,7 +82,6 @@ char* HdfsOrcScanner::OrcMemPool::malloc(uint64_t size) {
throw ResourceError(Status(TErrorCode::MEM_ALLOC_FAILED, size));
}
chunk_sizes_[addr] = size;
- ImpaladMetrics::MEM_POOL_TOTAL_BYTES->Increment(size);
return addr;
}
@@ -93,7 +91,6 @@ void HdfsOrcScanner::OrcMemPool::free(char* p) {
std::free(p);
int64_t size = chunk_sizes_[p];
mem_tracker_->Release(size);
- ImpaladMetrics::MEM_POOL_TOTAL_BYTES->Increment(-size);
chunk_sizes_.erase(p);
}
diff --git a/be/src/runtime/mem-pool.cc b/be/src/runtime/mem-pool.cc
index c0aa438..da8e865 100644
--- a/be/src/runtime/mem-pool.cc
+++ b/be/src/runtime/mem-pool.cc
@@ -52,9 +52,6 @@ MemPool::ChunkInfo::ChunkInfo(int64_t size, uint8_t* buf)
: data(buf),
size(size),
allocated_bytes(0) {
- if (ImpaladMetrics::MEM_POOL_TOTAL_BYTES != NULL) {
- ImpaladMetrics::MEM_POOL_TOTAL_BYTES->Increment(size);
- }
}
MemPool::~MemPool() {
@@ -65,9 +62,6 @@ MemPool::~MemPool() {
}
DCHECK(chunks_.empty()) << "Must call FreeAll() or AcquireData() for this pool";
- if (ImpaladMetrics::MEM_POOL_TOTAL_BYTES != NULL) {
- ImpaladMetrics::MEM_POOL_TOTAL_BYTES->Increment(-total_bytes_released);
- }
DCHECK_EQ(zero_length_region_, MEM_POOL_POISON);
}
@@ -96,9 +90,6 @@ void MemPool::FreeAll() {
total_reserved_bytes_ = 0;
mem_tracker_->Release(total_bytes_released);
- if (ImpaladMetrics::MEM_POOL_TOTAL_BYTES != NULL) {
- ImpaladMetrics::MEM_POOL_TOTAL_BYTES->Increment(-total_bytes_released);
- }
}
bool MemPool::FindChunk(int64_t min_size, bool check_limits) noexcept {
diff --git a/be/src/util/impalad-metrics.cc b/be/src/util/impalad-metrics.cc
index e4386ca..6db82e0 100644
--- a/be/src/util/impalad-metrics.cc
+++ b/be/src/util/impalad-metrics.cc
@@ -46,10 +46,6 @@ const char* ImpaladMetricKeys::TOTAL_SCAN_RANGES_PROCESSED =
"impala-server.scan-ranges.total";
const char* ImpaladMetricKeys::NUM_SCAN_RANGES_MISSING_VOLUME_ID =
"impala-server.scan-ranges.num-missing-volume-id";
-const char* ImpaladMetricKeys::MEM_POOL_TOTAL_BYTES =
- "impala-server.mem-pool.total-bytes";
-const char* ImpaladMetricKeys::HASH_TABLE_TOTAL_BYTES =
- "impala-server.hash-table.total-bytes";
const char* ImpaladMetricKeys::IO_MGR_NUM_OPEN_FILES =
"impala-server.io-mgr.num-open-files";
const char* ImpaladMetricKeys::IO_MGR_BYTES_READ =
@@ -131,7 +127,6 @@ const char* ImpaladMetricKeys::HEDGED_READ_OPS_WIN =
// These are created by impala-server during startup.
// =======
// Counters
-IntGauge* ImpaladMetrics::HASH_TABLE_TOTAL_BYTES = NULL;
IntCounter* ImpaladMetrics::BACKEND_NUM_QUERIES_EXECUTED = NULL;
IntGauge* ImpaladMetrics::BACKEND_NUM_QUERIES_EXECUTING = NULL;
IntCounter* ImpaladMetrics::IMPALA_SERVER_NUM_QUERIES = NULL;
@@ -174,7 +169,6 @@ IntGauge* ImpaladMetrics::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING = NULL;
IntGauge* ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_HIT_COUNT = NULL;
IntGauge* ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_MISS_COUNT = NULL;
IntGauge* ImpaladMetrics::IO_MGR_TOTAL_BYTES = NULL;
-IntGauge* ImpaladMetrics::MEM_POOL_TOTAL_BYTES = NULL;
IntGauge* ImpaladMetrics::NUM_FILES_OPEN_FOR_INSERT = NULL;
IntGauge* ImpaladMetrics::NUM_QUERIES_REGISTERED = NULL;
IntGauge* ImpaladMetrics::RESULTSET_CACHE_TOTAL_NUM_ROWS = NULL;
@@ -280,12 +274,6 @@ void ImpaladMetrics::CreateMetrics(MetricGroup* m) {
NUM_RANGES_MISSING_VOLUME_ID = m->AddCounter(
ImpaladMetricKeys::NUM_SCAN_RANGES_MISSING_VOLUME_ID, 0);
- // Initialize memory usage metrics
- MEM_POOL_TOTAL_BYTES = m->AddGauge(
- ImpaladMetricKeys::MEM_POOL_TOTAL_BYTES, 0);
- HASH_TABLE_TOTAL_BYTES = m->AddGauge(
- ImpaladMetricKeys::HASH_TABLE_TOTAL_BYTES, 0);
-
// Initialize insert metrics
NUM_FILES_OPEN_FOR_INSERT = m->AddGauge(
ImpaladMetricKeys::NUM_FILES_OPEN_FOR_INSERT, 0);
diff --git a/be/src/util/impalad-metrics.h b/be/src/util/impalad-metrics.h
index 05babaa..2a25e86 100644
--- a/be/src/util/impalad-metrics.h
+++ b/be/src/util/impalad-metrics.h
@@ -64,12 +64,6 @@ class ImpaladMetricKeys {
/// Number of scan ranges with missing volume id metadata
static const char* NUM_SCAN_RANGES_MISSING_VOLUME_ID;
- /// Number of bytes currently in use across all mem pools
- static const char* MEM_POOL_TOTAL_BYTES;
-
- /// Number of bytes currently in use across all hash tables
- static const char* HASH_TABLE_TOTAL_BYTES;
-
/// Number of files currently opened by the io mgr
static const char* IO_MGR_NUM_OPEN_FILES;
@@ -210,7 +204,6 @@ class ImpaladMetricKeys {
class ImpaladMetrics {
public:
// Counters
- static IntGauge* HASH_TABLE_TOTAL_BYTES;
static IntCounter* BACKEND_NUM_QUERIES_EXECUTED;
/// BACKEND_NUM_QUERIES_EXECUTING is used to determine when the backend has quiesced
/// and can be safely shut down without causing query failures. See IMPALA-7931 for
@@ -261,7 +254,6 @@ class ImpaladMetrics {
static IntGauge* IO_MGR_CACHED_FILE_HANDLES_HIT_COUNT;
static IntGauge* IO_MGR_CACHED_FILE_HANDLES_MISS_COUNT;
static IntGauge* IO_MGR_TOTAL_BYTES;
- static IntGauge* MEM_POOL_TOTAL_BYTES;
static IntGauge* NUM_FILES_OPEN_FOR_INSERT;
static IntGauge* NUM_QUERIES_REGISTERED;
static IntGauge* RESULTSET_CACHE_TOTAL_NUM_ROWS;
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index ed34259..b18bc1f 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -360,16 +360,6 @@
"key": "impala-server.backends.client-cache.total-clients"
},
{
- "description": "The current size of all allocated hash tables.",
- "contexts": [
- "IMPALAD"
- ],
- "label": "Hash Tables Size",
- "units": "BYTES",
- "kind": "GAUGE",
- "key": "impala-server.hash-table.total-bytes"
- },
- {
"description": "The total number of bytes read by the IO manager.",
"contexts": [
"IMPALAD"
@@ -460,16 +450,6 @@
"key": "impala-server.io-mgr.total-bytes"
},
{
- "description": "The current size of the memory pool shared by all queries",
- "contexts": [
- "IMPALAD"
- ],
- "label": "Memory Pool Size",
- "units": "BYTES",
- "kind": "GAUGE",
- "key": "impala-server.mem-pool.total-bytes"
- },
- {
"description": "The number of HDFS files currently open for writing.",
"contexts": [
"IMPALAD"
diff --git a/tests/verifiers/metric_verifier.py b/tests/verifiers/metric_verifier.py
index 0c1777b..747c7ee 100644
--- a/tests/verifiers/metric_verifier.py
+++ b/tests/verifiers/metric_verifier.py
@@ -24,9 +24,7 @@ METRIC_LIST = [
# "impala-server.backends.client-cache.clients-in-use", disabled as a
# work-around due to IMPALA-3327.
#"impala-server.backends.client-cache.clients-in-use",
- "impala-server.hash-table.total-bytes",
"impala-server.io-mgr.num-open-files",
- "impala-server.mem-pool.total-bytes",
"impala-server.num-files-open-for-insert",
"impala-server.scan-ranges.num-missing-volume-id",
# Buffer pool pages belong to specific queries. Therefore there should be
[impala] 02/04: IMPALA-8252: Only write node_metadata if it's set
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 8b143f53036657fc74180def4fb8276d57ca179a
Author: Lars Volker <lv...@cloudera.com>
AuthorDate: Tue Feb 26 16:24:06 2019 -0800
IMPALA-8252: Only write node_metadata if it's set
Since IMPALA-1048 we write TRuntimeProfileNode.node_metadata
unconditionally, even when both its fields are unset. This trips up the
Thrift library Java reader code, which expects to find exactly one type
of a union to be set.
This change fixes the issue by checking whether at least one of the
fields is set before marking node_metadata as set.
This change also adds a test to runtime-profile-test.cc to make sure
that the __isset flag is set correctly. Additionally I wrote a profile
and then manually read it with a Java client.
Change-Id: Ice1efcf75daba5c023566eb63d4e921c228c26fd
Reviewed-on: http://gerrit.cloudera.org:8080/12616
Reviewed-by: Lars Volker <lv...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/util/CMakeLists.txt | 2 +-
be/src/util/runtime-profile-test.cc | 17 +++++++++++++++++
be/src/util/runtime-profile.cc | 6 +++++-
3 files changed, 23 insertions(+), 2 deletions(-)
diff --git a/be/src/util/CMakeLists.txt b/be/src/util/CMakeLists.txt
index 0a26372..8acb5ef 100644
--- a/be/src/util/CMakeLists.txt
+++ b/be/src/util/CMakeLists.txt
@@ -187,7 +187,7 @@ ADD_UNIFIED_BE_LSAN_TEST(redactor-config-parser-test ParserTest.*)
ADD_UNIFIED_BE_LSAN_TEST(redactor-test "RedactorTest.*")
ADD_UNIFIED_BE_LSAN_TEST(redactor-unconfigured-test "RedactorUnconfigTest.*")
ADD_UNIFIED_BE_LSAN_TEST(rle-test "BitArray.*:RleTest.*")
-ADD_UNIFIED_BE_LSAN_TEST(runtime-profile-test "CountersTest.*:TimerCounterTest.*:TimeSeriesCounterTest.*:VariousNumbers/TimeSeriesCounterResampleTest.*")
+ADD_UNIFIED_BE_LSAN_TEST(runtime-profile-test "CountersTest.*:TimerCounterTest.*:TimeSeriesCounterTest.*:VariousNumbers/TimeSeriesCounterResampleTest.*:ToThrift.*")
ADD_UNIFIED_BE_LSAN_TEST(string-parser-test "StringToInt.*:StringToIntWithBase.*:StringToFloat.*:StringToBool.*")
ADD_UNIFIED_BE_LSAN_TEST(string-util-test "TruncateDownTest.*:TruncateUpTest.*:CommaSeparatedContainsTest.*")
ADD_UNIFIED_BE_LSAN_TEST(symbols-util-test "SymbolsUtil.*")
diff --git a/be/src/util/runtime-profile-test.cc b/be/src/util/runtime-profile-test.cc
index e07994e..795ef29 100644
--- a/be/src/util/runtime-profile-test.cc
+++ b/be/src/util/runtime-profile-test.cc
@@ -1141,5 +1141,22 @@ INSTANTIATE_TEST_CASE_P(VariousNumbers, TimeSeriesCounterResampleTest,
"120, 123, 126 (Showing 43 of 129 values from Thrift Profile)"})
));
+// Tests that the __isset field for TRuntimeProfileNode.node_metadata is set correctly
+// (IMPALA-8252).
+TEST(ToThrift, NodeMetadataIsSetCorrectly) {
+ ObjectPool pool;
+ RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
+ TRuntimeProfileTree thrift_profile;
+ profile->ToThrift(&thrift_profile);
+ // Profile is empty, expect 0 nodes
+ EXPECT_EQ(thrift_profile.nodes.size(), 1);
+ EXPECT_FALSE(thrift_profile.nodes[0].__isset.node_metadata);
+
+ // Set the plan node ID and make sure that the field is marked correctly
+ profile->SetPlanNodeId(1);
+ profile->ToThrift(&thrift_profile);
+ EXPECT_TRUE(thrift_profile.nodes[0].__isset.node_metadata);
+}
+
} // namespace impala
diff --git a/be/src/util/runtime-profile.cc b/be/src/util/runtime-profile.cc
index ec7417b..8f91b81 100644
--- a/be/src/util/runtime-profile.cc
+++ b/be/src/util/runtime-profile.cc
@@ -937,7 +937,11 @@ void RuntimeProfile::ToThrift(vector<TRuntimeProfileNode>* nodes) const {
// Set the required metadata field to the plan node ID for compatibility with any tools
// that rely on the plan node id being set there.
node.metadata = metadata_.__isset.plan_node_id ? metadata_.plan_node_id : -1;
- node.__set_node_metadata(metadata_);
+ // Thrift requires exactly one field of a union to be set so we only mark node_metadata
+ // as set if that is the case.
+ if (metadata_.__isset.plan_node_id || metadata_.__isset.data_sink_id) {
+ node.__set_node_metadata(metadata_);
+ }
node.indent = true;
CounterMap counter_map;
[impala] 01/04: IMPALA-7645: Add a query option to set the default
table file format
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 46e03392c38299731d937898c4177b7baedc8f7d
Author: Fredy Wijaya <fw...@cloudera.com>
AuthorDate: Sat Feb 23 12:56:10 2019 -0800
IMPALA-7645: Add a query option to set the default table file format
This patch adds a query option "DEFAULT_FILE_FORMAT" to allow setting
the default table file format. Below is the list of supported
"DEFAULT_FILE_FORMAT" query option values:
- TEXT
- RC_FILE
- SEQUENCE_FILE
- AVRO
- PARQUET
- KUDU
- ORC
For backward compatibility, the default table file format remains as
TEXT.
Testing:
- Ran all FE tests
- Added BE test
- Added E2E tests
- Ran test_ddl.py and test_set.py
Change-Id: Ic857c38076d973ad749a41fecd1b470c7881db5e
Reviewed-on: http://gerrit.cloudera.org:8080/12568
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/service/query-options-test.cc | 2 ++
be/src/service/query-options.cc | 22 ++++++++++++++
be/src/service/query-options.h | 5 ++--
common/thrift/ImpalaInternalService.thrift | 4 +++
common/thrift/ImpalaService.thrift | 3 ++
fe/src/main/cup/sql-parser.cup | 5 ++--
.../java/org/apache/impala/analysis/TableDef.java | 18 ++++++++----
.../functional-query/queries/QueryTest/set.test | 8 +++++
tests/metadata/test_ddl.py | 34 ++++++++++++++++++++++
9 files changed, 92 insertions(+), 9 deletions(-)
diff --git a/be/src/service/query-options-test.cc b/be/src/service/query-options-test.cc
index b18b58e..1be2b81 100644
--- a/be/src/service/query-options-test.cc
+++ b/be/src/service/query-options-test.cc
@@ -211,6 +211,8 @@ TEST(QueryOptions, SetEnumOptions) {
(THREE_LEVEL, TWO_LEVEL, TWO_LEVEL_THEN_THREE_LEVEL)), true);
TestEnumCase(options, CASE(compression_codec, THdfsCompression,
(NONE, GZIP, BZIP2, DEFAULT, SNAPPY, SNAPPY_BLOCKED)), false);
+ TestEnumCase(options, CASE(default_file_format, THdfsFileFormat,
+ (TEXT, RC_FILE, SEQUENCE_FILE, AVRO, PARQUET, KUDU, ORC)), true);
#undef CASE
#undef ENTRIES
#undef ENTRY
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index 5720eac..9618d18 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -765,6 +765,28 @@ Status impala::SetQueryOption(const string& key, const string& value,
query_options->__set_num_rows_produced_limit(num_rows_produced_limit);
break;
}
+ case TImpalaQueryOptions::DEFAULT_FILE_FORMAT: {
+ if (iequals(value, "TEXT") || iequals(value, "0")) {
+ query_options->__set_default_file_format(THdfsFileFormat::TEXT);
+ } else if (iequals(value, "RC_FILE") || iequals(value, "1")) {
+ query_options->__set_default_file_format(THdfsFileFormat::RC_FILE);
+ } else if (iequals(value, "SEQUENCE_FILE") || iequals(value, "2")) {
+ query_options->__set_default_file_format(THdfsFileFormat::SEQUENCE_FILE);
+ } else if (iequals(value, "AVRO") || iequals(value, "3")) {
+ query_options->__set_default_file_format(THdfsFileFormat::AVRO);
+ } else if (iequals(value, "PARQUET") || iequals(value, "4")) {
+ query_options->__set_default_file_format(THdfsFileFormat::PARQUET);
+ } else if (iequals(value, "KUDU") || iequals(value, "5")) {
+ query_options->__set_default_file_format(THdfsFileFormat::KUDU);
+ } else if (iequals(value, "ORC") || iequals(value, "6")) {
+ query_options->__set_default_file_format(THdfsFileFormat::ORC);
+ } else {
+ return Status(Substitute("Invalid default_file_format '$0'. Valid values are "
+ "TEXT, RC_FILE, SEQUENCE_FILE, AVRO, PARQUET, KUDU "
+ "and ORC.", value));
+ }
+ break;
+ }
default:
if (IsRemovedQueryOption(key)) {
LOG(WARNING) << "Ignoring attempt to set removed query option '" << key << "'";
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index 7fee4d1..4ead0e5 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -41,7 +41,7 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
// the DCHECK.
#define QUERY_OPTS_TABLE\
DCHECK_EQ(_TImpalaQueryOptions_VALUES_TO_NAMES.size(),\
- TImpalaQueryOptions::PLANNER_TESTCASE_MODE + 1);\
+ TImpalaQueryOptions::DEFAULT_FILE_FORMAT + 1);\
REMOVED_QUERY_OPT_FN(abort_on_default_limit_exceeded, ABORT_ON_DEFAULT_LIMIT_EXCEEDED)\
QUERY_OPT_FN(abort_on_error, ABORT_ON_ERROR, TQueryOptionLevel::REGULAR)\
REMOVED_QUERY_OPT_FN(allow_unsupported_formats, ALLOW_UNSUPPORTED_FORMATS)\
@@ -151,7 +151,8 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
QUERY_OPT_FN(num_rows_produced_limit, NUM_ROWS_PRODUCED_LIMIT,\
TQueryOptionLevel::ADVANCED)\
QUERY_OPT_FN(\
- planner_testcase_mode, PLANNER_TESTCASE_MODE, TQueryOptionLevel::DEVELOPMENT)
+ planner_testcase_mode, PLANNER_TESTCASE_MODE, TQueryOptionLevel::DEVELOPMENT)\
+ QUERY_OPT_FN(default_file_format, DEFAULT_FILE_FORMAT, TQueryOptionLevel::REGULAR)
;
/// Enforce practical limits on some query options to avoid undesired query state.
diff --git a/common/thrift/ImpalaInternalService.thrift b/common/thrift/ImpalaInternalService.thrift
index 5238848..65de009 100644
--- a/common/thrift/ImpalaInternalService.thrift
+++ b/common/thrift/ImpalaInternalService.thrift
@@ -323,6 +323,10 @@ struct TQueryOptions {
// See comment in ImpalaService.thrift
78: optional bool planner_testcase_mode = false;
+
+ // See comment in ImpalaService.thrift.
+ 79: optional CatalogObjects.THdfsFileFormat default_file_format =
+ CatalogObjects.THdfsFileFormat.TEXT;
}
// Impala currently has two types of sessions: Beeswax and HiveServer2
diff --git a/common/thrift/ImpalaService.thrift b/common/thrift/ImpalaService.thrift
index 71fc732..7c46bd8 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -377,6 +377,9 @@ enum TImpalaQueryOptions {
// debugging a testcase. Should not be set in user clusters. If set, a warning
// is emitted in the query runtime profile.
PLANNER_TESTCASE_MODE = 77
+
+ // Specifies the default table file format.
+ DEFAULT_FILE_FORMAT = 78
}
// The summary of a DML statement.
diff --git a/fe/src/main/cup/sql-parser.cup b/fe/src/main/cup/sql-parser.cup
index 9d80ced..3216600 100644
--- a/fe/src/main/cup/sql-parser.cup
+++ b/fe/src/main/cup/sql-parser.cup
@@ -1319,7 +1319,7 @@ create_tbl_stmt ::=
if (tbl_def.isExternal()) {
parser.parseError("external", SqlParserSymbols.KW_EXTERNAL);
}
- tbl_def.setOptions(new TableDef.Options(comment));
+ tbl_def.setOptions(new TableDef.Options(comment, parser.getQueryOptions()));
RESULT = new CreateTableDataSrcStmt(new CreateTableStmt(tbl_def),
data_src_name, init_string);
:}
@@ -1400,8 +1400,9 @@ tbl_options ::=
{:
CreateTableStmt.unescapeProperties(serde_props);
CreateTableStmt.unescapeProperties(tbl_props);
+
RESULT = new TableDef.Options(sort_cols, comment, row_format, serde_props,
- file_format, location, cache_op, tbl_props);
+ file_format, location, cache_op, tbl_props, parser.getQueryOptions());
:}
;
diff --git a/fe/src/main/java/org/apache/impala/analysis/TableDef.java b/fe/src/main/java/org/apache/impala/analysis/TableDef.java
index 0c79083..e9e5399 100644
--- a/fe/src/main/java/org/apache/impala/analysis/TableDef.java
+++ b/fe/src/main/java/org/apache/impala/analysis/TableDef.java
@@ -37,6 +37,7 @@ import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.thrift.TAccessEvent;
import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.THdfsFileFormat;
+import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.util.MetaStoreUtil;
import com.google.common.base.Preconditions;
@@ -126,22 +127,29 @@ class TableDef {
Options(List<String> sortCols, String comment, RowFormat rowFormat,
Map<String, String> serdeProperties, THdfsFileFormat fileFormat, HdfsUri location,
- HdfsCachingOp cachingOp, Map<String, String> tblProperties) {
+ HdfsCachingOp cachingOp, Map<String, String> tblProperties,
+ TQueryOptions queryOptions) {
this.sortCols = sortCols;
this.comment = comment;
this.rowFormat = rowFormat;
Preconditions.checkNotNull(serdeProperties);
this.serdeProperties = serdeProperties;
- this.fileFormat = fileFormat == null ? THdfsFileFormat.TEXT : fileFormat;
+ // The file format passed via STORED AS <file format> has a higher precedence than
+ // the one set in query options.
+ this.fileFormat = (fileFormat != null) ?
+ fileFormat : queryOptions.getDefault_file_format();
this.location = location;
this.cachingOp = cachingOp;
Preconditions.checkNotNull(tblProperties);
this.tblProperties = tblProperties;
}
- public Options(String comment) {
- this(ImmutableList.<String>of(), comment, RowFormat.DEFAULT_ROW_FORMAT,
- new HashMap<>(), THdfsFileFormat.TEXT, null, null, new HashMap<>());
+ public Options(String comment, TQueryOptions queryOptions) {
+ // Passing null to file format so that it uses the file format from the query option
+ // if specified, otherwise it will use the default file format, which is TEXT.
+ this(ImmutableList.of(), comment, RowFormat.DEFAULT_ROW_FORMAT,
+ new HashMap<>(), /*file format*/ null, null, null, new HashMap<>(),
+ queryOptions);
}
}
diff --git a/testdata/workloads/functional-query/queries/QueryTest/set.test b/testdata/workloads/functional-query/queries/QueryTest/set.test
index cabc356..aa74ee1 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/set.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/set.test
@@ -25,6 +25,7 @@ set all;
'PARQUET_FILE_SIZE','0','ADVANCED'
'REQUEST_POOL','','REGULAR'
'SYNC_DDL','0','REGULAR'
+'DEFAULT_FILE_FORMAT','TEXT','REGULAR'
---- TYPES
STRING, STRING, STRING
====
@@ -50,6 +51,7 @@ set all;
'PARQUET_FILE_SIZE','0','ADVANCED'
'REQUEST_POOL','','REGULAR'
'SYNC_DDL','0','REGULAR'
+'DEFAULT_FILE_FORMAT','TEXT','REGULAR'
---- TYPES
STRING, STRING, STRING
====
@@ -75,6 +77,7 @@ set all;
'PARQUET_FILE_SIZE','0','ADVANCED'
'REQUEST_POOL','','REGULAR'
'SYNC_DDL','0','REGULAR'
+'DEFAULT_FILE_FORMAT','TEXT','REGULAR'
---- TYPES
STRING, STRING, STRING
====
@@ -228,3 +231,8 @@ set SCAN_NODE_CODEGEN_THRESHOLD = "foo";
set max_io_buffers="foo";
---- RESULTS
====
+---- QUERY
+set default_file_format=foo;
+---- CATCH
+Invalid default_file_format 'foo'. Valid values are TEXT, RC_FILE, SEQUENCE_FILE, AVRO, PARQUET, KUDU and ORC.
+====
\ No newline at end of file
diff --git a/tests/metadata/test_ddl.py b/tests/metadata/test_ddl.py
index 1bf0e7d..af078d2 100644
--- a/tests/metadata/test_ddl.py
+++ b/tests/metadata/test_ddl.py
@@ -689,6 +689,40 @@ class TestDdlStatements(TestDdlBase):
self.run_test_case('QueryTest/partition-ddl-predicates-hdfs-only', vector,
use_db=unique_database, multiple_impalad=self._use_multiple_impalad(vector))
+ def test_create_table_file_format(self, vector, unique_database):
+ # When default_file_format query option is not specified, the default table file
+ # format is TEXT.
+ text_table = "{0}.text_tbl".format(unique_database)
+ self.execute_query_expect_success(
+ self.client, "create table {0}(i int)".format(text_table))
+ result = self.execute_query_expect_success(
+ self.client, "show create table {0}".format(text_table))
+ assert any("TEXTFILE" in x for x in result.data)
+
+ self.execute_query_expect_failure(
+ self.client, "create table {0}.foobar_tbl".format(unique_database),
+ {"default_file_format": "foobar"})
+
+ parquet_table = "{0}.parquet_tbl".format(unique_database)
+ self.execute_query_expect_success(
+ self.client, "create table {0}(i int)".format(parquet_table),
+ {"default_file_format": "parquet"})
+ result = self.execute_query_expect_success(
+ self.client, "show create table {0}".format(parquet_table))
+ assert any("PARQUET" in x for x in result.data)
+
+ # The table created should still be ORC even though the default_file_format query
+ # option is set to parquet.
+ orc_table = "{0}.orc_tbl".format(unique_database)
+ self.execute_query_expect_success(
+ self.client,
+ "create table {0}(i int) stored as orc".format(orc_table),
+ {"default_file_format": "parquet"})
+ result = self.execute_query_expect_success(
+ self.client, "show create table {0}".format(orc_table))
+ assert any("ORC" in x for x in result.data)
+
+
# IMPALA-2002: Tests repeated adding/dropping of .jar and .so in the lib cache.
class TestLibCache(TestDdlBase):
@classmethod
[impala] 03/04: IMPALA-7975 : Improve supportability of the
automatic invalidate feature
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 360b23bfa4eeb337cce380512f106a692bf3c49b
Author: Vihang Karajgaonkar <vi...@cloudera.com>
AuthorDate: Wed Feb 13 14:58:21 2019 -0800
IMPALA-7975 : Improve supportability of the automatic invalidate feature
This patch adds supportability metrics to make it easier to find the
state of Metastore events processor. It adds the following metrics
1. Total number of events received so far
This is not super useful in itself but is helpful to see if there are
events being received or not. A more usable metric which publishes the
rate of such events received is also published. See details for it
below.
2. 1min, 5min and 15min rate of events received
Exponential moving average of number of events received in last 1min,
5min and 15min. This can be used to determine if there are spikes in
event processor activity during certain hours of the day
3. Total number of events skipped so far
Events can be skipped based on certain flags are table and database
level. This metric keeps a count of total number of events skipped so
far. Useful to make decisions like - If most of the events are being
skipped, may be just turn off the event processing. If most of the
events are not skipped, do we need to add flags on certain databases?
4. Status of events processor
Easy way to determine what is the current state of metastore events
processor. Possible states could be STOPPED, ACTIVE, ERROR,
NEEDS_INVALIDATE
5. Events fetch and process durations in seconds
Average duration to fetch a batch of events and process it.
These metrics are displayed in the metrics page under events
metricsgroup. The refresh logic piggybacks on the catalogservermetrics
refresh thread.
Additionally, it also registers a url handler specific for events at
/events. The /events page provides a more detailed view of the metrics
of the event processor which include (min, max, mean, median) of the
durations and rate metrics for all the counters. In the future this page
can be enhanced to show the top-n tables with most number of invalidates
due to the events, etc.
Patch adds few tests to make sure the value of these event metrics are
as expected.
Change-Id: I23cb3aa866879eca03c64ab881796eaa9caa0337
Reviewed-on: http://gerrit.cloudera.org:8080/12549
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/catalog/catalog-server.cc | 30 ++++-
be/src/catalog/catalog-server.h | 4 +
be/src/catalog/catalog.cc | 6 +
be/src/catalog/catalog.h | 5 +
be/src/catalog/catalogd-main.cc | 4 +-
be/src/util/CMakeLists.txt | 4 +-
be/src/util/event-metrics.cc | 108 ++++++++++++++++
be/src/util/event-metrics.h | 86 +++++++++++++
common/thrift/JniCatalog.thrift | 38 ++++++
common/thrift/metrics.json | 82 +++++++++++-
.../impala/catalog/CatalogServiceCatalog.java | 26 +++-
.../catalog/events/ExternalEventsProcessor.java | 14 ++
.../impala/catalog/events/MetastoreEvents.java | 114 ++++++++--------
.../catalog/events/MetastoreEventsProcessor.java | 143 ++++++++++++++++++---
.../impala/catalog/events/NoOpEventProcessor.java | 22 ++++
.../java/org/apache/impala/common/Metrics.java | 62 +++++++--
.../java/org/apache/impala/service/JniCatalog.java | 6 +
.../events/MetastoreEventsProcessorTest.java | 94 +++++++++++++-
www/events.tmpl | 25 ++++
19 files changed, 776 insertions(+), 97 deletions(-)
diff --git a/be/src/catalog/catalog-server.cc b/be/src/catalog/catalog-server.cc
index 02b2703..da73424 100644
--- a/be/src/catalog/catalog-server.cc
+++ b/be/src/catalog/catalog-server.cc
@@ -22,13 +22,14 @@
#include "catalog/catalog-util.h"
#include "exec/read-write-util.h"
+#include "gen-cpp/CatalogInternalService_types.h"
+#include "gen-cpp/CatalogObjects_types.h"
+#include "gen-cpp/CatalogService_types.h"
#include "statestore/statestore-subscriber.h"
#include "util/debug-util.h"
+#include "util/event-metrics.h"
#include "util/logging-support.h"
#include "util/webserver.h"
-#include "gen-cpp/CatalogInternalService_types.h"
-#include "gen-cpp/CatalogObjects_types.h"
-#include "gen-cpp/CatalogService_types.h"
#include "common/names.h"
@@ -90,6 +91,8 @@ const string CATALOG_OBJECT_WEB_PAGE = "/catalog_object";
const string CATALOG_OBJECT_TEMPLATE = "catalog_object.tmpl";
const string TABLE_METRICS_WEB_PAGE = "/table_metrics";
const string TABLE_METRICS_TEMPLATE = "table_metrics.tmpl";
+const string EVENT_WEB_PAGE = "/events";
+const string EVENT_METRICS_TEMPLATE = "events.tmpl";
const int REFRESH_METRICS_INTERVAL_MS = 1000;
@@ -288,6 +291,9 @@ void CatalogServer::RegisterWebpages(Webserver* webserver) {
webserver->RegisterUrlCallback(TABLE_METRICS_WEB_PAGE, TABLE_METRICS_TEMPLATE,
[this](const auto& args, auto* doc) { this->TableMetricsUrlCallback(args, doc); },
false);
+ webserver->RegisterUrlCallback(EVENT_WEB_PAGE, EVENT_METRICS_TEMPLATE,
+ [this](const auto& args, auto* doc) { this->EventMetricsUrlCallback(args, doc); },
+ false);
RegisterLogLevelCallbacks(webserver, true);
}
@@ -390,6 +396,8 @@ void CatalogServer::UpdateCatalogTopicCallback(
}
partial_fetch_rpc_queue_len_metric_->SetValue(
response.catalog_partial_fetch_rpc_queue_len);
+ TEventProcessorMetrics eventProcessorMetrics = response.event_metrics;
+ MetastoreEventMetrics::refresh(&eventProcessorMetrics);
}
}
@@ -505,6 +513,22 @@ void CatalogServer::GetCatalogUsage(Document* document) {
document->GetAllocator());
}
+void CatalogServer::EventMetricsUrlCallback(
+ const Webserver::ArgumentMap& args, Document* document) {
+ TEventProcessorMetricsSummaryResponse event_processor_summary_response;
+ Status status = catalog_->GetEventProcessorSummary(&event_processor_summary_response);
+ if (!status.ok()) {
+ Value error(status.GetDetail().c_str(), document->GetAllocator());
+ document->AddMember("error", error, document->GetAllocator());
+ return;
+ }
+
+ Value event_processor_summary(
+ event_processor_summary_response.summary.c_str(), document->GetAllocator());
+ document->AddMember(
+ "event_processor_metrics", event_processor_summary, document->GetAllocator());
+}
+
void CatalogServer::CatalogObjectsUrlCallback(const Webserver::ArgumentMap& args,
Document* document) {
Webserver::ArgumentMap::const_iterator object_type_arg = args.find("object_type");
diff --git a/be/src/catalog/catalog-server.h b/be/src/catalog/catalog-server.h
index 76ec2be..4d3cd3e 100644
--- a/be/src/catalog/catalog-server.h
+++ b/be/src/catalog/catalog-server.h
@@ -210,6 +210,10 @@ class CatalogServer {
void TableMetricsUrlCallback(const Webserver::ArgumentMap& args,
rapidjson::Document* document);
+ // url handler for the metastore events page. It calls into JniCatalog to get the latest
+ // metastore event processor metrics and adds it to the document
+ void EventMetricsUrlCallback(
+ const Webserver::ArgumentMap& args, rapidjson::Document* document);
};
}
diff --git a/be/src/catalog/catalog.cc b/be/src/catalog/catalog.cc
index e5bf35c..8efebd9 100644
--- a/be/src/catalog/catalog.cc
+++ b/be/src/catalog/catalog.cc
@@ -68,6 +68,7 @@ Catalog::Catalog() {
{"getCatalogUsage", "()[B", &get_catalog_usage_id_},
{"getCatalogVersion", "()J", &get_catalog_version_id_},
{"getCatalogServerMetrics", "()[B", &get_catalog_server_metrics_},
+ {"getEventProcessorSummary", "()[B", &get_event_processor_summary_},
{"prioritizeLoad", "([B)V", &prioritize_load_id_},
{"getPartitionStats", "([B)[B", &get_partition_stats_id_},
{"updateTableUsage", "([B)V", &update_table_usage_id_},
@@ -159,6 +160,11 @@ Status Catalog::GetCatalogUsage(TGetCatalogUsageResponse* response) {
return JniUtil::CallJniMethod(catalog_, get_catalog_usage_id_, response);
}
+Status Catalog::GetEventProcessorSummary(
+ TEventProcessorMetricsSummaryResponse* response) {
+ return JniUtil::CallJniMethod(catalog_, get_event_processor_summary_, response);
+}
+
Status Catalog::GetCatalogServerMetrics(TGetCatalogServerMetricsResponse* response) {
return JniUtil::CallJniMethod(catalog_, get_catalog_server_metrics_, response);
}
diff --git a/be/src/catalog/catalog.h b/be/src/catalog/catalog.h
index c703bcb..4096732 100644
--- a/be/src/catalog/catalog.h
+++ b/be/src/catalog/catalog.h
@@ -108,6 +108,10 @@ class Catalog {
/// tables as well as the tables with the highest memory requirements.
Status GetCatalogUsage(TGetCatalogUsageResponse* response);
+ /// Returns the metastore event processor summary view. The summary string
+ /// in the response can contain detailed metrics along with status
+ Status GetEventProcessorSummary(TEventProcessorMetricsSummaryResponse* response);
+
/// Gets all functions in the catalog matching the parameters in the given
/// TFunctionsRequest.
Status GetFunctions(const TGetFunctionsRequest& request,
@@ -144,6 +148,7 @@ class Catalog {
jmethodID get_catalog_version_id_; // JniCatalog.getCatalogVersion()
jmethodID get_catalog_usage_id_; // JniCatalog.getCatalogUsage()
jmethodID get_catalog_server_metrics_; // JniCatalog.getCatalogServerMetrics()
+ jmethodID get_event_processor_summary_; // JniCatalog.getEventProcessorMetrics()
jmethodID get_dbs_id_; // JniCatalog.getDbs()
jmethodID get_table_names_id_; // JniCatalog.getTableNames()
jmethodID get_table_metrics_id_; // JniCatalog.getTableMetrics()
diff --git a/be/src/catalog/catalogd-main.cc b/be/src/catalog/catalogd-main.cc
index 250f27a..21fce3c 100644
--- a/be/src/catalog/catalogd-main.cc
+++ b/be/src/catalog/catalogd-main.cc
@@ -30,8 +30,9 @@
#include "service/fe-support.h"
#include "util/common-metrics.h"
#include "util/debug-util.h"
-#include "util/jni-util.h"
#include "util/default-path-handlers.h"
+#include "util/event-metrics.h"
+#include "util/jni-util.h"
#include "util/memory-metrics.h"
#include "util/metrics.h"
#include "util/network-util.h"
@@ -82,6 +83,7 @@ int CatalogdMain(int argc, char** argv) {
metrics->AddProperty<string>("catalog.version", GetVersionString(true));
CommonMetrics::InitCommonMetrics(metrics.get());
+ MetastoreEventMetrics::InitMetastoreEventMetrics(metrics.get());
CatalogServer catalog_server(metrics.get());
ABORT_IF_ERROR(catalog_server.Start());
diff --git a/be/src/util/CMakeLists.txt b/be/src/util/CMakeLists.txt
index 8acb5ef..6744da1 100644
--- a/be/src/util/CMakeLists.txt
+++ b/be/src/util/CMakeLists.txt
@@ -48,6 +48,7 @@ add_library(Util
default-path-handlers.cc
disk-info.cc
error-util.cc
+ event-metrics.cc
filesystem-util.cc
flat_buffer.cc
hdfs-util.cc
@@ -92,8 +93,7 @@ add_library(Util
webserver.cc
zip-util.cc
${MUSTACHE_SRC_DIR}/mustache.cc
- ${MPFIT_SRC_DIR}/mpfit.c
-)
+ ${MPFIT_SRC_DIR}/mpfit.c)
add_dependencies(Util gen-deps gen_ir_descriptions)
add_library(UtilTests
diff --git a/be/src/util/event-metrics.cc b/be/src/util/event-metrics.cc
new file mode 100644
index 0000000..61151c3
--- /dev/null
+++ b/be/src/util/event-metrics.cc
@@ -0,0 +1,108 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "util/event-metrics.h"
+
+DECLARE_int32(hms_event_polling_interval_s);
+
+namespace impala {
+string MetastoreEventMetrics::NUMBER_EVENTS_RECEIVED_METRIC_NAME =
+ "events-processor.events-received";
+string MetastoreEventMetrics::NUMBER_EVENTS_SKIPPED_METRIC_NAME =
+ "events-processor.events-skipped";
+string MetastoreEventMetrics::EVENT_PROCESSOR_STATUS_METRIC_NAME =
+ "events-processor.status";
+string MetastoreEventMetrics::EVENTS_FETCH_DURATION_MEAN_METRIC_NAME =
+ "events-processor.avg-events-fetch-duration";
+string MetastoreEventMetrics::EVENTS_PROCESS_DURATION_MEAN_METRIC_NAME =
+ "events-processor.avg-events-process-duration";
+
+string MetastoreEventMetrics::EVENTS_RECEIVED_1MIN_METRIC_NAME =
+ "events-processor.events-received-1min-rate";
+string MetastoreEventMetrics::EVENTS_RECEIVED_5MIN_METRIC_NAME =
+ "events-processor.events-received-5min-rate";
+string MetastoreEventMetrics::EVENTS_RECEIVED_15MIN_METRIC_NAME =
+ "events-processor.events-received-15min-rate";
+
+IntCounter* MetastoreEventMetrics::NUM_EVENTS_RECEIVED_COUNTER = nullptr;
+IntCounter* MetastoreEventMetrics::NUM_EVENTS_SKIPPED_COUNTER = nullptr;
+
+DoubleGauge* MetastoreEventMetrics::EVENTS_FETCH_DURATION_MEAN = nullptr;
+DoubleGauge* MetastoreEventMetrics::EVENTS_PROCESS_DURATION_MEAN = nullptr;
+
+StringProperty* MetastoreEventMetrics::EVENT_PROCESSOR_STATUS = nullptr;
+
+DoubleGauge* MetastoreEventMetrics::EVENTS_RECEIVED_1MIN_RATE = nullptr;
+DoubleGauge* MetastoreEventMetrics::EVENTS_RECEIVED_5MIN_RATE = nullptr;
+DoubleGauge* MetastoreEventMetrics::EVENTS_RECEIVED_15MIN_RATE = nullptr;
+
+// Initialize all the metrics for the events metric group
+void MetastoreEventMetrics::InitMetastoreEventMetrics(MetricGroup* metric_group) {
+ MetricGroup* event_metrics = metric_group->GetOrCreateChildGroup("events");
+ EVENT_PROCESSOR_STATUS = event_metrics->AddProperty<string>(
+ EVENT_PROCESSOR_STATUS_METRIC_NAME, "Unavailable");
+
+ // if event processing is disabled no need to initialized the other metrics
+ if (FLAGS_hms_event_polling_interval_s <= 0) return;
+
+ NUM_EVENTS_RECEIVED_COUNTER =
+ event_metrics->AddCounter(NUMBER_EVENTS_RECEIVED_METRIC_NAME, 0);
+ NUM_EVENTS_SKIPPED_COUNTER =
+ event_metrics->AddCounter(NUMBER_EVENTS_SKIPPED_METRIC_NAME, 0);
+ EVENTS_FETCH_DURATION_MEAN =
+ event_metrics->AddDoubleGauge(EVENTS_FETCH_DURATION_MEAN_METRIC_NAME, 0.0);
+ EVENTS_PROCESS_DURATION_MEAN =
+ event_metrics->AddDoubleGauge(EVENTS_PROCESS_DURATION_MEAN_METRIC_NAME, 0.0);
+ EVENTS_RECEIVED_1MIN_RATE =
+ event_metrics->AddDoubleGauge(EVENTS_RECEIVED_1MIN_METRIC_NAME, 0.0);
+ EVENTS_RECEIVED_5MIN_RATE =
+ event_metrics->AddDoubleGauge(EVENTS_RECEIVED_5MIN_METRIC_NAME, 0.0);
+ EVENTS_RECEIVED_15MIN_RATE =
+ event_metrics->AddDoubleGauge(EVENTS_RECEIVED_15MIN_METRIC_NAME, 0.0);
+}
+
+void MetastoreEventMetrics::refresh(TEventProcessorMetrics* response) {
+ if (!response) {
+ LOG(ERROR)
+ << "Received a null response when trying to refresh metastore event metrics";
+ return;
+ }
+ EVENT_PROCESSOR_STATUS->SetValue(response->status.c_str());
+
+ if (response->__isset.events_received) {
+ NUM_EVENTS_RECEIVED_COUNTER->SetValue(response->events_received);
+ }
+ if (response->__isset.events_skipped) {
+ NUM_EVENTS_SKIPPED_COUNTER->SetValue(response->events_skipped);
+ }
+ if (response->__isset.events_fetch_duration_mean) {
+ EVENTS_FETCH_DURATION_MEAN->SetValue(response->events_fetch_duration_mean);
+ }
+ if (response->__isset.events_process_duration_mean) {
+ EVENTS_PROCESS_DURATION_MEAN->SetValue(response->events_process_duration_mean);
+ }
+ if (response->__isset.events_received_1min_rate) {
+ EVENTS_RECEIVED_1MIN_RATE->SetValue(response->events_received_1min_rate);
+ }
+ if (response->__isset.events_received_5min_rate) {
+ EVENTS_RECEIVED_5MIN_RATE->SetValue(response->events_received_1min_rate);
+ }
+ if (response->__isset.events_received_15min_rate) {
+ EVENTS_RECEIVED_15MIN_RATE->SetValue(response->events_received_15min_rate);
+ }
+}
+} // namespace impala
\ No newline at end of file
diff --git a/be/src/util/event-metrics.h b/be/src/util/event-metrics.h
new file mode 100644
index 0000000..359ebb8
--- /dev/null
+++ b/be/src/util/event-metrics.h
@@ -0,0 +1,86 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "util/metrics.h"
+
+namespace impala {
+
+/// class which is used to refresh the metastore event related metrics from catalog
+class MetastoreEventMetrics {
+ public:
+ /// Registers and initializes the Metastore event metrics
+ static void InitMetastoreEventMetrics(MetricGroup* metric_group);
+
+ /// refresh all the metrics which are used to display on webui based on the given
+ /// response this method should be called at regular intervals to update the metrics
+ /// information on the webui
+ static void refresh(TEventProcessorMetrics* response);
+
+ /// Total number of events received so far
+ static IntCounter* NUM_EVENTS_RECEIVED_COUNTER;
+
+ /// Total number of events skipped so far
+ static IntCounter* NUM_EVENTS_SKIPPED_COUNTER;
+
+ /// Mean duration required to fetch a batch of events
+ static DoubleGauge* EVENTS_FETCH_DURATION_MEAN;
+
+ /// Mean duration required to process the fetched batch of events
+ static DoubleGauge* EVENTS_PROCESS_DURATION_MEAN;
+
+ /// The current status of Metastore events processor.
+ /// See MetastoreEventProcessor.EventProcessorStatus for possible state values
+ static StringProperty* EVENT_PROCESSOR_STATUS;
+
+ /// Exponentially weighted moving avg (EWMA) of number of events received in last 1 min
+ static DoubleGauge* EVENTS_RECEIVED_1MIN_RATE;
+
+ /// EWMA of number of events received in last 5 min
+ static DoubleGauge* EVENTS_RECEIVED_5MIN_RATE;
+
+ /// EWMA of number of events received in last 15 min
+ static DoubleGauge* EVENTS_RECEIVED_15MIN_RATE;
+
+ private:
+ /// Following metric names must match with the key in metrics.json
+
+ /// metric name for events received counter.
+ static string NUMBER_EVENTS_RECEIVED_METRIC_NAME;
+
+ /// metric name for events skipped counter
+ static string NUMBER_EVENTS_SKIPPED_METRIC_NAME;
+
+ /// metric name for event processor status
+ static string EVENT_PROCESSOR_STATUS_METRIC_NAME;
+
+ /// metric name for the mean time taken for events fetch metric
+ static string EVENTS_FETCH_DURATION_MEAN_METRIC_NAME;
+
+ /// metric name for the mean time taken for events processing metric
+ static string EVENTS_PROCESS_DURATION_MEAN_METRIC_NAME;
+
+ /// metric name for EWMA of number of events in last 1 min
+ static string EVENTS_RECEIVED_1MIN_METRIC_NAME;
+
+ /// metric name for EWMA of number of events in last 5 min
+ static string EVENTS_RECEIVED_5MIN_METRIC_NAME;
+
+ /// metric name for EWMA of number of events in last 15 min
+ static string EVENTS_RECEIVED_15MIN_METRIC_NAME;
+};
+
+} // namespace impala
diff --git a/common/thrift/JniCatalog.thrift b/common/thrift/JniCatalog.thrift
index 2489321..24571a8 100644
--- a/common/thrift/JniCatalog.thrift
+++ b/common/thrift/JniCatalog.thrift
@@ -744,13 +744,51 @@ struct TCommentOnParams {
4: optional TColumnName column_name
}
+struct TEventProcessorMetrics {
+ // status of event processor
+ 1: required string status
+
+ // Total number of events received so far
+ 2: optional i64 events_received
+
+ // Total number of events skipped so far
+ 3: optional i64 events_skipped
+
+ // Mean time in sec for the fetching metastore events
+ 4: optional double events_fetch_duration_mean
+
+ // Mean time in sec for processing a given batch of events
+ 5: optional double events_process_duration_mean
+
+ // Average number of events received in 1 min
+ 6: optional double events_received_1min_rate
+
+ // Average number of events received in 1 min
+ 7: optional double events_received_5min_rate
+
+ // Average number of events received in 1 min
+ 8: optional double events_received_15min_rate
+
+ // Average number events skipped in a polling interval
+ 9: optional double events_skipped_per_poll_mean
+}
+
// Response to GetCatalogServerMetrics() call.
struct TGetCatalogServerMetricsResponse {
// Partial fetch RPC queue length.
1: required i32 catalog_partial_fetch_rpc_queue_len
+
+ // gets the events processor metrics if configured
+ 2: optional TEventProcessorMetrics event_metrics;
}
// Request to copy the generated testcase from a given input path.
struct TCopyTestCaseReq {
1: required string input_path
}
+
+struct TEventProcessorMetricsSummaryResponse {
+ // summary view of the events processor which can include status,
+ // metrics and other details
+ 1: required string summary
+}
\ No newline at end of file
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index 371372c..ed34259 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -2000,5 +2000,85 @@
"units": "NONE",
"kind": "GAUGE",
"key": "catalog.partial-fetch-rpc.queue-len"
+ },
+ {
+ "description": "Metastore event processor status",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "The status of Metastore event processor",
+ "units": "NONE",
+ "kind": "PROPERTY",
+ "key": "events-processor.status"
+ },
+ {
+ "description": "Total number of metastore events received",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Total number of metastore events received",
+ "units": "NONE",
+ "kind": "COUNTER",
+ "key": "events-processor.events-received"
+ },
+ {
+ "description": "Total number of metastore events skipped",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Total number of metastore events skipped",
+ "units": "NONE",
+ "kind": "COUNTER",
+ "key": "events-processor.events-skipped"
+ },
+ {
+ "description": "Average time taken to fetch a batch of metastore events",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Average duration to fetch metastore events",
+ "units": "TIME_S",
+ "kind": "GAUGE",
+ "key": "events-processor.avg-events-fetch-duration"
+ },
+ {
+ "description": "Average time taken to process a batch of events received from metastore",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Average duration to process a batch of metastore events",
+ "units": "TIME_S",
+ "kind": "GAUGE",
+ "key": "events-processor.avg-events-process-duration"
+ },
+ {
+ "description": "Exponentially weighted moving average (EWMA) of number of events received in last 1 min",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "EWMA of events received in last 1 min",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "events-processor.events-received-1min-rate"
+ },
+ {
+ "description": "Exponentially weighted moving average (EWMA) of number of events received in last 5 min",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "EWMA of number of events received in last 5 min",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "events-processor.events-received-5min-rate"
+ },
+ {
+ "description": "Exponentially weighted moving average (EWMA) of number of events received in last 15 min",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "EWMA of number of events received in last 15 min",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "events-processor.events-received-15min-rate"
}
-]
+]
\ No newline at end of file
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index fee27dc..63c66f2 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -65,6 +65,8 @@ import org.apache.impala.thrift.TCatalogObject;
import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.TCatalogUpdateResult;
import org.apache.impala.thrift.TDatabase;
+import org.apache.impala.thrift.TEventProcessorMetrics;
+import org.apache.impala.thrift.TEventProcessorMetricsSummaryResponse;
import org.apache.impala.thrift.TFunction;
import org.apache.impala.thrift.TGetCatalogUsageResponse;
import org.apache.impala.thrift.TGetPartialCatalogObjectRequest;
@@ -331,12 +333,17 @@ public class CatalogServiceCatalog extends Catalog {
}
}
+ @VisibleForTesting
+ public ExternalEventsProcessor getMetastoreEventProcessor() {
+ return metastoreEventProcessor_;
+ }
+
/**
* Initializes the Catalog using the default MetastoreClientPool impl.
* @param initialHmsCnxnTimeoutSec Time (in seconds) CatalogServiceCatalog will wait
- * to establish an initial connection to the HMS before
- * giving up.
+ * to establish an initial connection to the HMS before giving up.
*/
+
public CatalogServiceCatalog(boolean loadInBackground, int numLoadingThreads,
int initialHmsCnxnTimeoutSec, AuthorizationConfig authConfig,
TUniqueId catalogServiceId, String kerberosPrincipal, String localLibraryPath)
@@ -2382,6 +2389,21 @@ public class CatalogServiceCatalog extends Catalog {
}
/**
+ * Gets the events processor metrics. Used for publishing metrics on the webUI
+ */
+ public TEventProcessorMetrics getEventProcessorMetrics() {
+ return metastoreEventProcessor_.getEventProcessorMetrics();
+ }
+
+ /**
+ * Gets the events processor summary. Used for populating the contents of the events
+ * processor detailed view page
+ */
+ public TEventProcessorMetricsSummaryResponse getEventProcessorSummary() {
+ return metastoreEventProcessor_.getEventProcessorSummary();
+ }
+
+ /**
* Retrieves the stored metrics of the specified table and returns a pretty-printed
* string representation. Throws an exception if table metrics were not available
* because the table was not loaded or because another concurrent operation was holding
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/ExternalEventsProcessor.java b/fe/src/main/java/org/apache/impala/catalog/events/ExternalEventsProcessor.java
index eda52b9..742d020 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/ExternalEventsProcessor.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/ExternalEventsProcessor.java
@@ -18,6 +18,8 @@
package org.apache.impala.catalog.events;
import org.apache.impala.catalog.CatalogException;
+import org.apache.impala.thrift.TEventProcessorMetrics;
+import org.apache.impala.thrift.TEventProcessorMetricsSummaryResponse;
/**
* Interface to process external events
@@ -51,4 +53,16 @@ public interface ExternalEventsProcessor {
* Implements the core logic of processing external events
*/
void processEvents();
+
+ /**
+ * Gets the event processor status and metrics. This method is used to show up the
+ * metrics on the metrics UI page
+ */
+ TEventProcessorMetrics getEventProcessorMetrics();
+
+ /**
+ * Gets a detailed view of the event processor which can be used to populate the
+ * content of a dedicated page for the event processor
+ */
+ TEventProcessorMetricsSummaryResponse getEventProcessorSummary();
}
\ No newline at end of file
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
index b69a74b..f94ddc3 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
@@ -37,6 +37,7 @@ import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogServiceCatalog;
import org.apache.impala.catalog.Db;
import org.apache.impala.catalog.Table;
+import org.apache.impala.common.Metrics;
import org.apache.impala.common.Pair;
import org.apache.impala.common.Reference;
import org.apache.impala.thrift.TTableName;
@@ -103,9 +104,12 @@ public class MetastoreEvents {
// catalog service instance to be used for creating eventHandlers
private final CatalogServiceCatalog catalog_;
+ // metrics registry to be made available for each events to publish metrics
+ private final Metrics metrics_;
- public MetastoreEventFactory(CatalogServiceCatalog catalog) {
+ public MetastoreEventFactory(CatalogServiceCatalog catalog, Metrics metrics) {
this.catalog_ = Preconditions.checkNotNull(catalog);
+ this.metrics_ = Preconditions.checkNotNull(metrics);
}
/**
@@ -118,31 +122,26 @@ public class MetastoreEvents {
MetastoreEventType metastoreEventType =
MetastoreEventType.from(event.getEventType());
switch (metastoreEventType) {
- case CREATE_TABLE:
- return new CreateTableEvent(catalog_, event);
- case DROP_TABLE:
- return new DropTableEvent(catalog_, event);
- case ALTER_TABLE:
- return new AlterTableEvent(catalog_, event);
- case CREATE_DATABASE:
- return new CreateDatabaseEvent(catalog_, event);
- case DROP_DATABASE:
- return new DropDatabaseEvent(catalog_, event);
+ case CREATE_TABLE: return new CreateTableEvent(catalog_, metrics_, event);
+ case DROP_TABLE: return new DropTableEvent(catalog_, metrics_, event);
+ case ALTER_TABLE: return new AlterTableEvent(catalog_, metrics_, event);
+ case CREATE_DATABASE: return new CreateDatabaseEvent(catalog_, metrics_, event);
+ case DROP_DATABASE: return new DropDatabaseEvent(catalog_, metrics_, event);
case ALTER_DATABASE:
// alter database events are currently ignored
- return new IgnoredEvent(catalog_, event);
+ return new IgnoredEvent(catalog_, metrics_, event);
case ADD_PARTITION:
// add partition events triggers invalidate table currently
- return new AddPartitionEvent(catalog_, event);
+ return new AddPartitionEvent(catalog_, metrics_, event);
case DROP_PARTITION:
// drop partition events triggers invalidate table currently
- return new DropPartitionEvent(catalog_, event);
+ return new DropPartitionEvent(catalog_, metrics_, event);
case ALTER_PARTITION:
// alter partition events triggers invalidate table currently
- return new AlterPartitionEvent(catalog_, event);
+ return new AlterPartitionEvent(catalog_, metrics_, event);
default:
// ignore all the unknown events by creating a IgnoredEvent
- return new IgnoredEvent(catalog_, event);
+ return new IgnoredEvent(catalog_, metrics_, event);
}
}
@@ -226,15 +225,21 @@ public class MetastoreEvents {
// eventType from the NotificationEvent
protected final MetastoreEventType eventType_;
+ // Actual notificationEvent object received from Metastore
protected final NotificationEvent metastoreNotificationEvent_;
- MetastoreEvent(CatalogServiceCatalog catalogServiceCatalog, NotificationEvent event) {
+ // metrics registry so that events can add metrics
+ protected final Metrics metrics_;
+
+ MetastoreEvent(CatalogServiceCatalog catalogServiceCatalog, Metrics metrics,
+ NotificationEvent event) {
this.catalog_ = catalogServiceCatalog;
this.event_ = event;
this.eventId_ = event_.getEventId();
this.eventType_ = MetastoreEventType.from(event.getEventType());
- dbName_ = Preconditions.checkNotNull(event.getDbName());
- metastoreNotificationEvent_ = event;
+ this.dbName_ = Preconditions.checkNotNull(event.getDbName());
+ this.metastoreNotificationEvent_ = event;
+ this.metrics_ = metrics;
}
/**
@@ -244,7 +249,10 @@ public class MetastoreEvents {
*/
public void processIfEnabled()
throws CatalogException, MetastoreNotificationException {
- if (isEventProcessingDisabled()) return;
+ if (isEventProcessingDisabled()) {
+ metrics_.getCounter(MetastoreEventsProcessor.EVENTS_SKIPPED_METRIC).inc();
+ return;
+ }
process();
}
@@ -352,8 +360,8 @@ public class MetastoreEvents {
protected org.apache.hadoop.hive.metastore.api.Table msTbl_;
private MetastoreTableEvent(CatalogServiceCatalog catalogServiceCatalog,
- NotificationEvent event) {
- super(catalogServiceCatalog, event);
+ Metrics metrics, NotificationEvent event) {
+ super(catalogServiceCatalog, metrics, event);
tblName_ = Preconditions.checkNotNull(event.getTableName());
debugLog("Creating event {} of type {} on table {}", eventId_, eventType_,
getFullyQualifiedTblName());
@@ -429,10 +437,9 @@ public class MetastoreEvents {
* Base class for all the database events
*/
public static abstract class MetastoreDatabaseEvent extends MetastoreEvent {
-
- MetastoreDatabaseEvent(CatalogServiceCatalog catalogServiceCatalog,
+ MetastoreDatabaseEvent(CatalogServiceCatalog catalogServiceCatalog, Metrics metrics,
NotificationEvent event) {
- super(catalogServiceCatalog, event);
+ super(catalogServiceCatalog, metrics, event);
debugLog("Creating event {} of type {} on database {}", eventId_,
eventType_, dbName_);
}
@@ -457,9 +464,9 @@ public class MetastoreEvents {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private CreateTableEvent(CatalogServiceCatalog catalog, NotificationEvent event)
- throws MetastoreNotificationException {
- super(catalog, event);
+ private CreateTableEvent(CatalogServiceCatalog catalog, Metrics metrics,
+ NotificationEvent event) throws MetastoreNotificationException {
+ super(catalog, metrics, event);
Preconditions.checkArgument(MetastoreEventType.CREATE_TABLE.equals(eventType_));
Preconditions
.checkNotNull(event.getMessage(), debugString("Event message is null"));
@@ -561,9 +568,9 @@ public class MetastoreEvents {
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
@VisibleForTesting
- AlterTableEvent(CatalogServiceCatalog catalog, NotificationEvent event)
- throws MetastoreNotificationException {
- super(catalog, event);
+ AlterTableEvent(CatalogServiceCatalog catalog, Metrics metrics,
+ NotificationEvent event) throws MetastoreNotificationException {
+ super(catalog, metrics, event);
Preconditions.checkArgument(MetastoreEventType.ALTER_TABLE.equals(eventType_));
JSONAlterTableMessage alterTableMessage =
(JSONAlterTableMessage) MetastoreEventsProcessor.getMessageFactory()
@@ -699,9 +706,9 @@ public class MetastoreEvents {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private DropTableEvent(CatalogServiceCatalog catalog, NotificationEvent event)
- throws MetastoreNotificationException {
- super(catalog, event);
+ private DropTableEvent(CatalogServiceCatalog catalog, Metrics metrics,
+ NotificationEvent event) throws MetastoreNotificationException {
+ super(catalog, metrics, event);
Preconditions.checkArgument(MetastoreEventType.DROP_TABLE.equals(eventType_));
JSONDropTableMessage dropTableMessage =
(JSONDropTableMessage) MetastoreEventsProcessor.getMessageFactory()
@@ -752,9 +759,9 @@ public class MetastoreEvents {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private CreateDatabaseEvent(CatalogServiceCatalog catalog, NotificationEvent event)
- throws MetastoreNotificationException {
- super(catalog, event);
+ private CreateDatabaseEvent(CatalogServiceCatalog catalog, Metrics metrics,
+ NotificationEvent event) throws MetastoreNotificationException {
+ super(catalog, metrics, event);
Preconditions.checkArgument(MetastoreEventType.CREATE_DATABASE.equals(eventType_));
JSONCreateDatabaseMessage createDatabaseMessage =
(JSONCreateDatabaseMessage) MetastoreEventsProcessor.getMessageFactory()
@@ -818,8 +825,9 @@ public class MetastoreEvents {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private DropDatabaseEvent(CatalogServiceCatalog catalog, NotificationEvent event) {
- super(catalog, event);
+ private DropDatabaseEvent(
+ CatalogServiceCatalog catalog, Metrics metrics, NotificationEvent event) {
+ super(catalog, metrics, event);
}
/**
@@ -853,9 +861,9 @@ public class MetastoreEvents {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private TableInvalidatingEvent(CatalogServiceCatalog catalog,
- NotificationEvent event) {
- super(catalog, event);
+ private TableInvalidatingEvent(
+ CatalogServiceCatalog catalog, Metrics metrics, NotificationEvent event) {
+ super(catalog, metrics, event);
}
/**
@@ -881,9 +889,9 @@ public class MetastoreEvents {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private AddPartitionEvent(CatalogServiceCatalog catalog, NotificationEvent event)
- throws MetastoreNotificationException {
- super(catalog, event);
+ private AddPartitionEvent(CatalogServiceCatalog catalog, Metrics metrics,
+ NotificationEvent event) throws MetastoreNotificationException {
+ super(catalog, metrics, event);
Preconditions.checkState(eventType_.equals(MetastoreEventType.ADD_PARTITION));
if (event.getMessage() == null) {
throw new IllegalStateException(debugString("Event message is null"));
@@ -904,9 +912,9 @@ public class MetastoreEvents {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private AlterPartitionEvent(CatalogServiceCatalog catalog, NotificationEvent event)
- throws MetastoreNotificationException {
- super(catalog, event);
+ private AlterPartitionEvent(CatalogServiceCatalog catalog, Metrics metrics,
+ NotificationEvent event) throws MetastoreNotificationException {
+ super(catalog, metrics, event);
Preconditions.checkState(eventType_.equals(MetastoreEventType.ALTER_PARTITION));
Preconditions.checkNotNull(event.getMessage());
AlterPartitionMessage alterPartitionMessage =
@@ -926,8 +934,9 @@ public class MetastoreEvents {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private DropPartitionEvent(CatalogServiceCatalog catalog, NotificationEvent event) {
- super(catalog, event);
+ private DropPartitionEvent(
+ CatalogServiceCatalog catalog, Metrics metrics, NotificationEvent event) {
+ super(catalog, metrics, event);
Preconditions.checkState(eventType_.equals(MetastoreEventType.DROP_PARTITION));
Preconditions.checkNotNull(event.getMessage());
//TODO we should use DropPartitionMessage to get the table object here but
@@ -950,8 +959,9 @@ public class MetastoreEvents {
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
*/
- private IgnoredEvent(CatalogServiceCatalog catalog, NotificationEvent event) {
- super(catalog, event);
+ private IgnoredEvent(
+ CatalogServiceCatalog catalog, Metrics metrics, NotificationEvent event) {
+ super(catalog, metrics, event);
}
@Override
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java
index 7bcae29..1a4e735 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEventsProcessor.java
@@ -17,6 +17,8 @@
package org.apache.impala.catalog.events;
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Timer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -25,6 +27,7 @@ import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.NotificationEvent;
import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
@@ -35,8 +38,12 @@ import org.apache.impala.catalog.CatalogServiceCatalog;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEvent;
import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventFactory;
-import org.apache.log4j.Logger;
+import org.apache.impala.common.Metrics;
+import org.apache.impala.thrift.TEventProcessorMetrics;
+import org.apache.impala.thrift.TEventProcessorMetricsSummaryResponse;
import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* A metastore event is a instance of the class
@@ -131,7 +138,9 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
public static final String HMS_ADD_THRIFT_OBJECTS_IN_EVENTS_CONFIG_KEY =
"hive.metastore.notifications.add.thrift.objects";
- private static final Logger LOG = Logger.getLogger(MetastoreEventsProcessor.class);
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(MetastoreEventsProcessor.class);
// Use ExtendedJSONMessageFactory to deserialize the event messages.
// ExtendedJSONMessageFactory adds additional information over JSONMessageFactory so
// that events are compatible with Sentry
@@ -145,6 +154,20 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
// maximum number of events to poll in each RPC
private static final int EVENTS_BATCH_SIZE_PER_RPC = 1000;
+ // time taken to fetch events during each poll
+ public static final String EVENTS_FETCH_DURATION_METRIC = "events-fetch-duration";
+ // time taken to apply all the events received duration each poll
+ public static final String EVENTS_PROCESS_DURATION_METRIC = "events-apply-duration";
+ // rate of events received per unit time
+ public static final String EVENTS_RECEIVED_METRIC = "events-received";
+ // total number of events which are skipped because of the flag setting
+ public static final String EVENTS_SKIPPED_METRIC = "events-skipped";
+ // name of the event processor status metric
+ public static final String STATUS_METRIC = "status";
+ // last sycned event id
+ public static final String LAST_SYNCED_ID_METRIC = "last-synced-event"
+ + "-id";
+
// possible status of event processor
public enum EventProcessorStatus {
STOPPED, // event processor is instantiated but not yet scheduled
@@ -161,7 +184,7 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
private final MetastoreEventFactory metastoreEventFactory_;
// keeps track of the last event id which we have synced to
- private long lastSyncedEventId_;
+ private final AtomicLong lastSyncedEventId_ = new AtomicLong(-1);
// polling interval in seconds. Note this is a time we wait AFTER each fetch call
private final long pollingFrequencyInSec_;
@@ -174,14 +197,31 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setDaemon(true)
.setNameFormat("MetastoreEventsProcessor").build());
+ // metrics registry to keep track of metrics related to event processing
+ //TODO create a separate singleton class which wraps around this so that we don't
+ // have to pass it around as a argument in constructor in MetastoreEvents
+ private final Metrics metrics_ = new Metrics();
+
@VisibleForTesting
MetastoreEventsProcessor(CatalogServiceCatalog catalog, long startSyncFromId,
long pollingFrequencyInSec) {
Preconditions.checkState(pollingFrequencyInSec > 0);
this.catalog_ = Preconditions.checkNotNull(catalog);
- lastSyncedEventId_ = startSyncFromId;
- metastoreEventFactory_ = new MetastoreEventFactory(catalog_);
+ lastSyncedEventId_.set(startSyncFromId);
+ metastoreEventFactory_ = new MetastoreEventFactory(catalog_, metrics_);
pollingFrequencyInSec_ = pollingFrequencyInSec;
+ initMetrics();
+ }
+
+ private void initMetrics() {
+ metrics_.addTimer(EVENTS_FETCH_DURATION_METRIC);
+ metrics_.addTimer(EVENTS_PROCESS_DURATION_METRIC);
+ metrics_.addMeter(EVENTS_RECEIVED_METRIC);
+ metrics_.addCounter(EVENTS_SKIPPED_METRIC);
+ metrics_.addGauge(STATUS_METRIC,
+ (Gauge<String>) () -> getStatus().toString());
+ metrics_.addGauge(LAST_SYNCED_ID_METRIC,
+ (Gauge<Long>) () -> lastSyncedEventId_.get());
}
/**
@@ -215,7 +255,7 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
*/
@VisibleForTesting
public long getLastSyncedEventId() {
- return lastSyncedEventId_;
+ return lastSyncedEventId_.get();
}
@VisibleForTesting
@@ -238,7 +278,7 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
Preconditions.checkState(eventProcessorStatus_ != EventProcessorStatus.STOPPED);
eventProcessorStatus_ = EventProcessorStatus.STOPPED;
LOG.info(String.format("Event processing is stopped. Last synced event id is %d",
- lastSyncedEventId_));
+ lastSyncedEventId_.get()));
}
/**
@@ -262,12 +302,12 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
Preconditions.checkArgument(fromEventId >= 0);
Preconditions.checkState(eventProcessorStatus_ != EventProcessorStatus.ACTIVE,
"Event processing start called when it is already active");
- long prevLastSyncedEventId = lastSyncedEventId_;
- lastSyncedEventId_ = fromEventId;
+ long prevLastSyncedEventId = lastSyncedEventId_.get();
+ lastSyncedEventId_.set(fromEventId);
eventProcessorStatus_ = EventProcessorStatus.ACTIVE;
LOG.info(String.format(
"Metastore event processing restarted. Last synced event id was updated "
- + "from %d to %d", prevLastSyncedEventId, lastSyncedEventId_));
+ + "from %d to %d", prevLastSyncedEventId, lastSyncedEventId_.get()));
}
/**
@@ -277,6 +317,8 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
@VisibleForTesting
protected List<NotificationEvent> getNextMetastoreEvents()
throws MetastoreNotificationFetchException {
+ final Timer.Context context = metrics_.getTimer(EVENTS_FETCH_DURATION_METRIC).time();
+ long lastSyncedEventId = lastSyncedEventId_.get();
try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
// fetch the current notification event id. We assume that the polling interval
// is small enough that most of these polling operations result in zero new
@@ -287,20 +329,22 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
long currentEventId = currentNotificationEventId.getEventId();
// no new events since we last polled
- if (currentEventId <= lastSyncedEventId_) {
+ if (currentEventId <= lastSyncedEventId) {
return Collections.emptyList();
}
NotificationEventResponse response = msClient.getHiveClient()
- .getNextNotification(lastSyncedEventId_, EVENTS_BATCH_SIZE_PER_RPC, null);
+ .getNextNotification(lastSyncedEventId, EVENTS_BATCH_SIZE_PER_RPC, null);
LOG.info(String
.format("Received %d events. Start event id : %d", response.getEvents().size(),
- lastSyncedEventId_));
+ lastSyncedEventId));
return response.getEvents();
} catch (TException e) {
throw new MetastoreNotificationFetchException(
"Unable to fetch notifications from metastore. Last synced event id is "
- + lastSyncedEventId_, e);
+ + lastSyncedEventId, e);
+ } finally {
+ context.stop();
}
}
@@ -318,7 +362,7 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
if (currentStatus != EventProcessorStatus.ACTIVE) {
LOG.warn(String.format(
"Event processing is skipped since status is %s. Last synced event id is %d",
- currentStatus, lastSyncedEventId_));
+ currentStatus, lastSyncedEventId_.get()));
return;
}
@@ -338,6 +382,59 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
}
/**
+ * Gets the current event processor metrics along with its status. If the status is
+ * not active the metrics are skipped. Only the status is sent
+ */
+ @Override
+ public TEventProcessorMetrics getEventProcessorMetrics() {
+ TEventProcessorMetrics eventProcessorMetrics = new TEventProcessorMetrics();
+ EventProcessorStatus currentStatus = getStatus();
+ eventProcessorMetrics.setStatus(currentStatus.toString());
+ if (currentStatus != EventProcessorStatus.ACTIVE) return eventProcessorMetrics;
+
+ long eventsReceived = metrics_.getMeter(EVENTS_RECEIVED_METRIC).getCount();
+ long eventsSkipped = metrics_.getCounter(EVENTS_SKIPPED_METRIC).getCount();
+ double avgFetchDuration =
+ metrics_.getTimer(EVENTS_FETCH_DURATION_METRIC).getMeanRate();
+ double avgProcessDuration =
+ metrics_.getTimer(EVENTS_PROCESS_DURATION_METRIC).getMeanRate();
+ double avgNumberOfEventsReceived1Min =
+ metrics_.getMeter(EVENTS_RECEIVED_METRIC).getOneMinuteRate();
+ double avgNumberOfEventsReceived5Min =
+ metrics_.getMeter(EVENTS_RECEIVED_METRIC).getFiveMinuteRate();
+ double avgNumberOfEventsReceived15Min =
+ metrics_.getMeter(EVENTS_RECEIVED_METRIC).getFifteenMinuteRate();
+
+
+ eventProcessorMetrics.setEvents_received(eventsReceived);
+ eventProcessorMetrics.setEvents_skipped(eventsSkipped);
+ eventProcessorMetrics.setEvents_fetch_duration_mean(avgFetchDuration);
+ eventProcessorMetrics.setEvents_process_duration_mean(avgProcessDuration);
+ eventProcessorMetrics.setEvents_received_1min_rate(avgNumberOfEventsReceived1Min);
+ eventProcessorMetrics.setEvents_received_5min_rate(avgNumberOfEventsReceived5Min);
+ eventProcessorMetrics.setEvents_received_15min_rate(avgNumberOfEventsReceived15Min);
+
+ LOG.trace("Events Received: {} Events skipped: {} Avg fetch duration: {} Avg process "
+ + "duration: {} Events received rate (1min) : {}",
+ eventsReceived, eventsSkipped, avgFetchDuration, avgProcessDuration,
+ avgNumberOfEventsReceived1Min);
+ return eventProcessorMetrics;
+ }
+
+ @Override
+ public TEventProcessorMetricsSummaryResponse getEventProcessorSummary() {
+ TEventProcessorMetricsSummaryResponse summaryResponse =
+ new TEventProcessorMetricsSummaryResponse();
+ summaryResponse.setSummary(metrics_.toString());
+ return summaryResponse;
+ }
+
+ @VisibleForTesting
+ Metrics getMetrics() {
+ return metrics_;
+ }
+
+ /**
* Process the given list of notification events. Useful for tests which provide a list
* of events
*
@@ -346,10 +443,13 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
@VisibleForTesting
protected void processEvents(List<NotificationEvent> events)
throws MetastoreNotificationException {
- List<MetastoreEvent> filteredEvents =
- metastoreEventFactory_.getFilteredEvents(events);
NotificationEvent lastProcessedEvent = null;
+ final Timer.Context context =
+ metrics_.getTimer(EVENTS_PROCESS_DURATION_METRIC).time();
+ metrics_.getMeter(EVENTS_RECEIVED_METRIC).mark(events.size());
try {
+ List<MetastoreEvent> filteredEvents =
+ metastoreEventFactory_.getFilteredEvents(events);
for (MetastoreEvent event : filteredEvents) {
// synchronizing each event processing reduces the scope of the lock so the a
// potential reset() during event processing is not blocked for longer than
@@ -360,14 +460,15 @@ public class MetastoreEventsProcessor implements ExternalEventsProcessor {
}
lastProcessedEvent = event.metastoreNotificationEvent_;
event.processIfEnabled();
- lastSyncedEventId_ = event.eventId_;
+ lastSyncedEventId_.set(event.eventId_);
}
}
} catch (CatalogException e) {
throw new MetastoreNotificationException(String.format(
- "Unable to process event %d of type %s. "
- + "Event processing will be stopped.", lastProcessedEvent.getEventId(),
- lastProcessedEvent.getEventType()), e);
+ "Unable to process event %d of type %s. Event processing will be stopped.",
+ lastProcessedEvent.getEventId(), lastProcessedEvent.getEventType()), e);
+ } finally {
+ context.stop();
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/NoOpEventProcessor.java b/fe/src/main/java/org/apache/impala/catalog/events/NoOpEventProcessor.java
index 44150d1..950fdf9 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/NoOpEventProcessor.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/NoOpEventProcessor.java
@@ -17,6 +17,10 @@
package org.apache.impala.catalog.events;
+import org.apache.impala.catalog.events.MetastoreEventsProcessor.EventProcessorStatus;
+import org.apache.impala.thrift.TEventProcessorMetrics;
+import org.apache.impala.thrift.TEventProcessorMetricsSummaryResponse;
+
/**
* A simple no-op events processor which does nothing. Used to plugin to the catalog
* when event processing is disabled so that we don't have to do a null check every
@@ -25,6 +29,12 @@ package org.apache.impala.catalog.events;
public class NoOpEventProcessor implements ExternalEventsProcessor {
private static final ExternalEventsProcessor INSTANCE = new NoOpEventProcessor();
+ private final TEventProcessorMetrics DEFAULT_METRICS_RESPONSE =
+ new TEventProcessorMetrics();
+
+ private final TEventProcessorMetricsSummaryResponse DEFAULT_SUMMARY_RESPONSE =
+ new TEventProcessorMetricsSummaryResponse();
+
/**
* Gets the instance of NoOpEventProcessor
*/
@@ -32,6 +42,8 @@ public class NoOpEventProcessor implements ExternalEventsProcessor {
private NoOpEventProcessor() {
// prevents instantiation
+ DEFAULT_METRICS_RESPONSE.setStatus(EventProcessorStatus.STOPPED.toString());
+ DEFAULT_SUMMARY_RESPONSE.setSummary("");
}
@Override
@@ -59,4 +71,14 @@ public class NoOpEventProcessor implements ExternalEventsProcessor {
public void processEvents() {
// no-op
}
+
+ @Override
+ public TEventProcessorMetrics getEventProcessorMetrics() {
+ return DEFAULT_METRICS_RESPONSE;
+ }
+
+ @Override
+ public TEventProcessorMetricsSummaryResponse getEventProcessorSummary() {
+ return DEFAULT_SUMMARY_RESPONSE;
+ }
}
\ No newline at end of file
diff --git a/fe/src/main/java/org/apache/impala/common/Metrics.java b/fe/src/main/java/org/apache/impala/common/Metrics.java
index cf8621f..7e60586 100644
--- a/fe/src/main/java/org/apache/impala/common/Metrics.java
+++ b/fe/src/main/java/org/apache/impala/common/Metrics.java
@@ -17,6 +17,7 @@
package org.apache.impala.common;
+import java.text.DecimalFormat;
import java.util.Map.Entry;
import com.codahale.metrics.Counter;
@@ -34,6 +35,9 @@ import com.codahale.metrics.Timer;
*/
public final class Metrics {
+ // formatter to round the double values of the rate counters to 4 digits after decimal
+ private static final DecimalFormat decimalFormatter_ = new DecimalFormat("#.####");
+
private final MetricRegistry registry_ = new MetricRegistry();
public Metrics() {}
@@ -115,6 +119,10 @@ public final class Metrics {
snapshotToString(entry.getValue().getSnapshot()));
result.append("\n");
}
+ for (Entry<String, Meter> entry : registry_.getMeters().entrySet()) {
+ result.append(entry.getKey()).append(":").append(meterToString(entry.getValue()));
+ result.append("\n");
+ }
return result.toString();
}
@@ -123,27 +131,57 @@ public final class Metrics {
*/
private String timerToString(Timer timer) {
StringBuilder builder = new StringBuilder();
- return builder.append("\n\tCount: " + timer.getCount() + "\n")
- .append("\tMean rate: " + timer.getMeanRate() + "\n")
- .append("\t1min rate: " + timer.getOneMinuteRate() + "\n")
- .append("\t5min rate: " + timer.getFiveMinuteRate() + "\n")
- .append("\t15min rate: " + timer.getFifteenMinuteRate() + "\n")
+ return builder.append("\n Count: " + timer.getCount())
+ .append("\n Mean rate: ")
+ .append(decimalFormatter_.format(timer.getMeanRate()))
+ .append("\n 1 min. rate: ")
+ .append(decimalFormatter_.format(timer.getOneMinuteRate()))
+ .append("\n 5 min. rate: ")
+ .append(decimalFormatter_.format(timer.getFiveMinuteRate()))
+ .append("\n 15 min. rate: ")
+ .append(decimalFormatter_.format(timer.getFifteenMinuteRate()))
.append(snapshotToString(timer.getSnapshot()))
.toString();
}
/**
+ * Helper method to pretty print the contents of a Meter
+ */
+ private String meterToString(Meter meter) {
+ StringBuilder builder = new StringBuilder();
+ return builder.append("\n Count: ")
+ .append(meter.getCount())
+ .append("\n Mean rate: ")
+ .append(decimalFormatter_.format(meter.getMeanRate()))
+ .append("\n 1 min. rate: ")
+ .append(decimalFormatter_.format(meter.getOneMinuteRate()))
+ .append("\n 5 min. rate: ")
+ .append(decimalFormatter_.format(meter.getFiveMinuteRate()))
+ .append("\n 15 min. rate: ")
+ .append(decimalFormatter_.format(meter.getFifteenMinuteRate()))
+ .toString();
+ }
+
+ /**
* Helper function that pretty prints the contents of a metric snapshot.
*/
private String snapshotToString(Snapshot snapshot) {
StringBuilder builder = new StringBuilder();
- return builder.append("\n\tMin (msec): " + snapshot.getMin() / 1000000 + "\n")
- .append("\tMax (msec): " + snapshot.getMax() / 1000000 + "\n")
- .append("\tMean (msec): " + snapshot.getMean() / 1000000 + "\n")
- .append("\tMedian (msec): " + snapshot.getMedian() / 1000000 + "\n")
- .append("\t75th-% (msec): " + snapshot.get75thPercentile() / 1000000 + "\n")
- .append("\t95th-% (msec): " + snapshot.get95thPercentile() / 1000000 + "\n")
- .append("\t99th-% (msec): " + snapshot.get99thPercentile() / 1000000 + "\n")
+ return builder.append("\n Min (msec): ")
+ .append(decimalFormatter_.format(snapshot.getMin() / 1000000))
+ .append("\n Max (msec): ")
+ .append(decimalFormatter_.format(snapshot.getMax() / 1000000))
+ .append("\n Mean (msec): ")
+ .append(decimalFormatter_.format(snapshot.getMean() / 1000000))
+ .append("\n Median (msec): ")
+ .append(decimalFormatter_.format(snapshot.getMedian() / 1000000))
+ .append("\n 75th-% (msec): ")
+ .append(decimalFormatter_.format(snapshot.get75thPercentile() / 1000000))
+ .append("\n 95th-% (msec): ")
+ .append(decimalFormatter_.format(snapshot.get95thPercentile() / 1000000))
+ .append("\n 99th-% (msec): ")
+ .append(decimalFormatter_.format(snapshot.get99thPercentile() / 1000000))
+ .append("\n")
.toString();
}
}
diff --git a/fe/src/main/java/org/apache/impala/service/JniCatalog.java b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
index 6e29f42..bdcb139 100644
--- a/fe/src/main/java/org/apache/impala/service/JniCatalog.java
+++ b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
@@ -314,6 +314,11 @@ public class JniCatalog {
return serializer.serialize(catalog_.getCatalogUsage());
}
+ public byte[] getEventProcessorSummary() throws TException {
+ TSerializer serializer = new TSerializer(protocolFactory_);
+ return serializer.serialize(catalog_.getEventProcessorSummary());
+ }
+
public void updateTableUsage(byte[] req) throws ImpalaException {
TUpdateTableUsageRequest thriftReq = new TUpdateTableUsageRequest();
JniUtil.deserializeThrift(protocolFactory_, thriftReq, req);
@@ -324,6 +329,7 @@ public class JniCatalog {
TGetCatalogServerMetricsResponse response = new TGetCatalogServerMetricsResponse();
response.setCatalog_partial_fetch_rpc_queue_len(
catalog_.getPartialFetchRpcQueueLength());
+ response.setEvent_metrics(catalog_.getEventProcessorMetrics());
TSerializer serializer = new TSerializer(protocolFactory_);
return serializer.serialize(response);
}
diff --git a/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java b/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
index e6977ce..1212de4 100644
--- a/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
@@ -19,6 +19,7 @@ package org.apache.impala.catalog.events;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -75,6 +76,8 @@ import org.apache.impala.thrift.TDdlExecRequest;
import org.apache.impala.thrift.TDdlType;
import org.apache.impala.thrift.TDropDbParams;
import org.apache.impala.thrift.TDropTableOrViewParams;
+import org.apache.impala.thrift.TEventProcessorMetrics;
+import org.apache.impala.thrift.TEventProcessorMetricsSummaryResponse;
import org.apache.impala.thrift.THdfsFileFormat;
import org.apache.impala.thrift.TPrimitiveType;
import org.apache.impala.thrift.TScalarType;
@@ -159,6 +162,7 @@ public class MetastoreEventsProcessorTest {
// reset the event processor to the current eventId
eventsProcessor_.stop();
eventsProcessor_.start(eventsProcessor_.getCurrentEventId());
+ eventsProcessor_.processEvents();
assertEquals(EventProcessorStatus.ACTIVE, eventsProcessor_.getStatus());
}
@@ -882,8 +886,8 @@ public class MetastoreEventsProcessorTest {
NotificationEvent fakeAlterTableNotification =
createFakeAlterTableNotification(dbName, tblName, tableBefore, tableAfter);
- AlterTableEvent alterTableEvent =
- new AlterTableEvent(fakeCatalog, fakeAlterTableNotification);
+ AlterTableEvent alterTableEvent = new AlterTableEvent(
+ fakeCatalog, eventsProcessor_.getMetrics(), fakeAlterTableNotification);
Assert.assertFalse("Alter table which " + "changes the flags should not be skipped. "
+ printFlagTransistions(dbFlag, tblFlagTransition),
alterTableEvent.isEventProcessingDisabled());
@@ -894,7 +898,8 @@ public class MetastoreEventsProcessorTest {
getTestTable(dbName, tblName, afterParams, false);
NotificationEvent nextNotification =
createFakeAlterTableNotification(dbName, tblName, tableAfter, nextTable);
- alterTableEvent = new AlterTableEvent(fakeCatalog, nextNotification);
+ alterTableEvent =
+ new AlterTableEvent(fakeCatalog, eventsProcessor_.getMetrics(), nextNotification);
if (shouldNextEventBeSkipped) {
assertTrue("Alter table event should not skipped following this table flag "
+ "transition. " + printFlagTransistions(dbFlag, tblFlagTransition),
@@ -933,6 +938,89 @@ public class MetastoreEventsProcessorTest {
}
/**
+ * Test generates some events and makes sure that the metrics match with expected
+ * number of events
+ */
+ @Test
+ public void testEventProcessorMetrics() throws TException {
+ TEventProcessorMetrics responseBefore = eventsProcessor_.getEventProcessorMetrics();
+ long numEventsReceivedBefore = responseBefore.getEvents_received();
+ long numEventsSkippedBefore = responseBefore.getEvents_skipped();
+ // event 1
+ createDatabase();
+ Map<String, String> tblParams = new HashMap<>(1);
+ tblParams.put(MetastoreEvents.DISABLE_EVENT_HMS_SYNC_KEY, "true");
+ // event 2
+ createTable(TEST_DB_NAME, "tbl_should_skipped", tblParams, true);
+ // event 3
+ createTable(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, null, true);
+ List<List<String>> partitionVals = new ArrayList<>();
+ partitionVals.add(Arrays.asList("1"));
+ partitionVals.add(Arrays.asList("2"));
+ partitionVals.add(Arrays.asList("3"));
+ // event 4
+ addPartitions(TEST_DB_NAME, "tbl_should_skipped", partitionVals);
+ // event 5
+ addPartitions(TEST_DB_NAME, TEST_TABLE_NAME_PARTITIONED, partitionVals);
+ eventsProcessor_.processEvents();
+ TEventProcessorMetrics response = eventsProcessor_.getEventProcessorMetrics();
+ assertEquals(EventProcessorStatus.ACTIVE.toString(), response.getStatus());
+ assertEquals(numEventsReceivedBefore + 5, response.getEvents_received());
+ // two events on tbl which is skipped
+ assertEquals(numEventsSkippedBefore + 2, response.getEvents_skipped());
+ assertTrue("Event fetch duration should be greater than zero",
+ response.getEvents_fetch_duration_mean() > 0);
+ assertTrue("Event process duration should be greater than zero",
+ response.getEvents_process_duration_mean() > 0);
+ TEventProcessorMetricsSummaryResponse summaryResponse =
+ catalog_.getEventProcessorSummary();
+ assertNotNull(summaryResponse);
+ }
+
+ /**
+ * Test makes sure that the event metrics are not set when event processor is not active
+ */
+ @Test
+ public void testEventProcessorWhenNotActive() throws TException {
+ try {
+ eventsProcessor_.stop();
+ assertEquals(EventProcessorStatus.STOPPED, eventsProcessor_.getStatus());
+ TEventProcessorMetrics response = eventsProcessor_.getEventProcessorMetrics();
+ assertNotNull(response);
+ assertEquals(EventProcessorStatus.STOPPED.toString(), response.getStatus());
+ assertFalse(response.isSetEvents_fetch_duration_mean());
+ assertFalse(response.isSetEvents_process_duration_mean());
+ assertFalse(response.isSetEvents_received());
+ assertFalse(response.isSetEvents_skipped());
+ assertFalse(response.isSetEvents_received_1min_rate());
+ assertFalse(response.isSetEvents_received_5min_rate());
+ assertFalse(response.isSetEvents_received_15min_rate());
+ TEventProcessorMetricsSummaryResponse summaryResponse =
+ eventsProcessor_.getEventProcessorSummary();
+ assertNotNull(summaryResponse);
+ } finally {
+ // reset the state of event process once the test completes
+ eventsProcessor_.start();
+ }
+ }
+
+ /**
+ * Tests makes sure that event metrics show valid state when event processing is not
+ * configured
+ */
+ @Test
+ public void testEventMetricsWhenNotConfigured() {
+ CatalogServiceCatalog testCatalog = CatalogServiceTestCatalog.create();
+ assertTrue("Events processed is not expected to be configured for this test",
+ testCatalog.getMetastoreEventProcessor() instanceof NoOpEventProcessor);
+ TEventProcessorMetrics response = testCatalog.getEventProcessorMetrics();
+ assertNotNull(response);
+ assertEquals(EventProcessorStatus.STOPPED.toString(), response.getStatus());
+ TEventProcessorMetricsSummaryResponse summaryResponse =
+ testCatalog.getEventProcessorSummary();
+ assertNotNull(summaryResponse);
+ }
+ /**
* Test runs all the supported DDL operations with the given value of flag at
* database and table level
*/
diff --git a/www/events.tmpl b/www/events.tmpl
new file mode 100644
index 0000000..484cfea
--- /dev/null
+++ b/www/events.tmpl
@@ -0,0 +1,25 @@
+<!--
+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.
+-->
+{{> www/common-header.tmpl }}
+{{^error}}
+
+<h3>Event Processor Summary</h3>
+<pre>{{event_processor_metrics}}</pre>
+
+{{> www/common-footer.tmpl }}