Support Cgroupsv2

lib/galaxy/config/sample/job_metrics_conf.xml.sample

@@ -47,6 +47,10 @@
   <!-- <cgroup verbose="true" /> -->
   <!-- Or, specific params can be recorded. -->
   <!-- <cgroup params="cpuacct.usage,memory.max_usage_in_bytes,memory.memsw.max_usage_in_bytes" /> -->
+  <!-- cgroupsv1 and cgroupsv2 are supported, by default metrics will be
+       collected for whichever version is mounted on your system ("auto"), but
+       you can explicitly specify the version -->
+  <!-- <cgroup version="2" /> -->
 
   <!-- Uncomment to record hostname - *nix only -->
   <!-- <hostname /> -->

lib/galaxy/job_metrics/instrumenters/cgroup.py

@@ -1,4 +1,5 @@
 """The module describes the ``cgroup`` job metrics plugin."""
+import decimal
 import logging
 import numbers
 from collections import namedtuple
@@ -17,7 +18,9 @@
 
 log = logging.getLogger(__name__)
 
+VALID_VERSIONS = ("auto", "1", "2")
 TITLES = {
+    # cgroupsv1
     "memory.memsw.max_usage_in_bytes": "Max memory usage (MEM+SWP)",
     "memory.max_usage_in_bytes": "Max memory usage (MEM)",
     "memory.limit_in_bytes": "Memory limit on cgroup (MEM)",
@@ -27,14 +30,37 @@
     "memory.oom_control.oom_kill_disable": "OOM Control enabled",
     "memory.oom_control.under_oom": "Was OOM Killer active?",
     "cpuacct.usage": "CPU Time",
+    # cgroupsv2
+    "memory.events.low":
+        "Number of times the cgroup was reclaimed due to high memory pressure even though its usage is under the low "
+        "boundary",
+    "memory.events.high":
+        "Number of times processes of the cgroup were throttled and routed to perform direct memory reclaim because "
+        "the high memory boundary was exceeded",
+    "memory.events.max": "Number of times the cgroup's memory usage was about to go over the max boundary",
+    "memory.events.oom": "Number of time the cgroup's memory usage reached the limit and allocation was about to fail",
+    "memory.events.oom_kill": "Number of processes belonging to this cgroup killed by any kind of OOM killer",
+    "memory.events.oom_group_kill": "Number of times a group OOM has occurred",
+    "memory.high": "Memory usage throttle limit",
+    "memory.low": "Best-effort memory protection",
+    "memory.max": "Memory usage hard limit",
+    "memory.min": "Hard memory protection",
+    "memory.peak": "Max memory usage recorded",
+    "cpu.stat.system_usec": "CPU system time (seconds)",
+    "cpu.stat.usage_usec": "CPU usage time (seconds)",
+    "cpu.stat.user_usec": "CPU user time (seconds)",
 }
 CONVERSION = {
     "memory.oom_control.oom_kill_disable": lambda x: "No" if x == 1 else "Yes",
     "memory.oom_control.under_oom": lambda x: "Yes" if x == 1 else "No",
+    "memory.peak": lambda x: nice_size(x),
     "cpuacct.usage": lambda x: formatting.seconds_to_str(x / 10**9),  # convert nanoseconds
+    "cpu.stat.system_usec": lambda x: formatting.seconds_to_str(x / 10**6),  # convert microseconds
+    "cpu.stat.usage_usec": lambda x: formatting.seconds_to_str(x / 10**6),  # convert microseconds
+    "cpu.stat.user_usec": lambda x: formatting.seconds_to_str(x / 10**6),  # convert microseconds
 }
-CPU_USAGE_TEMPLATE = r"""
-if [ -e "/proc/$$/cgroup" -a -d "{cgroup_mount}" ]; then
+CGROUPSV1_TEMPLATE = r"""
+if [ -e "/proc/$$/cgroup" -a -d "{cgroup_mount}" -a ! -f "{cgroup_mount}/cgroup.controllers" ]; then
     cgroup_path=$(cat "/proc/$$/cgroup" | awk -F':' '($2=="cpuacct,cpu") || ($2=="cpu,cpuacct") {{print $3}}');
     if [ ! -e "{cgroup_mount}/cpu$cgroup_path/cpuacct.usage" ]; then
         cgroup_path="";
@@ -44,12 +70,6 @@
             echo "__$(basename $f)__" >> {metrics}; cat "$f" >> {metrics} 2>/dev/null;
         fi;
     done;
-fi
-""".replace(
-    "\n", " "
-).strip()
-MEMORY_USAGE_TEMPLATE = """
-if [ -e "/proc/$$/cgroup" -a -d "{cgroup_mount}" ]; then
     cgroup_path=$(cat "/proc/$$/cgroup" | awk -F':' '$2=="memory"{{print $3}}');
     if [ ! -e "{cgroup_mount}/memory$cgroup_path/memory.max_usage_in_bytes" ]; then
         cgroup_path="";
@@ -61,6 +81,16 @@
 """.replace(
     "\n", " "
 ).strip()
+CGROUPSV2_TEMPLATE = r"""
+if [ -e "/proc/$$/cgroup" -a -f "{cgroup_mount}/cgroup.controllers" ]; then
+    cgroup_path=$(cat "/proc/$$/cgroup" | awk -F':' '($1=="0") {{print $3}}');
+    for f in {cgroup_mount}/${{cgroup_path}}/{{cpu,memory}}.*; do
+        echo "__$(basename $f)__" >> {metrics}; cat "$f" >> {metrics} 2>/dev/null;
+    done;
+fi
+""".replace(
+    "\n", " "
+).strip()
 
 
 Metric = namedtuple("Metric", ("key", "subkey", "value"))
@@ -76,7 +106,7 @@ def format(self, key, value):
                 return title, nice_size(value)
             except ValueError:
                 pass
-        elif isinstance(value, (numbers.Integral, numbers.Real)) and value == int(value):
+        elif isinstance(value, (decimal.Decimal, numbers.Integral, numbers.Real)) and value == int(value):
             value = int(value)
         return title, value
 
@@ -90,6 +120,8 @@ class CgroupPlugin(InstrumentPlugin):
     def __init__(self, **kwargs):
         self.verbose = asbool(kwargs.get("verbose", False))
         self.cgroup_mount = kwargs.get("cgroup_mount", "/sys/fs/cgroup")
+        self.version = str(kwargs.get("version", "auto"))
+        assert self.version in VALID_VERSIONS, f"cgroup metric version option must be one of {VALID_VERSIONS}"
         params_str = kwargs.get("params", None)
         if params_str:
             params = [v.strip() for v in params_str.split(",")]
@@ -99,22 +131,23 @@ def __init__(self, **kwargs):
 
     def post_execute_instrument(self, job_directory: str) -> List[str]:
         commands: List[str] = []
-        commands.append(self.__record_cgroup_cpu_usage(job_directory))
-        commands.append(self.__record_cgroup_memory_usage(job_directory))
+        if self.version in ("auto", "1"):
+            commands.append(self.__record_cgroup_v1_usage(job_directory))
+        if self.version in ("auto", "2"):
+            commands.append(self.__record_cgroup_v2_usage(job_directory))
         return commands
 
     def job_properties(self, job_id, job_directory: str) -> Dict[str, Any]:
         metrics = self.__read_metrics(self.__cgroup_metrics_file(job_directory))
         return metrics
 
-    def __record_cgroup_cpu_usage(self, job_directory: str) -> str:
-        # comounted cgroups (which cpu and cpuacct are on the supported Linux distros) can appear in any order (cpu,cpuacct or cpuacct,cpu)
-        return CPU_USAGE_TEMPLATE.format(
+    def __record_cgroup_v1_usage(self, job_directory: str) -> str:
+        return CGROUPSV1_TEMPLATE.format(
             metrics=self.__cgroup_metrics_file(job_directory), cgroup_mount=self.cgroup_mount
         )
 
-    def __record_cgroup_memory_usage(self, job_directory: str) -> str:
-        return MEMORY_USAGE_TEMPLATE.format(
+    def __record_cgroup_v2_usage(self, job_directory: str) -> str:
+        return CGROUPSV2_TEMPLATE.format(
             metrics=self.__cgroup_metrics_file(job_directory), cgroup_mount=self.cgroup_mount
         )