CPU and Max RSS Analysis tools

.gitignore

@@ -11,6 +11,9 @@ __pycache__/
 # vscode
 .vscode
 
+# pycharm
+.idea
+
 # processed workflow configs
 *.rc.processed
 *.cylc.processed

changes.d/6663.feat.md

@@ -0,0 +1 @@
+Added a Cylc job profiler which captures CPU and memory information from job runners which use cgroups. This information can be reviewed in the Analysis view in the GUI.

cylc/flow/cfgspec/globalcfg.py

@@ -1477,6 +1477,38 @@ def default_for(
 
                 .. versionadded:: 8.0.0
             ''')
+
+            with Conf('profiler', desc='''
+                Configure the Cylc job profiler.
+
+                This tool can capture CPU and memory information from
+                job runners which use cgroups such as PBS and Slurm.
+
+                .. versionadded:: 8.7.0
+            '''):
+                Conf('activate', VDR.V_BOOLEAN, False, desc='''
+                    Enable the Cylc profiler for this platform.
+                ''')
+                Conf('cgroups path', VDR.V_STRING,
+                     default='/sys/fs/cgroup',
+                     desc='''
+                     Configure the path to the cgroups filesystem.
+
+                     The default value (``/sys/fs/cgroup``) is the standard
+                     location for cgroups on linux and should work in
+                     most circumstances
+                     ''')
+                Conf('polling interval', VDR.V_INTEGER,
+                     default=10,
+                     desc='''
+                     Configure the profiler polling interval.
+
+                     The interval (in seconds) at which the profiler will
+                     poll the cgroups filesystem for resource usage data.
+                     The default value of 10 seconds should be sufficient for
+                     most use cases, but can be adjusted as needed.
+                ''')
+
             Conf('job runner', VDR.V_STRING, 'background', desc=f'''
                 The system used to run jobs on the platform.
 

cylc/flow/etc/job.sh

@@ -139,6 +139,12 @@ cylc__job__main() {
     mkdir -p "$(dirname "${CYLC_TASK_WORK_DIR}")" || true
     mkdir -p "${CYLC_TASK_WORK_DIR}"
     cd "${CYLC_TASK_WORK_DIR}"
+
+    if [[ "${CYLC_PROFILER}" == "True" ]] ; then
+       cylc profile -m "${CYLC_CGROUP}" -i "${CYLC_PROFILER_POLL_INTERVAL}" &
+       export profiler_pid="$!"
+    fi
+
     # Env-Script, User Environment, Pre-Script, Script and Post-Script
     # Run user scripts in subshell to protect cylc job script from interference.
     # Waiting on background process allows signal traps to trigger immediately.
@@ -157,11 +163,15 @@ cylc__job__main() {
             cylc__set_return "$ret_code"
         fi
     }
+    # Grab the max rss and cpu_time and clean up before changing directory
+    cylc__kill_profiler
     # Empty work directory remove
     cd
     rmdir "${CYLC_TASK_WORK_DIR}" 2>'/dev/null' || true
     # Send task succeeded message
+
     wait "${CYLC_TASK_MESSAGE_STARTED_PID}" 2>'/dev/null' || true
+
     cylc message -- "${CYLC_WORKFLOW_ID}" "${CYLC_TASK_JOB}" 'succeeded' || true
     # (Ignore shellcheck "globbing and word splitting" warning here).
     # shellcheck disable=SC2086
@@ -187,6 +197,14 @@ cylc__set_return() {
     return "${1:-0}"
 }
 
+###############################################################################
+# Save the data using cylc message and exit the profiler
+cylc__kill_profiler() {
+    if [[ -n "${profiler_pid:-}" ]] && ps -p "$profiler_pid" > /dev/null; then
+      kill -s SIGINT "${profiler_pid}" || true
+    fi
+}
+
 ###############################################################################
 # Disable selected or all (if no arguments given) fail traps.
 # Globals:
@@ -268,6 +286,9 @@ cylc__job_finish_err() {
     # (Ignore shellcheck "globbing and word splitting" warning here).
     # shellcheck disable=SC2086
     trap '' ${CYLC_VACATION_SIGNALS:-} ${CYLC_FAIL_SIGNALS}
+
+    cylc__kill_profiler
+
     if [[ -n "${CYLC_TASK_MESSAGE_STARTED_PID:-}" ]]; then
         wait "${CYLC_TASK_MESSAGE_STARTED_PID}" 2>'/dev/null' || true
     fi

cylc/flow/exceptions.py

@@ -529,3 +529,16 @@ def __init__(self, message, expr=None):
 
     def __str__(self):
         return self.message
+
+
+class CylcProfilerError(CylcError):
+    """Exception for errors raised from the cylc profiler. These errors do not
+    affect workflows functionally, just stats gathering. We don't want to
+    panic users."""
+    def __init__(self, exc: Exception, error_msg: str) -> None:
+        CylcError.__init__(
+            self,
+            f"{exc}. {error_msg}. This error came from the Cylc profiler"
+            f" and is not a problem with your workflow. Statistics gathering "
+            f"for the analysis view may be incomplete."
+        )

cylc/flow/job_file.py

@@ -224,8 +224,16 @@ def _write_task_environment(self, handle, job_conf):
             '\n    export CYLC_TASK_TRY_NUMBER=%s' % job_conf['try_num'])
         handle.write(
             "\n    export CYLC_TASK_FLOW_NUMBERS="
-            f"{','.join(str(f) for f in job_conf['flow_nums'])}"
-        )
+            f"{','.join(str(f) for f in job_conf['flow_nums'])}")
+        handle.write(
+            "\n    export CYLC_PROFILER="
+            f"{job_conf['platform']['profiler']['activate']}")
+        handle.write(
+            "\n    export CYLC_CGROUP="
+            f"{job_conf['platform']['profiler']['cgroups path']}")
+        handle.write(
+            "\n    export CYLC_PROFILER_POLL_INTERVAL="
+            f"{job_conf['platform']['profiler']['polling interval']}")
         # Standard parameter environment variables
         for var, val in job_conf['param_var'].items():
             handle.write('\n    export CYLC_TASK_PARAM_%s="%s"' % (var, val))

cylc/flow/scripts/message.py

@@ -93,12 +93,13 @@
 """
 
 
+import asyncio
 from logging import getLevelName, INFO
 import os
 import sys
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Sequence
 
-from cylc.flow.id_cli import parse_id
+from cylc.flow.id_cli import parse_id_async
 from cylc.flow.option_parsers import (
     WORKFLOW_ID_ARG_DOC,
     CylcOptionParser as COP
@@ -142,6 +143,10 @@ def main(parser: COP, options: 'Values', *args: str) -> None:
     if not args:
         parser.error('No message supplied')
         return
+    asyncio.run(_main(options, args))
+
+
+async def _main(options: 'Values', args: Sequence[str]) -> None:
     if len(args) <= 2:
         # BACK COMPAT: args <= 2
         # from:
@@ -160,7 +165,7 @@ def main(parser: COP, options: 'Values', *args: str) -> None:
         message_strs = list(args)
     else:
         workflow_id, job_id, *message_strs = args
-        workflow_id, *_ = parse_id(
+        workflow_id, *_ = await parse_id_async(
             workflow_id,
             constraint='workflows',
         )
@@ -198,4 +203,4 @@ def main(parser: COP, options: 'Values', *args: str) -> None:
             messages.append([options.severity, message_str.strip()])
         else:
             messages.append([getLevelName(INFO), message_str.strip()])
-    record_messages(workflow_id, job_id, messages)
+    await record_messages(workflow_id, job_id, messages)