Wed, Nov 14, 2018 at 04:04:45PM CET, olichtne@redhat.com wrote:
From: Ondrej Lichtner olichtne@redhat.com
This test module can be used to periodically sample the /proc/stat file for statistics and report back a list of differences between the individual samples as well as the raw data.
Can be used to calculate per-cpu and system wide cpu utilization.
Currently the test module samples until interrupted, so it should be run in the background and stopped with a job.kill(signal.SIGINT) call.
v2:
- added a comment explaining the interval parameter
- added a default value for the old_handler variable. In case the signal.signal call fails this will be used to avoid a NameError exception in the finally block
Signed-off-by: Ondrej Lichtner olichtne@redhat.com
lnst/Tests/CPUStatMonitor.py | 116 +++++++++++++++++++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 lnst/Tests/CPUStatMonitor.py
diff --git a/lnst/Tests/CPUStatMonitor.py b/lnst/Tests/CPUStatMonitor.py new file mode 100644 index 0000000..aabec06 --- /dev/null +++ b/lnst/Tests/CPUStatMonitor.py @@ -0,0 +1,116 @@ +import re +import time +import signal +from time import sleep +from lnst.Common.Parameters import IntParam +from lnst.Tests.BaseTestModule import BaseTestModule, TestModuleError, InterruptException
+def sigint_handler(signum, frame):
- raise InterruptException()
+class CPUStatMonitor(BaseTestModule):
- #number of miliseconds to sleep between each sample
- interval = IntParam(default=1000)
- def run(self):
self._res_data = {}
raw_samples = []
old_handler = None
try:
old_handler = signal.signal(signal.SIGINT, sigint_handler)
with open("/proc/stat") as stat:
while True:
stat.seek(0)
timestamp = time.time()
stat_lines = "".join(stat.readlines())
raw_samples.append({
"timestamp": timestamp,
"stat": stat_lines
})
sleep(self.params.interval / float(1000))
except InterruptException:
pass
finally:
if old_handler is not None:
signal.signal(signal.SIGINT, old_handler)
self._res_data["raw_data"] = raw_samples
self._res_data["data"] = self._process_samples(raw_samples)
return True
- def _process_samples(self, samples):
result = []
prev_sample = None
How about this?
prev_sample = samples[0] for sample in samples[1:]: ...
With this you can remove the 'if prev_sample is not None:' condition.
for sample in samples:
if prev_sample is not None:
parsed_prev = self._parse_stat_lines(prev_sample["stat"])
parsed_cur = self._parse_stat_lines(sample["stat"])
interval = self._subtract_nested_dicts(parsed_cur, parsed_prev)
interval["duration"] = (sample["timestamp"] -
prev_sample["timestamp"])
result.append(interval)
prev_sample = sample
return result
- def _subtract_nested_dicts(self, first, second):
result = {}
for key, val in first.items():
if isinstance(val, dict):
result[key] = self._subtract_nested_dicts(val, second[key])
else:
result[key] = val - second[key]
return result
- def _parse_stat_lines(self, stat):
result = {}
for line in stat.split("\n"):
cpu_data = self._parse_cpu_stats(line)
if cpu_data:
result[cpu_data[0]] = cpu_data[1]
continue
intr_data = self._parse_intr_stats(line)
if intr_data:
result[intr_data[0]] = intr_data[1]
continue
m = re.match(r"^(.*?) (\d+)$", line)
if m:
result[m.group(1)] = int(m.group(2))
return result
- def _parse_cpu_stats(self, stat_line):
result = {}
m = re.match(r"^(cpu\d*)\s+(\d+) (\d+) (\d+) (\d+) (\d+) (\d+) (\d+) (\d+) (\d+) (\d+)$",
stat_line)
if m:
cpu = m.group(1)
result["user"] = int(m.group(2))
result["nice"] = int(m.group(3))
result["system"] = int(m.group(4))
result["idle"] = int(m.group(5))
result["iowait"] = int(m.group(6))
result["irq"] = int(m.group(7))
result["softirq"] = int(m.group(8))
result["steal"] = int(m.group(9))
result["guest"] = int(m.group(10))
result["guest_nice"] = int(m.group(11))
return cpu, result
else:
return None
- def _parse_intr_stats(self, stat_line):
result = {}
m = re.match(r"^(intr|softirq) (\d+) (.*)$", stat_line)
if m:
result["total"] = int(m.group(2))
for i, irq in enumerate(m.group(3).split(" ")):
result[i] = int(irq)
return m.group(1), result
else:
return None
-- 2.19.1 _______________________________________________ LNST-developers mailing list -- lnst-developers@lists.fedorahosted.org To unsubscribe send an email to lnst-developers-leave@lists.fedorahosted.org Fedora Code of Conduct: https://getfedora.org/code-of-conduct.html List Guidelines: https://fedoraproject.org/wiki/Mailing_list_guidelines List Archives: https://lists.fedorahosted.org/archives/list/lnst-developers@lists.fedorahos...