Add detection for exceeded time limits for slurm jobs (#929)

* Add detection for exceeded time limits for slurm jobs.

Add tests to check that job kills due to memory or time are
detected by the cluster_tools.
Use a new version of the slurm docker image.

* use newest dockered-slurm version

* update changelog

* differentiate between memory and time limit exceptions

* Update cluster_tools/cluster_tools/schedulers/slurm.py

Co-authored-by: Philipp Otto <[email protected]>

* apply PR feedback

* rename multiprocessing module to avoid clash with standard lib module which led to a crash in pylint

* fix linting setup and fix all linting errors

* fix formatting

* Print docker logs in CI

* only execute cluster tools slurm tests in CI

* use cgroup/v1 plugin as v2 fails to initialize correctly in the CI

* revert changes to ci.yml

* skip slurm time limit test, because it takes >1min

---------

Co-authored-by: Philipp Otto <[email protected]>

Author: daniel-wer

.github/workflows/ci.yml

@@ -64,6 +64,11 @@ jobs:
             sleep 10
           done
 
+          # Show log output for debugging
+          docker logs slurmctld
+          docker logs c1
+          docker logs c2
+
           # Run setup.py on all three nodes
           docker exec -w /cluster_tools slurmctld bash -c "poetry install" &
           docker exec -w /cluster_tools c1 bash -c "poetry install" &

cluster_tools/Changelog.md

@@ -12,6 +12,7 @@ For upgrade instructions, please check the respective *Breaking Changes* section
 ### Breaking Changes
 
 ### Added
+- From now on the cluster tools will detect if slurm jobs were killed, because they exceeded the configured time limit, and throw a `RemoteTimeLimitException` indicating the issue instead of a generic `RemoteException`. If it is unclear whether a job was killed because of the time or memory limit, a `RemoteResourceLimitException` is thrown. [#929](https://github.com/scalableminds/webknossos-libs/pull/929)
 
 ### Changed
 

cluster_tools/cluster_tools/__init__.py

@@ -1,24 +1,16 @@
-import logging
-import multiprocessing
-import os
-import tempfile
-from concurrent import futures
-from concurrent.futures import ProcessPoolExecutor
-from functools import partial
-from pathlib import Path
-from shutil import rmtree
 from typing import Any, Union, overload
 
 from typing_extensions import Literal
 
-from cluster_tools._utils.warning import enrich_future_with_uncaught_warning
 from cluster_tools.executors.debug_sequential import DebugSequentialExecutor
-from cluster_tools.executors.multiprocessing import MultiprocessingExecutor
-from cluster_tools.executors.pickle import PickleExecutor
+from cluster_tools.executors.multiprocessing_ import MultiprocessingExecutor
+from cluster_tools.executors.pickle_ import PickleExecutor
 from cluster_tools.executors.sequential import SequentialExecutor
 from cluster_tools.schedulers.cluster_executor import (
     ClusterExecutor,
     RemoteOutOfMemoryException,
+    RemoteResourceLimitException,
+    RemoteTimeLimitException,
 )
 from cluster_tools.schedulers.kube import KubernetesExecutor
 from cluster_tools.schedulers.pbs import PBSExecutor

cluster_tools/cluster_tools/_utils/call.py

@@ -13,6 +13,8 @@ def call(command: str, stdin: Optional[str] = None) -> Tuple[str, str, int]:
         stdin_flag = None
     p = subprocess.run(
         command,
+        stdin=stdin_flag,
+        check=False,
         shell=True,
         capture_output=True,
         text=True,

cluster_tools/cluster_tools/_utils/file_wait_thread.py

@@ -2,7 +2,7 @@
 import os
 import threading
 import time
-from typing import TYPE_CHECKING, Callable, Dict, Tuple
+from typing import TYPE_CHECKING, Callable, Dict
 
 if TYPE_CHECKING:
     from cluster_tools.schedulers.cluster_executor import ClusterExecutor

cluster_tools/cluster_tools/_utils/multiprocessing_logging_handler.py

@@ -9,7 +9,7 @@
 from logging.handlers import QueueHandler
 from queue import Empty as QueueEmpty
 from queue import Queue
-from typing import Any, Callable, List, Tuple
+from typing import Any, Callable, List
 
 # Inspired by https://stackoverflow.com/a/894284
 

cluster_tools/cluster_tools/_utils/reflection.py

@@ -1,6 +1,4 @@
 import os
-import pickle
-import sys
 from typing import Callable
 
 WARNING_TIMEOUT = 10 * 60  # seconds

cluster_tools/cluster_tools/_utils/string.py renamed to cluster_tools/cluster_tools/_utils/string_.py

@@ -2,6 +2,8 @@
 import random
 import string
 
+# The module name includes a _-suffix to avoid name clashes with the standard library string module.
+
 
 def local_filename(filename: str = "") -> str:
     return os.path.join(os.getenv("CFUT_DIR", ".cfut"), filename)

cluster_tools/cluster_tools/_utils/tailf.py

@@ -33,7 +33,7 @@ def follow(self, seconds: int = 1) -> None:
         """
 
         self.check_file_validity(self.tailed_file)
-        with open(self.tailed_file, errors="replace") as file_:
+        with open(self.tailed_file, encoding="utf-8", errors="replace") as file_:
             # Don't seek, since we want to print the entire file here.
             while True:
                 line = file_.readline()

cluster_tools/cluster_tools/_utils/warning.py

@@ -1,6 +1,4 @@
 import logging
-import os
-import string
 import threading
 import time
 from concurrent.futures import Future