You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by et...@apache.org on 2019/06/04 18:35:48 UTC
[storm] branch master updated: [STORM-3395] Placeholder Topology
page components show task and executor counts
This is an automated email from the ASF dual-hosted git repository.
ethanli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new 90d7c02 [STORM-3395] Placeholder Topology page components show task and executor counts
new 08468c5 Merge pull request #3011 from dandsager1/STORM-3395
90d7c02 is described below
commit 90d7c0288d5dfbdeb02e9742d3f9215ce3178e5b
Author: dandsager <da...@verizonmedia.com>
AuthorDate: Tue May 21 10:27:43 2019 -0500
[STORM-3395] Placeholder Topology page components show task and executor counts
---
.../org/apache/storm/daemon/nimbus/Nimbus.java | 48 ++++++++++++++--------
1 file changed, 31 insertions(+), 17 deletions(-)
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index 0f662de..6d9e9f2 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -4144,19 +4144,13 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
*/
private void maybeAddPlaceholderSpoutAggStats(TopologyPageInfo topoPageInfo, StormTopology topology) {
if (topoPageInfo.get_id_to_spout_agg_stats().isEmpty()) {
- Map<String, SpoutSpec> spouts = topology.get_spouts();
- for (String spoutName : spouts.keySet()) {
+ for (Entry<String, SpoutSpec> entry : topology.get_spouts().entrySet()) {
// component
ComponentAggregateStats placeholderComponentStats = new ComponentAggregateStats();
placeholderComponentStats.set_type(ComponentType.SPOUT);
// common aggregate
- CommonAggregateStats commonStats = new CommonAggregateStats();
- commonStats.set_num_executors(0);
- commonStats.set_num_tasks(0);
- commonStats.set_emitted(0);
- commonStats.set_transferred(0);
- commonStats.set_acked(0);
+ CommonAggregateStats commonStats = getPlaceholderCommonAggregateStats(entry.getValue());
placeholderComponentStats.set_common_stats(commonStats);
// spout aggregate
@@ -4166,7 +4160,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
specificStats.set_spout(spoutAggStats);
placeholderComponentStats.set_specific_stats(specificStats);
- topoPageInfo.get_id_to_spout_agg_stats().put(spoutName, placeholderComponentStats);
+ topoPageInfo.get_id_to_spout_agg_stats().put(entry.getKey(), placeholderComponentStats);
}
}
}
@@ -4180,8 +4174,8 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
*/
private void maybeAddPlaceholderBoltAggStats(TopologyPageInfo topoPageInfo, StormTopology topology, boolean includeSys) {
if (topoPageInfo.get_id_to_bolt_agg_stats().isEmpty()) {
- Map<String, Bolt> bolts = topology.get_bolts();
- for (String boltName : bolts.keySet()) {
+ for (Entry<String, Bolt> entry : topology.get_bolts().entrySet()) {
+ String boltName = entry.getKey();
if ((!includeSys && Utils.isSystemId(boltName)) || boltName.equals(Constants.SYSTEM_COMPONENT_ID)) {
continue;
}
@@ -4191,12 +4185,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
placeholderComponentStats.set_type(ComponentType.BOLT);
// common aggregate
- CommonAggregateStats commonStats = new CommonAggregateStats();
- commonStats.set_num_executors(0);
- commonStats.set_num_tasks(0);
- commonStats.set_emitted(0);
- commonStats.set_transferred(0);
- commonStats.set_acked(0);
+ CommonAggregateStats commonStats = getPlaceholderCommonAggregateStats(entry.getValue());
placeholderComponentStats.set_common_stats(commonStats);
// bolt aggregate
@@ -4214,6 +4203,31 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
}
}
+ private CommonAggregateStats getPlaceholderCommonAggregateStats(Object component) {
+ // common aggregate
+ CommonAggregateStats commonStats = new CommonAggregateStats();
+
+ // get num_executors
+ int numExecutors = 0;
+ try {
+ numExecutors = StormCommon.numStartExecutors(component);
+ } catch (InvalidTopologyException e) {
+ // ignore
+ }
+
+ // get num_tasks
+ Map<String, Object> jsonMap = StormCommon.componentConf(component);
+ int numTasks = ObjectReader.getInt(jsonMap.getOrDefault(Config.TOPOLOGY_TASKS, numExecutors));
+
+ commonStats.set_num_executors(numExecutors);
+ commonStats.set_num_tasks(numTasks);
+ commonStats.set_emitted(0);
+ commonStats.set_transferred(0);
+ commonStats.set_acked(0);
+
+ return commonStats;
+ }
+
@Override
public SupervisorPageInfo getSupervisorPageInfo(String superId, String host, boolean includeSys)
throws NotAliveException, AuthorizationException, TException {