Merge pull request #740 from simvue-io/wk9874/2.0.0_a3

Bug fixes for 2.0.0a3

Author: kzscisoft

simvue/api/objects/base.py

@@ -171,6 +171,7 @@ def __init__(
             {
                 "Authorization": f"Bearer {self._user_config.server.token.get_secret_value()}",
                 "User-Agent": _user_agent or f"Simvue Python client {__version__}",
+                "Accept-Encoding": "gzip",
             }
             if not self._offline
             else {}

simvue/client.py

@@ -84,7 +84,8 @@ def __init__(
                 logger.warning(f"No {label} specified")
 
         self._headers: dict[str, str] = {
-            "Authorization": f"Bearer {self._user_config.server.token.get_secret_value()}"
+            "Authorization": f"Bearer {self._user_config.server.token.get_secret_value()}",
+            "Accept-Encoding": "gzip",
         }
 
     @prettify_pydantic

simvue/executor.py

@@ -22,6 +22,7 @@
 import pathlib
 import time
 import typing
+from simvue.api.objects.alert.user import UserAlert
 
 if typing.TYPE_CHECKING:
     import simvue
@@ -113,6 +114,7 @@ def __init__(self, simvue_runner: "simvue.Run", keep_logs: bool = True) -> None:
         self._alert_ids: dict[str, str] = {}
         self.command_str: dict[str, str] = {}
         self._processes: dict[str, subprocess.Popen] = {}
+        self._all_processes: list[psutil.Process] = []
 
     def std_out(self, process_id: str) -> str | None:
         if not os.path.exists(out_file := f"{self._runner.name}_{process_id}.out"):
@@ -270,19 +272,48 @@ def processes(self) -> list[psutil.Process]:
         if not self._processes:
             return []
 
-        _all_processes: list[psutil.Process] = []
+        _current_processes: list[psutil.Process] = []
 
         for process in self._processes.values():
             with contextlib.suppress(psutil.NoSuchProcess):
-                _all_processes.append(psutil.Process(process.pid))
+                _current_processes.append(psutil.Process(process.pid))
 
         with contextlib.suppress(psutil.NoSuchProcess, psutil.ZombieProcess):
-            for process in _all_processes:
+            for process in _current_processes:
                 for child in process.children(recursive=True):
-                    if child not in _all_processes:
-                        _all_processes.append(child)
+                    if child not in _current_processes:
+                        _current_processes.append(child)
 
-        return list(set(_all_processes))
+        _current_pids = set([_process.pid for _process in _current_processes])
+        _previous_pids = set([_process.pid for _process in self._all_processes])
+
+        # Find processes which used to exist, which are no longer running
+        _expired_process_pids = _previous_pids - _current_pids
+
+        # Remove these processes from list of all processes
+        self._all_processes = [
+            _process
+            for _process in self._all_processes
+            if _process.pid not in _expired_process_pids
+        ]
+
+        # Find new processes
+        _new_process_pids = _current_pids - _previous_pids
+        _new_processes = [
+            _process
+            for _process in _current_processes
+            if _process.pid in _new_process_pids
+        ]
+
+        # Get CPU usage stats for each of those new processes, so that next time it's measured by the heartbeat the value is accurate
+        if _new_processes:
+            [_process.cpu_percent() for _process in _new_processes]
+            time.sleep(0.1)
+
+        # Add these to the list of all processes
+        self._all_processes += _new_processes
+
+        return self._all_processes
 
     @property
     def success(self) -> int:
@@ -342,17 +373,26 @@ def _update_alerts(self) -> None:
         # allowing the executor to finish (and as such the run instance to exit)
         _wait_limit: float = 1
         for proc_id, process in self._processes.items():
+            # We don't want to override the user's setting for the alert status
+            # This is so that if a process incorrectly reports its return code,
+            # the user can manually set the correct status depending on logs etc.
+            _alert = UserAlert(identifier=self._alert_ids[proc_id])
+            _is_set = _alert.get_status(run_id=self._runner._id)
+
             if process.returncode != 0:
                 # If the process fails then purge the dispatcher event queue
                 # and ensure that the stderr event is sent before the run closes
                 if self._runner._dispatcher:
                     self._runner._dispatcher.purge()
-
-                self._runner.log_alert(
-                    identifier=self._alert_ids[proc_id], state="critical"
-                )
+                if not _is_set:
+                    self._runner.log_alert(
+                        identifier=self._alert_ids[proc_id], state="critical"
+                    )
             else:
-                self._runner.log_alert(identifier=self._alert_ids[proc_id], state="ok")
+                if not _is_set:
+                    self._runner.log_alert(
+                        identifier=self._alert_ids[proc_id], state="ok"
+                    )
 
             _current_time: float = 0
             while (

simvue/run.py

@@ -186,7 +186,8 @@ def __init__(
         )
         self._headers: dict[str, str] = (
             {
-                "Authorization": f"Bearer {self._user_config.server.token.get_secret_value()}"
+                "Authorization": f"Bearer {self._user_config.server.token.get_secret_value()}",
+                "Accept-Encoding": "gzip",
             }
             if mode != "offline"
             else {}
@@ -313,16 +314,11 @@ def processes(self) -> list[psutil.Process]:
             return process_list
 
         process_list += [self._parent_process]
-
-        # Attach child processes relating to the process set by set_pid
-        with contextlib.suppress(psutil.NoSuchProcess, psutil.ZombieProcess):
-            for child in self._parent_process.children(recursive=True):
-                if child not in process_list:
-                    process_list.append(child)
+        process_list += self._child_processes
 
         return list(set(process_list))
 
-    def _get_sysinfo(self) -> dict[str, typing.Any]:
+    def _get_sysinfo(self, interval: float | None = None) -> dict[str, typing.Any]:
         """Retrieve system administration
 
         Parameters
@@ -336,7 +332,7 @@ def _get_sysinfo(self) -> dict[str, typing.Any]:
             retrieved system specifications
         """
         processes = self.processes
-        cpu = get_process_cpu(processes, interval=0.1)
+        cpu = get_process_cpu(processes, interval=interval)
         memory = get_process_memory(processes)
         gpu = get_gpu_metrics(processes)
         data: dict[str, typing.Any] = {}
@@ -375,7 +371,10 @@ def _heartbeat(
             last_res_metric_call = time.time()
 
             if self._resources_metrics_interval:
-                self._add_metrics_to_dispatch(self._get_sysinfo(), join_on_fail=False)
+                self._add_metrics_to_dispatch(
+                    self._get_sysinfo(interval=1), join_on_fail=False, step=0
+                )
+                res_step = 1
 
             while not heartbeat_trigger.is_set():
                 time.sleep(0.1)
@@ -390,9 +389,10 @@ def _heartbeat(
                         # join would be called on this thread and a thread cannot
                         # join itself!
                         self._add_metrics_to_dispatch(
-                            self._get_sysinfo(), join_on_fail=False
+                            self._get_sysinfo(), join_on_fail=False, step=res_step
                         )
                         last_res_metric_call = res_time
+                        res_step += 1
 
                 if time.time() - last_heartbeat < self._heartbeat_interval:
                     continue
@@ -506,6 +506,9 @@ def _start(self, reconnect: bool = False) -> bool:
             self._pid = os.getpid()
 
         self._parent_process = psutil.Process(self._pid) if self._pid else None
+        self._child_processes = (
+            self._get_child_processes() if self._parent_process else None
+        )
 
         self._shutdown_event = threading.Event()
         self._heartbeat_termination_trigger = threading.Event()
@@ -920,6 +923,16 @@ def kill_all_processes(self) -> None:
             )
         self._executor.kill_all()
 
+    def _get_child_processes(self) -> list[psutil.Process]:
+        _process_list = []
+        # Attach child processes relating to the process set by set_pid
+        with contextlib.suppress(psutil.NoSuchProcess, psutil.ZombieProcess):
+            for child in self._parent_process.children(recursive=True):
+                if child not in _process_list:
+                    _process_list.append(child)
+
+        return list(set(_process_list))
+
     @property
     def executor(self) -> Executor:
         """Return the executor for this run"""
@@ -975,6 +988,13 @@ def set_pid(self, pid: int) -> None:
         """
         self._pid = pid
         self._parent_process = psutil.Process(self._pid)
+        self._child_processes = self._get_child_processes()
+        # Get CPU usage stats for each of those new processes, so that next time it's measured by the heartbeat the value is accurate
+        [
+            _process.cpu_percent()
+            for _process in self._child_processes + [self._parent_process]
+        ]
+        time.sleep(0.1)
 
     @skip_if_failed("_aborted", "_suppress_errors", False)
     @pydantic.validate_call
@@ -1686,6 +1706,11 @@ def add_alerts(
         names = names or []
 
         if names and not ids:
+            if self._user_config.run.mode == "offline":
+                self._error(
+                    "Cannot retrieve alerts based on names in offline mode - please use IDs instead."
+                )
+                return False
             try:
                 if alerts := Alert.get(offline=self._user_config.run.mode == "offline"):
                     ids += [id for id, alert in alerts if alert.name in names]
@@ -1986,6 +2011,12 @@ def log_alert(
             self._error("Please specify alert to update either by ID or by name.")
             return False
 
+        if name and self._user_config.run.mode == "offline":
+            self._error(
+                "Cannot retrieve alerts based on names in offline mode - please use IDs instead."
+            )
+            return False
+
         if name:
             try:
                 if alerts := Alert.get(offline=self._user_config.run.mode == "offline"):

simvue/sender.py

@@ -11,6 +11,7 @@
 from concurrent.futures import ThreadPoolExecutor
 import threading
 import requests
+import psutil
 from simvue.config.user import SimvueConfiguration
 
 import simvue.api.objects
@@ -167,13 +168,14 @@ def sender(
     cache_dir = cache_dir or _user_config.offline.cache
 
     cache_dir.joinpath("server_ids").mkdir(parents=True, exist_ok=True)
+    _lock_path = cache_dir.joinpath("sender.lock")
 
     # Check that no other sender is already currently running...
-    if cache_dir.joinpath("sender.lock").exists():
+    if _lock_path.exists() and psutil.pid_exists(int(_lock_path.read_text())):
         raise RuntimeError("A sender is already running for this cache!")
 
     # Create lock file to prevent other senders running while this one isn't finished
-    cache_dir.joinpath("sender.lock").touch()
+    _lock_path.write_text(str(psutil.Process().pid))
 
     _id_mapping: dict[str, str] = {
         file_path.name.split(".")[0]: file_path.read_text()
@@ -233,5 +235,5 @@ def sender(
                 _heartbeat_files,
             )
     # Remove lock file to allow another sender to start in the future
-    cache_dir.joinpath("sender.lock").unlink()
+    _lock_path.unlink()
     return _id_mapping