You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by zd-project <gi...@git.apache.org> on 2018/07/13 19:51:37 UTC
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary to StormMetricsReg...
GitHub user zd-project opened a pull request:
https://github.com/apache/storm/pull/2764
STORM-3147: Port ClusterSummary to StormMetricsRegistry
This PR depends on #2763
The implementation is kind of ugly right now due to caching and synchronization in metrics update. I hope you can give out some advice in improvement. @revans2 @HeartSaVioR @srdo
@kishorvpatil I've listed out all metrics I think worth implementing from ClusterSummay in Measured enum. I'm not sure if there's value in SUPERVISOR_TOTAL_RESOURCE, TOPOLOGY_STATUS, and TOPOLOGY_SCHED_STATUS. I'd like to hear what you think.
I have also discovered a potential bug of scheduler when working on this [STORM-3151](https://issues.apache.org/jira/browse/STORM-3151), can you help confirm it?
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zd-project/storm STORM-3147
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/2764.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2764
----
commit 802014c88a67ae5ef044c1f87889d976063eee71
Author: Zhengdai Hu <hu...@...>
Date: 2018-07-12T16:31:00Z
STORM-3150: Improve Gauge registration methods and refactored code
commit 4a8f63066dcede70a8df483383d6b2bcd616d299
Author: Zhengdai Hu <hu...@...>
Date: 2018-07-13T19:22:20Z
STORM-3147: Port ClusterSummary to StormMetricsRegistry
----
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209381686
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4744,206 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ //This histogram reflects the data distribution across only one ClusterSummary, i.e.,
+ // data distribution across all entities of a type (e.g., data from all nimbus/topologies) at one moment.
+ // Hence we use half of the CACHING_WINDOW time to ensure it retains only data from the most recent update
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ /**
+ * Constructor to put all items in ClusterSummary in MetricSet as a metric.
+ * All metrics are derived from a cached ClusterSummary object,
+ * expired {@link ClusterSummaryMetricSet#CACHING_WINDOW} seconds after first query in a while from reporters.
+ * In case of {@link com.codahale.metrics.ScheduledReporter}, CACHING_WINDOW should be set shorter than
+ * reporting interval to avoid outdated reporting.
+ */
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.debug("the new summary is {}", newSummary);
+ //Force update histogram upon each cache refresh
+ //This behavior relies on the fact that most common implementation of Reporter
--- End diff --
How about making a repeating timer task that fetches the cluster summary, updates the histograms, and updates fields that are then read by the gauges?
The reason I don't like this is that it makes a lot of assumptions about the internals of code we don't control. If the reporter looks at histograms before gauges, you get the issue you mention (and keep in mind anyone can implement a reporter). If the DerivativeGauge were implemented differently (e.g. if the parent gauge pushed the value to the child, rather than the other way around) this also would break.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209039674
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2918,6 +2943,10 @@ public void launchServer() throws Exception {
}
});
}
+
+ //Should we make the delaySecs and recurSecs in sync with any conf value?
+ // They should be around the reporting interval, but it's not configurable
+ timer.scheduleRecurring(5, 5, clusterMetricSet);
--- End diff --
It's a random number, really.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209337293
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4746,186 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.info("the new summary is {}", newSummary);
+ //This is ugly but I can't think of a better way to update histogram only once per caching
+ // It also kind of depends on the implementation that gauges gets updated before histograms
+ updateHistogram(newSummary);
+ return newSummary;
+ } else {
+ return null;
+ }
+ } catch (Exception e) {
+ LOG.warn("Get cluster info exception.", e);
+ throw new RuntimeException(e);
+ }
+ }
+ };
+
+ clusterSummaryMetrics.put("cluster:num-nimbus-leaders", new DerivativeGauge<ClusterSummary, Long>(cachedSummary) {
+ @Override
+ protected Long transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_nimbuses().stream().filter(NimbusSummary::is_isLeader).count();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-nimbuses", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_nimbuses_size();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-supervisors", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors_size();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-topologies", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_topologies_size();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-total-workers", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors().stream().mapToInt(SupervisorSummary::get_num_workers).sum();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-total-used-workers", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors().stream().mapToInt(SupervisorSummary::get_num_used_workers).sum();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:total-fragmented-memory-non-negative", new DerivativeGauge<ClusterSummary, Double>(cachedSummary) {
+ @Override
+ protected Double transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors().stream()
+ //Filtered negative value
+ .mapToDouble(supervisorSummary -> Math.max(supervisorSummary.get_fragmented_mem(), 0)).sum();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:total-fragmented-cpu-non-negative", new DerivativeGauge<ClusterSummary, Double>(cachedSummary) {
+ @Override
+ protected Double transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors().stream()
+ //Filtered negative value
+ .mapToDouble(supervisorSummary -> Math.max(supervisorSummary.get_fragmented_cpu(), 0)).sum();
+ }
+ });
+ }
+
+ private void updateHistogram(ClusterSummary newSummary) {
+ for (NimbusSummary nimbusSummary : newSummary.get_nimbuses()) {
+ nimbusUptime.update(nimbusSummary.get_uptime_secs());
+ }
+ for (SupervisorSummary summary : newSummary.get_supervisors()) {
+ supervisorsUptime.update(summary.get_uptime_secs());
+ supervisorsNumWorkers.update(summary.get_num_workers());
+ supervisorsNumUsedWorkers.update(summary.get_num_used_workers());
+ supervisorsUsedMem.update(Math.round(summary.get_used_mem()));
+ supervisorsUsedCpu.update(Math.round(summary.get_used_cpu()));
+ supervisorsFragmentedMem.update(Math.round(summary.get_fragmented_mem()));
+ supervisorsFragmentedCpu.update(Math.round(summary.get_fragmented_cpu()));
+ }
+ for (TopologySummary summary : newSummary.get_topologies()) {
+ topologiesNumTasks.update(summary.get_num_tasks());
+ topologiesNumExecutors.update(summary.get_num_executors());
+ topologiesNumWorker.update(summary.get_num_workers());
+ topologiesUptime.update(summary.get_uptime_secs());
+ topologiesReplicationCount.update(summary.get_replication_count());
+ topologiesRequestedMemOnHeap.update(Math.round(summary.get_requested_memonheap()));
+ topologiesRequestedMemOffHeap.update(Math.round(summary.get_requested_memoffheap()));
+ topologiesRequestedCpu.update(Math.round(summary.get_requested_cpu()));
+ topologiesAssignedMemOnHeap.update(Math.round(summary.get_assigned_memonheap()));
+ topologiesAssignedMemOffHeap.update(Math.round(summary.get_assigned_memoffheap()));
+ topologiesAssignedCpu.update(Math.round(summary.get_assigned_cpu()));
+ }
+ }
+
+ //This is not thread safe
--- End diff --
Like I said. This implementation isn't pretty, and I hope it can be replace with some better implementation later. It is not thread safe but with current implementation it is guaranteed to be called only by one thread at a time, and I don't see anywhere else we'll be using it. So I think it should be good.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208986783
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
--- End diff --
`ported` is not clear out of context of this PR. better to use something like `clusterSummaryMetrics`
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209341987
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2865,16 +2890,14 @@ public void launchServer() throws Exception {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
- StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", this::fragmentedMemory);
- StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", this::fragmentedCpu);
- StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values()
+ // Num supervisor, and fragmented resources have been included in cluster summary
--- End diff --
Please remove, unless you're looking at the diff, you won't know what this means
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208998511
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
--- End diff --
Sure, sounds good.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209333088
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4746,186 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.info("the new summary is {}", newSummary);
+ //This is ugly but I can't think of a better way to update histogram only once per caching
+ // It also kind of depends on the implementation that gauges gets updated before histograms
+ updateHistogram(newSummary);
+ return newSummary;
+ } else {
+ return null;
+ }
+ } catch (Exception e) {
+ LOG.warn("Get cluster info exception.", e);
+ throw new RuntimeException(e);
+ }
+ }
+ };
+
+ clusterSummaryMetrics.put("cluster:num-nimbus-leaders", new DerivativeGauge<ClusterSummary, Long>(cachedSummary) {
+ @Override
+ protected Long transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_nimbuses().stream().filter(NimbusSummary::is_isLeader).count();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-nimbuses", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_nimbuses_size();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-supervisors", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors_size();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-topologies", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_topologies_size();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-total-workers", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors().stream().mapToInt(SupervisorSummary::get_num_workers).sum();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:num-total-used-workers", new DerivativeGauge<ClusterSummary, Integer>(cachedSummary) {
+ @Override
+ protected Integer transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors().stream().mapToInt(SupervisorSummary::get_num_used_workers).sum();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:total-fragmented-memory-non-negative", new DerivativeGauge<ClusterSummary, Double>(cachedSummary) {
+ @Override
+ protected Double transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors().stream()
+ //Filtered negative value
+ .mapToDouble(supervisorSummary -> Math.max(supervisorSummary.get_fragmented_mem(), 0)).sum();
+ }
+ });
+ clusterSummaryMetrics.put("cluster:total-fragmented-cpu-non-negative", new DerivativeGauge<ClusterSummary, Double>(cachedSummary) {
+ @Override
+ protected Double transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_supervisors().stream()
+ //Filtered negative value
+ .mapToDouble(supervisorSummary -> Math.max(supervisorSummary.get_fragmented_cpu(), 0)).sum();
+ }
+ });
+ }
+
+ private void updateHistogram(ClusterSummary newSummary) {
+ for (NimbusSummary nimbusSummary : newSummary.get_nimbuses()) {
+ nimbusUptime.update(nimbusSummary.get_uptime_secs());
+ }
+ for (SupervisorSummary summary : newSummary.get_supervisors()) {
+ supervisorsUptime.update(summary.get_uptime_secs());
+ supervisorsNumWorkers.update(summary.get_num_workers());
+ supervisorsNumUsedWorkers.update(summary.get_num_used_workers());
+ supervisorsUsedMem.update(Math.round(summary.get_used_mem()));
+ supervisorsUsedCpu.update(Math.round(summary.get_used_cpu()));
+ supervisorsFragmentedMem.update(Math.round(summary.get_fragmented_mem()));
+ supervisorsFragmentedCpu.update(Math.round(summary.get_fragmented_cpu()));
+ }
+ for (TopologySummary summary : newSummary.get_topologies()) {
+ topologiesNumTasks.update(summary.get_num_tasks());
+ topologiesNumExecutors.update(summary.get_num_executors());
+ topologiesNumWorker.update(summary.get_num_workers());
+ topologiesUptime.update(summary.get_uptime_secs());
+ topologiesReplicationCount.update(summary.get_replication_count());
+ topologiesRequestedMemOnHeap.update(Math.round(summary.get_requested_memonheap()));
+ topologiesRequestedMemOffHeap.update(Math.round(summary.get_requested_memoffheap()));
+ topologiesRequestedCpu.update(Math.round(summary.get_requested_cpu()));
+ topologiesAssignedMemOnHeap.update(Math.round(summary.get_assigned_memonheap()));
+ topologiesAssignedMemOffHeap.update(Math.round(summary.get_assigned_memoffheap()));
+ topologiesAssignedCpu.update(Math.round(summary.get_assigned_cpu()));
+ }
+ }
+
+ //This is not thread safe
--- End diff --
Why not make this thread safe, isn't the only issue here that `active` isn't an AtomicBoolean?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208749367
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2219,21 +2305,16 @@ private void mkAssignments(String scratchTopoId) throws Exception {
newAssignments.put(topoId, newAssignment);
}
- if (!newAssignments.equals(existingAssignments)) {
+ boolean assignmentChanged = inspectSchduling(existingAssignments, newAssignments);
+ if (assignmentChanged) {
LOG.debug("RESETTING id->resources and id->worker-resources cache!");
- LOG.info("Fragmentation after scheduling is: {} MB, {} PCore CPUs", fragmentedMemory(), fragmentedCpu());
- nodeIdToResources.get().forEach((id, node) ->
- LOG.info(
- "Node Id: {} Total Mem: {}, Used Mem: {}, Available Mem: {}, Total CPU: {}, Used "
- + "CPU: {}, Available CPU: {}, fragmented: {}",
- id, node.getTotalMem(), node.getUsedMem(), node.getAvailableMem(),
- node.getTotalCpu(), node.getUsedCpu(), node.getAvailableCpu(), isFragmented(node)));
idToResources.set(new HashMap<>());
idToWorkerResources.set(new HashMap<>());
}
//tasks figure out what tasks to talk to by looking at topology at runtime
// only log/set when there's been a change to the assignment
+ // TODO: why do we have loop fission here
--- End diff --
Will log in Jira
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208718858
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnLeft().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
}
- Map<List<Long>, NodeInfo> old = assignment.get_executor_node_port();
- Map<List<Long>, List<Object>> reassigned = new HashMap<>();
- for (Entry<List<Long>, List<Object>> execAndNodePort : execToNodePort.entrySet()) {
- NodeInfo oldAssigned = old.get(execAndNodePort.getKey());
- String node = (String) execAndNodePort.getValue().get(0);
- Long port = (Long) execAndNodePort.getValue().get(1);
- if (oldAssigned == null || !oldAssigned.get_node().equals(node)
- || !port.equals(oldAssigned.get_port_iterator().next())) {
- reassigned.put(execAndNodePort.getKey(), execAndNodePort.getValue());
- }
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnRight().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
}
- if (!reassigned.isEmpty()) {
- int count = (new HashSet<>(execToNodePort.values())).size();
- Set<List<Long>> reExecs = reassigned.keySet();
- LOG.info("Reassigning {} to {} slots", topoId, count);
- LOG.info("Reassign executors: {}", reExecs);
+ for (Entry<String, MapDifference.ValueDifference<Assignment>> entry : difference.entriesDiffering().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToSlot = entry.getValue().rightValue().get_executor_node_port();
+ final Set<NodeInfo> slots = new HashSet<>(execToSlot.values());
+ LOG.info("Reassigning {} to {} slots", entry.getKey(), slots.size());
+ LOG.info("Reassign executors: {}", execToSlot.keySet());
+
+ //We can simplify this part if scheduler does't partially remove or add executor
+ final Map<List<Long>, NodeInfo> oldExecToSlot = entry.getValue().leftValue().get_executor_node_port();
+
+ final Map<List<Long>, NodeInfo> smaller = execToSlot.size() > oldExecToSlot.size() ? oldExecToSlot : execToSlot;
--- End diff --
I think you can get the same result by getting the common keys for execToSlot and oldExecToSlot, and iterating that set and comparing the values.
```
Set<List<Long>> common = new HashSet<>(execToSlot.keySet());
common.retainAll(oldExecToSlot.keySet());
for (List<Long> exec : common) {
if (execToSlot.get(exec).equals(oldExecToSlot.get(exec))) {
...
}
}
```
Disregard if it's written this way for performance, but I think using the set intersection would be easier to read.
---
[GitHub] storm issue #2764: STORM-3147: Port ClusterSummary as metrics to StormMetric...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on the issue:
https://github.com/apache/storm/pull/2764
cachedSummary shouldn't be refreshed too often based on current implementation. So I don't think we should risk the race condition just for negligible gain in performance.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209154387
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4652,4 +4745,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
--- End diff --
Then `isLeader` can be made public or protected, and Nimbus can be injected into this class via a constructor.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209367723
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4744,206 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ //This histogram reflects the data distribution across only one ClusterSummary, i.e.,
+ // data distribution across all entities of a type (e.g., data from all nimbus/topologies) at one moment.
+ // Hence we use half of the CACHING_WINDOW time to ensure it retains only data from the most recent update
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ /**
+ * Constructor to put all items in ClusterSummary in MetricSet as a metric.
+ * All metrics are derived from a cached ClusterSummary object,
+ * expired {@link ClusterSummaryMetricSet#CACHING_WINDOW} seconds after first query in a while from reporters.
+ * In case of {@link com.codahale.metrics.ScheduledReporter}, CACHING_WINDOW should be set shorter than
+ * reporting interval to avoid outdated reporting.
+ */
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.debug("the new summary is {}", newSummary);
+ //Force update histogram upon each cache refresh
+ //This behavior relies on the fact that most common implementation of Reporter
--- End diff --
This gauge is never registered. It simply serves as a cache for other gauge to use. Otherwise we'd have to come up with our own caching mechanism.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208703156
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -780,15 +870,15 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
key.add(ni.get_node());
key.add(ni.get_port_iterator().next());
List<List<Long>> value = new ArrayList<>(entry.getValue());
- value.sort((a, b) -> a.get(0).compareTo(b.get(0)));
+ value.sort(Comparator.comparing(a -> a.get(0)));
--- End diff --
Neat, didn't know about this method.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208724033
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2826,9 +2915,22 @@ public void launchServer() throws Exception {
.parallelStream()
.mapToDouble(SupervisorResources::getTotalCpu)
.sum());
-
+ StormMetricsRegistry.registerGauge("nimbus:longest-scheduling-time-ms", () -> {
+ Long currTime = Time.nanoTime();
+ Long startTime = schedulingStartTime.get();
+ //There could be race condition here but seems trivial, elapsed is
+ // guaranteed to be no longer than real elapsed time of scheduling
+ Long longest = longestSchedulingTime.get();
+ if (startTime != null) {
+ longest = currTime - startTime > longest ? currTime - startTime : longest;
--- End diff --
I don't understand what is happening here. Why are we potentially calculating a new longest here?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209363589
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4744,206 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ //This histogram reflects the data distribution across only one ClusterSummary, i.e.,
+ // data distribution across all entities of a type (e.g., data from all nimbus/topologies) at one moment.
+ // Hence we use half of the CACHING_WINDOW time to ensure it retains only data from the most recent update
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ /**
+ * Constructor to put all items in ClusterSummary in MetricSet as a metric.
+ * All metrics are derived from a cached ClusterSummary object,
+ * expired {@link ClusterSummaryMetricSet#CACHING_WINDOW} seconds after first query in a while from reporters.
+ * In case of {@link com.codahale.metrics.ScheduledReporter}, CACHING_WINDOW should be set shorter than
+ * reporting interval to avoid outdated reporting.
+ */
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.debug("the new summary is {}", newSummary);
+ //Force update histogram upon each cache refresh
+ //This behavior relies on the fact that most common implementation of Reporter
--- End diff --
1. I think it'd be better to have all metrics in sync. Having a StormTimer is also problematic because there's no way to guarantee when histogram gets updated vs. when reporters decide to report metrics. We would then get out of sync metrics for gauge and histogram, which can be confusing.
2. Since for histogram we track distribution for only one ClusterSummary measurement. The out of sync behavior give us no guarantee if newer value will override older ones that haven't been reported, and it'd be hard to figure out the correct caching window.
This ultimately boils down to the difference between active update pattern of histogram vs. passive update pattern of gauge, and this is the best kind of hack I can think of as of now to mediate the conflict. An alternative would be to convert all histogram to gauge, which means manually calculate avg/min/max/percentile. This will bloat the code greatly and I don't see a very good reason doing it, almost like trading one hack for another.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209302512
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
--- End diff --
Okay
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209084882
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4652,4 +4745,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
--- End diff --
Can this class be moved to another file? Nimbus is already huge.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208743919
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java ---
@@ -52,6 +52,7 @@ public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf
public void start() {
if (reporter != null) {
LOG.debug("Starting...");
+ //TODO: will we make the period customizable?
--- End diff --
Will log in Jira.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208970103
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2807,16 +2897,15 @@ public void launchServer() throws Exception {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
- StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", this::fragmentedMemory);
- StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", this::fragmentedCpu);
- StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values()
+ //Be cautious using method reference instead of lambda. subexpression preceding :: will be evaluated only upon evaluation
+ // Num supervisor, and fragmented resources have been included in cluster summary
+ StormMetricsRegistry.registerGauge("nimbus:total-available-memory (nonegative)", () -> nodeIdToResources.get().values()
.parallelStream()
- .mapToDouble(SupervisorResources::getAvailableMem)
+ .mapToDouble(supervisorResources -> Math.max(supervisorResources.getAvailableMem(), 0))
.sum());
- StormMetricsRegistry.registerGauge("nimbus:available-cpu", () -> nodeIdToResources.get().values()
+ StormMetricsRegistry.registerGauge("nimbus:available-cpu (nonnegative)", () -> nodeIdToResources.get().values()
--- End diff --
Thanks. Sorry I was unclear. What I meant what I don't know if we need to worry about not changing the names of existing metrics (backward compatibility). I'm not sure why we want to change the name to include `(nonnegative)`, and if we want to include it, why not use the same syntax as the rest of the name (words separated by `-`)?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209312212
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4723,4 +4754,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
--- End diff --
fixed
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209304885
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4652,4 +4745,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
--- End diff --
Okay, that makes sense.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208714561
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnLeft().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
}
- Map<List<Long>, NodeInfo> old = assignment.get_executor_node_port();
- Map<List<Long>, List<Object>> reassigned = new HashMap<>();
- for (Entry<List<Long>, List<Object>> execAndNodePort : execToNodePort.entrySet()) {
- NodeInfo oldAssigned = old.get(execAndNodePort.getKey());
- String node = (String) execAndNodePort.getValue().get(0);
- Long port = (Long) execAndNodePort.getValue().get(1);
- if (oldAssigned == null || !oldAssigned.get_node().equals(node)
- || !port.equals(oldAssigned.get_port_iterator().next())) {
- reassigned.put(execAndNodePort.getKey(), execAndNodePort.getValue());
- }
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnRight().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
}
- if (!reassigned.isEmpty()) {
- int count = (new HashSet<>(execToNodePort.values())).size();
- Set<List<Long>> reExecs = reassigned.keySet();
- LOG.info("Reassigning {} to {} slots", topoId, count);
- LOG.info("Reassign executors: {}", reExecs);
+ for (Entry<String, MapDifference.ValueDifference<Assignment>> entry : difference.entriesDiffering().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToSlot = entry.getValue().rightValue().get_executor_node_port();
+ final Set<NodeInfo> slots = new HashSet<>(execToSlot.values());
+ LOG.info("Reassigning {} to {} slots", entry.getKey(), slots.size());
+ LOG.info("Reassign executors: {}", execToSlot.keySet());
+
+ //We can simplify this part if scheduler does't partially remove or add executor
--- End diff --
This seems like a TODO, remember to remove this.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209044589
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
--- End diff --
Right, that makes sense. I don't think it requires reordering though, the sequence you posted can happen without the compiler reordering anything.
I don't think it really hurts anything to have this race, but maybe we should consider removing the longest update in the gauge. Unless the scheduler is outright hanging, I'm not sure what the value is for us to report a partial measurement like that?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208746266
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
+ long elapsed = -schedulingStartTime.getAndSet(null) + Time.nanoTime();
--- End diff --
Oh my bad. This is probably premature optimization.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208706303
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
--- End diff --
Which race are you worried about here?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209034728
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
--- End diff --
If compiler is allowed to hoist code, we can ended up with the case:
1) gauge "longest-scheduling-time-ms" acquires scheduling start time
2) elpased capture end time, update `schedulingDuration` Timer and `longestSchedulingTime`
3) gauge acquires longest time from `longestSchedulingTime` and `currTime`
4) since start time isn't null, program execute statement
```java
longest = currTime - startTime > longest ? currTime - startTime : longest;
```
So it'll be slightly longer than the value of `longestSchedulingTime` for this round. Which means we might saw jiggles in gauge value
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209059400
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
+ long elapsed = -schedulingStartTime.getAndSet(null) + Time.nanoTime();
--- End diff --
Now I remembered. I did this to alter the evaluation order so that `#getAndSet` is evaluated first.
---
[GitHub] storm issue #2764: STORM-3147: Port ClusterSummary as metrics to StormMetric...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on the issue:
https://github.com/apache/storm/pull/2764
Please notice that only the last commit is under #2764, which was previously out of sync with #2754 because I don't want to have too many dependencies between PR. For issue specific to #2754 I think it's better to review it there. @srdo
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209311426
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4652,4 +4745,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
--- End diff --
Yes, I agree. The current implementation isn't ideal.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208721202
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2219,21 +2305,16 @@ private void mkAssignments(String scratchTopoId) throws Exception {
newAssignments.put(topoId, newAssignment);
}
- if (!newAssignments.equals(existingAssignments)) {
+ boolean assignmentChanged = inspectSchduling(existingAssignments, newAssignments);
+ if (assignmentChanged) {
LOG.debug("RESETTING id->resources and id->worker-resources cache!");
- LOG.info("Fragmentation after scheduling is: {} MB, {} PCore CPUs", fragmentedMemory(), fragmentedCpu());
- nodeIdToResources.get().forEach((id, node) ->
- LOG.info(
- "Node Id: {} Total Mem: {}, Used Mem: {}, Available Mem: {}, Total CPU: {}, Used "
- + "CPU: {}, Available CPU: {}, fragmented: {}",
- id, node.getTotalMem(), node.getUsedMem(), node.getAvailableMem(),
- node.getTotalCpu(), node.getUsedCpu(), node.getAvailableCpu(), isFragmented(node)));
idToResources.set(new HashMap<>());
idToWorkerResources.set(new HashMap<>());
}
//tasks figure out what tasks to talk to by looking at topology at runtime
// only log/set when there's been a change to the assignment
+ // TODO: why do we have loop fission here
--- End diff --
My guess would be for readability. If you want to refactor go ahead.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209336348
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2905,6 +2929,10 @@ public void launchServer() throws Exception {
throw new RuntimeException(e);
}
});
+
+ //Should we make the delaySecs and recurSecs in sync with any conf value?
--- End diff --
Okay, we can do it later. Filed https://issues.apache.org/jira/browse/STORM-3192. Please remove the comment since it seems like it wouldn't be useful outside this PR.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208748414
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
--- End diff --
Good catch. I'll use `areEqual` instead. This method is for inspecting the performance (throughput) of scheduler, specifically, how many assignments, workers and executors are changed. It doesn't affect any actual assignments results.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208749175
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnLeft().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
}
- Map<List<Long>, NodeInfo> old = assignment.get_executor_node_port();
- Map<List<Long>, List<Object>> reassigned = new HashMap<>();
- for (Entry<List<Long>, List<Object>> execAndNodePort : execToNodePort.entrySet()) {
- NodeInfo oldAssigned = old.get(execAndNodePort.getKey());
- String node = (String) execAndNodePort.getValue().get(0);
- Long port = (Long) execAndNodePort.getValue().get(1);
- if (oldAssigned == null || !oldAssigned.get_node().equals(node)
- || !port.equals(oldAssigned.get_port_iterator().next())) {
- reassigned.put(execAndNodePort.getKey(), execAndNodePort.getValue());
- }
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnRight().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
}
- if (!reassigned.isEmpty()) {
- int count = (new HashSet<>(execToNodePort.values())).size();
- Set<List<Long>> reExecs = reassigned.keySet();
- LOG.info("Reassigning {} to {} slots", topoId, count);
- LOG.info("Reassign executors: {}", reExecs);
+ for (Entry<String, MapDifference.ValueDifference<Assignment>> entry : difference.entriesDiffering().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToSlot = entry.getValue().rightValue().get_executor_node_port();
+ final Set<NodeInfo> slots = new HashSet<>(execToSlot.values());
+ LOG.info("Reassigning {} to {} slots", entry.getKey(), slots.size());
+ LOG.info("Reassign executors: {}", execToSlot.keySet());
+
+ //We can simplify this part if scheduler does't partially remove or add executor
+ final Map<List<Long>, NodeInfo> oldExecToSlot = entry.getValue().leftValue().get_executor_node_port();
+
+ final Map<List<Long>, NodeInfo> smaller = execToSlot.size() > oldExecToSlot.size() ? oldExecToSlot : execToSlot;
--- End diff --
Yeah this is probably premature optimization as well. I don't have an idea how drastically old and new assignments will vary in size. But if you don't think it's too big of a problem, I don't mind removing it.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209367613
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4744,206 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ //This histogram reflects the data distribution across only one ClusterSummary, i.e.,
+ // data distribution across all entities of a type (e.g., data from all nimbus/topologies) at one moment.
+ // Hence we use half of the CACHING_WINDOW time to ensure it retains only data from the most recent update
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ /**
+ * Constructor to put all items in ClusterSummary in MetricSet as a metric.
+ * All metrics are derived from a cached ClusterSummary object,
+ * expired {@link ClusterSummaryMetricSet#CACHING_WINDOW} seconds after first query in a while from reporters.
+ * In case of {@link com.codahale.metrics.ScheduledReporter}, CACHING_WINDOW should be set shorter than
+ * reporting interval to avoid outdated reporting.
+ */
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.debug("the new summary is {}", newSummary);
+ //Force update histogram upon each cache refresh
+ //This behavior relies on the fact that most common implementation of Reporter
--- End diff --
Nevermind, I should read a little closer.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209365696
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4744,206 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ //This histogram reflects the data distribution across only one ClusterSummary, i.e.,
+ // data distribution across all entities of a type (e.g., data from all nimbus/topologies) at one moment.
+ // Hence we use half of the CACHING_WINDOW time to ensure it retains only data from the most recent update
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ /**
+ * Constructor to put all items in ClusterSummary in MetricSet as a metric.
+ * All metrics are derived from a cached ClusterSummary object,
+ * expired {@link ClusterSummaryMetricSet#CACHING_WINDOW} seconds after first query in a while from reporters.
+ * In case of {@link com.codahale.metrics.ScheduledReporter}, CACHING_WINDOW should be set shorter than
+ * reporting interval to avoid outdated reporting.
+ */
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.debug("the new summary is {}", newSummary);
+ //Force update histogram upon each cache refresh
+ //This behavior relies on the fact that most common implementation of Reporter
--- End diff --
We can also develop a passive histogram that simply wraps a round a gauge. It's better to file another Jira then.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209029787
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4723,4 +4754,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
--- End diff --
There has been inconsistency on how I should naming the metrics. In previous commits I have all of the metrics in metric set starting with prefix "summary.". But I'm not sure.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208720133
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
--- End diff --
The name is a little vague. How about `calculateAssignmentChanged` or something like that? Also there's a missing e in scheduling.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208986545
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
--- End diff --
Don't need constant since it's only an initial capacity of map
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209007420
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4723,4 +4754,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
--- End diff --
this is not used
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209028432
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2871,16 +2897,15 @@ public void launchServer() throws Exception {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
- StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", this::fragmentedMemory);
- StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", this::fragmentedCpu);
- StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values()
+ //Be cautious using method reference instead of lambda. subexpression preceding :: will be evaluated only upon evaluation
+ // Num supervisor, and fragmented resources have been included in cluster summary
+ StormMetricsRegistry.registerGauge("nimbus:total-available-memory (nonegative)", () -> nodeIdToResources.get().values()
--- End diff --
See https://issues.apache.org/jira/browse/STORM-3151?filter=-2
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208704942
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2826,9 +2915,22 @@ public void launchServer() throws Exception {
.parallelStream()
.mapToDouble(SupervisorResources::getTotalCpu)
.sum());
-
+ StormMetricsRegistry.registerGauge("nimbus:longest-scheduling-time-ms", () -> {
+ Long currTime = Time.nanoTime();
+ Long startTime = schedulingStartTime.get();
+ //There could be race condition here but seems trivial, elapsed is
+ // guaranteed to be no longer than real elapsed time of scheduling
+ Long longest = longestSchedulingTime.get();
+ if (startTime != null) {
+ longest = currTime - startTime > longest ? currTime - startTime : longest;
+ }
+ //To millis. How should I put the constant for magic numbers?
--- End diff --
You can use TimeUnit instead to do conversions, removing the need for magic numbers https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/TimeUnit.html
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209367121
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4744,206 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ //This histogram reflects the data distribution across only one ClusterSummary, i.e.,
+ // data distribution across all entities of a type (e.g., data from all nimbus/topologies) at one moment.
+ // Hence we use half of the CACHING_WINDOW time to ensure it retains only data from the most recent update
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ /**
+ * Constructor to put all items in ClusterSummary in MetricSet as a metric.
+ * All metrics are derived from a cached ClusterSummary object,
+ * expired {@link ClusterSummaryMetricSet#CACHING_WINDOW} seconds after first query in a while from reporters.
+ * In case of {@link com.codahale.metrics.ScheduledReporter}, CACHING_WINDOW should be set shorter than
+ * reporting interval to avoid outdated reporting.
+ */
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.debug("the new summary is {}", newSummary);
+ //Force update histogram upon each cache refresh
+ //This behavior relies on the fact that most common implementation of Reporter
--- End diff --
What gauged are you talking about? Basically my envisioning is that we subclass a histogram that simply wraps around many derivative gauges which computes avg/min/max on their own. They are encapsulated as a ReservoirSnapshot upon the histogram is queried. On the other hand we disable the `update` method so they can't be actively updated.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208972356
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnLeft().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
}
- Map<List<Long>, NodeInfo> old = assignment.get_executor_node_port();
- Map<List<Long>, List<Object>> reassigned = new HashMap<>();
- for (Entry<List<Long>, List<Object>> execAndNodePort : execToNodePort.entrySet()) {
- NodeInfo oldAssigned = old.get(execAndNodePort.getKey());
- String node = (String) execAndNodePort.getValue().get(0);
- Long port = (Long) execAndNodePort.getValue().get(1);
- if (oldAssigned == null || !oldAssigned.get_node().equals(node)
- || !port.equals(oldAssigned.get_port_iterator().next())) {
- reassigned.put(execAndNodePort.getKey(), execAndNodePort.getValue());
- }
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnRight().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
}
- if (!reassigned.isEmpty()) {
- int count = (new HashSet<>(execToNodePort.values())).size();
- Set<List<Long>> reExecs = reassigned.keySet();
- LOG.info("Reassigning {} to {} slots", topoId, count);
- LOG.info("Reassign executors: {}", reExecs);
+ for (Entry<String, MapDifference.ValueDifference<Assignment>> entry : difference.entriesDiffering().entrySet()) {
--- End diff --
Thanks, I didn't look at the Javadoc and misunderstood what this did.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209303203
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4652,4 +4745,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
--- End diff --
The intention is we fixed it in development stage so it'll just work in production, since thrift protocol is compiled, not evaluated at runtime.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208705183
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
+ long elapsed = -schedulingStartTime.getAndSet(null) + Time.nanoTime();
--- End diff --
This is a weird way to write this. Why `-a + b` rather than `b - a`?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208722273
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2807,16 +2897,15 @@ public void launchServer() throws Exception {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
- StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", this::fragmentedMemory);
- StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", this::fragmentedCpu);
- StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values()
+ //Be cautious using method reference instead of lambda. subexpression preceding :: will be evaluated only upon evaluation
+ // Num supervisor, and fragmented resources have been included in cluster summary
+ StormMetricsRegistry.registerGauge("nimbus:total-available-memory (nonegative)", () -> nodeIdToResources.get().values()
.parallelStream()
- .mapToDouble(SupervisorResources::getAvailableMem)
+ .mapToDouble(supervisorResources -> Math.max(supervisorResources.getAvailableMem(), 0))
.sum());
- StormMetricsRegistry.registerGauge("nimbus:available-cpu", () -> nodeIdToResources.get().values()
+ StormMetricsRegistry.registerGauge("nimbus:available-cpu (nonnegative)", () -> nodeIdToResources.get().values()
--- End diff --
I don't know whether we need to worry about this, but do we need to change the metric names here? If so, I'd rather build the non-negativity into the name, e.g. `nimbus:available-cpu-non-negative`.
---
[GitHub] storm issue #2764: STORM-3147: Port ClusterSummary as metrics to StormMetric...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on the issue:
https://github.com/apache/storm/pull/2764
What do you think @srdo
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208749692
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2807,16 +2897,15 @@ public void launchServer() throws Exception {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
- StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", this::fragmentedMemory);
- StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", this::fragmentedCpu);
- StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values()
+ //Be cautious using method reference instead of lambda. subexpression preceding :: will be evaluated only upon evaluation
+ // Num supervisor, and fragmented resources have been included in cluster summary
+ StormMetricsRegistry.registerGauge("nimbus:total-available-memory (nonegative)", () -> nodeIdToResources.get().values()
.parallelStream()
- .mapToDouble(SupervisorResources::getAvailableMem)
+ .mapToDouble(supervisorResources -> Math.max(supervisorResources.getAvailableMem(), 0))
.sum());
- StormMetricsRegistry.registerGauge("nimbus:available-cpu", () -> nodeIdToResources.get().values()
+ StormMetricsRegistry.registerGauge("nimbus:available-cpu (nonnegative)", () -> nodeIdToResources.get().values()
--- End diff --
See Jira [STORM-3151](https://issues.apache.org/jira/browse/STORM-3151?filter=-2)
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208968787
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2807,16 +2897,15 @@ public void launchServer() throws Exception {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
- StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", this::fragmentedMemory);
- StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", this::fragmentedCpu);
- StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values()
+ //Be cautious using method reference instead of lambda. subexpression preceding :: will be evaluated only upon evaluation
--- End diff --
Yes, I think I've seen it before too :)
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208747514
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
--- End diff --
Okay will come up with a better name.
---
[GitHub] storm issue #2764: STORM-3147: Port ClusterSummary as metrics to StormMetric...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2764
I'd like https://github.com/apache/storm/pull/2754 to go in first, so we're sure the review from there is reflected here. When that one is merged, I'd like to take another look at these changes.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209081264
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
+ long elapsed = -schedulingStartTime.getAndSet(null) + Time.nanoTime();
--- End diff --
Okay, but I don't see how that fixes the race?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208746090
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2826,9 +2915,22 @@ public void launchServer() throws Exception {
.parallelStream()
.mapToDouble(SupervisorResources::getTotalCpu)
.sum());
-
+ StormMetricsRegistry.registerGauge("nimbus:longest-scheduling-time-ms", () -> {
+ Long currTime = Time.nanoTime();
+ Long startTime = schedulingStartTime.get();
+ //There could be race condition here but seems trivial, elapsed is
+ // guaranteed to be no longer than real elapsed time of scheduling
+ Long longest = longestSchedulingTime.get();
+ if (startTime != null) {
+ longest = currTime - startTime > longest ? currTime - startTime : longest;
+ }
+ //To millis. How should I put the constant for magic numbers?
--- End diff --
I'm concerned of loss of precision here since all of them return long instead of double after conversion.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208705728
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
+ long elapsed = -schedulingStartTime.getAndSet(null) + Time.nanoTime();
+ longestSchedulingTime.updateAndGet(t -> t > elapsed ? t : elapsed);
--- End diff --
Nit: Could use Math.max for this.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208965019
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2918,6 +2943,10 @@ public void launchServer() throws Exception {
}
});
}
+
+ //Should we make the delaySecs and recurSecs in sync with any conf value?
+ // They should be around the reporting interval, but it's not configurable
+ timer.scheduleRecurring(5, 5, clusterMetricSet);
--- End diff --
why do we set to 5 and 5
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209100932
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4652,4 +4745,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
--- End diff --
ClusterSummaryMetricSet rely on `isLeader` from Nimbus.
---
[GitHub] storm issue #2764: STORM-3147: Port ClusterSummary as metrics to StormMetric...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on the issue:
https://github.com/apache/storm/pull/2764
Has been rebased on 3133 now.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209041653
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4723,4 +4754,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-CPU");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-CPU");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-CPU");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-CPU");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.info("the new summary is {}", newSummary);
+ //This is ugly but I can't think of a better way to update histogram only once per caching
+ // It also kind of depends on the implementation that gauges gets updated before histograms
+ updateHistogram(newSummary);
--- End diff --
Yes, because I want to collect the distribution of value (let's say the assigned cpu) across all entities (daemons, or topologies) only within one clusterSummary (so it's the distribution of assigned cpu among all supervisors at one moment, snapshotted in ClusterSummary). It doesn't make sense to me to accumulate both across entities and time.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208967564
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
--- End diff --
Makes sense, but what I'm asking is why is this check here at all? if the maps are equal, the for loops below will all be skipped anyway, because entriesOnlyOnLeft/Right and entriesDiffering are all empty, right?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208721916
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2807,16 +2897,15 @@ public void launchServer() throws Exception {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
- StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", this::fragmentedMemory);
- StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", this::fragmentedCpu);
- StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values()
+ //Be cautious using method reference instead of lambda. subexpression preceding :: will be evaluated only upon evaluation
--- End diff --
Nit: This is a hint about the language, I'd rather not have it here.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208707855
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2131,17 +2223,13 @@ private void mkAssignments(String scratchTopoId) throws Exception {
}
}
// make the new assignments for topologies
- Map<String, SchedulerAssignment> newSchedulerAssignments = null;
synchronized (schedLock) {
- newSchedulerAssignments = computeNewSchedulerAssignments(existingAssignments, topologies, bases, scratchTopoId);
+ Map<String, SchedulerAssignment> newSchedulerAssignments =
+ computeNewSchedulerAssignments(existingAssignments, topologies, bases, scratchTopoId);
+ //Should probably change List<Object> to Tuple<String, Long> for better readability
--- End diff --
If you want to do it, go ahead IMO.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209333494
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2905,6 +2929,10 @@ public void launchServer() throws Exception {
throw new RuntimeException(e);
}
});
+
+ //Should we make the delaySecs and recurSecs in sync with any conf value?
--- End diff --
Why not update the metricset active flag when Nimbus becomes active instead of polling for changes?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209064682
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
--- End diff --
You're right. I don't see a good fix. I agree that the race is not that important, I'd maybe just remove the comments, since users won't see them here.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209366165
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4744,206 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ //This histogram reflects the data distribution across only one ClusterSummary, i.e.,
+ // data distribution across all entities of a type (e.g., data from all nimbus/topologies) at one moment.
+ // Hence we use half of the CACHING_WINDOW time to ensure it retains only data from the most recent update
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ /**
+ * Constructor to put all items in ClusterSummary in MetricSet as a metric.
+ * All metrics are derived from a cached ClusterSummary object,
+ * expired {@link ClusterSummaryMetricSet#CACHING_WINDOW} seconds after first query in a while from reporters.
+ * In case of {@link com.codahale.metrics.ScheduledReporter}, CACHING_WINDOW should be set shorter than
+ * reporting interval to avoid outdated reporting.
+ */
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.debug("the new summary is {}", newSummary);
+ //Force update histogram upon each cache refresh
+ //This behavior relies on the fact that most common implementation of Reporter
--- End diff --
I have another potentially related question: What does this gauge output to the reporter? Is it going to be the toString on the ClusterSummary, or what happens when the value is delivered to the reporter? Is the gauge just here to update the other metrics?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209029068
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-CPU");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-CPU");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-CPU");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-CPU");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
--- End diff --
People probably won't notice this part of code when they change thrift protocol. We should break the code to remind them to update this part as well.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r203727117
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2222,13 +2231,23 @@ private void mkAssignments(String scratchTopoId) throws Exception {
if (!newAssignments.equals(existingAssignments)) {
LOG.debug("RESETTING id->resources and id->worker-resources cache!");
+ //Should we change these logs from info to debug after they are port to metrics?
--- End diff --
Same as above
---
[GitHub] storm issue #2764: STORM-3147: Port ClusterSummary as metrics to StormMetric...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on the issue:
https://github.com/apache/storm/pull/2764
All nits addressed?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208982864
--- Diff: storm-client/src/jvm/org/apache/storm/scheduler/WorkerSlot.java ---
@@ -39,6 +42,11 @@ public String getId() {
return getNodeId() + ":" + getPort();
}
+ public List<Object> toList() {
+ //For compatibility to call in Nimbus#mkAssignment
--- End diff --
Will remove the comment and log it in Jira
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208709991
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
--- End diff --
Is this check correct? If existing is `A B C` and new is `B C`, entriesInCommon is `B C`, so this would be false.
Why is the check even needed? Don't we always want to remove all existing that aren't in new, and add all new that aren't in existing?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r203726617
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java ---
@@ -52,6 +52,7 @@ public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf
public void start() {
if (reporter != null) {
LOG.debug("Starting...");
+ //TODO: will we make the period customizable?
--- End diff --
Please either raise an issue for this or remove the TODO. They have a habit of getting put in the code and then forgotten.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209367421
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4744,206 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ //This histogram reflects the data distribution across only one ClusterSummary, i.e.,
+ // data distribution across all entities of a type (e.g., data from all nimbus/topologies) at one moment.
+ // Hence we use half of the CACHING_WINDOW time to ensure it retains only data from the most recent update
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ /**
+ * Constructor to put all items in ClusterSummary in MetricSet as a metric.
+ * All metrics are derived from a cached ClusterSummary object,
+ * expired {@link ClusterSummaryMetricSet#CACHING_WINDOW} seconds after first query in a while from reporters.
+ * In case of {@link com.codahale.metrics.ScheduledReporter}, CACHING_WINDOW should be set shorter than
+ * reporting interval to avoid outdated reporting.
+ */
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.debug("the new summary is {}", newSummary);
+ //Force update histogram upon each cache refresh
+ //This behavior relies on the fact that most common implementation of Reporter
--- End diff --
The `cachedSummary` gauge
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208746736
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
--- End diff --
This is probably trivial, but I think it's possible to have longestSchdulingTime slightly higher max of scheduleTopologyTimeMs.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208975129
--- Diff: storm-client/src/jvm/org/apache/storm/scheduler/WorkerSlot.java ---
@@ -39,6 +42,11 @@ public String getId() {
return getNodeId() + ":" + getPort();
}
+ public List<Object> toList() {
+ //For compatibility to call in Nimbus#mkAssignment
--- End diff --
Eh, it's probably fine. If you want to keep this method, I don't think the comment should be here though. It won't make sense outside the context of this PR.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209311621
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
--- End diff --
Jira issued.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208703338
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -411,6 +431,10 @@
private final StormTimer timer;
private final IScheduler scheduler;
private final IScheduler underlyingScheduler;
+ //Metrics related
+ private final AtomicReference<Long> schedulingStartTime = new AtomicReference<>(null);
--- End diff --
Nit: Consider renaming these so it's obvious what time unit it is, e.g. `schedulingStartTimeNanos`
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208748740
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnLeft().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
}
- Map<List<Long>, NodeInfo> old = assignment.get_executor_node_port();
- Map<List<Long>, List<Object>> reassigned = new HashMap<>();
- for (Entry<List<Long>, List<Object>> execAndNodePort : execToNodePort.entrySet()) {
- NodeInfo oldAssigned = old.get(execAndNodePort.getKey());
- String node = (String) execAndNodePort.getValue().get(0);
- Long port = (Long) execAndNodePort.getValue().get(1);
- if (oldAssigned == null || !oldAssigned.get_node().equals(node)
- || !port.equals(oldAssigned.get_port_iterator().next())) {
- reassigned.put(execAndNodePort.getKey(), execAndNodePort.getValue());
- }
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnRight().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
}
- if (!reassigned.isEmpty()) {
- int count = (new HashSet<>(execToNodePort.values())).size();
- Set<List<Long>> reExecs = reassigned.keySet();
- LOG.info("Reassigning {} to {} slots", topoId, count);
- LOG.info("Reassign executors: {}", reExecs);
+ for (Entry<String, MapDifference.ValueDifference<Assignment>> entry : difference.entriesDiffering().entrySet()) {
--- End diff --
`entriesDiffering` Returns an unmodifiable map describing keys that appear in both maps, but with different values.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208749524
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2807,16 +2897,15 @@ public void launchServer() throws Exception {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
- StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", this::fragmentedMemory);
- StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", this::fragmentedCpu);
- StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values()
+ //Be cautious using method reference instead of lambda. subexpression preceding :: will be evaluated only upon evaluation
--- End diff --
Interesting, I thought I removed this already. Maybe it's rebased back in accidentally.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208965124
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
--- End diff --
you can file a separate jira to discuss about it and remove it from this pr
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208744273
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2222,13 +2231,23 @@ private void mkAssignments(String scratchTopoId) throws Exception {
if (!newAssignments.equals(existingAssignments)) {
LOG.debug("RESETTING id->resources and id->worker-resources cache!");
+ //Should we change these logs from info to debug after they are port to metrics?
--- End diff --
Will log in Jira.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209048016
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
--- End diff --
If no code reordering happens, gauge should always evaluate `currTime` first, and has to get startTime from `schedulingStartTimeNs` is set to null. So if we guarantee that elpased is evaluated after that, longest-scheduling-time-ms will not exceeds the real longest scheduling time.
That being said, I think the race here should be pretty negligible especially if we discard the decimals in ns-to-ms conversion. Meanwhile I did hear about complaints of hanging schedulers before, so I say we keep partial measurement and remove the comments, or just "please be noticed that it's normal to see minor jiggling in the longest scheduling time due to race condition."
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208745736
--- Diff: storm-client/src/jvm/org/apache/storm/scheduler/WorkerSlot.java ---
@@ -39,6 +42,11 @@ public String getId() {
return getNodeId() + ":" + getPort();
}
+ public List<Object> toList() {
+ //For compatibility to call in Nimbus#mkAssignment
--- End diff --
I thought it looks cleaner to on Nimbus side. The legacy implementation isn't pretty to start with. (a lot of List<Object> or List<Long>) If you want I can file a Jira to improve `mkAssignment` in general, to eliminate them altogether.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208970512
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2826,9 +2915,22 @@ public void launchServer() throws Exception {
.parallelStream()
.mapToDouble(SupervisorResources::getTotalCpu)
.sum());
-
+ StormMetricsRegistry.registerGauge("nimbus:longest-scheduling-time-ms", () -> {
+ Long currTime = Time.nanoTime();
+ Long startTime = schedulingStartTime.get();
+ //There could be race condition here but seems trivial, elapsed is
+ // guaranteed to be no longer than real elapsed time of scheduling
+ Long longest = longestSchedulingTime.get();
+ if (startTime != null) {
+ longest = currTime - startTime > longest ? currTime - startTime : longest;
--- End diff --
Okay, that makes sense. Thanks for explaining. Could you add a comment to this effect here?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208967428
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2826,9 +2915,22 @@ public void launchServer() throws Exception {
.parallelStream()
.mapToDouble(SupervisorResources::getTotalCpu)
.sum());
-
+ StormMetricsRegistry.registerGauge("nimbus:longest-scheduling-time-ms", () -> {
+ Long currTime = Time.nanoTime();
+ Long startTime = schedulingStartTime.get();
+ //There could be race condition here but seems trivial, elapsed is
+ // guaranteed to be no longer than real elapsed time of scheduling
+ Long longest = longestSchedulingTime.get();
+ if (startTime != null) {
+ longest = currTime - startTime > longest ? currTime - startTime : longest;
+ }
+ //To millis. How should I put the constant for magic numbers?
--- End diff --
Because the decimals are too trivial?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209348428
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4745,194 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.info("the new summary is {}", newSummary);
--- End diff --
This better to be `LOG.debug`?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209305125
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4652,4 +4745,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
--- End diff --
Additionally, we can't call `getClusterInfoImpl` directly, so we have to use the public thrift API `getClusterInfo`. This will disrupt the accuracy of its meter "nimbus:num-getClusterInfo-calls", because I think it's for tracking public API invocation only.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208984273
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2871,16 +2897,15 @@ public void launchServer() throws Exception {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
- StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", this::fragmentedMemory);
- StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", this::fragmentedCpu);
- StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values()
+ //Be cautious using method reference instead of lambda. subexpression preceding :: will be evaluated only upon evaluation
+ // Num supervisor, and fragmented resources have been included in cluster summary
+ StormMetricsRegistry.registerGauge("nimbus:total-available-memory (nonegative)", () -> nodeIdToResources.get().values()
--- End diff --
why do we need `non-negative`
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209045056
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
--- End diff --
If we keep the update and race, I think we need to either remove the comments about the race, or expand them to provide the example you posted. Leaving the comment as is won't help someone reading it later understand what race is there.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209016305
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4723,4 +4754,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-CPU");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-CPU");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-CPU");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-CPU");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.info("the new summary is {}", newSummary);
+ //This is ugly but I can't think of a better way to update histogram only once per caching
+ // It also kind of depends on the implementation that gauges gets updated before histograms
+ updateHistogram(newSummary);
--- End diff --
If I understand this correctly, updateHistogram will only be called every `CACHING_WINDOW` i.e. 5s. And histogram is `new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS)`, which means it only collects data in most recent 2.5s. The histogram will be dealing with basically the same data in the most of time.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208699933
--- Diff: storm-client/src/jvm/org/apache/storm/scheduler/WorkerSlot.java ---
@@ -39,6 +42,11 @@ public String getId() {
return getNodeId() + ":" + getPort();
}
+ public List<Object> toList() {
+ //For compatibility to call in Nimbus#mkAssignment
--- End diff --
This seems to only be used in one place. Why is this better than invoking `Arrays.asList(slot.getNodeId(), slot.getPort())`?
The reason I'm asking is that to me, there isn't an obvious WorkerSlot to List translation, so I think a generic toList like this decreases readability compared to just writing out this code in Nimbus.
Same comment for ExecutorDetails.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209083160
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
--- End diff --
I'd still rather get rid of it, because it's a redundant branch point, and less indentation is great. Personal preference though, so up to you.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208746949
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2131,17 +2223,13 @@ private void mkAssignments(String scratchTopoId) throws Exception {
}
}
// make the new assignments for topologies
- Map<String, SchedulerAssignment> newSchedulerAssignments = null;
synchronized (schedLock) {
- newSchedulerAssignments = computeNewSchedulerAssignments(existingAssignments, topologies, bases, scratchTopoId);
+ Map<String, SchedulerAssignment> newSchedulerAssignments =
+ computeNewSchedulerAssignments(existingAssignments, topologies, bases, scratchTopoId);
+ //Should probably change List<Object> to Tuple<String, Long> for better readability
--- End diff --
Again, probably better with another Jira for general code cleanup
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209306763
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4652,4 +4745,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
--- End diff --
That can be solved by making `getClusterInfoImpl` package private, but I see your point. Maybe we just leave it here for now, and I'll see if we can do something with it when I look at the non-static registry branch. It's not like it hurts anything to have it here, I'd just rather have it in another file if possible.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208988762
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-CPU");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-CPU");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-CPU");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-CPU");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
--- End diff --
assertion is good. Just curious in which case this will happen
---
[GitHub] storm issue #2764: STORM-3147: Port ClusterSummary as metrics to StormMetric...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on the issue:
https://github.com/apache/storm/pull/2764
This can be merged after #2789
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208714291
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnLeft().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
}
- Map<List<Long>, NodeInfo> old = assignment.get_executor_node_port();
- Map<List<Long>, List<Object>> reassigned = new HashMap<>();
- for (Entry<List<Long>, List<Object>> execAndNodePort : execToNodePort.entrySet()) {
- NodeInfo oldAssigned = old.get(execAndNodePort.getKey());
- String node = (String) execAndNodePort.getValue().get(0);
- Long port = (Long) execAndNodePort.getValue().get(1);
- if (oldAssigned == null || !oldAssigned.get_node().equals(node)
- || !port.equals(oldAssigned.get_port_iterator().next())) {
- reassigned.put(execAndNodePort.getKey(), execAndNodePort.getValue());
- }
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnRight().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
}
- if (!reassigned.isEmpty()) {
- int count = (new HashSet<>(execToNodePort.values())).size();
- Set<List<Long>> reExecs = reassigned.keySet();
- LOG.info("Reassigning {} to {} slots", topoId, count);
- LOG.info("Reassign executors: {}", reExecs);
+ for (Entry<String, MapDifference.ValueDifference<Assignment>> entry : difference.entriesDiffering().entrySet()) {
--- End diff --
Won't this also catch all removed assignments? The old reassigned set only contained assignments that were either changed or new, not ones that were removed.
---
[GitHub] storm issue #2764: [WIP] STORM-3147: Port ClusterSummary as metrics to Storm...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on the issue:
https://github.com/apache/storm/pull/2764
Should be changed after #2771 is approved.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209331742
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4746,186 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.info("the new summary is {}", newSummary);
+ //This is ugly but I can't think of a better way to update histogram only once per caching
+ // It also kind of depends on the implementation that gauges gets updated before histograms
+ updateHistogram(newSummary);
+ return newSummary;
+ } else {
+ return null;
+ }
+ } catch (Exception e) {
+ LOG.warn("Get cluster info exception.", e);
+ throw new RuntimeException(e);
+ }
+ }
+ };
+
+ clusterSummaryMetrics.put("cluster:num-nimbus-leaders", new DerivativeGauge<ClusterSummary, Long>(cachedSummary) {
+ @Override
+ protected Long transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_nimbuses().stream().filter(NimbusSummary::is_isLeader).count();
--- End diff --
Please make this and other similar code below use a new line for each step.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208751049
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2826,9 +2915,22 @@ public void launchServer() throws Exception {
.parallelStream()
.mapToDouble(SupervisorResources::getTotalCpu)
.sum());
-
+ StormMetricsRegistry.registerGauge("nimbus:longest-scheduling-time-ms", () -> {
+ Long currTime = Time.nanoTime();
+ Long startTime = schedulingStartTime.get();
+ //There could be race condition here but seems trivial, elapsed is
+ // guaranteed to be no longer than real elapsed time of scheduling
+ Long longest = longestSchedulingTime.get();
+ if (startTime != null) {
+ longest = currTime - startTime > longest ? currTime - startTime : longest;
--- End diff --
We would like to compute the distribution of scheduler latency as well as the longest scheduling iteration. If a scheduler is stuck in the middle of a scheduling iteration, the histogram won't reflect that until the scheduling iteration has ended because timer only report the time for a complete cycle. Hence I added this gauge to track the longest scheduling iteration in real time.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208970052
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
--- End diff --
We don't know whether two non empty maps are equal or not until we do `Maps#difference`. Then I can assign the variable `anyChanged` to remember if any changes in ScheduledAssignments have been made, which is the return value as well as the determinant if I should add more loggings
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208968475
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -1984,11 +2074,13 @@ private int fragmentedCpu() {
Cluster cluster = new Cluster(inimbus, supervisors, topoToSchedAssignment, topologies, conf);
cluster.setStatusMap(idToSchedStatus.get());
- long beforeSchedule = System.currentTimeMillis();
+ schedulingStartTime.set(Time.nanoTime());
scheduler.schedule(topologies, cluster);
- long scheduleTimeElapsedMs = System.currentTimeMillis() - beforeSchedule;
- LOG.debug("Scheduling took {} ms for {} topologies", scheduleTimeElapsedMs, topologies.getTopologies().size());
- scheduleTopologyTimeMs.update(scheduleTimeElapsedMs);
+ //Will compiler optimize the order of evalutation and cause race condition?
--- End diff --
How?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208966535
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2826,9 +2915,22 @@ public void launchServer() throws Exception {
.parallelStream()
.mapToDouble(SupervisorResources::getTotalCpu)
.sum());
-
+ StormMetricsRegistry.registerGauge("nimbus:longest-scheduling-time-ms", () -> {
+ Long currTime = Time.nanoTime();
+ Long startTime = schedulingStartTime.get();
+ //There could be race condition here but seems trivial, elapsed is
+ // guaranteed to be no longer than real elapsed time of scheduling
+ Long longest = longestSchedulingTime.get();
+ if (startTime != null) {
+ longest = currTime - startTime > longest ? currTime - startTime : longest;
+ }
+ //To millis. How should I put the constant for magic numbers?
--- End diff --
Sorry, I get what you're saying. Disregard the above. If you're making the gauge measure milliseconds it probably still makes sense to use whole milliseconds rather than a floating point representation.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208976776
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
--- End diff --
I think we're talking past each other a little.
What I'm saying is that this code block looks to me like
```
if (A == B) {
for (a only in A) {}
for (b only in B) {}
for (c in A or B where the value is different) {}
}
```
What I'm saying is that the `if` guard is not necessary, because if the `if` returns false, we'd be skipping all 3 for loops anyway, because the iterated sets are empty. If that's the case, we can still update `anyChanged`, but it shouldn't be happening in an `if`.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208987578
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
--- End diff --
I see what you mean. It appears to me that this improved the readability when I first wrote this block. Like you said if `if` here is evaluated true, all there loops will be skipped anyway, with or without if guard. So I thought we can just wrap it in an `if` statement just to make this clear, and also get rid of some loop overhead.
---
[GitHub] storm issue #2764: STORM-3147: Port ClusterSummary as metrics to StormMetric...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on the issue:
https://github.com/apache/storm/pull/2764
Added clarifications to ClusterSummaryMetricSet
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208971197
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2826,9 +2915,22 @@ public void launchServer() throws Exception {
.parallelStream()
.mapToDouble(SupervisorResources::getTotalCpu)
.sum());
-
+ StormMetricsRegistry.registerGauge("nimbus:longest-scheduling-time-ms", () -> {
+ Long currTime = Time.nanoTime();
+ Long startTime = schedulingStartTime.get();
+ //There could be race condition here but seems trivial, elapsed is
+ // guaranteed to be no longer than real elapsed time of scheduling
+ Long longest = longestSchedulingTime.get();
+ if (startTime != null) {
+ longest = currTime - startTime > longest ? currTime - startTime : longest;
+ }
+ //To millis. How should I put the constant for magic numbers?
--- End diff --
Yes, or if the decimals are meaningful the gauge should be using nanoseconds instead of millis.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209029163
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-CPU");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-CPU");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-CPU");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-CPU");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.info("the new summary is {}", newSummary);
+ //This is ugly but I can't think of a better way to update histogram only once per caching
+ // It also kind of depends on the implementation that gauges gets updated before histograms
+ updateHistogram(newSummary);
+ return newSummary;
+ } else {
+ return null;
+ }
+ } catch (Exception e) {
+ LOG.warn("Get cluster info exception.", e);
+ throw new RuntimeException(e);
+ }
+ }
+ };
+
+ ported.put("cluster:num-nimbus-leaders", new DerivativeGauge<ClusterSummary, Long>(cachedSummary) {
+ @Override
+ protected Long transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_nimbuses().stream().filter(NimbusSummary::is_isLeader).count();
--- End diff --
This is to guard against the two nimbi bug
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209042043
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2918,6 +2943,10 @@ public void launchServer() throws Exception {
}
});
}
+
+ //Should we make the delaySecs and recurSecs in sync with any conf value?
+ // They should be around the reporting interval, but it's not configurable
+ timer.scheduleRecurring(5, 5, clusterMetricSet);
--- End diff --
Just to be in sync with the caching frequency
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208968113
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
+ Map<String, Assignment> newAssignments) {
+ assert existingAssignments != null && newAssignments != null;
+ boolean anyChanged = existingAssignments.isEmpty() ^ newAssignments.isEmpty();
+ long numRemovedExec = 0;
+ long numRemovedSlot = 0;
+ long numAddedExec = 0;
+ long numAddedSlot = 0;
+ if (existingAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : newAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
+ }
+ } else if (newAssignments.isEmpty()) {
+ for (Entry<String, Assignment> entry : existingAssignments.entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
+ }
+ } else {
+ MapDifference<String, Assignment> difference = Maps.difference(existingAssignments, newAssignments);
+ if (anyChanged = (difference.entriesInCommon().size() != newAssignments.size())) {
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnLeft().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Removing {} from {} slots", entry.getKey(), count);
+ LOG.info("Remove executors: {}", execToPort.keySet());
+ numRemovedSlot += count;
+ numRemovedExec += execToPort.size();
}
- Map<List<Long>, NodeInfo> old = assignment.get_executor_node_port();
- Map<List<Long>, List<Object>> reassigned = new HashMap<>();
- for (Entry<List<Long>, List<Object>> execAndNodePort : execToNodePort.entrySet()) {
- NodeInfo oldAssigned = old.get(execAndNodePort.getKey());
- String node = (String) execAndNodePort.getValue().get(0);
- Long port = (Long) execAndNodePort.getValue().get(1);
- if (oldAssigned == null || !oldAssigned.get_node().equals(node)
- || !port.equals(oldAssigned.get_port_iterator().next())) {
- reassigned.put(execAndNodePort.getKey(), execAndNodePort.getValue());
- }
+ for (Entry<String, Assignment> entry : difference.entriesOnlyOnRight().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToPort = entry.getValue().get_executor_node_port();
+ final long count = new HashSet<>(execToPort.values()).size();
+ LOG.info("Assigning {} to {} slots", entry.getKey(), count);
+ LOG.info("Assign executors: {}", execToPort.keySet());
+ numAddedSlot += count;
+ numAddedExec += execToPort.size();
}
- if (!reassigned.isEmpty()) {
- int count = (new HashSet<>(execToNodePort.values())).size();
- Set<List<Long>> reExecs = reassigned.keySet();
- LOG.info("Reassigning {} to {} slots", topoId, count);
- LOG.info("Reassign executors: {}", reExecs);
+ for (Entry<String, MapDifference.ValueDifference<Assignment>> entry : difference.entriesDiffering().entrySet()) {
+ final Map<List<Long>, NodeInfo> execToSlot = entry.getValue().rightValue().get_executor_node_port();
+ final Set<NodeInfo> slots = new HashSet<>(execToSlot.values());
+ LOG.info("Reassigning {} to {} slots", entry.getKey(), slots.size());
+ LOG.info("Reassign executors: {}", execToSlot.keySet());
+
+ //We can simplify this part if scheduler does't partially remove or add executor
+ final Map<List<Long>, NodeInfo> oldExecToSlot = entry.getValue().leftValue().get_executor_node_port();
+
+ final Map<List<Long>, NodeInfo> smaller = execToSlot.size() > oldExecToSlot.size() ? oldExecToSlot : execToSlot;
--- End diff --
I would lean toward the simpler way to write it unless you know that this performs better.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208987159
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-CPU");
--- End diff --
use lowercase for "CPU" to be consistent with other names?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208997165
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -735,39 +756,108 @@ private static int numUsedWorkers(SchedulerAssignment assignment) {
return ret;
}
- private static Map<String, Map<List<Long>, List<Object>>> computeNewTopoToExecToNodePort(
- Map<String, SchedulerAssignment> schedAssignments, Map<String, Assignment> existingAssignments) {
- Map<String, Map<List<Long>, List<Object>>> ret = computeTopoToExecToNodePort(schedAssignments);
- // Print some useful information
- if (existingAssignments != null && !existingAssignments.isEmpty()) {
- for (Entry<String, Map<List<Long>, List<Object>>> entry : ret.entrySet()) {
- String topoId = entry.getKey();
- Map<List<Long>, List<Object>> execToNodePort = entry.getValue();
- Assignment assignment = existingAssignments.get(topoId);
- if (assignment == null) {
- continue;
+ private boolean inspectSchduling(Map<String, Assignment> existingAssignments,
--- End diff --
I don't think we should use `calculateAssignmentChanged` as it doesn't really calculate anything for scheduling process so much as inspect/audit the throughput of scheduling. I would prefer naming `auditAssignmentChanges` or `reviewAssignmentChanges`, etc.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208965327
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2826,9 +2915,22 @@ public void launchServer() throws Exception {
.parallelStream()
.mapToDouble(SupervisorResources::getTotalCpu)
.sum());
-
+ StormMetricsRegistry.registerGauge("nimbus:longest-scheduling-time-ms", () -> {
+ Long currTime = Time.nanoTime();
+ Long startTime = schedulingStartTime.get();
+ //There could be race condition here but seems trivial, elapsed is
+ // guaranteed to be no longer than real elapsed time of scheduling
+ Long longest = longestSchedulingTime.get();
+ if (startTime != null) {
+ longest = currTime - startTime > longest ? currTime - startTime : longest;
+ }
+ //To millis. How should I put the constant for magic numbers?
--- End diff --
Why would there be a loss of precision? The input number is a long, which is an integral type (not floating point).
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209354005
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4716,4 +4744,206 @@ public IScheduler getForcedScheduler() {
}
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> clusterSummaryMetrics = new HashMap<String, com.codahale.metrics.Metric>() {
+ @Override
+ public com.codahale.metrics.Metric put(String key, com.codahale.metrics.Metric value) {
+ return super.put(StormMetricsRegistry.name(SUMMARY, key), value);
+ }
+ };
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ //This histogram reflects the data distribution across only one ClusterSummary, i.e.,
+ // data distribution across all entities of a type (e.g., data from all nimbus/topologies) at one moment.
+ // Hence we use half of the CACHING_WINDOW time to ensure it retains only data from the most recent update
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ clusterSummaryMetrics.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ /**
+ * Constructor to put all items in ClusterSummary in MetricSet as a metric.
+ * All metrics are derived from a cached ClusterSummary object,
+ * expired {@link ClusterSummaryMetricSet#CACHING_WINDOW} seconds after first query in a while from reporters.
+ * In case of {@link com.codahale.metrics.ScheduledReporter}, CACHING_WINDOW should be set shorter than
+ * reporting interval to avoid outdated reporting.
+ */
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.debug("the new summary is {}", newSummary);
+ //Force update histogram upon each cache refresh
+ //This behavior relies on the fact that most common implementation of Reporter
--- End diff --
Sorry, I didn't notice this before.
This feels like a hack that will break in case the Dropwizard guys decide to report in a different order. Unless it's guaranteed somewhere in the API that this order will be used, we shouldn't rely on it. Why not just update the histograms periodically via e.g. the StormTimer?
Alternatively, do we even need this behavior for this code to work? If the reporter was reporting in a different order, we'd just get counters that are updated slightly late.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by zd-project <gi...@git.apache.org>.
Github user zd-project commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209335050
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -2905,6 +2929,10 @@ public void launchServer() throws Exception {
throw new RuntimeException(e);
}
});
+
+ //Should we make the delaySecs and recurSecs in sync with any conf value?
--- End diff --
I would very much love to. However based on current Nimbus implementation there's no way to do it. There's is however one leadership change callback buried in Zookeeper side code (See `LeaderListenerCallback`), which communicates through IStormClusterState. It'll be great if you can open a PR to improve Nimbus in general, along with [STORM-3187](https://issues.apache.org/jira/browse/STORM-3187). Otherwise we kind of have to using busy waiting here.
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by Ethanlm <gi...@git.apache.org>.
Github user Ethanlm commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r208991078
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4729,4 +4760,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-CPU");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-CPU");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-CPU");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-CPU");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
+ && ClusterSummary._Fields.findByName("supervisors") == ClusterSummary._Fields.SUPERVISORS
+ && ClusterSummary._Fields.findByName("topologies") == ClusterSummary._Fields.TOPOLOGIES
+ && ClusterSummary._Fields.findByName("nimbuses") == ClusterSummary._Fields.NIMBUSES;
+
+ final CachedGauge<ClusterSummary> cachedSummary = new CachedGauge<ClusterSummary>(CACHING_WINDOW, TimeUnit.SECONDS) {
+ @Override
+ protected ClusterSummary loadValue() {
+ try {
+ if (active) {
+ ClusterSummary newSummary = getClusterInfoImpl();
+ LOG.info("the new summary is {}", newSummary);
+ //This is ugly but I can't think of a better way to update histogram only once per caching
+ // It also kind of depends on the implementation that gauges gets updated before histograms
+ updateHistogram(newSummary);
+ return newSummary;
+ } else {
+ return null;
+ }
+ } catch (Exception e) {
+ LOG.warn("Get cluster info exception.", e);
+ throw new RuntimeException(e);
+ }
+ }
+ };
+
+ ported.put("cluster:num-nimbus-leaders", new DerivativeGauge<ClusterSummary, Long>(cachedSummary) {
+ @Override
+ protected Long transform(ClusterSummary clusterSummary) {
+ return clusterSummary.get_nimbuses().stream().filter(NimbusSummary::is_isLeader).count();
--- End diff --
Will this always be 1?
---
[GitHub] storm pull request #2764: STORM-3147: Port ClusterSummary as metrics to Stor...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2764#discussion_r209154625
--- Diff: storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java ---
@@ -4652,4 +4745,192 @@ public IScheduler getForcedScheduler() {
}
+ //enum NotPorted {
+ // //Declared in StormConf. I don't see the value in reporting so.
+ // SUPERVISOR_TOTAL_RESOURCE,
+ // //May be able to aggregate based on status;
+ // TOPOLOGY_STATUS,
+ // TOPOLOGY_SCHED_STATUS,
+ // //May be aggregated, as well as other distinct values
+ // NUM_DISTINCT_NIMBUS_VERSION;
+ //}
+
+ private class ClusterSummaryMetricSet implements MetricSet, Runnable {
+ static final int CACHING_WINDOW = 5;
+ static final int PORTED_METRICS = 25;
+ static final String SUMMARY = "summary";
+
+ private final Map<String, com.codahale.metrics.Metric> ported = new HashMap<>(PORTED_METRICS);
+ private final Function<String, Histogram> registerHistogram = (name) -> {
+ final Histogram histogram = new Histogram(new SlidingTimeWindowReservoir(CACHING_WINDOW / 2, TimeUnit.SECONDS));
+ ported.put(name, histogram);
+ return histogram;
+ };
+ private volatile boolean active = false;
+
+ //NImbus metrics distribution
+ private final Histogram nimbusUptime = registerHistogram.apply("nimbuses:uptime-secs");
+
+ //Supervisor metrics distribution
+ private final Histogram supervisorsUptime = registerHistogram.apply("supervisors:uptime-secs");
+ private final Histogram supervisorsNumWorkers = registerHistogram.apply("supervisors:num-workers");
+ private final Histogram supervisorsNumUsedWorkers = registerHistogram.apply("supervisors:num-used-workers");
+ private final Histogram supervisorsUsedMem = registerHistogram.apply("supervisors:used-mem");
+ private final Histogram supervisorsUsedCpu = registerHistogram.apply("supervisors:used-cpu");
+ private final Histogram supervisorsFragmentedMem = registerHistogram.apply("supervisors:fragmented-mem");
+ private final Histogram supervisorsFragmentedCpu = registerHistogram.apply("supervisors:fragmented-cpu");
+
+ //Topology metrics distribution
+ private final Histogram topologiesNumTasks = registerHistogram.apply("topologies:num-tasks");
+ private final Histogram topologiesNumExecutors = registerHistogram.apply("topologies:num-executors");
+ private final Histogram topologiesNumWorker = registerHistogram.apply("topologies:num-workers");
+ private final Histogram topologiesUptime = registerHistogram.apply("topologies:uptime-secs");
+ private final Histogram topologiesReplicationCount = registerHistogram.apply("topologies:replication-count");
+ private final Histogram topologiesRequestedMemOnHeap = registerHistogram.apply("topologies:requested-mem-on-heap");
+ private final Histogram topologiesRequestedMemOffHeap = registerHistogram.apply("topologies:requested-mem-off-heap");
+ private final Histogram topologiesRequestedCpu = registerHistogram.apply("topologies:requested-cpu");
+ private final Histogram topologiesAssignedMemOnHeap = registerHistogram.apply("topologies:assigned-mem-on-heap");
+ private final Histogram topologiesAssignedMemOffHeap = registerHistogram.apply("topologies:assigned-mem-off-heap");
+ private final Histogram topologiesAssignedCpu = registerHistogram.apply("topologies:assigned-cpu");
+
+ ClusterSummaryMetricSet() {
+ //Break the code if out of sync to thrift protocol
+ assert ClusterSummary._Fields.values().length == 3
--- End diff --
Keep in mind that asserts are only evaluated in test mode, or if people run with assertions enabled. If you want to always validate this, use e.g. `Validate`.
---