feat: add support for live job statistics, like sstat (#355)

* add code to support live job stats retrieval
* switch linking to libslurmfull
   - will make a few things easier, without so many function
  reimplementations
   - is required to implement getting live-performance stats of jobs in the
  first place
  - nothing changes from a user perspective

* fix [user|system|total]_cpu_time not being calculated correctly
* add tests for live job statistics
* xcollections: add to_dict method
* separate Job and Step statistics
* core.job: rename cpu_time to elapsed_cpu_time
* db.job: return 1 for cpus() when the job is still pending
* stats: fix logic how collection/step stats are added up
* setup: enhanced search for slurm lib
* stats: make sure cpu time values are int
* update CHANGELOG

Author: tazend

pyslurm/__init__.py

@@ -4,10 +4,10 @@
 """
 from __future__ import absolute_import
 
-import ctypes
+import os
 import sys
 
-sys.setdlopenflags(sys.getdlopenflags() | ctypes.RTLD_GLOBAL)
+sys.setdlopenflags(sys.getdlopenflags() | os.RTLD_GLOBAL | os.RTLD_DEEPBIND )
 
 # Initialize slurm api
 from pyslurm.api import slurm_init, slurm_fini

pyslurm/core/job/job.pxd

@@ -25,6 +25,7 @@
 from pyslurm.utils cimport cstr, ctime
 from pyslurm.utils.uint cimport *
 from pyslurm.utils.ctime cimport time_t
+from pyslurm.db.stats cimport JobStatistics
 from libc.string cimport memcpy, memset
 from libc.stdint cimport uint8_t, uint16_t, uint32_t, uint64_t, int64_t
 from libc.stdlib cimport free
@@ -77,16 +78,16 @@ cdef class Jobs(MultiClusterMap):
 
     Attributes:
         memory (int):
-            Total amount of memory for all Jobs in this collection, in
-            Mebibytes
+            Total amount of memory requested for all Jobs in this collection,
+            in Mebibytes
         cpus (int):
-            Total amount of cpus for all Jobs in this collection.
+            Total amount of cpus requested for all Jobs in this collection.
         ntasks (int):
-            Total amount of tasks for all Jobs in this collection.
-        cpu_time (int):
+            Total amount of tasks requested for all Jobs in this collection.
+        elapsed_cpu_time (int):
             Total amount of CPU-Time used by all the Jobs in the collection.
             This is the result of multiplying the run_time with the amount of
-            cpus for each job.
+            cpus requested for each job.
         frozen (bool):
             If this is set to True and the `reload()` method is called, then
             *ONLY* Jobs that already exist in this collection will be
@@ -95,13 +96,18 @@ cdef class Jobs(MultiClusterMap):
             Slurm controllers memory will not be removed either.
             The default is False, so old jobs will be removed, and new Jobs
             will be added - basically the same behaviour as doing Jobs.load().
+        stats (JobStatistics):
+            Real-time statistics of all Jobs in this collection.
+            Before you can access the stats data for this, you have to call
+            the `load_stats` method on this collection.
     """
     cdef:
         job_info_msg_t *info
         slurm_job_info_t tmp_info
 
     cdef public:
         frozen
+        JobStatistics stats
 
 
 cdef class Job:
@@ -119,6 +125,14 @@ cdef class Job:
             Before you can access the Steps data for a Job, you have to call
             the `reload()` method of a Job instance or the `load_steps()`
             method of a Jobs collection.
+        stats (JobStatistics):
+            Real-time statistics of a Job.
+            Before you can access the stats data for a Job, you have to call
+            the `load_stats` method of a Job instance or the Jobs collection.
+        pids (dict[str, list]):
+            Current Process-IDs of the Job, organized by node name.
+            Before you can access the pids data for a Job, you have to call
+            the `load_stats` method of a Job instance or the Jobs collection.
         name (str):
             Name of the Job
         id (int):
@@ -362,17 +376,20 @@ cdef class Job:
             Whether this Job is a cronjob.
         cronjob_time (str):
             The time specification for the Cronjob.
-        cpu_time (int):
+        elapsed_cpu_time (int):
             Amount of CPU-Time used by the Job so far.
             This is the result of multiplying the run_time with the amount of
-            cpus.
+            cpus requested.
     """
     cdef:
         slurm_job_info_t *ptr
         dict passwd
         dict groups
 
-    cdef public JobSteps steps
+    cdef public:
+        JobSteps steps
+        JobStatistics stats
+        dict pids
 
     cdef _calc_run_time(self)
 

pyslurm/core/job/job.pyx

@@ -63,6 +63,7 @@ cdef class Jobs(MultiClusterMap):
 
     def __init__(self, jobs=None, frozen=False):
         self.frozen = frozen
+        self.stats = JobStatistics()
         super().__init__(data=jobs,
                          typ="Jobs",
                          val_type=Job,
@@ -161,15 +162,48 @@ cdef class Jobs(MultiClusterMap):
             Pending Jobs will be ignored, since they don't have any Steps yet.
 
         Raises:
-            RPCError: When retrieving the Job information for all the Steps
-                failed.
+            RPCError: When retrieving the information for all the Steps failed.
         """
         cdef dict steps = JobSteps.load_all()
         for job in self.values():
             jid = job.id
             if jid in steps:
                 job.steps = steps[jid]
 
+    def load_stats(self):
+        """Load realtime stats for this collection of Jobs.
+
+        This function additionally fills in the `stats` attribute for all Jobs
+        in the collection, and also populates its own `stats` attribute.
+        Implicitly calls `load_steps()`.
+
+        !!! note
+
+            Pending Jobs will be ignored, since they don't have any Stats yet.
+
+        Returns:
+            (JobStatistics): The statistics of this job collection.
+
+        Raises:
+            RPCError: When retrieving the stats for all the Jobs failed.
+
+        Examples:
+            >>> import pyslurm
+            >>> jobs = pyslurm.Jobs.load()
+            >>> stats = jobs.load_stats()
+            >>>
+            >>> # Print the CPU Time Used
+            >>> print(stats.total_cpu_time)
+        """
+        self.load_steps()
+        stats = JobStatistics()
+        for job in self.values():
+            job.load_stats()
+            stats.add(job.stats)
+
+        self.stats = stats
+        return self.stats
+
     @property
     def memory(self):
         return xcollections.sum_property(self, Job.memory)
@@ -183,7 +217,7 @@ cdef class Jobs(MultiClusterMap):
         return xcollections.sum_property(self, Job.ntasks)
 
     @property
-    def cpu_time(self):
+    def elapsed_cpu_time(self):
         return xcollections.sum_property(self, Job.cpu_time)
 
 
@@ -199,6 +233,8 @@ cdef class Job:
         self.groups = {}
         cstr.fmalloc(&self.ptr.cluster, LOCAL_CLUSTER)
         self.steps = JobSteps()
+        self.stats = JobStatistics()
+        self.pids = {}
 
     def _alloc_impl(self):
         if not self.ptr:
@@ -225,7 +261,7 @@ cdef class Job:
         !!! note
 
             If the Job is not pending, the related Job steps will also be
-            loaded.
+            loaded. Job statistics are however not loaded automatically.
 
         Args:
             job_id (int):
@@ -276,6 +312,8 @@ cdef class Job:
         wrap.passwd = {}
         wrap.groups = {}
         wrap.steps = JobSteps.__new__(JobSteps)
+        wrap.stats = JobStatistics()
+        wrap.pids = {}
         memcpy(wrap.ptr, in_ptr, sizeof(slurm_job_info_t))
         return wrap
 
@@ -297,6 +335,8 @@ cdef class Job:
         """
         cdef dict out = instance_to_dict(self)
         out["steps"] = self.steps.to_dict()
+        out["stats"] = self.stats.to_dict()
+        out["pids"] = self.pids
         return out
 
     def send_signal(self, signal, steps="children", hurry=False):
@@ -516,6 +556,49 @@ cdef class Job:
         """
         verify_rpc(slurm_notify_job(self.id, msg))
 
+    def load_stats(self):
+        """Load realtime statistics for a Job and its steps.
+
+        Calling this function returns the Job statistics, and additionally
+        populates the `stats` and `pids` attribute of the instance.
+
+        Returns:
+            (JobStatistics): The statistics of the job.
+
+        Raises:
+            RPCError: When receiving the Statistics was not successful.
+
+        Examples:
+            >>> import pyslurm
+            >>> job = pyslurm.Job.load(9999)
+            >>> stats = job.load_stats()
+            >>>
+            >>> # Print the CPU Time Used
+            >>> print(stats.total_cpu_time)
+            >>>
+            >>> # Print the Process-IDs for the whole Job, organized by hostname
+            >>> print(job.pids)
+        """
+        if not self.steps:
+            job = Job.load(self.id)
+            self.steps = job.steps
+
+        all_pids = {}
+        for step in self.steps.values():
+            step.load_stats()
+            self.stats._sum_steps(step.stats)
+
+            for node, pids in step.pids.items():
+                if node not in all_pids:
+                    all_pids[node] = []
+
+                all_pids[node].extend(pids)
+
+        self.stats.elapsed_cpu_time = self.run_time * self.cpus
+
+        self.pids = all_pids
+        return self.stats
+
     def get_batch_script(self):
         """Return the content of the script for a Batch-Job.
 
@@ -1181,7 +1264,7 @@ cdef class Job:
         return cstr.to_unicode(self.ptr.cronspec)
 
     @property
-    def cpu_time(self):
+    def elapsed_cpu_time(self):
         return self.cpus * self.run_time
 
     @property