add more averaging

Author: lcniel

sams/sampler/SlurmCGroup.py

@@ -39,6 +39,7 @@
 import logging
 import os
 import re
+import time
 
 import sams.base
 
@@ -51,6 +52,13 @@ def __init__(self, id, outQueue, config):
         self.processes = {}
         self.cgroup = None
         self.cgroup_base = self.config.get([self.id, "cgroup_base"], "/cgroup")
+        self.create_time = time.time()
+        self.last_sample_time = self.create_time
+        self.metrics_to_average = self.config.get(
+            [self.id, "metrics_to_average"],
+            ["memory_usage"])
+        self._average_values = {k: 0 for k in self.metrics_to_average}
+        self._last_averaged_values = {k: 0 for k in self.metrics_to_average}
 
     def do_sample(self):
         return self._get_cgroup()
@@ -73,6 +81,7 @@ def sample(self):
             "memory_max_usage": memory_max_usage,
             "memory_swap": str(int(memory_usage_and_swap) - int(memory_usage)),
         }
+        self.compute_sample_averages(entry)
         self._most_recent_sample = [self._storage_wrapping(entry)]
         self.store(entry)
 
@@ -81,6 +90,30 @@ def _get_cgroup_regex():
         """Version-specific regular expression to find correct cgroup path."""
         return r"^/(slurm/uid_\d+/job_\d+)/"
 
+    def compute_sample_averages(self, data):
+        """ Computes averages of selected measurements by
+        means of trapezoidal quadrature, approximating
+        that the time this function is called is the actual
+        time of sampling. This is not completely correct but simplifies
+        the implementation.
+        """
+        sample_time = time.time()
+        elapsed_time = sample_time - self.last_sample_time
+        total_elapsed_time = sample_time - self.create_time
+        for key, item in data.items():
+            if key in self.metrics_to_average:
+                # Trapezoidal quadrature
+                weighted_item = (
+                        0.5 * (float(item) + float(self._last_averaged_values[key])) * elapsed_time)
+                self._last_averaged_values[key] = item
+                previous_integral = self._average_values[key] * (total_elapsed_time - elapsed_time)
+                new_integral = previous_integral + weighted_item
+                self._average_values[key] = new_integral / total_elapsed_time
+
+        for key, item in self._average_values.items():
+            data[key + '_average'] = item
+        self.last_sample_time = time.time()
+
     def _get_cgroup(self):
         """Get the cgroup base path for the slurm job"""
         if self.cgroup:

sams/sampler/Software.py

@@ -158,6 +158,11 @@ def __init__(self, id, outQueue, config):
         self.last_sample_time = None
         self.last_total = None
         self.software_mapper = None
+        self.metrics_to_average = self.config.get(
+            [self.id, "metrics_to_average"],
+            ["system", "user"])
+        self._average_values = {k: 0 for k in self.metrics_to_average}
+        self._last_averaged_values = {k: 0 for k in self.metrics_to_average}
 
         software_mapper = self.config.get([id, "software_mapper"], None)
         if software_mapper is not None:
@@ -195,35 +200,62 @@ def sample(self):
 
         for pid in self.pids:
             logger.debug("evaluate pid: %d", pid)
-            if not pid in self.processes.keys():
+            if pid not in self.processes.keys():
                 logger.debug("Create new instance of Process for pid: %d", pid)
                 self.processes[pid] = Process(pid, self.jobid)
             self.processes[pid].update(uptime)
 
         # Send information about current usage
         aggr, total = self._aggregate()
-        if self.last_sample_time:
-            time_diff = time.time() - self.last_sample_time
-            if time_diff > self.sampler_interval / 2:
-                entry = {
-                    "current": {
-                        "software": self.map_software(aggr),
-                        "total_user": total["user"],
-                        "total_system": total["system"],
-                        "user": (total["user"] - self.last_total["user"])
-                        / time_diff,
-                        "system": (total["system"] - self.last_total["system"])
-                        / time_diff,
-                        }
+
+        if self.last_sample_time is None:
+            self.last_total = total
+            self.last_sample_time = time.time()
+            return
+
+        time_diff = time.time() - self.last_sample_time
+        if time_diff > self.sampler_interval / 2:
+            entry = {
+                "current": {
+                    "software": self.map_software(aggr),
+                    "total_user": total["user"],
+                    "total_system": total["system"],
+                    "user": (total["user"] - self.last_total["user"])
+                    / time_diff,
+                    "system": (total["system"] - self.last_total["system"])
+                    / time_diff,
                     }
-                self._most_recent_sample = [self._storage_wrapping(entry)]
-                self.store(entry)
-                self.last_total = total
-                self.last_sample_time = time.time()
-        else:
+                }
+            self.compute_sample_averages(entry["current"])
+            self._most_recent_sample = [self._storage_wrapping(entry)]
+            self.store(entry)
             self.last_total = total
             self.last_sample_time = time.time()
 
+    def compute_sample_averages(self, data):
+        """ Computes averages of selected measurements by
+        means of trapezoidal quadrature, approximating
+        that the time this function is called is the actual
+        time of sampling. This is not completely correct but simplifies
+        the implementation.
+        """
+        sample_time = time.time()
+        elapsed_time = sample_time - self.last_sample_time
+        total_elapsed_time = sample_time - self.create_time
+        for key, item in data.items():
+            if key in self.metrics_to_average:
+                # Trapezoidal quadrature
+                weighted_item = (
+                        0.5 * (float(item) + float(self._last_averaged_values[key])) * elapsed_time)
+                self._last_averaged_values[key] = item
+                previous_integral = self._average_values[key] * (total_elapsed_time - elapsed_time)
+                new_integral = previous_integral + weighted_item
+                self._average_values[key] = new_integral / total_elapsed_time
+
+        for key, item in self._average_values.items():
+            data[key + '_average'] = item
+        data['elapsed_time'] = total_elapsed_time
+
     def last_updated(self):
         procs = list(filter(lambda p: not p.ignore, self.processes.values()))
         if not procs:
@@ -250,7 +282,7 @@ def _aggregate(self):
                 a["system"],
             )
             exe = a["exe"]
-            if not exe in aggr:
+            if exe not in aggr:
                 aggr[exe] = {"user": 0.0, "system": 0.0}
             aggr[exe]["user"] += a["user"]
             aggr[exe]["system"] += a["system"]

sams/sampler/ZFSStats.py

@@ -42,6 +42,7 @@
 
 import logging
 import subprocess
+import time
 
 import sams.base
 
@@ -79,12 +80,19 @@ def __init__(self, id, outQueue, config):
         self.volumes = self.config.get([self.id, "volumes"])
         self.zfs_command = self.config.get([self.id, "zfs_command"], "/sbin/zfs")
         self.jobid = self.config.get(["options", "jobid"], 0)
+        self.create_time = time.time()
+        self.last_sample_time = self.create_time
+        self.metrics_to_average = self.config.get(
+            [self.id, "metrics_to_average"],
+            ["used"])
 
         if not self.volumes:
             raise sams.base.SamplerException("volumes not configured")
 
         volumes = [volume % dict(jobid=self.jobid) for volume in self.volumes]
 
+        self._average_values = {v: {k: 0 for k in self.metrics_to_average} for v in volumes}
+        self._last_averaged_values = {v: {k: 0 for k in self.metrics_to_average} for v in volumes}
         self.zfsstat = None
         if volumes:
             self.zfsstat = ZFSStats(volumes=volumes, zfs_command=self.zfs_command)
@@ -98,9 +106,34 @@ def sample(self):
         logger.debug("sample()")
         if self.zfsstat:
             entry = self.zfsstat.sample()
+            self.compute_sample_averages(entry)
             self._most_recent_sample = [self._storage_wrapping(entry)]
             self.store(entry)
 
+    def compute_sample_averages(self, volume_data):
+        """ Computes averages of selected measurements by
+        means of trapezoidal quadrature, approximating
+        that the time this function is called is the actual
+        time of sampling. This is not completely correct but simplifies
+        the implementation.
+        """
+        sample_time = time.time()
+        elapsed_time = sample_time - self.last_sample_time
+        total_elapsed_time = sample_time - self.create_time
+        for v, data in volume_data.items():
+            for key, item in data.items():
+                if key in self.metrics_to_average:
+                    # Trapezoidal quadrature
+                    weighted_item = (
+                            0.5 * (float(item) + float(self._last_averaged_values[v][key])) * elapsed_time)
+                    self._last_averaged_values[v][key] = item
+                    previous_integral = self._average_values[v][key] * (total_elapsed_time - elapsed_time)
+                    new_integral = previous_integral + weighted_item
+                    self._average_values[v][key] = new_integral / total_elapsed_time
+            for key, item in self._average_values[v].items():
+                data[key + '_average'] = item
+        self.last_sample_time = sample_time
+
     @classmethod
     def final_data(cls):
         return {}