You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2018/10/01 16:09:22 UTC
[1/2] storm git commit: STORM-3237 track Nimbus mkAssignment failures
Repository: storm
Updated Branches:
refs/heads/master c367f9af7 -> c47c23cf8
STORM-3237 track Nimbus mkAssignment failures
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d7d2c623
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d7d2c623
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d7d2c623
Branch: refs/heads/master
Commit: d7d2c62312d2e3ed73273a88359d1e2f35a0ecbc
Parents: 4c3a9bd
Author: Aaron Gresch <ag...@yahoo-inc.com>
Authored: Thu Sep 27 13:30:59 2018 -0500
Committer: Aaron Gresch <ag...@yahoo-inc.com>
Committed: Fri Sep 28 15:23:58 2018 -0500
----------------------------------------------------------------------
docs/ClusterMetrics.md | 1 +
.../org/apache/storm/daemon/nimbus/Nimbus.java | 108 +++++++++++--------
2 files changed, 63 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d7d2c623/docs/ClusterMetrics.md
----------------------------------------------------------------------
diff --git a/docs/ClusterMetrics.md b/docs/ClusterMetrics.md
index 17c4ca1..4e4d0f1 100644
--- a/docs/ClusterMetrics.md
+++ b/docs/ClusterMetrics.md
@@ -101,6 +101,7 @@ These are metrics that are specific to a nimbus instance. In many instances onl
| nimbus:num-uploadChunk-calls | meter | calls to uploadChunk thrift method. |
| nimbus:num-uploadNewCredentials-calls | meter | calls to uploadNewCredentials thrift method. |
| nimbus:process-worker-metric-calls | meter | calls to processWorkerMetrics thrift method. |
+| nimbus:mkAssignments-Errors | meter | tracks exceptions from mkAssignments |
| nimbus:topology-scheduling-duration-ms | timer | time it takes to do a scheduling run. |
| nimbus:total-available-memory-non-negative | gauge | available memory on the cluster MB |
| nimbuses:uptime-secs | histogram | uptime of nimbuses |
http://git-wip-us.apache.org/repos/asf/storm/blob/d7d2c623/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------
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 d66c0ef..98ecfe3 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
@@ -265,6 +265,8 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
private final Meter getOwnerResourceSummariesCalls;
private final Meter shutdownCalls;
private final Meter processWorkerMetricsCalls;
+ private final Meter mkAssignmentsErrors;
+
//Timer
private final Timer fileUploadDuration;
private final Timer schedulingDuration;
@@ -511,6 +513,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
"nimbus:num-getOwnerResourceSummaries-calls");
this.shutdownCalls = metricsRegistry.registerMeter("nimbus:num-shutdown-calls");
this.processWorkerMetricsCalls = metricsRegistry.registerMeter("nimbus:process-worker-metric-calls");
+ this.mkAssignmentsErrors = metricsRegistry.registerMeter("nimbus:mkAssignments-Errors");
this.fileUploadDuration = metricsRegistry.registerTimer("nimbus:files-upload-duration-ms");
this.schedulingDuration = metricsRegistry.registerTimer("nimbus:topology-scheduling-duration-ms");
this.numAddedExecPerScheduling = metricsRegistry.registerHistogram("nimbus:num-added-executors-per-scheduling");
@@ -2166,64 +2169,77 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
}
private void mkAssignments(String scratchTopoId) throws Exception {
- if (!isReadyForMKAssignments()) {
- return;
- }
- // get existing assignment (just the topologyToExecutorToNodePort map) -> default to {}
- // filter out ones which have a executor timeout
- // figure out available slots on cluster. add to that the used valid slots to get total slots. figure out how many executors
- // should be in each slot (e.g., 4, 4, 4, 5)
- // only keep existing slots that satisfy one of those slots. for rest, reassign them across remaining slots
- // edge case for slots with no executor timeout but with supervisor timeout... just treat these as valid slots that can be
- // reassigned to. worst comes to worse the executor will timeout and won't assign here next time around
+ try {
+ if (!isReadyForMKAssignments()) {
+ return;
+ }
+ // get existing assignment (just the topologyToExecutorToNodePort map) -> default to {}
+ // filter out ones which have a executor timeout
+ // figure out available slots on cluster. add to that the used valid slots to get total slots. figure out how many executors
+ // should be in each slot (e.g., 4, 4, 4, 5)
+ // only keep existing slots that satisfy one of those slots. for rest, reassign them across remaining slots
+ // edge case for slots with no executor timeout but with supervisor timeout... just treat these as valid slots that can be
+ // reassigned to. worst comes to worse the executor will timeout and won't assign here next time around
- IStormClusterState state = stormClusterState;
- //read all the topologies
- Map<String, StormBase> bases;
- Map<String, TopologyDetails> tds = new HashMap<>();
- synchronized (submitLock) {
- // should promote: only fetch storm bases of topologies that need scheduling.
- bases = state.topologyBases();
+ IStormClusterState state = stormClusterState;
+ //read all the topologies
+ Map<String, StormBase> bases;
+ Map<String, TopologyDetails> tds = new HashMap<>();
+ synchronized (submitLock) {
+ // should promote: only fetch storm bases of topologies that need scheduling.
+ bases = state.topologyBases();
- for (Iterator<Entry<String, StormBase>> it = bases.entrySet().iterator(); it.hasNext(); ) {
- Entry<String, StormBase> entry = it.next();
- String id = entry.getKey();
- try {
- tds.put(id, readTopologyDetails(id, entry.getValue()));
- } catch (KeyNotFoundException e) {
- //A race happened and it is probably not running
- it.remove();
+ for (Iterator<Entry<String, StormBase>> it = bases.entrySet().iterator(); it.hasNext(); ) {
+ Entry<String, StormBase> entry = it.next();
+ String id = entry.getKey();
+ try {
+ tds.put(id, readTopologyDetails(id, entry.getValue()));
+ } catch (KeyNotFoundException e) {
+ //A race happened and it is probably not running
+ it.remove();
+ }
}
}
- }
- Topologies topologies = new Topologies(tds);
- List<String> assignedTopologyIds = state.assignments(null);
- Map<String, Assignment> existingAssignments = new HashMap<>();
- for (String id : assignedTopologyIds) {
- //for the topology which wants rebalance (specified by the scratchTopoId)
- // we exclude its assignment, meaning that all the slots occupied by its assignment
- // will be treated as free slot in the scheduler code.
- if (!id.equals(scratchTopoId)) {
- Assignment currentAssignment = state.assignmentInfo(id, null);
- if (!currentAssignment.is_set_owner()) {
- TopologyDetails td = tds.get(id);
- if (td != null) {
- currentAssignment.set_owner(td.getTopologySubmitter());
- state.setAssignment(id, currentAssignment, td.getConf());
+ List<String> assignedTopologyIds = state.assignments(null);
+ Map<String, Assignment> existingAssignments = new HashMap<>();
+ for (String id : assignedTopologyIds) {
+ //for the topology which wants rebalance (specified by the scratchTopoId)
+ // we exclude its assignment, meaning that all the slots occupied by its assignment
+ // will be treated as free slot in the scheduler code.
+ if (!id.equals(scratchTopoId)) {
+ Assignment currentAssignment = state.assignmentInfo(id, null);
+ if (!currentAssignment.is_set_owner()) {
+ TopologyDetails td = tds.get(id);
+ if (td != null) {
+ currentAssignment.set_owner(td.getTopologySubmitter());
+ state.setAssignment(id, currentAssignment, td.getConf());
+ }
}
+ existingAssignments.put(id, currentAssignment);
}
- existingAssignments.put(id, currentAssignment);
}
+
+ // make the new assignments for topologies
+ lockingMkAssignments(existingAssignments, bases, scratchTopoId, assignedTopologyIds, state, tds);
+ } catch (Exception e) {
+ this.mkAssignmentsErrors.mark();
+ throw e;
}
- // make the new assignments for topologies
+ }
+
+ private void lockingMkAssignments(Map<String, Assignment> existingAssignments, Map<String, StormBase> bases,
+ String scratchTopoId, List<String> assignedTopologyIds, IStormClusterState state,
+ Map<String, TopologyDetails> tds) throws Exception {
+ Topologies topologies = new Topologies(tds);
+
synchronized (schedLock) {
Map<String, SchedulerAssignment> newSchedulerAssignments =
- computeNewSchedulerAssignments(existingAssignments, topologies, bases, scratchTopoId);
+ computeNewSchedulerAssignments(existingAssignments, topologies, bases, scratchTopoId);
Map<String, Map<List<Long>, List<Object>>> topologyToExecutorToNodePort =
- computeTopoToExecToNodePort(newSchedulerAssignments, assignedTopologyIds);
+ computeTopoToExecToNodePort(newSchedulerAssignments, assignedTopologyIds);
Map<String, Map<WorkerSlot, WorkerResources>> newAssignedWorkerToResources =
- computeTopoToNodePortToResources(newSchedulerAssignments);
+ computeTopoToNodePortToResources(newSchedulerAssignments);
int nowSecs = Time.currentTimeSecs();
Map<String, SupervisorDetails> basicSupervisorDetailsMap = basicSupervisorDetailsMap(state);
//construct the final Assignments by adding start-times etc into it
@@ -2330,7 +2346,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
totalAssignmentsChangedNodes.putAll(assignmentChangedNodes(existingAssignment, assignment));
}
notifySupervisorsAssignments(newAssignments, assignmentsDistributer, totalAssignmentsChangedNodes,
- basicSupervisorDetailsMap);
+ basicSupervisorDetailsMap);
Map<String, Collection<WorkerSlot>> addedSlots = new HashMap<>();
for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
[2/2] storm git commit: Merge branch 'agresch_mkAssignments_fail' of
https://github.com/agresch/storm into STORM-3237
Posted by bo...@apache.org.
Merge branch 'agresch_mkAssignments_fail' of https://github.com/agresch/storm into STORM-3237
STORM-3237: track Nimbus mkAssignment failures
This closes #2852
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/c47c23cf
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/c47c23cf
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/c47c23cf
Branch: refs/heads/master
Commit: c47c23cf8140db59c2ad76bd3ac5ef17ceb8fad4
Parents: c367f9a d7d2c62
Author: Robert Evans <ev...@yahoo-inc.com>
Authored: Mon Oct 1 09:27:44 2018 -0500
Committer: Robert Evans <ev...@yahoo-inc.com>
Committed: Mon Oct 1 09:27:44 2018 -0500
----------------------------------------------------------------------
docs/ClusterMetrics.md | 1 +
.../org/apache/storm/daemon/nimbus/Nimbus.java | 108 +++++++++++--------
2 files changed, 63 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/c47c23cf/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------