You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@heron.apache.org by hu...@apache.org on 2018/04/09 18:44:31 UTC
[incubator-heron] branch master updated: Align MetricsCacheMgr
location discovery in Stmgr (#2797)
This is an automated email from the ASF dual-hosted git repository.
huijun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-heron.git
The following commit(s) were added to refs/heads/master by this push:
new a667fcd Align MetricsCacheMgr location discovery in Stmgr (#2797)
a667fcd is described below
commit a667fcd740e02fa9ff66361059ac2e098e01aa97
Author: Huijun Wu <hu...@twitter.com>
AuthorDate: Mon Apr 9 11:44:28 2018 -0700
Align MetricsCacheMgr location discovery in Stmgr (#2797)
* toggle metricscache connection according to metricscache switch config in stmgr
* update str
* update .h
* fix compile
* added one line comment
* fix unit
---
heron/executor/src/python/heron_executor.py | 3 ++-
heron/executor/tests/python/heron_executor_unittest.py | 10 ++++++----
heron/stmgr/src/cpp/manager/stmgr.cpp | 14 ++++++++++----
heron/stmgr/src/cpp/manager/stmgr.h | 6 +++++-
heron/stmgr/src/cpp/server/stmgr-main.cpp | 3 ++-
heron/stmgr/tests/cpp/server/stmgr_unittest.cpp | 2 +-
6 files changed, 26 insertions(+), 12 deletions(-)
diff --git a/heron/executor/src/python/heron_executor.py b/heron/executor/src/python/heron_executor.py
index 3f36e5d..df63c35 100755
--- a/heron/executor/src/python/heron_executor.py
+++ b/heron/executor/src/python/heron_executor.py
@@ -725,7 +725,8 @@ class HeronExecutor(object):
'--config_file=%s' % self.heron_internals_config_file,
'--override_config_file=%s' % self.override_config_file,
'--ckptmgr_port=%s' % str(self.checkpoint_manager_port),
- '--ckptmgr_id=%s' % self.ckptmgr_ids[self.shard]]
+ '--ckptmgr_id=%s' % self.ckptmgr_ids[self.shard],
+ '--metricscachemgr_mode=%s' % self.metricscache_manager_mode.lower()]
retval[self.stmgr_ids[self.shard]] = stmgr_cmd
# metricsmgr_metrics_sink_config_file = 'metrics_sinks.yaml'
diff --git a/heron/executor/tests/python/heron_executor_unittest.py b/heron/executor/tests/python/heron_executor_unittest.py
index 5b661e7..3dd8a62 100644
--- a/heron/executor/tests/python/heron_executor_unittest.py
+++ b/heron/executor/tests/python/heron_executor_unittest.py
@@ -175,7 +175,8 @@ class HeronExecutorTest(unittest.TestCase):
'--myhost=%s --data_port=master_port '
'--local_data_port=tmaster_controller_port --metricsmgr_port=metricsmgr_port '
'--shell_port=shell-port --config_file=%s --override_config_file=%s '
- '--ckptmgr_port=ckptmgr-port --ckptmgr_id=ckptmgr-1'
+ '--ckptmgr_port=ckptmgr-port --ckptmgr_id=ckptmgr-1 '
+ '--metricscachemgr_mode=cluster'
% (HOSTNAME, INTERNAL_CONF_PATH, OVERRIDE_PATH)),
ProcessInfo(MockPOpen(), 'container_1_word_3', get_expected_instance_command('word', 3, 1)),
ProcessInfo(MockPOpen(), 'container_1_exclaim1_1',
@@ -199,7 +200,8 @@ class HeronExecutorTest(unittest.TestCase):
'--data_port=master_port '
'--local_data_port=tmaster_controller_port --metricsmgr_port=metricsmgr_port '
'--shell_port=shell-port --config_file=%s --override_config_file=%s '
- '--ckptmgr_port=ckptmgr-port --ckptmgr_id=ckptmgr-7'
+ '--ckptmgr_port=ckptmgr-port --ckptmgr_id=ckptmgr-7 '
+ '--metricscachemgr_mode=cluster'
% (HOSTNAME, INTERNAL_CONF_PATH, OVERRIDE_PATH)),
ProcessInfo(MockPOpen(), 'metricsmgr-7', get_expected_metricsmgr_command(7)),
ProcessInfo(MockPOpen(), 'heron-shell-7', get_expected_shell_command(7)),
@@ -268,9 +270,9 @@ class HeronExecutorTest(unittest.TestCase):
("--checkpoint-manager-classpath", "ckptmgr_classpath"),
("--checkpoint-manager-port", "ckptmgr-port"),
("--stateful-config-file", "stateful_config_file"),
- ("--health-manager-mode", "healthmgr_mode"),
+ ("--health-manager-mode", "cluster"),
("--health-manager-classpath", "healthmgr_classpath"),
- ("--metricscache-manager-mode", "metricscache_mode")
+ ("--metricscache-manager-mode", "cluster")
]
args = ("%s=%s" % (arg[0], (str(arg[1]))) for arg in executor_args)
diff --git a/heron/stmgr/src/cpp/manager/stmgr.cpp b/heron/stmgr/src/cpp/manager/stmgr.cpp
index de12189..3fd805b 100644
--- a/heron/stmgr/src/cpp/manager/stmgr.cpp
+++ b/heron/stmgr/src/cpp/manager/stmgr.cpp
@@ -70,7 +70,8 @@ StMgr::StMgr(EventLoop* eventLoop, const sp_string& _myhost, sp_int32 _data_port
const std::vector<sp_string>& _instances, const sp_string& _zkhostport,
const sp_string& _zkroot, sp_int32 _metricsmgr_port, sp_int32 _shell_port,
sp_int32 _ckptmgr_port, const sp_string& _ckptmgr_id,
- sp_int64 _high_watermark, sp_int64 _low_watermark)
+ sp_int64 _high_watermark, sp_int64 _low_watermark,
+ const sp_string& _metricscachemgr_mode)
: pplan_(NULL),
topology_name_(_topology_name),
@@ -96,7 +97,8 @@ StMgr::StMgr(EventLoop* eventLoop, const sp_string& _myhost, sp_int32 _data_port
ckptmgr_port_(_ckptmgr_port),
ckptmgr_id_(_ckptmgr_id),
high_watermark_(_high_watermark),
- low_watermark_(_low_watermark) {}
+ low_watermark_(_low_watermark),
+ metricscachemgr_mode_(_metricscachemgr_mode) {}
void StMgr::Init() {
LOG(INFO) << "Init Stmgr" << std::endl;
@@ -119,8 +121,10 @@ void StMgr::Init() {
metrics_manager_client_->register_metric(METRIC_TIME_SPENT_BACK_PRESSURE_INIT,
back_pressure_metric_initiated_);
state_mgr_->SetTMasterLocationWatch(topology_name_, [this]() { this->FetchTMasterLocation(); });
- state_mgr_->SetMetricsCacheLocationWatch(
+ if (0 != metricscachemgr_mode_.compare("disabled")) {
+ state_mgr_->SetMetricsCacheLocationWatch(
topology_name_, [this]() { this->FetchMetricsCacheLocation(); });
+ }
reliability_mode_ = heron::config::TopologyConfigHelper::GetReliabilityMode(*hydrated_topology_);
if (reliability_mode_ == config::TopologyConfigVars::EFFECTIVELY_ONCE) {
@@ -156,7 +160,9 @@ void StMgr::Init() {
// constructor needs actual stmgr ports, thus put FetchTMasterLocation()
// has to be after after StartStmgrServer and StartInstanceServer()
FetchTMasterLocation();
- FetchMetricsCacheLocation();
+ if (0 != metricscachemgr_mode_.compare("disabled")) {
+ FetchMetricsCacheLocation();
+ }
if (reliability_mode_ == config::TopologyConfigVars::EFFECTIVELY_ONCE) {
// Now start the stateful restorer
diff --git a/heron/stmgr/src/cpp/manager/stmgr.h b/heron/stmgr/src/cpp/manager/stmgr.h
index fa8602f..cd0d0da 100644
--- a/heron/stmgr/src/cpp/manager/stmgr.h
+++ b/heron/stmgr/src/cpp/manager/stmgr.h
@@ -64,7 +64,8 @@ class StMgr {
const std::vector<sp_string>& _instances, const sp_string& _zkhostport,
const sp_string& _zkroot, sp_int32 _metricsmgr_port, sp_int32 _shell_port,
sp_int32 _ckptmgr_port, const sp_string& _ckptmgr_id,
- sp_int64 _high_watermark, sp_int64 _low_watermark);
+ sp_int64 _high_watermark, sp_int64 _low_watermark,
+ const sp_string& _metricscachemgr_mode);
virtual ~StMgr();
// All kinds of initialization like starting servers and clients
@@ -253,6 +254,9 @@ class StMgr {
sp_int64 high_watermark_;
sp_int64 low_watermark_;
+
+ // whether MetricsCacheMgr is running
+ sp_string metricscachemgr_mode_;
};
} // namespace stmgr
diff --git a/heron/stmgr/src/cpp/server/stmgr-main.cpp b/heron/stmgr/src/cpp/server/stmgr-main.cpp
index d4703bb..0273506 100644
--- a/heron/stmgr/src/cpp/server/stmgr-main.cpp
+++ b/heron/stmgr/src/cpp/server/stmgr-main.cpp
@@ -43,6 +43,7 @@ DEFINE_string(config_file, "", "The heron internals config file");
DEFINE_string(override_config_file, "", "The override heron internals config file");
DEFINE_string(ckptmgr_id, "", "The id of the local ckptmgr");
DEFINE_int32(ckptmgr_port, 0, "The port of the local ckptmgr");
+DEFINE_string(metricscachemgr_mode, "disabled", "MetricsCacheMgr mode, default `disabled`");
int main(int argc, char* argv[]) {
gflags::ParseCommandLineFlags(&argc, &argv, true);
@@ -79,7 +80,7 @@ int main(int argc, char* argv[]) {
FLAGS_topology_name, FLAGS_topology_id, topology, FLAGS_stmgr_id,
instances, FLAGS_zkhostportlist, FLAGS_zkroot, FLAGS_metricsmgr_port,
FLAGS_shell_port, FLAGS_ckptmgr_port, FLAGS_ckptmgr_id,
- high_watermark, low_watermark);
+ high_watermark, low_watermark, FLAGS_metricscachemgr_mode);
mgr.Init();
ss.loop();
return 0;
diff --git a/heron/stmgr/tests/cpp/server/stmgr_unittest.cpp b/heron/stmgr/tests/cpp/server/stmgr_unittest.cpp
index a40de27..8670304 100644
--- a/heron/stmgr/tests/cpp/server/stmgr_unittest.cpp
+++ b/heron/stmgr/tests/cpp/server/stmgr_unittest.cpp
@@ -283,7 +283,7 @@ void StartStMgr(EventLoopImpl*& ss, heron::stmgr::StMgr*& mgr, std::thread*& stm
topology_id,
stmgr_topology, stmgr_id, workers, zkhostportlist, dpath,
metricsmgr_port, shell_port, ckptmgr_port, ckptmgr_id,
- _high_watermark, _low_watermark);
+ _high_watermark, _low_watermark, "disabled");
EXPECT_EQ(0, stmgr_port);
EXPECT_EQ(0, local_data_port);
mgr->Init();
--
To stop receiving notification emails like this one, please contact
huijun@apache.org.