You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2017/10/10 18:27:39 UTC
[1/5] storm git commit: Adding cluster level fragmented CPU/Memory
metrics to nimbus
Repository: storm
Updated Branches:
refs/heads/master 352cd4665 -> f867e3ae5
Adding cluster level fragmented CPU/Memory metrics to nimbus
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/22427d0e
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/22427d0e
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/22427d0e
Branch: refs/heads/master
Commit: 22427d0e793dd67c16498e59f4da70ed174e6cc9
Parents: c607d04
Author: Kishor Patil <kp...@yahoo-inc.com>
Authored: Wed Oct 4 11:25:51 2017 -0400
Committer: Kishor Patil <kp...@yahoo-inc.com>
Committed: Wed Oct 4 11:25:51 2017 -0400
----------------------------------------------------------------------
.../org/apache/storm/daemon/nimbus/Nimbus.java | 32 ++++++++++++++++++--
.../org/apache/storm/scheduler/Cluster.java | 10 +++++-
2 files changed, 39 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/22427d0e/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index e54aaeb..438434c 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -1592,7 +1592,24 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
}
return ret;
}
-
+
+ private boolean isFragmented(SupervisorResources supervisorResources) {
+ double minMemory = ObjectReader.getInt(conf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB), 256)
+ + ObjectReader.getInt(conf.get(Config.TOPOLOGY_ACKER_RESOURCES_ONHEAP_MEMORY_MB), 128);
+ double minCPU = ObjectReader.getInt(conf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), 50)
+ + ObjectReader.getInt(conf.get(Config.TOPOLOGY_ACKER_CPU_PCORE_PERCENT), 50);
+
+ return minMemory > supervisorResources.getAvailableMem() || minCPU > supervisorResources.getAvailableCpu();
+ }
+
+ private double fragmentedMemory() {
+ return nodeIdToResources.get().values().parallelStream().filter(x -> isFragmented(x) == true).mapToDouble(SupervisorResources::getAvailableMem).sum();
+ }
+
+ private double fragmentedCpu() {
+ return nodeIdToResources.get().values().parallelStream().filter(x -> isFragmented(x) == true).mapToDouble(SupervisorResources::getAvailableCpu).sum();
+ }
+
private Map<String, SchedulerAssignment> computeNewSchedulerAssignments(Map<String, Assignment> existingAssignments,
Topologies topologies, Map<String, StormBase> bases, String scratchTopologyId) throws KeyNotFoundException, AuthorizationException, InvalidTopologyException, IOException {
Map<String, Set<List<Integer>>> topoToExec = computeTopologyToExecutors(bases);
@@ -1825,6 +1842,11 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
if (!newAssignments.equals(existingAssignments)) {
LOG.debug("RESETTING id->resources and id->worker-resources cache!");
+ LOG.info("Fragmentation after scheduling is: {} MB, {} PCore CPUs", fragmentedMemory(), fragmentedCpu());
+ //map.forEach((x, y) -> System.out.println(x + ": " + y));
+ nodeIdToResources.get().forEach((id, node) ->
+ LOG.info("Node Id: {} Total Mem: {}, Used Mem: {}, Avialble Mem: {}, Total CPU: {}, Used CPU: {}, Available CPU: {}, fragmented: {}",
+ id, node.getTotalMem(), node.getUsedMem(), node.getAvailableMem(), node.getTotalCpu(), node.getUsedCpu(), node.getAvailableCpu()));
idToResources.set(new HashMap<>());
idToWorkerResources.set(new HashMap<>());
}
@@ -2445,7 +2467,13 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
}
});
- StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null));
+ StormMetricsRegistry.registerGauge("nimbus:num-supervisors", () -> state.supervisors(null).size());
+ StormMetricsRegistry.registerGauge("nimbus:fragmented-memory", () -> fragmentedMemory());
+ StormMetricsRegistry.registerGauge("nimbus:fragmented-cpu", () -> fragmentedCpu());
+ StormMetricsRegistry.registerGauge("nimbus:available-memory", () -> nodeIdToResources.get().values().parallelStream().mapToDouble(SupervisorResources::getAvailableMem).sum());
+ StormMetricsRegistry.registerGauge("nimbus:available-cpu", () -> nodeIdToResources.get().values().parallelStream().mapToDouble(SupervisorResources::getAvailableCpu).sum());
+ StormMetricsRegistry.registerGauge("nimbus:total-memory", () -> nodeIdToResources.get().values().parallelStream().mapToDouble(SupervisorResources::getTotalMem).sum());
+ StormMetricsRegistry.registerGauge("nimbus:total-cpu", () -> nodeIdToResources.get().values().parallelStream().mapToDouble(SupervisorResources::getTotalCpu).sum());
StormMetricsRegistry.startMetricsReporters(conf);
if (clusterConsumerExceutors != null) {
http://git-wip-us.apache.org/repos/asf/storm/blob/22427d0e/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
index b5f105a..6cb53e4 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
@@ -81,6 +81,14 @@ public class Cluster implements ISchedulingState {
return totalCpu;
}
+ public double getAvailableCpu() {
+ return totalCpu - usedCpu;
+ }
+
+ public double getAvailableMem() {
+ return totalMem - usedMem;
+ }
+
private SupervisorResources add(WorkerResources wr) {
return new SupervisorResources(
totalMem,
@@ -934,7 +942,7 @@ public class Cluster implements ISchedulingState {
public Map<String, SupervisorResources> getSupervisorsResourcesMap() {
Map<String, SupervisorResources> ret = new HashMap<>();
for (SupervisorDetails sd : supervisors.values()) {
- ret.put(sd.getId(), new SupervisorResources(sd.getTotalMemory(), sd.getTotalMemory(), 0, 0));
+ ret.put(sd.getId(), new SupervisorResources(sd.getTotalMemory(), sd.getTotalCPU(), 0, 0));
}
for (SchedulerAssignmentImpl assignment : assignments.values()) {
for (Entry<WorkerSlot, WorkerResources> entry :
[5/5] storm git commit: Merge branch 'storm2770' of
https://github.com/kishorvpatil/incubator-storm into STORM-2770
Posted by bo...@apache.org.
Merge branch 'storm2770' of https://github.com/kishorvpatil/incubator-storm into STORM-2770
STORM-2770: Add fragmentation metrics for CPU and Memory
This closes #2358
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/f867e3ae
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/f867e3ae
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/f867e3ae
Branch: refs/heads/master
Commit: f867e3ae577b4ac61d6941af0373253accd38ff5
Parents: 352cd46 0b3930d
Author: Robert Evans <ev...@yahoo-inc.com>
Authored: Tue Oct 10 13:10:21 2017 -0500
Committer: Robert Evans <ev...@yahoo-inc.com>
Committed: Tue Oct 10 13:10:21 2017 -0500
----------------------------------------------------------------------
.../storm/generated/SupervisorSummary.java | 206 ++++++++++++++++++-
storm-client/src/py/storm/ttypes.py | 28 ++-
storm-client/src/storm.thrift | 2 +
storm-core/src/clj/org/apache/storm/ui/core.clj | 14 +-
.../public/templates/index-page-template.html | 12 ++
.../org/apache/storm/daemon/nimbus/Nimbus.java | 37 +++-
.../org/apache/storm/scheduler/Cluster.java | 10 +-
7 files changed, 297 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/f867e3ae/storm-core/src/clj/org/apache/storm/ui/core.clj
----------------------------------------------------------------------
[2/5] storm git commit: Add fragmented cpu/memory to SupervisorSummary
Posted by bo...@apache.org.
Add fragmented cpu/memory to SupervisorSummary
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/e0e9bcb3
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/e0e9bcb3
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/e0e9bcb3
Branch: refs/heads/master
Commit: e0e9bcb30c5d6dff8f2fe08322312374fb7d63f2
Parents: 22427d0
Author: Kishor Patil <kp...@yahoo-inc.com>
Authored: Wed Oct 4 17:21:16 2017 +0000
Committer: Kishor Patil <kp...@yahoo-inc.com>
Committed: Wed Oct 4 17:21:27 2017 +0000
----------------------------------------------------------------------
.../storm/generated/SupervisorSummary.java | 206 ++++++++++++++++++-
storm-client/src/py/storm/ttypes.py | 28 ++-
storm-client/src/storm.thrift | 2 +
3 files changed, 231 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/e0e9bcb3/storm-client/src/jvm/org/apache/storm/generated/SupervisorSummary.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/generated/SupervisorSummary.java b/storm-client/src/jvm/org/apache/storm/generated/SupervisorSummary.java
index 64eafeb..049ebea 100644
--- a/storm-client/src/jvm/org/apache/storm/generated/SupervisorSummary.java
+++ b/storm-client/src/jvm/org/apache/storm/generated/SupervisorSummary.java
@@ -64,6 +64,8 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
private static final org.apache.thrift.protocol.TField TOTAL_RESOURCES_FIELD_DESC = new org.apache.thrift.protocol.TField("total_resources", org.apache.thrift.protocol.TType.MAP, (short)7);
private static final org.apache.thrift.protocol.TField USED_MEM_FIELD_DESC = new org.apache.thrift.protocol.TField("used_mem", org.apache.thrift.protocol.TType.DOUBLE, (short)8);
private static final org.apache.thrift.protocol.TField USED_CPU_FIELD_DESC = new org.apache.thrift.protocol.TField("used_cpu", org.apache.thrift.protocol.TType.DOUBLE, (short)9);
+ private static final org.apache.thrift.protocol.TField FRAGMENTED_MEM_FIELD_DESC = new org.apache.thrift.protocol.TField("fragmented_mem", org.apache.thrift.protocol.TType.DOUBLE, (short)10);
+ private static final org.apache.thrift.protocol.TField FRAGMENTED_CPU_FIELD_DESC = new org.apache.thrift.protocol.TField("fragmented_cpu", org.apache.thrift.protocol.TType.DOUBLE, (short)11);
private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
static {
@@ -80,6 +82,8 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
private Map<String,Double> total_resources; // optional
private double used_mem; // optional
private double used_cpu; // optional
+ private double fragmented_mem; // optional
+ private double fragmented_cpu; // optional
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
public enum _Fields implements org.apache.thrift.TFieldIdEnum {
@@ -91,7 +95,9 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
VERSION((short)6, "version"),
TOTAL_RESOURCES((short)7, "total_resources"),
USED_MEM((short)8, "used_mem"),
- USED_CPU((short)9, "used_cpu");
+ USED_CPU((short)9, "used_cpu"),
+ FRAGMENTED_MEM((short)10, "fragmented_mem"),
+ FRAGMENTED_CPU((short)11, "fragmented_cpu");
private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
@@ -124,6 +130,10 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
return USED_MEM;
case 9: // USED_CPU
return USED_CPU;
+ case 10: // FRAGMENTED_MEM
+ return FRAGMENTED_MEM;
+ case 11: // FRAGMENTED_CPU
+ return FRAGMENTED_CPU;
default:
return null;
}
@@ -169,8 +179,10 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
private static final int __NUM_USED_WORKERS_ISSET_ID = 2;
private static final int __USED_MEM_ISSET_ID = 3;
private static final int __USED_CPU_ISSET_ID = 4;
+ private static final int __FRAGMENTED_MEM_ISSET_ID = 5;
+ private static final int __FRAGMENTED_CPU_ISSET_ID = 6;
private byte __isset_bitfield = 0;
- private static final _Fields optionals[] = {_Fields.VERSION,_Fields.TOTAL_RESOURCES,_Fields.USED_MEM,_Fields.USED_CPU};
+ private static final _Fields optionals[] = {_Fields.VERSION,_Fields.TOTAL_RESOURCES,_Fields.USED_MEM,_Fields.USED_CPU,_Fields.FRAGMENTED_MEM,_Fields.FRAGMENTED_CPU};
public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
static {
Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
@@ -194,6 +206,10 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.DOUBLE)));
tmpMap.put(_Fields.USED_CPU, new org.apache.thrift.meta_data.FieldMetaData("used_cpu", org.apache.thrift.TFieldRequirementType.OPTIONAL,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.DOUBLE)));
+ tmpMap.put(_Fields.FRAGMENTED_MEM, new org.apache.thrift.meta_data.FieldMetaData("fragmented_mem", org.apache.thrift.TFieldRequirementType.OPTIONAL,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.DOUBLE)));
+ tmpMap.put(_Fields.FRAGMENTED_CPU, new org.apache.thrift.meta_data.FieldMetaData("fragmented_cpu", org.apache.thrift.TFieldRequirementType.OPTIONAL,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.DOUBLE)));
metaDataMap = Collections.unmodifiableMap(tmpMap);
org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(SupervisorSummary.class, metaDataMap);
}
@@ -244,6 +260,8 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
}
this.used_mem = other.used_mem;
this.used_cpu = other.used_cpu;
+ this.fragmented_mem = other.fragmented_mem;
+ this.fragmented_cpu = other.fragmented_cpu;
}
public SupervisorSummary deepCopy() {
@@ -267,6 +285,10 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
this.used_mem = 0.0;
set_used_cpu_isSet(false);
this.used_cpu = 0.0;
+ set_fragmented_mem_isSet(false);
+ this.fragmented_mem = 0.0;
+ set_fragmented_cpu_isSet(false);
+ this.fragmented_cpu = 0.0;
}
public String get_host() {
@@ -482,6 +504,50 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
__isset_bitfield = EncodingUtils.setBit(__isset_bitfield, __USED_CPU_ISSET_ID, value);
}
+ public double get_fragmented_mem() {
+ return this.fragmented_mem;
+ }
+
+ public void set_fragmented_mem(double fragmented_mem) {
+ this.fragmented_mem = fragmented_mem;
+ set_fragmented_mem_isSet(true);
+ }
+
+ public void unset_fragmented_mem() {
+ __isset_bitfield = EncodingUtils.clearBit(__isset_bitfield, __FRAGMENTED_MEM_ISSET_ID);
+ }
+
+ /** Returns true if field fragmented_mem is set (has been assigned a value) and false otherwise */
+ public boolean is_set_fragmented_mem() {
+ return EncodingUtils.testBit(__isset_bitfield, __FRAGMENTED_MEM_ISSET_ID);
+ }
+
+ public void set_fragmented_mem_isSet(boolean value) {
+ __isset_bitfield = EncodingUtils.setBit(__isset_bitfield, __FRAGMENTED_MEM_ISSET_ID, value);
+ }
+
+ public double get_fragmented_cpu() {
+ return this.fragmented_cpu;
+ }
+
+ public void set_fragmented_cpu(double fragmented_cpu) {
+ this.fragmented_cpu = fragmented_cpu;
+ set_fragmented_cpu_isSet(true);
+ }
+
+ public void unset_fragmented_cpu() {
+ __isset_bitfield = EncodingUtils.clearBit(__isset_bitfield, __FRAGMENTED_CPU_ISSET_ID);
+ }
+
+ /** Returns true if field fragmented_cpu is set (has been assigned a value) and false otherwise */
+ public boolean is_set_fragmented_cpu() {
+ return EncodingUtils.testBit(__isset_bitfield, __FRAGMENTED_CPU_ISSET_ID);
+ }
+
+ public void set_fragmented_cpu_isSet(boolean value) {
+ __isset_bitfield = EncodingUtils.setBit(__isset_bitfield, __FRAGMENTED_CPU_ISSET_ID, value);
+ }
+
public void setFieldValue(_Fields field, Object value) {
switch (field) {
case HOST:
@@ -556,6 +622,22 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
}
break;
+ case FRAGMENTED_MEM:
+ if (value == null) {
+ unset_fragmented_mem();
+ } else {
+ set_fragmented_mem((Double)value);
+ }
+ break;
+
+ case FRAGMENTED_CPU:
+ if (value == null) {
+ unset_fragmented_cpu();
+ } else {
+ set_fragmented_cpu((Double)value);
+ }
+ break;
+
}
}
@@ -588,6 +670,12 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
case USED_CPU:
return get_used_cpu();
+ case FRAGMENTED_MEM:
+ return get_fragmented_mem();
+
+ case FRAGMENTED_CPU:
+ return get_fragmented_cpu();
+
}
throw new IllegalStateException();
}
@@ -617,6 +705,10 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
return is_set_used_mem();
case USED_CPU:
return is_set_used_cpu();
+ case FRAGMENTED_MEM:
+ return is_set_fragmented_mem();
+ case FRAGMENTED_CPU:
+ return is_set_fragmented_cpu();
}
throw new IllegalStateException();
}
@@ -715,6 +807,24 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
return false;
}
+ boolean this_present_fragmented_mem = true && this.is_set_fragmented_mem();
+ boolean that_present_fragmented_mem = true && that.is_set_fragmented_mem();
+ if (this_present_fragmented_mem || that_present_fragmented_mem) {
+ if (!(this_present_fragmented_mem && that_present_fragmented_mem))
+ return false;
+ if (this.fragmented_mem != that.fragmented_mem)
+ return false;
+ }
+
+ boolean this_present_fragmented_cpu = true && this.is_set_fragmented_cpu();
+ boolean that_present_fragmented_cpu = true && that.is_set_fragmented_cpu();
+ if (this_present_fragmented_cpu || that_present_fragmented_cpu) {
+ if (!(this_present_fragmented_cpu && that_present_fragmented_cpu))
+ return false;
+ if (this.fragmented_cpu != that.fragmented_cpu)
+ return false;
+ }
+
return true;
}
@@ -767,6 +877,16 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
if (present_used_cpu)
list.add(used_cpu);
+ boolean present_fragmented_mem = true && (is_set_fragmented_mem());
+ list.add(present_fragmented_mem);
+ if (present_fragmented_mem)
+ list.add(fragmented_mem);
+
+ boolean present_fragmented_cpu = true && (is_set_fragmented_cpu());
+ list.add(present_fragmented_cpu);
+ if (present_fragmented_cpu)
+ list.add(fragmented_cpu);
+
return list.hashCode();
}
@@ -868,6 +988,26 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
return lastComparison;
}
}
+ lastComparison = Boolean.valueOf(is_set_fragmented_mem()).compareTo(other.is_set_fragmented_mem());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (is_set_fragmented_mem()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.fragmented_mem, other.fragmented_mem);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ lastComparison = Boolean.valueOf(is_set_fragmented_cpu()).compareTo(other.is_set_fragmented_cpu());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (is_set_fragmented_cpu()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.fragmented_cpu, other.fragmented_cpu);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
return 0;
}
@@ -947,6 +1087,18 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
sb.append(this.used_cpu);
first = false;
}
+ if (is_set_fragmented_mem()) {
+ if (!first) sb.append(", ");
+ sb.append("fragmented_mem:");
+ sb.append(this.fragmented_mem);
+ first = false;
+ }
+ if (is_set_fragmented_cpu()) {
+ if (!first) sb.append(", ");
+ sb.append("fragmented_cpu:");
+ sb.append(this.fragmented_cpu);
+ first = false;
+ }
sb.append(")");
return sb.toString();
}
@@ -1096,6 +1248,22 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
+ case 10: // FRAGMENTED_MEM
+ if (schemeField.type == org.apache.thrift.protocol.TType.DOUBLE) {
+ struct.fragmented_mem = iprot.readDouble();
+ struct.set_fragmented_mem_isSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ case 11: // FRAGMENTED_CPU
+ if (schemeField.type == org.apache.thrift.protocol.TType.DOUBLE) {
+ struct.fragmented_cpu = iprot.readDouble();
+ struct.set_fragmented_cpu_isSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
default:
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
@@ -1160,6 +1328,16 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
oprot.writeDouble(struct.used_cpu);
oprot.writeFieldEnd();
}
+ if (struct.is_set_fragmented_mem()) {
+ oprot.writeFieldBegin(FRAGMENTED_MEM_FIELD_DESC);
+ oprot.writeDouble(struct.fragmented_mem);
+ oprot.writeFieldEnd();
+ }
+ if (struct.is_set_fragmented_cpu()) {
+ oprot.writeFieldBegin(FRAGMENTED_CPU_FIELD_DESC);
+ oprot.writeDouble(struct.fragmented_cpu);
+ oprot.writeFieldEnd();
+ }
oprot.writeFieldStop();
oprot.writeStructEnd();
}
@@ -1195,7 +1373,13 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
if (struct.is_set_used_cpu()) {
optionals.set(3);
}
- oprot.writeBitSet(optionals, 4);
+ if (struct.is_set_fragmented_mem()) {
+ optionals.set(4);
+ }
+ if (struct.is_set_fragmented_cpu()) {
+ optionals.set(5);
+ }
+ oprot.writeBitSet(optionals, 6);
if (struct.is_set_version()) {
oprot.writeString(struct.version);
}
@@ -1215,6 +1399,12 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
if (struct.is_set_used_cpu()) {
oprot.writeDouble(struct.used_cpu);
}
+ if (struct.is_set_fragmented_mem()) {
+ oprot.writeDouble(struct.fragmented_mem);
+ }
+ if (struct.is_set_fragmented_cpu()) {
+ oprot.writeDouble(struct.fragmented_cpu);
+ }
}
@Override
@@ -1230,7 +1420,7 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
struct.set_num_used_workers_isSet(true);
struct.supervisor_id = iprot.readString();
struct.set_supervisor_id_isSet(true);
- BitSet incoming = iprot.readBitSet(4);
+ BitSet incoming = iprot.readBitSet(6);
if (incoming.get(0)) {
struct.version = iprot.readString();
struct.set_version_isSet(true);
@@ -1258,6 +1448,14 @@ public class SupervisorSummary implements org.apache.thrift.TBase<SupervisorSumm
struct.used_cpu = iprot.readDouble();
struct.set_used_cpu_isSet(true);
}
+ if (incoming.get(4)) {
+ struct.fragmented_mem = iprot.readDouble();
+ struct.set_fragmented_mem_isSet(true);
+ }
+ if (incoming.get(5)) {
+ struct.fragmented_cpu = iprot.readDouble();
+ struct.set_fragmented_cpu_isSet(true);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/e0e9bcb3/storm-client/src/py/storm/ttypes.py
----------------------------------------------------------------------
diff --git a/storm-client/src/py/storm/ttypes.py b/storm-client/src/py/storm/ttypes.py
index 0daf3ff..33c5faa 100644
--- a/storm-client/src/py/storm/ttypes.py
+++ b/storm-client/src/py/storm/ttypes.py
@@ -3025,6 +3025,8 @@ class SupervisorSummary:
- total_resources
- used_mem
- used_cpu
+ - fragmented_mem
+ - fragmented_cpu
"""
thrift_spec = (
@@ -3038,9 +3040,11 @@ class SupervisorSummary:
(7, TType.MAP, 'total_resources', (TType.STRING,None,TType.DOUBLE,None), None, ), # 7
(8, TType.DOUBLE, 'used_mem', None, None, ), # 8
(9, TType.DOUBLE, 'used_cpu', None, None, ), # 9
+ (10, TType.DOUBLE, 'fragmented_mem', None, None, ), # 10
+ (11, TType.DOUBLE, 'fragmented_cpu', None, None, ), # 11
)
- def __init__(self, host=None, uptime_secs=None, num_workers=None, num_used_workers=None, supervisor_id=None, version=thrift_spec[6][4], total_resources=None, used_mem=None, used_cpu=None,):
+ def __init__(self, host=None, uptime_secs=None, num_workers=None, num_used_workers=None, supervisor_id=None, version=thrift_spec[6][4], total_resources=None, used_mem=None, used_cpu=None, fragmented_mem=None, fragmented_cpu=None,):
self.host = host
self.uptime_secs = uptime_secs
self.num_workers = num_workers
@@ -3050,6 +3054,8 @@ class SupervisorSummary:
self.total_resources = total_resources
self.used_mem = used_mem
self.used_cpu = used_cpu
+ self.fragmented_mem = fragmented_mem
+ self.fragmented_cpu = fragmented_cpu
def read(self, iprot):
if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
@@ -3111,6 +3117,16 @@ class SupervisorSummary:
self.used_cpu = iprot.readDouble()
else:
iprot.skip(ftype)
+ elif fid == 10:
+ if ftype == TType.DOUBLE:
+ self.fragmented_mem = iprot.readDouble()
+ else:
+ iprot.skip(ftype)
+ elif fid == 11:
+ if ftype == TType.DOUBLE:
+ self.fragmented_cpu = iprot.readDouble()
+ else:
+ iprot.skip(ftype)
else:
iprot.skip(ftype)
iprot.readFieldEnd()
@@ -3161,6 +3177,14 @@ class SupervisorSummary:
oprot.writeFieldBegin('used_cpu', TType.DOUBLE, 9)
oprot.writeDouble(self.used_cpu)
oprot.writeFieldEnd()
+ if self.fragmented_mem is not None:
+ oprot.writeFieldBegin('fragmented_mem', TType.DOUBLE, 10)
+ oprot.writeDouble(self.fragmented_mem)
+ oprot.writeFieldEnd()
+ if self.fragmented_cpu is not None:
+ oprot.writeFieldBegin('fragmented_cpu', TType.DOUBLE, 11)
+ oprot.writeDouble(self.fragmented_cpu)
+ oprot.writeFieldEnd()
oprot.writeFieldStop()
oprot.writeStructEnd()
@@ -3189,6 +3213,8 @@ class SupervisorSummary:
value = (value * 31) ^ hash(self.total_resources)
value = (value * 31) ^ hash(self.used_mem)
value = (value * 31) ^ hash(self.used_cpu)
+ value = (value * 31) ^ hash(self.fragmented_mem)
+ value = (value * 31) ^ hash(self.fragmented_cpu)
return value
def __repr__(self):
http://git-wip-us.apache.org/repos/asf/storm/blob/e0e9bcb3/storm-client/src/storm.thrift
----------------------------------------------------------------------
diff --git a/storm-client/src/storm.thrift b/storm-client/src/storm.thrift
index 11879d1..bc411cb 100644
--- a/storm-client/src/storm.thrift
+++ b/storm-client/src/storm.thrift
@@ -188,6 +188,8 @@ struct SupervisorSummary {
7: optional map<string, double> total_resources;
8: optional double used_mem;
9: optional double used_cpu;
+ 10: optional double fragmented_mem;
+ 11: optional double fragmented_cpu;
}
struct NimbusSummary {
[3/5] storm git commit: Show fragmented cpu/memory on cluster summary
Posted by bo...@apache.org.
Show fragmented cpu/memory on cluster summary
Fix Configuration porting
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2269e202
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2269e202
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2269e202
Branch: refs/heads/master
Commit: 2269e2028db6999c19c887116d0fcc95448a9a6d
Parents: e0e9bcb
Author: Kishor Patil <kp...@yahoo-inc.com>
Authored: Wed Oct 4 13:59:51 2017 -0400
Committer: Kishor Patil <kp...@yahoo-inc.com>
Committed: Wed Oct 4 16:05:06 2017 -0400
----------------------------------------------------------------------
storm-core/src/clj/org/apache/storm/ui/core.clj | 14 +++++++++----
.../public/templates/index-page-template.html | 12 +++++++++++
.../org/apache/storm/daemon/nimbus/Nimbus.java | 22 +++++++++++++-------
3 files changed, 36 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/2269e202/storm-core/src/clj/org/apache/storm/ui/core.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/ui/core.clj b/storm-core/src/clj/org/apache/storm/ui/core.clj
index 1d2ea85..1740d25 100644
--- a/storm-core/src/clj/org/apache/storm/ui/core.clj
+++ b/storm-core/src/clj/org/apache/storm/ui/core.clj
@@ -428,13 +428,17 @@
:let [sup-total-mem (get (.get_total_resources s) Config/SUPERVISOR_MEMORY_CAPACITY_MB)
sup-total-cpu (get (.get_total_resources s) Config/SUPERVISOR_CPU_CAPACITY)
sup-avail-mem (max (- sup-total-mem (.get_used_mem s)) 0.0)
- sup-avail-cpu (max (- sup-total-cpu (.get_used_cpu s)) 0.0)]]
- [sup-total-mem sup-total-cpu sup-avail-mem sup-avail-cpu]))
- [0.0 0.0 0.0 0.0])
+ sup-avail-cpu (max (- sup-total-cpu (.get_used_cpu s)) 0.0)
+ sup-fragmented-cpu (.get_fragmented_cpu s)
+ sup-fragmented-mem (.get_fragmented_mem s)]]
+ [sup-total-mem sup-total-cpu sup-avail-mem sup-avail-cpu sup-fragmented-cpu sup-fragmented-mem]))
+ [0.0 0.0 0.0 0.0 0.0 0.0])
total-mem (nth resourceSummary 0)
total-cpu (nth resourceSummary 1)
avail-mem (nth resourceSummary 2)
- avail-cpu (nth resourceSummary 3)]
+ avail-cpu (nth resourceSummary 3)
+ fragmented-cpu (nth resourceSummary 4)
+ fragmented-mem (nth resourceSummary 5)]
{"user" user
"stormVersion" STORM-VERSION
"supervisors" (count sups)
@@ -449,6 +453,8 @@
"totalCpu" total-cpu
"availMem" avail-mem
"availCpu" avail-cpu
+ "fragmentedMem" fragmented-mem
+ "fragmentedCpu" fragmented-cpu
"memAssignedPercentUtil" (if (and (not (nil? total-mem)) (> total-mem 0.0)) (format "%.1f" (* (/ (- total-mem avail-mem) total-mem) 100.0)) 0.0)
"cpuAssignedPercentUtil" (if (and (not (nil? total-cpu)) (> total-cpu 0.0)) (format "%.1f" (* (/ (- total-cpu avail-cpu) total-cpu) 100.0)) 0.0)})))
http://git-wip-us.apache.org/repos/asf/storm/blob/2269e202/storm-core/src/ui/public/templates/index-page-template.html
----------------------------------------------------------------------
diff --git a/storm-core/src/ui/public/templates/index-page-template.html b/storm-core/src/ui/public/templates/index-page-template.html
index f1f8e66..48658a9 100644
--- a/storm-core/src/ui/public/templates/index-page-template.html
+++ b/storm-core/src/ui/public/templates/index-page-template.html
@@ -119,6 +119,11 @@
</span>
</th>
<th>
+ <span data-toggle="tooltip" data-placement="right" title="Total Fragmented Memory in Cluster in MB.">
+ Total Fragmented Memory (MB)
+ </span>
+ </th>
+ <th>
<span data-toggle="tooltip" data-placement="top" title="Percent Utilization of Memory Resources in Cluster.">
Memory Utilization (%)
</span>
@@ -134,6 +139,11 @@
</span>
</th>
<th>
+ <span data-toggle="tooltip" data-placement="top" title="Total Fragmented CPU Resources in Cluster. Every 100 means 1 core.">
+ Fragmented CPU (%)
+ </span>
+ </th>
+ <th>
<span data-toggle="tooltip" data-placement="top" title="Percent Utilization of CPU Resources in Cluster.">
CPU Utilization (%)
</span>
@@ -144,9 +154,11 @@
<tr>
<td>{{totalMem}}</td>
<td>{{availMem}}</td>
+ <td>{{fragmentedMem}}</td>
<td>{{memAssignedPercentUtil}}</td>
<td>{{totalCpu}}</td>
<td>{{availCpu}}</td>
+ <td>{{fragmentedCpu}}</td>
<td>{{cpuAssignedPercentUtil}}</td>
</tr>
</tbody>
http://git-wip-us.apache.org/repos/asf/storm/blob/2269e202/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index 438434c..d2529c4 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -1594,20 +1594,22 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
}
private boolean isFragmented(SupervisorResources supervisorResources) {
- double minMemory = ObjectReader.getInt(conf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB), 256)
- + ObjectReader.getInt(conf.get(Config.TOPOLOGY_ACKER_RESOURCES_ONHEAP_MEMORY_MB), 128);
- double minCPU = ObjectReader.getInt(conf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), 50)
- + ObjectReader.getInt(conf.get(Config.TOPOLOGY_ACKER_CPU_PCORE_PERCENT), 50);
+ double minMemory = ObjectReader.getDouble(conf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB), 256.0)
+ + ObjectReader.getDouble(conf.get(Config.TOPOLOGY_ACKER_RESOURCES_ONHEAP_MEMORY_MB), 128.0);
+ double minCPU = ObjectReader.getDouble(conf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), 50.0)
+ + ObjectReader.getDouble(conf.get(Config.TOPOLOGY_ACKER_CPU_PCORE_PERCENT), 50.0);
return minMemory > supervisorResources.getAvailableMem() || minCPU > supervisorResources.getAvailableCpu();
}
private double fragmentedMemory() {
- return nodeIdToResources.get().values().parallelStream().filter(x -> isFragmented(x) == true).mapToDouble(SupervisorResources::getAvailableMem).sum();
+ Double res = nodeIdToResources.get().values().parallelStream().filter(x -> isFragmented(x) == true).mapToDouble(SupervisorResources::getAvailableMem).filter(x -> x > 0).sum();
+ return res.intValue();
}
- private double fragmentedCpu() {
- return nodeIdToResources.get().values().parallelStream().filter(x -> isFragmented(x) == true).mapToDouble(SupervisorResources::getAvailableCpu).sum();
+ private int fragmentedCpu() {
+ Double res = nodeIdToResources.get().values().parallelStream().filter(x -> isFragmented(x) == true).mapToDouble(SupervisorResources::getAvailableCpu).filter(x -> x > 0).sum();
+ return res.intValue();
}
private Map<String, SchedulerAssignment> computeNewSchedulerAssignments(Map<String, Assignment> existingAssignments,
@@ -1846,7 +1848,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
//map.forEach((x, y) -> System.out.println(x + ": " + y));
nodeIdToResources.get().forEach((id, node) ->
LOG.info("Node Id: {} Total Mem: {}, Used Mem: {}, Avialble Mem: {}, Total CPU: {}, Used CPU: {}, Available CPU: {}, fragmented: {}",
- id, node.getTotalMem(), node.getUsedMem(), node.getAvailableMem(), node.getTotalCpu(), node.getUsedCpu(), node.getAvailableCpu()));
+ id, node.getTotalMem(), node.getUsedMem(), node.getAvailableMem(), node.getTotalCpu(), node.getUsedCpu(), node.getAvailableCpu(), isFragmented(node)));
idToResources.set(new HashMap<>());
idToWorkerResources.set(new HashMap<>());
}
@@ -2251,6 +2253,10 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
if (resources != null) {
ret.set_used_mem(resources.getUsedMem());
ret.set_used_cpu(resources.getUsedCpu());
+ if (isFragmented(resources)) {
+ ret.set_fragmented_cpu(resources.getAvailableCpu());
+ ret.set_fragmented_mem(resources.getAvailableMem());
+ }
}
if (info.is_set_version()) {
ret.set_version(info.get_version());
[4/5] storm git commit: Remove unwanted comment
Posted by bo...@apache.org.
Remove unwanted comment
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/0b3930dc
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/0b3930dc
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/0b3930dc
Branch: refs/heads/master
Commit: 0b3930dc4506b0613e30d9d736becfac77b9a41b
Parents: 2269e20
Author: Kishor Patil <kp...@yahoo-inc.com>
Authored: Mon Oct 9 10:15:29 2017 -0400
Committer: Kishor Patil <kp...@yahoo-inc.com>
Committed: Mon Oct 9 10:15:29 2017 -0400
----------------------------------------------------------------------
.../src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/0b3930dc/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index d2529c4..5ab0ea1 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -1845,7 +1845,6 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
if (!newAssignments.equals(existingAssignments)) {
LOG.debug("RESETTING id->resources and id->worker-resources cache!");
LOG.info("Fragmentation after scheduling is: {} MB, {} PCore CPUs", fragmentedMemory(), fragmentedCpu());
- //map.forEach((x, y) -> System.out.println(x + ": " + y));
nodeIdToResources.get().forEach((id, node) ->
LOG.info("Node Id: {} Total Mem: {}, Used Mem: {}, Avialble Mem: {}, Total CPU: {}, Used CPU: {}, Available CPU: {}, fragmented: {}",
id, node.getTotalMem(), node.getUsedMem(), node.getAvailableMem(), node.getTotalCpu(), node.getUsedCpu(), node.getAvailableCpu(), isFragmented(node)));