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/02/27 16:04:06 UTC
[3/8] storm git commit: STORM-2333: Addressed some review comments
STORM-2333: Addressed some review comments
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/86e0f526
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/86e0f526
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/86e0f526
Branch: refs/heads/master
Commit: 86e0f5268e892a8c897576e9c63254414757a9cf
Parents: b89984c
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Tue Feb 21 15:25:18 2017 -0600
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Tue Feb 21 15:25:18 2017 -0600
----------------------------------------------------------------------
docs/cgroups_in_storm.md | 11 +++-
.../container/cgroup/CgroupCoreFactory.java | 58 +++++++++---------
.../storm/container/cgroup/Hierarchy.java | 2 +-
.../apache/storm/metric/cgroup/CGroupCpu.java | 64 ++++++++++----------
.../storm/metric/cgroup/CGroupCpuGuarantee.java | 34 +++++++----
.../storm/metric/cgroup/CGroupMemoryLimit.java | 10 ++-
.../storm/metric/cgroup/CGroupMemoryUsage.java | 10 ++-
.../storm/metric/cgroup/CGroupMetricsBase.java | 50 +++++++--------
8 files changed, 133 insertions(+), 106 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/86e0f526/docs/cgroups_in_storm.md
----------------------------------------------------------------------
diff --git a/docs/cgroups_in_storm.md b/docs/cgroups_in_storm.md
index 445ae83..5eac9dc 100644
--- a/docs/cgroups_in_storm.md
+++ b/docs/cgroups_in_storm.md
@@ -22,7 +22,7 @@ A sample/default cgconfig.conf file is supplied in the <stormroot>/conf director
mount {
cpuset = /cgroup/cpuset;
cpu = /cgroup/storm_resources;
- cpuacct = /cgroup/cpuacct;
+ cpuacct = /cgroup/storm_resources;
memory = /cgroup/storm_resources;
devices = /cgroup/devices;
freezer = /cgroup/freezer;
@@ -54,6 +54,13 @@ For a more detailed explanation of the format and configs for the cgconfig.conf
https://access.redhat.com/documentation/en-US/Red_Hat_Enterprise_Linux/6/html/Resource_Management_Guide/ch-Using_Control_Groups.html#The_cgconfig.conf_File
+To let storm manage the cgroups for indavidual workers you need to make sure that the resources you want to control are mounted under the same directory as in the example above.
+If they are not in the same directory the supervisor will throw an exception.
+
+The perm section needs to be configured so that the user the supervisor is running as can modify the group.
+
+If run as user is enabled so the supervisor spawns other processes as the user that launched the topology make sure that the permissions are such that indavidual users have read access but not write access.
+
# Settings Related To CGroups in Storm
| Setting | Function |
@@ -89,6 +96,8 @@ org.apache.storm.metric.cgroup.CGroupCPU reports back metrics similar to org.apa
CGroup reports these as CLK_TCK counts, and not milliseconds so the accuracy is determined by what CLK_TCK is set to. On most systems it is 100 times a second so at most the accuracy is 10 ms.
+To make this metric work cpuacct must be mounted.
+
## CGroupCpuGuarantee
org.apache.storm.metric.cgroup.CGroupCpuGuarantee reports back an approximate number of ms of CPU time that this worker is guaranteed to get. This is calculated from the resources requested by the tasks in that given worker.
http://git-wip-us.apache.org/repos/asf/storm/blob/86e0f526/storm-core/src/jvm/org/apache/storm/container/cgroup/CgroupCoreFactory.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/container/cgroup/CgroupCoreFactory.java b/storm-core/src/jvm/org/apache/storm/container/cgroup/CgroupCoreFactory.java
index 53a8a7f..4d35c77 100755
--- a/storm-core/src/jvm/org/apache/storm/container/cgroup/CgroupCoreFactory.java
+++ b/storm-core/src/jvm/org/apache/storm/container/cgroup/CgroupCoreFactory.java
@@ -34,39 +34,37 @@ import java.util.Set;
public class CgroupCoreFactory {
+ public static CgroupCore getInstance(SubSystemType type, String dir) {
+ switch (type) {
+ case blkio:
+ return new BlkioCore(dir);
+ case cpuacct:
+ return new CpuacctCore(dir);
+ case cpuset:
+ return new CpusetCore(dir);
+ case cpu:
+ return new CpuCore(dir);
+ case devices:
+ return new DevicesCore(dir);
+ case freezer:
+ return new FreezerCore(dir);
+ case memory:
+ return new MemoryCore(dir);
+ case net_cls:
+ return new NetClsCore(dir);
+ case net_prio:
+ return new NetPrioCore(dir);
+ default:
+ return null;
+ }
+ }
+
public static Map<SubSystemType, CgroupCore> getInstance(Set<SubSystemType> types, String dir) {
Map<SubSystemType, CgroupCore> result = new HashMap<SubSystemType, CgroupCore>();
for (SubSystemType type : types) {
- switch (type) {
- case blkio:
- result.put(SubSystemType.blkio, new BlkioCore(dir));
- break;
- case cpuacct:
- result.put(SubSystemType.cpuacct, new CpuacctCore(dir));
- break;
- case cpuset:
- result.put(SubSystemType.cpuset, new CpusetCore(dir));
- break;
- case cpu:
- result.put(SubSystemType.cpu, new CpuCore(dir));
- break;
- case devices:
- result.put(SubSystemType.devices, new DevicesCore(dir));
- break;
- case freezer:
- result.put(SubSystemType.freezer, new FreezerCore(dir));
- break;
- case memory:
- result.put(SubSystemType.memory, new MemoryCore(dir));
- break;
- case net_cls:
- result.put(SubSystemType.net_cls, new NetClsCore(dir));
- break;
- case net_prio:
- result.put(SubSystemType.net_prio, new NetPrioCore(dir));
- break;
- default:
- break;
+ CgroupCore inst = getInstance(type, dir);
+ if (inst != null) {
+ result.put(type, inst);
}
}
return result;
http://git-wip-us.apache.org/repos/asf/storm/blob/86e0f526/storm-core/src/jvm/org/apache/storm/container/cgroup/Hierarchy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/container/cgroup/Hierarchy.java b/storm-core/src/jvm/org/apache/storm/container/cgroup/Hierarchy.java
index 440531a..b2b245c 100755
--- a/storm-core/src/jvm/org/apache/storm/container/cgroup/Hierarchy.java
+++ b/storm-core/src/jvm/org/apache/storm/container/cgroup/Hierarchy.java
@@ -20,7 +20,7 @@ package org.apache.storm.container.cgroup;
import java.util.Set;
/**
- * A class that describes a cgroup hiearchy
+ * A class that describes a cgroup hierarchy
*/
public class Hierarchy {
http://git-wip-us.apache.org/repos/asf/storm/blob/86e0f526/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpu.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpu.java b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpu.java
index 0429c97..bf474be 100644
--- a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpu.java
+++ b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpu.java
@@ -17,52 +17,54 @@
*/
package org.apache.storm.metric.cgroup;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.Map;
+import org.apache.storm.container.cgroup.SubSystemType;
+import org.apache.storm.container.cgroup.core.CgroupCore;
+import org.apache.storm.container.cgroup.core.CpuacctCore;
+import org.apache.storm.container.cgroup.core.CpuacctCore.StatType;
+
/**
* Report CPU used in the cgroup
*/
public class CGroupCpu extends CGroupMetricsBase<Map<String,Long>> {
long previousSystem = 0;
long previousUser = 0;
+ private int userHz = -1;
public CGroupCpu(Map<String, Object> conf) {
- super(conf, "cpuacct.stat");
+ super(conf, SubSystemType.cpuacct);
}
- public int getUserHZ() {
- return 100; // On most systems (x86) this is fine.
- // If someone really does want full support
- // we need to run `getconf CLK_TCK` and cache the result.
+ public synchronized int getUserHZ() throws IOException {
+ if (userHz < 0) {
+ ProcessBuilder pb = new ProcessBuilder("getconf", "CLK_TCK");
+ Process p = pb.start();
+ BufferedReader in = new BufferedReader(new InputStreamReader(p.getInputStream()));
+ String line = in.readLine().trim();
+ userHz = Integer.valueOf(line);
+ }
+ return userHz;
}
@Override
- public Map<String, Long> parseFileContents(String contents) {
- try {
- long systemHz = 0;
- long userHz = 0;
- for (String line: contents.split("\n")) {
- if (!line.isEmpty()) {
- String [] parts = line.toLowerCase().split("\\s+");
- if (parts[0].contains("system")) {
- systemHz = Long.parseLong(parts[1].trim());
- } else if (parts[0].contains("user")) {
- userHz = Long.parseLong(parts[1].trim());
- }
- }
- }
- long user = userHz - previousUser;
- long sys = systemHz - previousSystem;
- previousUser = userHz;
- previousSystem = systemHz;
- long hz = getUserHZ();
- HashMap<String, Long> ret = new HashMap<>();
- ret.put("user-ms", user * 1000/hz); //Convert to millis
- ret.put("sys-ms", sys * 1000/hz);
- return ret;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
+ public Map<String, Long> getDataFrom(CgroupCore core) throws IOException {
+ CpuacctCore cpu = (CpuacctCore) core;
+ Map<StatType, Long> stat = cpu.getCpuStat();
+ long systemHz = stat.get(StatType.system);
+ long userHz = stat.get(StatType.user);
+ long user = userHz - previousUser;
+ long sys = systemHz - previousSystem;
+ previousUser = userHz;
+ previousSystem = systemHz;
+ long hz = getUserHZ();
+ HashMap<String, Long> ret = new HashMap<>();
+ ret.put("user-ms", user * 1000/hz); //Convert to millis
+ ret.put("sys-ms", sys * 1000/hz);
+ return ret;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/86e0f526/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpuGuarantee.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpuGuarantee.java b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpuGuarantee.java
index 1c924cd..47ac153 100644
--- a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpuGuarantee.java
+++ b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupCpuGuarantee.java
@@ -17,8 +17,11 @@
*/
package org.apache.storm.metric.cgroup;
+import java.io.IOException;
import java.util.Map;
+import org.apache.storm.container.cgroup.SubSystemType;
+import org.apache.storm.container.cgroup.core.CgroupCore;
import org.apache.storm.container.cgroup.core.CpuCore;
/**
@@ -26,23 +29,28 @@ import org.apache.storm.container.cgroup.core.CpuCore;
*/
public class CGroupCpuGuarantee extends CGroupMetricsBase<Long> {
long previousTime = -1;
-
+
public CGroupCpuGuarantee(Map<String, Object> conf) {
- super(conf, CpuCore.CPU_SHARES);
+ super(conf, SubSystemType.cpu);
}
@Override
- public Long parseFileContents(String contents) {
- Long msGuarantee = null;
- long now = System.currentTimeMillis();
- if (previousTime > 0) {
- long shares = Long.valueOf(contents.trim());
- //By convention each share corresponds to 1% of a CPU core
- // or 100 = 1 core full time. So the guaranteed number of ms
- // (approximately) should be ...
- msGuarantee = (shares * (now - previousTime))/100;
+ public Long getDataFrom(CgroupCore core) {
+ try {
+ CpuCore cpu = (CpuCore)core;
+ Long msGuarantee = null;
+ long now = System.currentTimeMillis();
+ if (previousTime > 0) {
+ long shares = cpu.getCpuShares();
+ //By convention each share corresponds to 1% of a CPU core
+ // or 100 = 1 core full time. So the guaranteed number of ms
+ // (approximately) should be ...
+ msGuarantee = (shares * (now - previousTime))/100;
+ }
+ previousTime = now;
+ return msGuarantee;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
}
- previousTime = now;
- return msGuarantee;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/86e0f526/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryLimit.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryLimit.java b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryLimit.java
index caa5c51..bfd87a9 100644
--- a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryLimit.java
+++ b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryLimit.java
@@ -19,17 +19,21 @@ package org.apache.storm.metric.cgroup;
import java.util.Map;
+import org.apache.storm.container.cgroup.SubSystemType;
+import org.apache.storm.container.cgroup.core.CgroupCore;
+import org.apache.storm.container.cgroup.core.MemoryCore;
+
/**
* Reports the current memory limit of the cgroup for this worker
*/
public class CGroupMemoryLimit extends CGroupMetricsBase<Long> {
public CGroupMemoryLimit(Map<String, Object> conf) {
- super(conf, "memory.limit_in_bytes");
+ super(conf, SubSystemType.memory);
}
@Override
- public Long parseFileContents(String contents) {
- return Long.parseLong(contents.trim());
+ public Long getDataFrom(CgroupCore core) throws Exception {
+ return ((MemoryCore) core).getPhysicalUsageLimit();
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/86e0f526/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryUsage.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryUsage.java b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryUsage.java
index bb00514..9462166 100644
--- a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryUsage.java
+++ b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMemoryUsage.java
@@ -19,17 +19,21 @@ package org.apache.storm.metric.cgroup;
import java.util.Map;
+import org.apache.storm.container.cgroup.SubSystemType;
+import org.apache.storm.container.cgroup.core.CgroupCore;
+import org.apache.storm.container.cgroup.core.MemoryCore;
+
/**
* Reports the current memory usage of the cgroup for this worker
*/
public class CGroupMemoryUsage extends CGroupMetricsBase<Long> {
public CGroupMemoryUsage(Map<String, Object> conf) {
- super(conf, "memory.usage_in_bytes");
+ super(conf, SubSystemType.memory);
}
@Override
- public Long parseFileContents(String contents) {
- return Long.parseLong(contents.trim());
+ public Long getDataFrom(CgroupCore core) throws Exception {
+ return ((MemoryCore) core).getPhysicalUsage();
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/86e0f526/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMetricsBase.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMetricsBase.java b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMetricsBase.java
index 74bb42c..3f29d0f 100644
--- a/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMetricsBase.java
+++ b/storm-core/src/jvm/org/apache/storm/metric/cgroup/CGroupMetricsBase.java
@@ -20,10 +20,13 @@ package org.apache.storm.metric.cgroup;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
-import java.io.IOException;
import java.util.Map;
import org.apache.storm.Config;
+import org.apache.storm.container.cgroup.CgroupCenter;
+import org.apache.storm.container.cgroup.CgroupCoreFactory;
+import org.apache.storm.container.cgroup.SubSystemType;
+import org.apache.storm.container.cgroup.core.CgroupCore;
import org.apache.storm.metric.api.IMetric;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,15 +37,27 @@ import org.slf4j.LoggerFactory;
public abstract class CGroupMetricsBase<T> implements IMetric {
private static final Logger LOG = LoggerFactory.getLogger(CGroupMetricsBase.class);
private boolean enabled;
- private File fullFile;
+ private CgroupCore core = null;
- public CGroupMetricsBase(Map<String, Object> conf, String fileName) {
+ public CGroupMetricsBase(Map<String, Object> conf, SubSystemType type) {
+ final String simpleName = getClass().getSimpleName();
enabled = false;
+ CgroupCenter center = CgroupCenter.getInstance();
+ if (center == null) {
+ LOG.warn("{} is diabled. cgroups do not appear to be enabled on this system", simpleName);
+ return;
+ }
+ if (!center.isSubSystemEnabled(type)) {
+ LOG.warn("{} is disabled. {} is not an enabled subsystem", simpleName, type);
+ return;
+ }
+
String hierarchyDir = (String)conf.get(Config.STORM_CGROUP_HIERARCHY_DIR);
if (hierarchyDir == null || hierarchyDir.isEmpty()) {
- LOG.warn("{} is disabled {} is not set", getClass().getSimpleName(), Config.STORM_CGROUP_HIERARCHY_DIR);
+ LOG.warn("{} is disabled {} is not set", simpleName, Config.STORM_CGROUP_HIERARCHY_DIR);
return;
}
+
//Good so far, check if we are in a CGroup
File cgroupFile = new File("/proc/self/cgroup");
if (!cgroupFile.exists()) {
@@ -58,18 +73,13 @@ public abstract class CGroupMetricsBase<T> implements IMetric {
//parts[0] == 0 for CGroup V2, else maps to hierarchy in /proc/cgroups
//parts[1] is empty for CGroups V2 else what is mapped that we are looking for
String cgroupPath = parts[2];
-
- fullFile = new File(new File(new File(hierarchyDir), cgroupPath), fileName);
- if (!fullFile.exists()) {
- LOG.warn("{} is disabled {} does not exist", getClass().getSimpleName(), fullFile);
- return;
- }
+ core = CgroupCoreFactory.getInstance(type, new File(hierarchyDir, cgroupPath).getAbsolutePath());
} catch (Exception e) {
- LOG.warn("{} is disabled error trying to read or parse {}", getClass().getSimpleName(), cgroupFile);
+ LOG.warn("{} is disabled error trying to read or parse {}", simpleName, cgroupFile);
return;
}
enabled = true;
- LOG.info("{} is ENABLED {} exists...", getClass().getSimpleName(), fullFile);
+ LOG.info("{} is ENABLED {} exists...", simpleName);
}
@Override
@@ -77,20 +87,12 @@ public abstract class CGroupMetricsBase<T> implements IMetric {
if (!enabled) {
return null;
}
- StringBuffer contents = new StringBuffer();
- try (BufferedReader reader = new BufferedReader(new FileReader(fullFile))) {
- char[] buf = new char[4096];
- int len;
- while((len = reader.read(buf)) > 0) {
- contents.append(buf, 0, len);
- }
- Object ret = parseFileContents(contents.toString());
- LOG.debug("{} is returning {} from {}", getClass().getSimpleName(), ret, fullFile);
- return ret;
- } catch (IOException e) {
+ try {
+ return getDataFrom(core);
+ } catch (Exception e) {
throw new RuntimeException(e);
}
}
- public abstract T parseFileContents(String contents);
+ public abstract T getDataFrom(CgroupCore core) throws Exception;
}