Google Git Google Git
Sign in
third-party-mirror/SchedMD/slurm/04ea0cdbdba2c80ca9dd0f6cb72a005d66d780bf/./testsuite/python/lib/atf.py
blob: 79739cd9a2daae2b60b054eb482c0dde3d3479d5 [file]
##############################################################################
# ATF (Automated Testing Framework) python testsuite module
# Copyright (C) SchedMD LLC.
##############################################################################
import collections
import copy
import datetime
import enum
import errno
import glob
import json
# import glob
import logging
import math
import os
import pathlib
import pwd
import re
import shlex
import shutil
import signal
import socket
import stat
import subprocess
import sys
import time
import traceback
import jsondiff
import jsonpatch
import pexpect
import pytest
# slurmrestd
import requests
import yaml
# This module will be (un)imported in require_openapi_generator()
openapi_client = None
import importlib
import threading
##############################################################################
# ATF module functions
##############################################################################
default_command_timeout = 60
default_polling_timeout = 45
default_sql_cmd_timeout = 120
PERIODIC_TIMEOUT = 30
def get_open_port():
"""Finds an open port.
Warning: Race conditions abound so be ready to retry calling function;
Example:
>>> while not some_test(port):
>>> port = get_open_port()
Shamelessly based on:
https://stackoverflow.com/questions/2838244/get-open-tcp-port-in-python
"""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("", 0))
s.listen(1)
port = s.getsockname()[1]
s.close()
return port
def node_range_to_list(node_expression):
"""Converts a node range expression into a list of node names.
Example:
>>> node_range_to_list('node[1,3-5]')
['node1', 'node3', 'node4', 'node5']
"""
node_list = []
output = run_command_output(
f"scontrol show hostnames {node_expression}", fatal=True, quiet=True
)
for line in output.rstrip().splitlines():
node_list.append(line)
return node_list
def node_list_to_range(node_list):
"""Converts a list of node names to a node range expression.
Example:
>>> node_list_to_range(['node1', 'node3', 'node4', 'node5'])
'node[1,3-5]'
"""
return run_command_output(
f"scontrol show hostlistsorted {','.join(node_list)}", fatal=True, quiet=True
).rstrip()
def range_to_list(range_expression):
"""Converts an integer range expression into a list of integers.
Example:
>>> range_to_list('1,3-5')
[1, 3, 4, 5]
"""
return list(map(int, node_range_to_list(f"[{range_expression}]")))
def list_to_range(numeric_list):
"""Converts a list of integers to an integer range expression.
Example:
>>> list_to_range([1, 3, 4, 5])
'1,3-5'
"""
node_range_expression = node_list_to_range(map(str, numeric_list))
return re.sub(r"^\[(.*)\]$", r"\1", node_range_expression)
def mask_to_list(mask):
"""Converts an integer bitmask to a sorted list of its set-bit indices.
Example:
>>> mask_to_list(0b1011)
[0, 1, 3]
"""
return [i for i in range(mask.bit_length()) if mask >> i & 1]
def get_coredumps():
"""
Return the coredumps with the expected pattern in the tmp dirs of the test
and in the logs dir (the logs dir is the cwd of the daemons).
"""
core_list = glob.glob(f"{properties['slurm-logs-dir']}/*.core*") + glob.glob(
f"{module_tmp_path}/../**/*.core*", recursive=True
)
# pytest creates a 'current' symlink alongside to tmp directories
core_set = set()
for core in core_list:
if core.endswith(".bt"):
continue
core_set.add(os.path.realpath(core))
return list(core_set)
def resolve_core_binary(core_path, execfn_path):
"""Return the on-disk binary whose build-id matches the executable
recorded in the coredump, or None if no match is found.
Handles the case where a symlink like may have been swapped between
in upgrades/mixed versions setups.
"""
# Get the build-ids of all the modules in the coredump.
# The executable is not necessarily the first module listed by eu-unstrip
core_bids = set()
result = run_command(
f"DEBUGINFOD_URLS= eu-unstrip -n --core {core_path}",
quiet=True,
user="root",
timeout=300,
)
for line in result["stdout"].splitlines():
m = re.match(r"\S+\s+([0-9a-f]+)", line)
if m:
core_bids.add(m.group(1))
if not core_bids:
logging.warning(f"Unable to extract build-ids from coredump {core_path}")
return None
candidates = [os.path.realpath(execfn_path)]
sp = properties.get("slurm-prefix")
if sp and execfn_path.startswith(sp + "/"):
rel = execfn_path[len(sp) + 1 :]
for key in ("old-slurm-prefix", "new-slurm-prefix"):
if key in properties:
candidates.append(f"{properties[key]}/{rel}")
for cand in candidates:
out = run_command_output(f"readelf -n {cand}", quiet=True, user="root")
m = re.search(r"Build ID:\s+([0-9a-f]+)", out)
if os.path.exists(cand) and m and m.group(1) in core_bids:
return cand
logging.warning(
f"No binary matches build-ids {sorted(core_bids)} for coredump {core_path} (candidates tried: {candidates})"
)
return None
def run_command(
command,
fatal=False,
timeout=default_command_timeout,
quiet=False,
chdir=None,
user=None,
input=None,
xfail=False,
env_vars=None,
background=False,
):
"""Executes a command and returns a dictionary result.
Args:
command (string): The command to execute. The command is run within a
bash subshell, so pipes, redirection, etc. are performed.
fatal (boolean): If True, a non-zero exit code (or zero if combined
with xfail) will result in the test failing.
timeout (integer): If the command does not exit before timeout number
of seconds, this function will return with an exit code of 110.
quiet (boolean): If True, logging is performed at the TRACE log level.
chdir (directory): Change to the specified directory before executing
the command.
user (user name): Run the command as the specified user. This requires
the invoking user to have unprompted sudo rights.
input (string): The specified input is supplied to the command as stdin.
xfail (boolean): If True, the command is expected to fail.
env_vars (string): A string to set environmental variables that is
prepended to the command when run.
background (boolean): If True, runs command in background and returns
immediately. Process can be accessed via results["process"]. Cannot
be used with fatal, xfail, timeout, capture_output, or check.
Returns:
A dictionary containing the following keys:
start_time: epoch start time
duration: number of seconds the command ran for
exit_code: exit code for the command
stdout: command stdout as a string
stderr: command stderr as a string
Example:
>>> run_command('ls -l', fatal=True)
{'command': 'ls -l', 'start_time': 1712268971.532, 'duration': 0.007, 'exit_code': 0, 'stdout': 'total 124\n-rw-rw-r-- 1 slurm slurm 118340 Apr 4 22:15 atf.py\n-rw-rw-r-- 1 slurm slurm 498 Apr 4 22:09 test.py\n-rw-rw-r-- 1 slurm slurm 1013 Apr 4 22:09 python_script.py\n', 'stderr': ''}
>>> run_command('ls /non/existent/path', xfail=True)
{'command': 'ls /non/existent/path', 'start_time': 1712269123.2, 'duration': 0.005, 'exit_code': 2, 'stdout': '', 'stderr': "ls: cannot access '/non/existent/path': No such file or directory\n"}
>>> run_command('sleep 5', timeout=2)
{'command': 'sleep 5', 'start_time': 1712269157.113, 'duration': 2.0, 'exit_code': 110, 'stdout': '', 'stderr': ''}
"""
additional_run_kwargs = {}
if chdir is not None:
additional_run_kwargs["cwd"] = chdir
if input is not None:
additional_run_kwargs["input"] = input
if timeout is not None:
additional_run_kwargs["timeout"] = timeout
if quiet:
log_command_level = logging.TRACE
log_details_level = logging.TRACE
else:
log_command_level = logging.NOTE
log_details_level = logging.DEBUG
if env_vars is not None:
command = env_vars.strip() + " " + command
# If user is not specified but test-user is set, then set user to test-user
if not user and properties["test-user-set"]:
user = properties["test-user"]
start_time = time.time()
invocation_message = "Running command"
if user is not None:
invocation_message += f" as user {user}"
invocation_message += f": {command}"
logging.log(log_command_level, invocation_message)
try:
if user is not None:
if not properties["sudo-rights"]:
pytest.skip(
"This test requires the test user to have unprompted sudo rights",
allow_module_level=True,
)
# Use su to honor ulimits, specially core
preserve = [
"PATH",
"SLURM_TESTSUITE_CONF",
*sorted(properties["lmod-touched-vars"]),
]
cmd = [
"sudo",
f"--preserve-env={','.join(preserve)}",
"-u",
user,
"/bin/bash",
"-lc",
command,
]
else:
cmd = command
# Run in background mode and return before completion if requested
if background:
# Create kwargs for Popen, excluding unsupported run() params
background_run_kwargs = additional_run_kwargs.copy()
# Check that parameters that don't make sense haven't been specified
if background_run_kwargs["timeout"] != default_command_timeout:
pytest.fail("Unsupported: timeout cannot be used with background=True")
del background_run_kwargs["timeout"]
if fatal:
pytest.fail("Unsupported: fatal cannot be used with background=True")
if xfail:
pytest.fail("Unsupported: xfail cannot be used with background=True")
if "input" in background_run_kwargs:
input_str = background_run_kwargs.pop("input")
else:
input_str = None
# Run command differently depending on if using sudo for security
if isinstance(cmd, list):
process = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
**background_run_kwargs,
)
else:
process = subprocess.Popen(
cmd,
shell=True,
executable="/bin/bash",
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
**background_run_kwargs,
)
# Handle input if provided
if input_str is not None:
process.stdin.write(input_str)
process.stdin.close()
# Return early with process object
results = {
"command": command,
"start_time": float(int(start_time * 1000)) / 1000,
"duration": float(int((time.time() - start_time) * 1000)) / 1000,
"exit_code": None,
"stdout": "",
"stderr": "",
"process": process,
}
return results
# Synchronous code path
if isinstance(cmd, list):
cp = subprocess.run(
cmd, capture_output=True, text=True, **additional_run_kwargs
)
else:
cp = subprocess.run(
cmd,
shell=True,
executable="/bin/bash",
capture_output=True,
text=True,
**additional_run_kwargs,
)
end_time = time.time()
duration = end_time - start_time
exit_code = cp.returncode
stdout = cp.stdout
stderr = cp.stderr
except subprocess.TimeoutExpired as e:
duration = e.timeout
exit_code = errno.ETIMEDOUT
# These are byte objects, not strings
stdout = e.stdout.decode("utf-8") if e.stdout else ""
stderr = e.stderr.decode("utf-8") if e.stderr else ""
if input is not None:
logging.log(log_details_level, f"Command input: {input}")
logging.log(log_details_level, f"Command exit code: {exit_code}")
logging.log(log_details_level, f"Command stdout: {stdout}")
logging.log(log_details_level, f"Command stderr: {stderr}")
logging.log(log_details_level, "Command duration: %.03f seconds", duration)
message = ""
if exit_code == errno.ETIMEDOUT:
message = f'Command "{command}" timed out after {duration} seconds'
elif exit_code != 0 and not xfail:
message = f'Command "{command}" failed with rc={exit_code}'
elif exit_code == 0 and xfail:
message = f'Command "{command}" was expected to fail but succeeded'
if (exit_code != 0 and not xfail) or (exit_code == 0 and xfail):
if stderr != "" or stdout != "":
message += ":"
if stderr != "":
message += f" {stderr}"
if stdout != "":
message += f" {stdout}"
if message != "":
message = message.rstrip()
if fatal:
pytest.fail(message)
elif not quiet:
logging.warning(message)
results = {}
results["command"] = command
results["start_time"] = float(int(start_time * 1000)) / 1000
results["duration"] = float(int(duration * 1000)) / 1000
results["exit_code"] = exit_code
results["stdout"] = stdout
results["stderr"] = stderr
return results
def run_command_error(command, **run_command_kwargs):
"""Executes a command and returns the standard error.
This function accepts the same arguments as run_command.
Args:
command (string): The command to execute. The command is run within a
bash subshell, so pipes, redirection, etc. are performed.
Returns:
The standard error (stderr) output of the command as a string.
Example:
>>> run_command_error('ls /non/existent/path')
"ls: cannot access '/non/existent/path': No such file or directory\n"
>>> run_command_error('grep foo /etc/passwd', quiet=True)
''
>>> run_command_error('echo error message >&2', xfail=True)
'error message\n'
"""
results = run_command(command, **run_command_kwargs)
return results["stderr"]
def run_command_output(command, **run_command_kwargs):
"""Executes a command and returns the standard output.
This function accepts the same arguments as run_command.
Args:
command (string): The command to execute. The command is run within a
Bash subshell, so pipes, redirection, etc. are performed.
Returns:
The standard output (stdout) of the command as a string.
Example:
>>> run_command_output('ls')
'file1.txt\nfile2.txt\nscript.py\n'
>>> run_command_output('echo Hello, World!')
'Hello, World!\n'
>>> run_command_output('grep foo /etc/passwd', xfail=True)
''
"""
results = run_command(command, **run_command_kwargs)
return results["stdout"]
def run_command_exit(command, **run_command_kwargs):
"""Executes a command and returns the exit code.
This function accepts the same arguments as run_command.
Args:
command (string): The command to execute. The command is run within a
Bash subshell, so pipes, redirection, etc. are performed.
Returns:
The exit code of the command as an integer.
Example:
>>> run_command_exit('ls')
0
>>> run_command_exit('grep foo /etc/passwd', xfail=True)
1
>>> run_command_exit('sleep 5', timeout=2)
110
"""
results = run_command(command, **run_command_kwargs)
return results["exit_code"]
def timer(
timeout=default_polling_timeout,
poll_interval=None,
fatal=False,
xfail=False,
quiet=False,
):
"""A timer to do-while a timeout is not triggered.
Iterates in a loop, sleeping between iterations, until the caller breaks
or the timeout is reached.
Args:
timeout (float): Maximum time in seconds to wait (default:
default_polling_timeout).
poll_interval (float): Seconds between iterations. Defaults to
timeout / 10.0
fatal (bool): If True, call pytest.fail() on timeout. Otherwise the
caller should use the for-else to detect the timeout.
xfail (bool): If True, a timeout is expected.
quiet (bool): If True, no progress is log.
Example:
>>> running = False
>>> for t in atf.timer(timeout=60, fatal=False):
... if get_job_state(job_id) == 'RUNNING':
... running = True
... break
... else:
... logging.warning("Job was not RUNNING before the timeout.")
... assert running, "Job should be running"
"""
if poll_interval is None:
poll_interval = timeout / 10.0
_not = " not" if not xfail else ""
message = f"Timer should{_not} timeout"
start = time.time()
remaining_time = timeout
while True:
# Run the caller loop (at least once, like a do-while)
yield int(remaining_time)
# Check for timeout
remaining_time = timeout - (time.time() - start)
if remaining_time <= 0:
msg = message + " (timeout)"
if not xfail:
if fatal:
pytest.fail(msg)
logging.warning(msg)
else:
logging.debug(msg)
return
# Wait for the next attempt
if not quiet:
logging.debug(message + f", remaining time: {remaining_time:.0f}s")
time.sleep(poll_interval)
def repeat_until(
callable,
condition,
timeout=default_polling_timeout,
poll_interval=None,
xfail=False,
fatal=False,
):
"""Repeats a callable until a condition is met or it times out.
The callable returns an object that the condition operates on.
Args:
callable (callable): Repeatedly called until the condition is met or
the timeout is reached.
condition (callable): A callable object that returns a boolean. This
function will return True when the condition call returns True.
timeout (integer): If timeout number of seconds expires before the
condition is met, return False.
poll_interval (float): Number of seconds to wait between condition
polls. This may be a decimal fraction. The default poll interval
depends on the timeout used, but varies between .1 and 1 seconds.
xfail (boolean): If True, a timeout is expected.
fatal (boolean): If True, the test will fail if condition is not met
(or if condition is met with xfail).
Returns:
True if the condition is met by the timeout, False otherwise.
Example:
>>> repeat_until(lambda : random.randint(1,10), lambda n: n == 5, timeout=30, poll_interval=1)
True
"""
begin_time = time.time()
if poll_interval is None:
if timeout <= 5:
poll_interval = 0.1
elif timeout <= 10:
poll_interval = 0.2
else:
poll_interval = 1
logging.debug(
f"Waiting until condition related to {condition.__code__.co_varnames} is met, or timeout after {timeout}s..."
)
condition_met = False
while time.time() < begin_time + timeout:
value = callable()
if condition(value):
logging.debug(f"Condition met, value: {value}")
condition_met = True
break
logging.debug(f"Condition not yet met, value: {value}")
time.sleep(poll_interval)
if not xfail and not condition_met:
if fatal:
pytest.fail(f"Condition was not met within the {timeout} second timeout")
else:
logging.warning(
f"Condition was not met within the {timeout} second timeout"
)
elif xfail and condition_met:
if fatal:
pytest.fail(
f"Condition was met within the {timeout} second timeout and wasn't expected"
)
else:
logging.warning(
f"Condition was met within the {timeout} second timeout and wasn't expected"
)
return condition_met
def repeat_command_until(command, condition, quiet=True, **repeat_until_kwargs):
"""Repeats a command until a condition is met or it times out.
This function accepts the same arguments as repeat_until.
Args:
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
True if the condition is met by the timeout, False otherwise.
Example:
>>> repeat_command_until("scontrol ping", lambda results: re.search(r'is UP', results['stdout']))
True
"""
return repeat_until(
lambda: run_command(command, quiet=quiet),
condition,
**repeat_until_kwargs,
)
def pids_from_exe(executable):
"""Finds process IDs (PIDs) of running processes with given executable name.
Args:
executable (string): The name of the executable for which to find running processes.
Returns:
A list of integer process IDs (PIDs) for running processes that match the given
executable name.
Example:
>>> pids_from_exe('/usr/bin/python3')
[12345, 67890]
>>> pids_from_exe('/usr/sbin/sshd')
[54321]
>>> pids_from_exe('/bin/non-existent')
[]
"""
# Avoid issues with links:
executable = os.path.realpath(executable)
# We have to elevate privileges here, but forking off thousands of sudo
# commands is expensive, so we will sudo a dynamic bash script for speed
script = f"""cd /proc
for pid in `ls -d1 [0-9]*`;
do if [[ "$(readlink $pid/exe)" = "{executable}" ]];
then echo $pid;
fi
done"""
pids = []
output = run_command_output(script, user="root", quiet=True)
for line in output.rstrip().splitlines():
pids.append(int(line))
return pids
def is_slurmrestd_running():
"""Checks if slurmrestd is running.
Needs to be run after the related properties are set.
"""
def safe_request_openapi():
try:
return request_slurmrestd("openapi/v3")
except Exception as err:
logging.warn(err)
return None
# TODO: We could check also if the required plugins/parsers in properties
# are in the returned specs, but the format still depends on the version.
# Once v0.0.39 is removed, we could add the extra check.
return repeat_until(safe_request_openapi, lambda r: r and r.status_code == 200)
def is_slurmctld_running(quiet=False):
"""Checks whether slurmctld is running.
Args:
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
True if the slurmctld is running, False otherwise.
Example:
>>> is_slurmctld_running()
True
>>> is_slurmctld_running(quiet=True)
True
"""
# Check whether slurmctld is running
if re.search(r"is UP", run_command_output("scontrol ping", quiet=quiet)):
return True
return False
def gcore(component, pid=None, sbin=True, core_name=None, sigprof=True):
"""Generates a gcore file for all pids running of a given Slurm component.
The gcore file will be save at slurm-logs-dir, where all the logs and kernel
coredumps should be generated too.
Args:
component (string): The name of the component. E.g. slurmctld, slurmdbd...
pid (integer): The specific PID of the process to gcore. All PIDs of component are gcored by default...
sbin: If True search for pids related to slurm-sbin-dir, or slurm-bin-dir otherwise.
core_name (string): Name prefix of the core files, component by default.
They are named <core_name>.core.<pid>.
sigprof (boolean): If True, also send SIGPROF after the gcore. Use
False when the process should keep running unaltered.
Returns:
None
"""
# Ensure that slurm-logs-dir is set.
if "slurm-logs-dir" not in properties:
properties["slurm-logs-dir"] = os.path.dirname(
get_config_parameter("SlurmctldLogFile", live=False, quiet=True)
)
if sbin:
prefix = properties["slurm-sbin-dir"]
else:
prefix = properties["slurm-bin-dir"]
pids = pids_from_exe(f"{prefix}/{component}")
if pid:
if pid not in pids:
logging.warning(
f"Requested PID {pid} is not in the obtained PIDs ({pids}), but using it anyway"
)
pids = [pid]
if not pids:
logging.warning(f"Process {prefix}/{component} not found")
if core_name is None:
core_name = component
logging.debug(f"Getting gcores (sigprof={sigprof}) of PIDs: {pids}")
for pid in pids:
run_command(
f"sudo gcore -o {properties['slurm-logs-dir']}/{core_name}.core {pid}"
)
# Sending SIGPROF should be the last thing, because if it's not handled
# properly it defaults to behave like SIGTERM.
if sigprof:
run_command(f"kill -SIGPROF {pid}", user="root")
def slurmd_pids(slurmd_name):
"""Returns the PIDs of the slurmds running for slurmd_name.
The slurmd's own SlurmdPidFile is used when it contains %n, so it is
per-node. This also finds a slurmd -F by the node name slurmctld assigned
to it, which is what slurmd expands %n with. The pid in it is checked to
still be a slurmd for slurmd_name, as a slurmd that was killed leaves a
stale pidfile whose pid may have been reused.
Otherwise, or without a slurmd_name, the processes of this installation's
slurmd binary are searched for an exact "-N slurmd_name" (no -N at all for
None), so node1 does not match node10, whatever the order of the slurmd
options.
Args:
slurmd_name (string or None): The node name, or None for a slurmd
started without -N whose node name is unknown (slurmd -F).
Returns:
A list of integer PIDs, empty if no such slurmd is running.
Example:
>>> slurmd_pids("node1")
[12345]
>>> slurmd_pids(None) # slurmd -F
[]
"""
def _args(pid):
try:
with open(f"/proc/{pid}/cmdline", "rb") as cmdline:
return cmdline.read().decode().split("\0")
except OSError:
# Already gone
return None
def _node_option(args):
return next((value for arg, value in zip(args, args[1:]) if arg == "-N"), None)
# get_config_parameter() casefolds values, and this is a path
pidfile = next(
(
str(value)
for key, value in get_config(live=False, quiet=True).items()
if key.casefold() == "slurmdpidfile"
),
None,
)
if slurmd_name and pidfile:
if "%n" in pidfile:
pidfile = pidfile.replace("%n", slurmd_name).replace(
"%h", socket.gethostname().split(".")[0]
)
pid = run_command_output(f"cat {pidfile}", user="root", quiet=True).strip()
if not pid.isdigit():
return []
args = _args(int(pid))
if (
not args
or os.path.basename(args[0]) != "slurmd"
or _node_option(args) not in (None, slurmd_name)
):
return []
return [int(pid)]
# If slurmd_name is not specified (-F)
pids = []
for pid in pids_from_exe(f"{properties['slurm-sbin-dir']}/slurmd"):
args = _args(pid)
if args is not None and _node_option(args) == slurmd_name:
pids.append(pid)
return pids
def start_slurmd(
slurmd_name=None, args="", wait="registered", fatal=True, xfail=False, quiet=False
):
"""Starts a slurmd, records it and waits for it to be usable.
Tests should start slurmds with this function instead of running slurmd
directly, so that what any slurmd needs is always in place (e.g. its TLS
node token with the s2n variant, or its spool directory).
Any slurmd already running for slurmd_name is stopped first, so this also
restarts a slurmd.
This function may only be used in auto-config mode.
Args:
slurmd_name (string): The node name, passed to slurmd as -N. With -F in
args, slurmd gets its node name from slurmctld (one of the FUTURE
nodes) and refuses -N, so slurmd_name is not passed and is only the
node expected to be assigned, used to find, wait for and clean up
the slurmd. It can only be None with -F, when that node is not
known.
args (string): Other slurmd options, e.g. "-Z -b --conf 'Feature=f1'".
wait (string): What to wait for after slurmd starts:
"registered": slurmctld has a registration from this slurmd (its
slurmd_start_time changed). This is the default. Without a
slurmd_name it is the same as "process".
"idle": registered and the node is IDLE.
"process": the slurmd process is running.
None: nothing, e.g. when slurmd is expected to die.
fatal (boolean): If True, fail the test if slurmd fails to start or
the wait times out. If False, a failure to start is returned and a
wait timeout is logged as a warning.
xfail (boolean): If True, slurmd is expected to fail to start.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
The run_command() results of starting slurmd.
Example:
>>> start_slurmd("node1")
>>> start_slurmd("dyn1", "-Z -b --conf 'Feature=f1'", wait="idle")
>>> start_slurmd("future1", "-F")
>>> start_slurmd(args="-F", wait="process")
>>> start_slurmd("node1", "--bad-option", fatal=False, xfail=True)
"""
def _slurmd_start_time():
"""The slurmd_start_time slurmctld has for slurmd_name, or None."""
node = get_nodes(quiet=True).get(slurmd_name)
if not node:
return None
start_time = node.get("slurmd_start_time")
if isinstance(start_time, dict):
start_time = start_time.get("number")
return start_time or None
# slurmd -F has no NodeName, so it derives one from its own hostname
# (slurmd.c) and %n/%h in SlurmdSpoolDir expand to that
dynamic_future = "-F" in args.split()
if slurmd_name is None and not dynamic_future:
pytest.fail("start_slurmd() needs a slurmd_name unless args include -F")
slurmd_label = slurmd_name or "-F"
if wait not in ("registered", "idle", "process", None):
pytest.fail(f"Invalid start_slurmd() wait value: {wait}")
if slurmd_name is None and wait in ("registered", "idle"):
wait = "process"
if slurmd_pids(slurmd_name):
logging.warning(f"slurmd for {slurmd_label} already running, stopping it")
if slurmd_name:
stop_slurmd(slurmd_name, quiet=quiet)
else:
# stop_slurmd(None) would stop every slurmd, not just this -F one
_stop_pids(
lambda: slurmd_pids(None),
"slurmd -F",
"TERM",
default_polling_timeout,
True,
quiet,
)
# Create the slurmd's spool and tmpfs directories, and remove at teardown
spool_name = slurmd_name or socket.gethostname().split(".")[0]
created_dirs = properties.setdefault("slurmd-dirs-created", [])
for dir_property in ("slurm-spool-dir", "slurm-tmpfs"):
if not properties.get(dir_property):
continue
path = properties[dir_property].replace("%n", spool_name)
run_command(f"mkdir -p {path}", user="root", fatal=True, quiet=True)
created_dirs.append(path)
# With TLS, a slurmd only starts if it can get its certificate
if slurmd_name:
_require_node_tls_credentials([slurmd_name], quiet=quiet)
# slurmd_start_time has one-second resolution, so a registration in the
# same second as the previous one would look like no registration at all.
# Start in a later second.
previous_start_time = None
if wait in ("registered", "idle"):
previous_start_time = _slurmd_start_time()
if previous_start_time:
for _ in timer(timeout=3, poll_interval=0.05, fatal=True, quiet=True):
if int(time.time()) > previous_start_time:
break
command = f"{properties['slurm-sbin-dir']}/slurmd"
if slurmd_name and not dynamic_future:
command += f" -N {slurmd_name}"
if args:
command += f" {args}"
logging.debug(f"Starting slurmd for {slurmd_label}...")
start_time = int(time.time())
results = run_command(command, user="root", fatal=fatal, xfail=xfail, quiet=quiet)
if results["exit_code"] != 0:
return results
properties.setdefault("slurmds-started", set()).add(slurmd_name)
def _wait_failed(message):
if fatal:
pytest.fail(message)
logging.warning(message)
# Poll often: the default interval (a tenth of the timeout) would add
# seconds to every start, as the first check usually comes too early
# (slurmd writes its pidfile, and registers, a bit after it daemonizes).
if wait == "process":
for _ in timer(poll_interval=0.2, quiet=quiet):
if slurmd_pids(slurmd_name):
break
else:
_wait_failed(f"slurmd for {slurmd_label} is not running")
elif wait in ("registered", "idle"):
for _ in timer(poll_interval=0.5, quiet=quiet):
registered = _slurmd_start_time()
if (
registered
and registered >= start_time
and registered != previous_start_time
):
break
else:
_wait_failed(f"slurmd for {slurmd_name} never registered with slurmctld")
return results
if wait == "idle" and not wait_for_node_state(slurmd_name, "IDLE"):
_wait_failed(f"Node {slurmd_name} never became IDLE")
return results
def _stop_pids(get_pids, label, signal, timeout, kill_on_timeout, quiet):
"""Signals the processes get_pids() returns and waits for them to exit.
See stop_slurmd() for the arguments and the return value.
"""
pids = get_pids()
if not pids:
return True
logging.debug(f"Sending SIG{signal} to {label} ({pids})...")
run_command(
f"kill -{signal} {' '.join(str(pid) for pid in pids)}",
user="root",
quiet=quiet,
)
if not timeout:
return True
if repeat_until(
get_pids, lambda pids: not pids, timeout=timeout, poll_interval=0.2
):
return True
if not kill_on_timeout or signal == "KILL":
logging.warning(f"{label} still running {timeout}s after SIG{signal}")
return False
logging.warning(f"{label} ignored SIG{signal} for {timeout}s, killing it")
pids = get_pids()
if pids:
run_command(
f"kill -KILL {' '.join(str(pid) for pid in pids)}",
user="root",
quiet=quiet,
)
return repeat_until(get_pids, lambda pids: not pids, poll_interval=0.2)
def stop_slurmd(
slurmd_name=None,
signal="TERM",
timeout=default_polling_timeout,
kill_on_timeout=True,
quiet=False,
):
"""Stops slurmds, or emulates a slurmd crash.
Works for any slurmd, also those started by Slurm itself (e.g. by a
ResumeProgram) rather than by start_slurmd().
Use slurmd_pids() instead to check whether a slurmd is running, as this
always signals it.
Args:
slurmd_name (string): The node to stop the slurmd of. If None, every
slurmd of this installation that is running.
signal (string): The signal to send. "TERM" (the default) asks
slurmd to shut down, "KILL" emulates a crash.
timeout (integer): Seconds to wait for the slurmds to exit. With 0,
only send the signal, e.g. to let the caller report a slurmd that
does not shut down instead of hiding it.
kill_on_timeout (boolean): If True, send SIGKILL to any slurmd still
running after the timeout, and wait for it again. If False, return
False instead, e.g. to check that slurmd shuts down in time.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
True if the slurmds exited (or timeout is 0), False otherwise.
Example:
>>> stop_slurmd("node1")
True
>>> stop_slurmd() # Every running slurmd
True
>>> stop_slurmd("node1", signal="KILL") # Emulate a crash
True
>>> stop_slurmd(timeout=30, kill_on_timeout=False) # Shuts down in 30s?
False
"""
started = properties.setdefault("slurmds-started", set())
if slurmd_name is None:
started.clear()
slurmd_exe = f"{properties['slurm-sbin-dir']}/slurmd"
return _stop_pids(
lambda: pids_from_exe(slurmd_exe),
"all slurmds",
signal,
timeout,
kill_on_timeout,
quiet,
)
started.discard(slurmd_name)
return _stop_pids(
lambda: slurmd_pids(slurmd_name),
f"slurmd for {slurmd_name}",
signal,
timeout,
kill_on_timeout,
quiet,
)
def start_dynamic_slurmd(
node_name,
extra_conf="",
wait_for_usable=True,
timeout=default_polling_timeout,
quiet=False,
):
"""Start a dynamic-normal slurmd (``slurmd -Z``) for a node.
Launches a dynamic-normal slurmd through start_slurmd() that self-registers
with the controller, optionally passing per-node configuration via slurmd's
``--conf`` option (e.g. ``"Gres=gpu:a100:2 MaxDrainedTres=gres/gpu:1"``).
Any slurmd already running for node_name is stopped first and the node
deleted from the controller, so the node is cleanly (re)registered. Pair
with stop_slurmd() and delete_node() to tear the node down.
This function may only be used in auto-config mode.
Args:
node_name (string): The name the dynamic node registers under.
extra_conf (string): Value passed to slurmd's ``--conf`` option (node
configuration such as ``Gres=`` or ``MaxDrainedTres=``). An empty
string omits ``--conf``.
wait_for_usable (boolean): If True, block until the node registers and
reaches a usable state (IDLE or DEGRADED, and not DRAIN). A node
that registers DRAINED (e.g. GRES missing beyond MaxDrainedTres)
will not satisfy this; pass False and inspect the state directly
for those cases.
timeout (integer): Seconds to wait for the node to become usable.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
The node name, for convenient assignment.
Example:
>>> node = start_dynamic_slurmd("node100", extra_conf="Gres=gpu:a100:2")
>>> start_dynamic_slurmd("node101", wait_for_usable=False)
"""
if not properties["auto-config"]:
require_auto_config("wants to start a dynamic slurmd")
# Only drop the controller's node record when a daemon was actually
# running: that is a restart, and the stale record would make the
# re-register look like an unexpected node reboot. If nothing was running
# the caller stopped the slurmd deliberately and kept the record, so the
# node state -- including any drained TRES -- must survive for the new
# daemon to re-register against.
if slurmd_pids(node_name):
stop_slurmd(node_name, timeout=timeout, quiet=quiet)
delete_node(node_name, quiet=quiet)
args = "-Z -v -b"
if extra_conf:
args += f" --conf {shlex.quote(extra_conf)}"
start_slurmd(node_name, args=args, wait="registered", quiet=quiet)
if wait_for_usable:
state = ""
for _ in timer(timeout=timeout, fatal=True):
state = (get_nodes(quiet=True).get(node_name) or {}).get("state") or ""
if ("IDLE" in state or "DEGRADED" in state) and "DRAIN" not in state:
break
assert (
"IDLE" in state or "DEGRADED" in state
), f"{node_name} registered as {state!r}, not IDLE or DEGRADED"
return node_name
def delete_node(node_name, quiet=False):
"""Delete a node from the controller with ``scontrol delete NodeName=``.
Not fatal: the record may already be gone, and this is also used from
fixture teardown, where aborting would mask the real failure. Use it after
stop_slurmd() so a later slurmd start for a dynamic node is a fresh
registration rather than an unexpected node reboot.
Args:
node_name (string): The node to delete.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
The run_command() results of the scontrol delete.
Example:
>>> delete_node("node100")
"""
return run_command(
f"scontrol delete NodeName={node_name}",
user=properties["slurm-user"],
fatal=False,
quiet=quiet,
)
def start_slurmctld(clean=False, quiet=False, also_slurmds=False, full_recovery=False):
"""Starts the Slurm controller daemon (slurmctld).
This function may only be used in auto-config mode.
Args:
clean (boolean): If True, clears previous slurmctld state.
quiet (boolean): If True, logging is performed at the TRACE log level.
also_slurmds (boolean): If True, also start all required slurmds.
full_recovery (boolean): If True, start slurmctld with -R to recover
all state (mutually exclusive with clean, which ignores state).
Returns:
None
Example:
>>> start_slurmctld() # Start slurmctld with default settings
>>> start_slurmctld(clean=True, quiet=True) # Start slurmctld with clean state and quiet logging
>>> start_slurmctld(full_recovery=True) # Start slurmctld with -R
"""
if clean and full_recovery:
pytest.fail("clean and full_recovery are mutually exclusive")
if not properties["auto-config"]:
require_auto_config("wants to start slurmctld")
# First start sackd if necessary, so clients can be run.
if _is_sackd_required(quiet=quiet):
start_sackd(quiet=quiet)
logging.debug("Starting slurmctld...")
if is_slurmctld_running(quiet=quiet):
logging.warning("Slurmctld was already started")
stop_slurmctld(quiet, also_slurmds)
# With TLS, every configured node must be able to get its certificate.
# This covers the slurmds that tests cannot provision in start_slurmd():
# those started by Slurm itself (e.g. a ResumeProgram for CLOUD nodes) and
# "slurmd -F", whose node name slurmctld assigns from the FUTURE nodes.
# All the nodes in slurm.conf, of any State
node_names = [
name for name in get_nodes(live=False, quiet=True) if name != "DEFAULT"
]
_require_node_tls_credentials(node_names, quiet=quiet)
# Start slurmctld
command = f"{properties['slurm-sbin-dir']}/slurmctld"
if clean:
command += " -c -i"
elif full_recovery:
command += " -R"
results = run_command(command, user=properties["slurm-user"], quiet=quiet)
if results["exit_code"] != 0:
pytest.fail(
f"Unable to start slurmctld (rc={results['exit_code']}): {results['stderr']}"
)
# Verify that slurmctld is running
if not repeat_command_until(
"scontrol ping", lambda results: re.search(r"is UP", results["stdout"])
):
logging.warning(
"scontrol ping is not responding, trying to get slurmctld core file..."
)
gcore("slurmctld")
pytest.fail("Slurmctld is not running")
else:
logging.debug("Slurmctld started successfully")
if also_slurmds:
slurmd_list = node_names
if not slurmd_list:
pytest.fail("Unable to determine the slurmd node names")
# (Multi)Slurmds, waiting below for all of them at once
for slurmd_name in slurmd_list:
start_slurmd(slurmd_name, wait="process", quiet=quiet)
# Verify that the slurmd is registered correctly
timeout = 30 + 5 * len(slurmd_list)
if not repeat_until(
lambda: get_nodes(quiet=True),
lambda nodes: all(
nodes[name]["state"][0] == "IDLE" for name in slurmd_list
),
timeout=timeout,
):
nodes = get_nodes(quiet=True)
non_idle = [
name for name in slurmd_list if nodes[name]["state"][0] != "IDLE"
]
logging.warning(
f"Getting the core files of the still not IDLE slurmds ({non_idle})"
)
for node in non_idle:
pid = run_command_output(
f"pgrep -f 'slurmd -N {slurmd_name}'", quiet=quiet
).strip()
gcore("slurmd", pid)
pytest.fail(f"Some nodes are not IDLE: {non_idle}")
logging.debug(f"All nodes are IDLE: {slurmd_list}")
def start_slurmdbd(clean=False, quiet=False):
"""Starts the Slurm DB daemon (slurmdbd).
This function may only be used in auto-config mode.
Args:
clean (boolean): If True, clears previous slurmdbd state.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
None
"""
if not properties["auto-config"]:
require_auto_config("wants to start slurmdbd")
# First start sackd if necessary, so clients can be run.
if _is_sackd_required(quiet=quiet):
start_sackd(quiet=quiet)
# Save and drop the original DB only once per module: a later call would
# save the DB already created or modified by the module.
if properties["accounting"] != "db-saved":
dump_accounting_database(properties["sql-db-backup"], drop=True)
properties["accounting"] = "db-saved"
logging.debug("Starting slurmdbd...")
if (
run_command_exit(
"sacctmgr show cluster", user=properties["slurm-user"], quiet=quiet
)
== 0
):
logging.warning("slurmdbd was already running")
else:
# Start slurmdbd
results = run_command(
f"{properties['slurm-sbin-dir']}/slurmdbd",
user=properties["slurm-user"],
quiet=quiet,
)
if results["exit_code"] != 0:
pytest.fail(
f"Unable to start slurmdbd (rc={results['exit_code']}): {results['stderr']}"
)
# Verify that slurmdbd is running
if not repeat_command_until(
"sacctmgr show cluster", lambda results: results["exit_code"] == 0
):
logging.warning(
"sacctmgr show cluster is not responding, trying to get slurmdbd core file..."
)
gcore("slurmdbd")
pytest.fail("Slurmdbd is not running")
else:
logging.debug("Slurmdbd started successfully")
def clean_slurm_conf_nodes(quiet=False):
"""Cleans the Slurm configuration file from unnecessary default node0.
This function may only be used in auto-config mode.
Args:
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
None
"""
if not properties["auto-config"]:
require_auto_config("wants to start slurmds")
# Remove unnecessary default node0 from config to avoid being used or reserved
output = run_command_output(
f"cat {properties['slurm-config-dir']}/slurm.conf",
user=properties["slurm-user"],
quiet=quiet,
)
if len(re.findall(r"NodeName=", output)) > 1:
run_command(
f"sed -i '/NodeName=node0 /d' {properties['slurm-config-dir']}/slurm.conf",
user=properties["slurm-user"],
quiet=quiet,
)
def _is_sackd_required(quiet=False):
"""True when slurm.conf sets AuthInfo=disable_sack.
In that mode slurmd no longer binds /run/slurm/sack.socket (so
multi-slurmd on the same host can coexist), and a standalone sackd
is required to provide that socket for client commands.
"""
info = get_config_parameter("AuthInfo", live=False, default="", quiet=quiet) or ""
return "disable_sack" in info
def start_sackd(quiet=False):
"""Starts the SACK daemon (sackd).
sackd has no log-file option, so we launch it under Popen in foreground
(`-D -vvv`) and redirect both streams to <slurm-logs-dir>/sackd.log.
"""
if not properties["auto-config"]:
require_auto_config("wants to start sackd")
sackd_exe = f"{properties['slurm-sbin-dir']}/sackd"
if pids_from_exe(sackd_exe):
logging.warning("Sackd was already running, stopping it first.")
stop_sackd(quiet)
if "slurm-logs-dir" not in properties:
properties["slurm-logs-dir"] = os.path.dirname(
get_config_parameter("SlurmctldLogFile", live=False, quiet=quiet)
)
sackd_log = f"{properties['slurm-logs-dir']}/sackd.log"
logging.debug(f"Starting sackd; logs at {sackd_log}")
properties["sackd_log"] = open(sackd_log, "w")
cmd = [
"sudo",
"-u",
properties["slurm-user"],
sackd_exe,
"-D",
"-vvv",
]
properties["sackd"] = subprocess.Popen(
cmd,
stdin=subprocess.DEVNULL,
stdout=properties["sackd_log"],
stderr=properties["sackd_log"],
start_new_session=True,
)
# Wait for either the SACK socket to appear
for t in timer():
if properties["sackd"].poll() is not None:
rc = properties["sackd"].returncode
properties["sackd_log"].flush()
properties["sackd_log"].close()
properties.pop("sackd_log", None)
properties.pop("sackd", None)
pytest.fail(f"sackd exited (rc={rc}) before binding the socket.")
if os.path.exists("/run/slurm/sack.socket"):
logging.debug("Sackd started successfully")
return
else:
pytest.fail("sackd never created /run/slurm/sack.socket")
def stop_sackd(quiet=False):
"""Stops the SACK daemon (sackd) if running."""
if not properties["auto-config"]:
require_auto_config("wants to stop sackd")
failures = []
proc = properties.pop("sackd", None)
if proc is None:
logging.debug("sackd was not running...")
else:
logging.debug("Stopping sackd...")
proc.send_signal(signal.SIGINT)
try:
proc.wait(timeout=default_polling_timeout)
logging.debug("No sackd is running.")
except subprocess.TimeoutExpired:
logging.warning("sackd didn't exit on SIGINT; sending SIGKILL")
proc.kill()
try:
proc.wait(timeout=default_polling_timeout)
logging.debug("No sackd is running.")
except Exception:
msg = "sackd didn't exit on SIGKILL"
failures.append(msg)
logging.warning(msg)
log_fh = properties.pop("sackd_log", None)
if log_fh is not None:
try:
log_fh.close()
except Exception:
msg = "Unable to close sackd logs"
failures.append(msg)
logging.warning(msg)
if failures:
return failures
return None
def start_slurm(clean=False, quiet=False):
"""Starts all applicable Slurm daemons.
This function examines the Slurm configuration files to determine which daemons
need to be started.
This function may only be used in auto-config mode.
Args:
clean (boolean): If True, clears previous slurmctld state.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
None
Example:
>>> start_slurm() # Start all Slurm daemons with default settings
>>> start_slurm(clean=True, quiet=True) # Start all Slurm daemons with clean state and quiet logging
"""
if not properties["auto-config"]:
require_auto_config("wants to start slurm")
# First start sackd if necessary, so clients can be run.
if _is_sackd_required(quiet=quiet):
start_sackd(quiet=quiet)
# Determine whether slurmdbd should be included
if properties["accounting"]:
start_slurmdbd(clean, quiet)
clean_slurm_conf_nodes(quiet)
# Start slurmctld
start_slurmctld(clean, quiet, also_slurmds=True)
# Start slurmrestd if required
if properties["slurmrestd-started"]:
start_slurmrestd()
def stop_slurmctld(quiet=False, also_slurmds=False, signal=None):
"""Stops the Slurm controller daemon (slurmctld), or emulates its crash.
This function may only be used in auto-config mode.
Args:
also_slurmds (boolean): If True, also stop all slurmds.
quiet (boolean): If True, logging is performed at the TRACE log level.
signal (string): If None (the default), slurmctld is shut down with
scontrol shutdown. Otherwise it is sent this signal instead, e.g.
"KILL" to emulate a crash. The slurmds, with also_slurmds, are
still shut down normally.
Returns:
None, or the list of failures if anything could not be stopped.
Example:
>>> stop_slurmctld() # Stop slurmctld with default logging
>>> stop_slurmctld(quiet=True) # Stop slurmctld with quiet logging
>>> stop_slurmctld(signal="KILL") # Emulate a slurmctld crash
"""
rc = None
failures = []
if not properties["auto-config"]:
require_auto_config("wants to stop slurmctld")
# Stop slurmctld
if signal:
logging.debug(f"Sending SIG{signal} to slurmctld...")
_stop_pids(
lambda: pids_from_exe(f"{properties['slurm-sbin-dir']}/slurmctld"),
"slurmctld",
signal,
0,
False,
quiet,
)
else:
command = "scontrol shutdown"
if not also_slurmds:
command += " slurmctld"
logging.debug("Stopping slurmctld...")
else:
logging.debug("Stopping slurmctld and slurmds...")
results = run_command(command, user=properties["slurm-user"], quiet=quiet)
if results["exit_code"] != 0:
failures.append(f"Command {command} failed: {results}")
# Verify that slurmctld is not running
gcore_timeout = 60
for t in timer(timeout=gcore_timeout, poll_interval=1):
if not pids_from_exe(f"{properties['slurm-sbin-dir']}/slurmctld"):
logging.debug("No slurmctld is running.")
break
else:
logging.warning(
f"Slurmctld is still stopping after {gcore_timeout}s, getting an early core file"
)
gcore("slurmctld", core_name="slurmctld.early_gcore", sigprof=False)
for t in timer(timeout=gcore_timeout, poll_interval=1):
if not pids_from_exe(f"{properties['slurm-sbin-dir']}/slurmctld"):
logging.debug("No slurmctld is running.")
break
else:
failures.append("Slurmctld is still running")
logging.warning("Getting the core files of the still running slurmctld")
gcore("slurmctld")
if also_slurmds:
if get_version("sbin/slurmd") < (24, 11):
# FIXED: t20764.
slurmd_list = [
name for name in get_nodes(live=False, quiet=True) if name != "DEFAULT"
]
if not slurmd_list:
failures.append("Unable to determine the slurmd node names")
for slurmd_name in slurmd_list:
run_command(
f"sudo systemctl stop {slurmd_name}_slurmstepd.scope", quiet=quiet
)
if signal:
# slurmctld did not shut down any slurmd
stop_slurmd(timeout=0, kill_on_timeout=False, quiet=quiet)
else:
# Only the recorded slurmds scontrol shutdown may not reach, e.g.
# those of deleted dynamic nodes or of an unregistered slurmd -F,
# not the ones already shutting down.
for slurmd_name in list(properties.get("slurmds-started", ())):
if slurmd_name:
stop_slurmd(
slurmd_name, timeout=0, kill_on_timeout=False, quiet=quiet
)
else:
_stop_pids(
lambda: slurmd_pids(None), "slurmd -F", "TERM", 0, False, quiet
)
properties["slurmds-started"] = set()
# Verify that slurmds are not running
if not repeat_until(
lambda: pids_from_exe(f"{properties['slurm-sbin-dir']}/slurmd"),
lambda pids: len(pids) == 0,
timeout=default_polling_timeout * 2,
):
logging.warning("Some slurmd was not stopped.")
logging.warning(
"Sometimes slurm may be just rebooting or similar when shutdown was run"
)
logging.debug("Shutting down slurmds manually")
for pid in pids_from_exe(f"{properties['slurm-sbin-dir']}/slurmd"):
run_command(f"kill {pid}", user="root")
if not repeat_until(
lambda: pids_from_exe(f"{properties['slurm-sbin-dir']}/slurmd"),
lambda pids: len(pids) == 0,
timeout=default_polling_timeout * 2,
):
failures.append("Some slurmds are still running")
logging.warning("Getting the core files of the still running slurmds")
gcore("slurmd")
else:
logging.debug("No slurmd is running.")
else:
logging.debug("No slurmd is running.")
if failures:
rc = failures
return rc
def stop_slurmdbd(quiet=False):
"""Stops the Slurm DB daemon (slurmdbd).
This function may only be used in auto-config mode.
Args:
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
None
"""
rc = None
failures = []
if not properties["auto-config"]:
require_auto_config("wants to stop slurmdbd")
logging.debug("Stopping slurmdbd...")
# Stop slurmdbd
# Some tests may fill up the DB and sacctmgr may need some extra time
results = run_command(
"sacctmgr shutdown",
user=properties["slurm-user"],
quiet=quiet,
timeout=default_command_timeout * 5,
)
if results["exit_code"] != 0:
failures.append(
f"Command \"sacctmgr shutdown\" failed with rc={results['exit_code']}"
)
# Verify that slurmdbd is not running (we might have to wait for rollups to complete)
if not repeat_until(
lambda: pids_from_exe(f"{properties['slurm-sbin-dir']}/slurmdbd"),
lambda pids: len(pids) == 0,
timeout=default_polling_timeout * 2,
):
failures.append("Slurmdbd is still running")
logging.warning("Getting the core files of the still running slurmdbd")
gcore("slurmdbd")
else:
logging.debug("No slurmdbd is running.")
if failures:
rc = failures
return rc
def stop_slurm(fatal=True, quiet=False):
"""Stops all applicable Slurm daemons.
This function examines the Slurm configuration files to determine which daemons
need to be stopped.
This function may only be used in auto-config mode.
Args:
fatal (boolean): If True, a failure to stop all daemons will result in the
test failing.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
True if all Slurm daemons were stopped, False otherwise.
Example:
>>> stop_slurm() # Stop all Slurm daemons with default settings
True
>>> stop_slurm(fatal=False, quiet=True) # Stop all Slurm daemons with non-fatal failures and quiet logging
False
"""
failures = []
if not properties["auto-config"]:
require_auto_config("wants to stop slurm")
# Determine whether slurmdbd should be included
if properties["accounting"]:
err = stop_slurmdbd(quiet)
if err:
failures.extend(err)
# Stop slurmctld and slurmds
err = stop_slurmctld(quiet=quiet, also_slurmds=True)
if err:
failures.extend(err)
# Stop slurmrestd if was started
if properties["slurmrestd-started"]:
properties["slurmrestd"].send_signal(signal.SIGINT)
try:
properties["slurmrestd"].wait(timeout=60)
except Exception:
properties["slurmrestd"].kill()
properties["slurmrestd_log"].close()
# Stop sackd if it was needed for this configuration.
if _is_sackd_required(quiet=quiet):
err = stop_sackd(quiet=quiet)
if err:
failures.extend(err)
if failures:
for fail in failures:
logging.warning(fail)
if fatal:
pytest.fail(failures[0])
return False
else:
return True
def restart_slurmctld(clean=False, quiet=False, full_recovery=False):
"""Restarts the Slurm controller daemon (slurmctld).
This function may only be used in auto-config mode.
Args:
clean (boolean): If True, clears previous slurmctld state.
quiet (boolean): If True, logging is performed at the TRACE log level.
full_recovery (boolean): If True, start slurmctld with -R to recover
all state (mutually exclusive with clean, which ignores state).
Returns:
None
Example:
>>> restart_slurmctld() # Restart slurmctld with default settings
>>> restart_slurmctld(clean=True, quiet=True) # Restart slurmctld with clean state and quiet logging
>>> restart_slurmctld(full_recovery=True) # Restart slurmctld with -R
"""
if clean and full_recovery:
pytest.fail("clean and full_recovery are mutually exclusive")
stop_slurmctld(quiet=quiet)
start_slurmctld(clean=clean, quiet=quiet, full_recovery=full_recovery)
def restart_slurm(clean=False, quiet=False):
"""Restarts all applicable Slurm daemons.
This function may only be used in auto-config mode.
Args:
clean (boolean): If True, clears previous slurmctld state.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
None
Example:
>>> restart_slurm() # Restart all Slurm daemons with default settings
>>> restart_slurm(clean=True, quiet=True) # Restart all Slurm daemons with clean state and quiet logging
"""
stop_slurm(quiet=quiet)
start_slurm(clean=clean, quiet=quiet)
def require_slurm_running():
"""Ensures that the Slurm daemons are running.
In local-config mode, the test is skipped if Slurm is not running.
In auto-config mode, Slurm is started if necessary.
In order to avoid multiple restarts of Slurm (in auto-config), this function
should be called at the end of the setup preconditions.
Args:
None
Returns:
None
Example:
>>> require_slurm_running() # Ensure Slurm is running or start it in auto-config mode
"""
global nodes
if properties["auto-config"]:
if not is_slurmctld_running(quiet=True):
# Check and report the mixed components
versions = dict()
versions["slurmdbd"] = get_version("sbin/slurmdbd")
versions["slurmctld"] = get_version("sbin/slurmctld")
versions["slurmd"] = get_version("sbin/slurmd")
versions["scontrol"] = get_version("bin/scontrol")
if len(set(versions.values())) == 1:
logging.info(
f"Starting Slurm with all components in the same version: {versions['slurmctld']}"
)
else:
logging.info(f"Starting Slurm in a mixed version setup: {versions}")
properties["slurm-started"] = True
start_slurm(clean=True, quiet=True)
else:
if not is_slurmctld_running(quiet=True):
pytest.skip(
"This test requires slurm to be running", allow_module_level=True
)
# As a side effect, build up initial nodes dictionary
nodes = get_nodes(quiet=True)
config = get_config()
properties["slurmctld_host"] = config["SlurmctldHost[0]"].split("(")[0]
properties["slurmctld_port"] = config["SlurmctldPort"]
def is_upgrade_setup(
old_slurm_prefix="/opt/slurm-old",
new_slurm_prefix="/opt/slurm-new",
old_build_prefix="",
new_build_prefix="",
old_source_prefix="",
new_source_prefix="",
force_old=False,
):
"""
Return True if we have two Slurms configured in the system.
"""
if not os.path.exists(f"{old_slurm_prefix}/bin"):
logging.debug(f"Old prefix {old_slurm_prefix} not exists.")
return False
if not os.path.exists(f"{new_slurm_prefix}/bin"):
logging.debug(f"New prefix {new_slurm_prefix} not exists.")
return False
# Add the right properties
setup_upgrades(
old_slurm_prefix,
new_slurm_prefix,
old_build_prefix,
new_build_prefix,
old_source_prefix,
new_source_prefix,
force_old,
)
return True
def require_upgrades(
old_slurm_prefix="/opt/slurm-old",
new_slurm_prefix="/opt/slurm-new",
old_build_prefix="",
new_build_prefix="",
old_source_prefix="",
new_source_prefix="",
force_old=True,
):
"""Checks if has two different versions installed.
If they are not, skip.
"""
if not properties["auto-config"]:
require_auto_config("to change/upgrade Slurm setup")
if not is_upgrade_setup(
old_slurm_prefix,
new_slurm_prefix,
old_build_prefix,
new_build_prefix,
old_source_prefix,
new_source_prefix,
force_old,
):
pytest.skip("This test needs an upgrade setup")
# Double-check that old_version <= new_version
old_version = get_version(slurm_prefix=old_slurm_prefix)
new_version = get_version(slurm_prefix=new_slurm_prefix)
if old_version > new_version:
pytest.skip(
f"Old version ({old_version}) has to be older than new version ({new_version})"
)
logging.info(f"Required upgrade setup found: {old_version} and {new_version}")
def setup_upgrades(
old_slurm_prefix="/opt/slurm-old",
new_slurm_prefix="/opt/slurm-new",
old_build_prefix="",
new_build_prefix="",
old_source_prefix="",
new_source_prefix="",
force_old=False,
):
"""
Adds the necessary atf.properties[] with the old/new paths.
If force_old is specified itt also update the links pointing to the old
paths, and they will be restored in the global teardown.
"""
# TODO: We should use slurm-new(-build) instead of slurm-git(-build)
if old_build_prefix == "":
old_build_prefix = properties["slurm-build-dir"]
if new_build_prefix == "":
new_build_prefix = f"{properties['slurm-build-dir']}/../slurm-build-new"
if old_source_prefix == "":
old_source_prefix = properties["slurm-source-dir"]
if new_source_prefix == "":
new_source_prefix = f"{properties['slurm-source-dir']}/../slurm-git"
properties["old-slurm-prefix"] = old_slurm_prefix
properties["new-slurm-prefix"] = new_slurm_prefix
properties["old-build-prefix"] = old_build_prefix
properties["new-build-prefix"] = new_build_prefix
properties["old-source-prefix"] = old_source_prefix
properties["new-source-prefix"] = new_source_prefix
properties["forced_upgrade_setup"] = force_old
if force_old:
logging.debug(
"Setting bin/ and sbin/ pointing to old version and saving a backup..."
)
run_command(
f"sudo mv {properties['slurm-sbin-dir']} {module_tmp_path}/upgrade-sbin",
quiet=True,
fatal=True,
)
run_command(
f"sudo mv {properties['slurm-bin-dir']} {module_tmp_path}/upgrade-bin",
quiet=True,
fatal=True,
)
run_command(
f"sudo mkdir {properties['slurm-sbin-dir']} {properties['slurm-bin-dir']}",
quiet=True,
fatal=True,
)
run_command(
f"sudo ln -s {properties['old-slurm-prefix']}/sbin/* {properties['slurm-sbin-dir']}/",
quiet=True,
fatal=True,
)
run_command(
f"sudo ln -s {properties['old-slurm-prefix']}/bin/* {properties['slurm-bin-dir']}/",
quiet=True,
fatal=True,
)
def upgrade_component(component, new_version=True):
"""Upgrades a component creating the required links, and restarts it if necessary.
This function needs require_upgrades() to be already run to work properly.
Args:
component (string): The bin/ or sbin/ component of Slurm to check.
new_version (boolean): Set it false to downgrade to the older version instead.
Returns:
A tuple representing the version. E.g. (25.05.0).
"""
if (
"old-slurm-prefix" not in properties.keys()
or "new-slurm-prefix" not in properties.keys()
):
pytest.fail("To upgrade_components() first we need to call require_upgrades()")
if not os.path.exists(f"{properties['slurm-prefix']}/{component}"):
pytest.fail(f"Unknown or not existing {component}")
if new_version:
upgrade_prefix = properties["new-slurm-prefix"]
else:
upgrade_prefix = properties["old-slurm-prefix"]
# Stop components when necessary
if component == "sbin/slurmdbd":
stop_slurmdbd()
elif component == "sbin/slurmctld":
stop_slurmctld()
run_command(
f"sudo rm -f {properties['slurm-prefix']}/{component}",
quiet=True,
fatal=True,
)
run_command(
f"sudo ln -s {upgrade_prefix}/{component} {properties['slurm-prefix']}/{component}",
quiet=True,
fatal=True,
)
# Restart components when necessary
if component == "sbin/slurmdbd":
start_slurmdbd()
elif component == "sbin/slurmctld":
start_slurmctld()
def get_slurmd_C():
"""Return a dict with the main values reported by 'slurmd -C'
Numeric fields (CPUs, Boards, SocketsPerBoard, CoresPerSocket,
ThreadsPerCore, RealMemory) are returned as integers. Other fields
(NodeName and Gres) are returned as strings.
"""
fields = [
"NodeName",
"CPUs",
"Boards",
"SocketsPerBoard",
"CoresPerSocket",
"ThreadsPerCore",
"RealMemory",
"Gres",
]
integer_fields = (
"CPUs",
"Boards",
"SocketsPerBoard",
"CoresPerSocket",
"ThreadsPerCore",
"RealMemory",
)
output = run_command_output("slurmd -C", fatal=True)
keys = re.findall(r"(" + "|".join(fields) + r")=(\S+)", output)
slurmd_c = dict(keys)
for field in integer_fields:
if field in slurmd_c:
slurmd_c[field] = int(slurmd_c[field])
return slurmd_c
def get_version(component="sbin/slurmctld", slurm_prefix=""):
"""Returns the version of the Slurm component as a tuple.
It calls the component with -V and converts the output into a tuple.
It skips if the component doesn't exist in the current Slurm version.
Args:
component (string): The bin/ or sbin/ component of Slurm to check.
It also supports "config.h" to obtain the VERSION in the header.
slurm_prefix (string): The path where the component is. By default the defined in testsuite.conf.
If component is "config.h", then it's the build dir.
Returns:
A tuple representing the version. E.g. (25.05.0).
"""
# Components that don't exist in all supported versions, with the version
# they were added in.
added_in = {
"bin/swait": (26, 5),
}
if component in added_in:
require_version(
added_in[component],
component="bin/scontrol",
slurm_prefix=slurm_prefix,
reason=f"{component} was added in {added_in[component]}",
)
# TODO: Ticket 25155 - Remove fatal=False once 25.11 is not supported
fatal = True
if component == "bin/sh5util":
logging.warning(
"Ticket 25155: Exit code of 'sh5util -V' is incorrect for versions older than 26.05"
)
fatal = False
if component == "config.h":
if slurm_prefix == "":
slurm_prefix = properties["slurm-build-dir"]
header = pathlib.Path(f"{slurm_prefix}/config.h")
if not header.exists():
pytest.fail("Unable to access to config.h to get Slurm version")
version_str = re.search(
r'#define\s+VERSION\s+"([^"]+)"', header.read_text()
).group(1)
else:
if slurm_prefix == "":
slurm_prefix = f"{properties['slurm-sbin-dir']}/.."
version_str = (
run_command_output(
f"{slurm_prefix}/{component} -V", quiet=True, user="root", fatal=fatal
)
.strip()
.replace("slurm ", "")
)
return tuple(int(n) for p in version_str.split(".") for n in [p.split("-")[0]])
def get_build_config(name, fatal=False):
"""Returns the value the config.h of the Slurm build defines name to.
Assumes upgrade setups build the old and new Slurm with the same config.h.
Args:
name (string): The config.h macro to look up.
fatal (bool): Fail when config.h is missing. Otherwise, log a warning.
Returns:
The value name is defined to, an empty string if it is defined without
a value, or None if it is not defined or config.h is missing.
Example:
>>> get_build_config("HAVE_NUMA")
'1'
>>> get_build_config("HAVE_NO_SUCH_FEATURE")
None
"""
header = pathlib.Path(f"{properties['slurm-build-dir']}/config.h")
if not header.exists():
msg = f"Unable to access {header} to check for {name}"
if fatal:
pytest.fail(msg)
logging.warning(msg)
return None
for line in header.read_text().splitlines():
if not line.startswith("#"):
continue
# Autoconf indents some directives, e.g. "# define WORDS_BIGENDIAN 1"
words = line[1:].split(maxsplit=2)
if words[:2] == ["define", name]:
return words[2].strip() if len(words) > 2 else ""
return None
def require_build_config(name, value=None, reason=None):
"""Skips unless the config.h of the Slurm build defines name.
Fails if config.h is missing, since that is a misconfigured testsuite
rather than a build without the feature.
Args:
name (string): The config.h macro to require.
value (string or None): If value is a string match the macro exactly, if
value is None allow any defined macro as #ifdef.
reason (string): The reason the macro is required.
Returns:
None
Example:
>>> require_build_config("HAVE_NUMA")
"""
macro_value = get_build_config(name, fatal=True)
equal = value is not None
if (equal and macro_value != value) or (not equal and macro_value is None):
if not reason:
reason = f"This test requires Slurm built with {name}"
if equal:
reason = f"{reason} {value}"
pytest.skip(reason, allow_module_level=True)
# Unique PS1 sentinel used to synchronize on the shell prompt with pexpect.
# The set form uses '\$' (literal backslash + dollar) so the *echoed input*
# of the PS1 assignment does not match the *expect regex* below, which has
# no backslash. Bash interprets '\$' in PS1 as the prompt-escape that renders
# as '$' for a normal user and '#' for root -- hence '[\$\#]' in the regex.
_PEXPECT_PROMPT_SET = "PS1='[TEST_PROMPT]\\$ '"
_PEXPECT_PROMPT_RE = re.compile(r"\[TEST_PROMPT\][\$\#] ")
def run_command_pexpect(cmd):
"""Spawn `cmd` under pexpect with live transcript logging.
Args:
cmd (string): the command line to pass to `pexpect.spawn`.
Returns:
A `pexpect.spawn` child with `logfile_read` set to `sys.stdout`.
Example:
>>> child = atf.run_command_pexpect("salloc")
>>> child.expect("Nodes.*are ready for job") # cmd-specific sync
>>> atf.setup_pexpect_prompt(child) # switch to prompt sync
>>> child.sendline("hostname")
>>> atf.wait_pexpect_prompt(child)
>>> # child.before now contains the output of `hostname`
"""
child = pexpect.spawn(cmd, encoding="utf-8")
child.logfile_read = sys.stdout
return child
def setup_pexpect_prompt(child):
"""Install a unique PS1 sentinel on a pexpect child shell.
Note: `wait_pexpect_prompt` consumes the buffer up to the sentinel
prompt, so any output that arrived before this call -- including
command startup messages -- will only be available via `child.before`
(not via subsequent `expect()` calls). Wait for any cmd-specific
startup output BEFORE calling this.
"""
child.sendline("unset PROMPT_COMMAND")
child.sendline(_PEXPECT_PROMPT_SET)
wait_pexpect_prompt(child)
def wait_pexpect_prompt(child):
"""Wait for the pexpect-managed shell prompt sentinel to appear.
Blocks until the unique PS1 sentinel configured by
`setup_pexpect_prompt()` is seen on the stream. Use this after any
`sendline()` to know the shell has finished the previous command and
is ready for the next one.
"""
child.expect(_PEXPECT_PROMPT_RE)
def require_expect():
"""Setup the expect directory to be used later with run_expect_test()"""
# Create a tmp expect dir from sources with the right permissions
properties["expect_dir"] = f"{module_tmp_path}/expect"
run_command(
f"rsync -a {properties['testsuite_base_dir']}/expect/ {properties['expect_dir']}/",
user=properties["slurm-user"],
fatal=True,
quiet=True,
)
# Create the globals.local
properties["expect_globals_file"] = f"{properties['expect_dir']}/globals.local"
colorize = 1
if "no-color" in properties and properties["no-color"]:
colorize = 0
globals_local_content = f"""\
set testsuite_user {get_user_name()}
set testsuite_colorize {colorize}
set testsuite_cleanup_on_failure false
# These are not necessary because we should use SLURM_TESTSUITE_CONF
# set slurm_dir {properties['slurm-prefix']}
# set build_dir {properties['slurm-build-dir']}
# set src_dir {properties['slurm-source-dir']}
"""
run_command(
f"cat > {properties['expect_globals_file']}",
input=globals_local_content,
user=properties["slurm-user"],
fatal=True,
quiet=True,
)
def run_expect_test(test_num=None, fail=True):
"""Run the expect test corresponding to the test_name.
Args:
test_num (string): The expect test number (e.g. "3.17"). If None, it is
derived from properties["test_name"].
fail (bool): If True (default), call pytest.fail in case of failure.
Returns:
(reason, returncode)
Note: pytest.skip() is always raised when rc > 127 (expect-level skip)
"""
reason = None
if test_num is None:
test_num = (
properties["test_name"]
.replace("test_", "")
.replace("_py", "")
.replace("_", ".")
)
# Most expect tests assume that are run by SlurmUser.
# We need to preserve any env vars that module_load/unload/purge touched, so the lmod-loaded
# and SLURM_TESTSUITE_CONF.
preserve = [
"PATH",
"SLURM_TESTSUITE_CONF",
*sorted(properties["lmod-touched-vars"]),
]
cmd = (
f"sudo"
f" -u {properties['slurm-user']}"
f" --preserve-env={','.join(preserve)}"
f" SLURM_LOCAL_GLOBALS_FILE={properties['expect_globals_file']}"
f" {properties['expect_dir']}/test{test_num}"
)
# Expect tests need to be launch in the expect_dir.
proc = subprocess.Popen(
cmd,
cwd=properties["expect_dir"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
shell=True,
text=True,
)
stdout = ""
for line in proc.stdout:
print(line, end="")
stdout += line
proc.wait()
# If it passed just end
if proc.returncode == 0:
return reason, 0
# Clean the stdout to search for the main reasons to not pass
ansi_escape = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")
stdout = ansi_escape.sub("", stdout)
# Get the actual body of the test
sections = [s for s in stdout.split("=" * 78 + "\n")]
if len(sections) < 3:
pytest.fail("Something failed while running the expect test")
body = sections[2]
# Search for the main reason to not pass (as in regression.py)
fatals = re.findall(r"(?ms)\[[^\]]+\][ \[]+Fatal[ \]:]+(.*?) \(fail[^\)]+\)$", body)
errors = re.findall(
r"(?ms)\[[^\]]+\][ \[]+Error[ \]:]+(.*?) \(subfail[^\)]+\)$", body
)
warnings = re.findall(
r"(?ms)\[[^\]]+\][ \[]+Warning[ \]:]+((?:(?!Warning).)*) \((?:sub)?skip[^\)]+\)$",
body,
)
if fatals:
reason = fatals[0]
elif errors:
reason = errors[0]
elif warnings:
reason = warnings[0]
# Forward the expect result to pytest
if proc.returncode > 127:
pytest.skip(reason)
elif fail:
pytest.fail(reason)
return reason, proc.returncode
def require_version(
min_version,
component="sbin/slurmctld",
slurm_prefix="",
reason=None,
max_version=None,
):
"""Checks if the component is at least the required version, or skips.
Args:
min_version (tuple): The tuple representing the minimal version required, version should be >=.
max_version (tuple): The tuple representing the maximal version required, version should be <.
component (string): The bin/ or sbin/ component of Slurm to check.
slurm_prefix (string): The path where the component is. By default the defined in testsuite.conf.
reason (string): The reason why the version of the component is required.
"""
component_version = get_version(component, slurm_prefix)
if component_version >= min_version and (
max_version is None or component_version < max_version
):
return
if not reason:
reason = f"The version of {component} is {component_version}, required is {min_version}"
if max_version:
reason += f" up to {max_version} (not included)"
pytest.skip(reason, allow_module_level=True)
def request_slurmctld(request, user=None):
"""
Returns the slurmctld response of a given request.
It needs slurmctld >= 25.11 listening HTTP requests.
"""
token = None
if user is not None:
token = "".join(
run_command_output(
f"scontrol token username={user}",
fatal=True,
quiet=True,
user=properties["slurm-user"],
)
.strip()
.split("\n")[-1]
.split("=")[1:]
)
headers = None
# run_command_output worked and returned in time, set JWT headers
if user is not None and token is not None:
headers = {"X-SLURM-USER-NAME": user, "X-SLURM-USER-TOKEN": token}
logging.debug(f"HTTP request to slurmctld with user: {user}, token: {token}")
resp = requests.get(
f"http://{properties['slurmctld_host']}:{properties['slurmctld_port']}/{request}",
headers=headers,
)
logging.debug(f"HTTP request to slurmctld returned: {resp.text}")
return resp
def request_slurmrestd(request):
"""Returns the slurmrestd response of a given request.
It needs slurmrestd to be running (see require_slurmrestd())
"""
return requests.get(
f"{properties['slurmrestd_url']}/{request}",
headers=properties["slurmrestd-headers"],
)
def get_deprecated_openapi_spec_patch(openapi_spec, undeprecate=False):
"""
Returns a JsonPatch object that can be applied to the openapi_spec to
fully deprecate it, as we should do for older specs in newer Slurm
versions.
If undeprecate is set to True, then it does the opposite, it returns a
JsonPatch to remove all deprecate marks of the openapi_specs. This
should NOT be used in general, but as we generated some openapi_specs
with newer Slurm versions, we may need to undeprecate them (see v41).
"""
ops = []
def _escape_json_pointer(s: str) -> str:
return s.replace("~", "~0").replace("/", "~1")
for raw_path, path_obj in openapi_spec.get("paths", {}).items():
escaped_path = _escape_json_pointer(raw_path)
for method, operation in path_obj.items():
if method.lower() not in (
"get",
"post",
"put",
"delete",
"patch",
"head",
"options",
):
continue
# 1. Mark the operation itself as deprecated
if not undeprecate:
ops.append(
{
"op": "add",
"path": f"/paths/{escaped_path}/{method}/deprecated",
"value": True,
}
)
else:
ops.append(
{
"op": "remove",
"path": f"/paths/{escaped_path}/{method}/deprecated",
}
)
# 2. Mark every parameter as deprecated
params = operation.get("parameters", [])
for i, _ in enumerate(params):
if not undeprecate:
ops.append(
{
"op": "add",
"path": f"/paths/{escaped_path}/{method}/parameters/{i}/deprecated",
"value": True,
}
)
else:
ops.append(
{
"op": "remove",
"path": f"/paths/{escaped_path}/{method}/parameters/{i}/deprecated",
}
)
return jsonpatch.JsonPatch(ops)
def assert_openapi_spec_eq(spec_a, spec_b):
"""
Asserts that two OpenAPI specifications are equal, ignoring some info and
description fields.
Note that atf.properties["openapi_spec"] saves the specs used to generate
the python openapi_client from slurmrestd and we have some json specs in
the testsuite_data_dir that can be read with the openapi_spec feature.
Usually we want to assert both are equal.
Args:
spec_a: one loaded json spec
spec_b: the other loaded json spec
Returns:
None, but a pytest assert call.
"""
# Copy specs so we can change some things
_spec_a = copy.deepcopy(spec_a)
_spec_b = copy.deepcopy(spec_b)
# Avoid checking versions that change with each new release
_spec_a["info"]["x-slurm"] = None
_spec_b["info"]["x-slurm"] = None
_spec_a["info"]["version"] = None
_spec_b["info"]["version"] = None
_spec_a["tags"] = None
_spec_b["tags"] = None
# Recursively remove all description fields
def _strip_descriptions(oas):
"""Recursively removes all description fields in oas
Returns:
oas with all description fields set to None
"""
if type(oas) is dict:
for key, value in oas.items():
if key.casefold() == "description".casefold():
oas[key] = None
else:
oas[key] = _strip_descriptions(value)
elif type(oas) is list:
for value in oas:
value = _strip_descriptions(value)
return oas
_spec_a = _strip_descriptions(_spec_a)
_spec_b = _strip_descriptions(_spec_b)
d = jsondiff.diff(_spec_a, _spec_b)
assert (
len(d) == 0
), f"OpenAPI specs should be equal(equivalent, but these differences were found: {d}"
def require_openapi_generator(version="7.3.0"):
"""Generates an OpenAPI client using OpenAPI-Generator, or skips if not available (even in auto-config).
It needs slurmrestd to be running (see require_slurmrestd()).
It also sets the necessary OPENAPI_GENERATOR_VERSION and JAVA_OPTS
environment variables.
Args:
version (string): the required version.
Returns:
None
"""
# Require specific testing version
os.environ["OPENAPI_GENERATOR_VERSION"] = version
# Work around: https://github.com/OpenAPITools/openapi-generator/issues/13684
os.environ["JAVA_OPTS"] = (
"--add-opens java.base/java.util=ALL-UNNAMED --add-opens java.base/java.lang=ALL-UNNAMED"
)
ogc_version = (
run_command_output("openapi-generator-cli version").strip().split("\n")[-1]
)
if ogc_version != version:
pytest.skip(
f"test requires openapi-generator-cli version {version} (not {ogc_version})",
allow_module_level=True,
)
# allow pointing to an existing OpenAPI generated client
opath = module_tmp_path
if "SLURM_TESTSUITE_OPENAPI_CLIENT" in os.environ:
opath = os.environ["SLURM_TESTSUITE_OPENAPI_CLIENT"]
pyapi_path = f"{opath}/pyapi/"
spec_path = f"{opath}/openapi.json"
# Always create path if needed
os.makedirs(opath, exist_ok=True)
if not os.path.exists(spec_path):
r = request_slurmrestd("openapi/v3")
if r.status_code != 200:
pytest.fail(f"Error requesting openapi specs from slurmrestd: {r}")
properties["openapi_spec"] = json.loads(r.text)
if properties["openapi_spec"] is None:
pytest.fail(f"Error parsing OpenAPI specs from slurmrestd: {r.text}")
with open(spec_path, "w") as f:
f.write(r.text)
f.close()
if not os.path.exists(pyapi_path):
run_command(
f"openapi-generator-cli generate -i '{spec_path}' -g python-pydantic-v1 --strict-spec=true -o '{pyapi_path}'",
fatal=True,
timeout=60,
)
sys.path.insert(0, pyapi_path)
# Re-import openapi_client
# Regular import doesn't work if was already imported by another test.
global openapi_client
module_name = "openapi_client"
module_prefix = module_name + "."
for mod in list(sys.modules):
if mod == module_name or mod.startswith(module_prefix):
del sys.modules[mod]
openapi_client = importlib.import_module(module_name)
importlib.reload(openapi_client)
properties["openapi_config"] = openapi_client.Configuration()
properties["openapi_config"].host = properties["slurmrestd_url"]
properties["openapi_config"].access_token = properties["slurmrestd-headers"][
"X-SLURM-USER-TOKEN"
]
def openapi_util():
"""
Returns a UtilApi client from OpenAPI.
It needs require_openapi_generator() to be run first.
"""
return openapi_client.UtilApi(
openapi_client.ApiClient(properties["openapi_config"])
)
def openapi_slurm():
"""
Returns a SlurmApi client from OpenAPI.
It needs require_openapi_generator() to be run first.
"""
return openapi_client.SlurmApi(
openapi_client.ApiClient(properties["openapi_config"])
)
def openapi_slurmdb():
"""
Returns a SlurmdbApi client from OpenAPI.
It needs require_openapi_generator() to be run first.
"""
return openapi_client.SlurmdbApi(
openapi_client.ApiClient(properties["openapi_config"])
)
def get_config(live=True, source="slurm", quiet=True, delimiter="="):
"""Returns the Slurm configuration as a dictionary.
Args:
live (boolean):
If True, the configuration information is obtained via
a query to the relevant Slurm daemon (e.g., scontrol show config).
If False, the configuration information is obtained by directly
parsing the relevant Slurm configuration file (e.g. slurm.conf).
source (string):
If live is True, source should be either scontrol or sacctmgr.
If live is False, source should be the name of the config file
without the .conf prefix (e.g. slurmdbd).
quiet (boolean): If True, logging is performed at the TRACE log level.
delimiter (string): The delimiter between the parameter name and the value.
Returns:
A dictionary comprised of the parameter names and their values.
For parameters that can have multiple lines and subparameters,
the dictionary value will be a dictionary of dictionaries.
Example:
>>> get_config()
{'AccountingStorageBackupHost': '(null)', 'AccountingStorageEnforce': 'none', 'AccountingStorageHost': 'localhost', 'AccountingStorageExternalHost': '(null)', 'AccountingStorageParameters': '(null)', 'AccountingStoragePort': '0', 'AccountingStorageTRES': 'cpu,mem,energy,node,billing,fs/disk,vmem,pages', ...}
>>> get_config(live=False, source='slurm')
{'SlurmctldHost': 'nathan-atf-docstrings', 'SlurmUser': 'slurm', 'SlurmctldLogFile': '/var/slurm/log/slurmctld.log', 'SlurmctldPidFile': '/var/slurm/run/slurmctld.pid', 'SlurmctldDebug': 'debug3', 'SlurmdLogFile': '/var/slurm/log/slurmd.%n.log', 'SlurmdPidFile': '/var/slurm/run/slurmd.%n.pid', ...}
>>> get_config(live=True, source='scontrol', quiet=True)
{'AccountingStorageBackupHost': '(null)', 'AccountingStorageEnforce': 'none', 'AccountingStorageHost': 'localhost', 'AccountingStorageExternalHost': '(null)', 'AccountingStorageParameters': '(null)', 'AccountingStoragePort': '0', 'AccountingStorageTRES': 'cpu,mem,energy,node,billing,fs/disk,vmem,pages', ...}
"""
slurm_dict = {}
if live:
if source == "slurm" or source == "controller" or source == "scontrol":
command = "scontrol"
elif source == "slurmdbd" or source == "dbd" or source == "sacctmgr":
command = "sacctmgr"
else:
pytest.fail(f"Invalid live source value ({source})")
output = run_command_output(f"{command} show config", fatal=True, quiet=quiet)
for line in output.splitlines():
if match := re.search(rf"^\s*(\S+)\s*{re.escape(delimiter)}\s*(.*)$", line):
slurm_dict[match.group(1)] = match.group(2).rstrip()
else:
config = source
config_file = f"{properties['slurm-config-dir']}/{config}.conf"
# We might be looking for parameters in a config file that has not
# been created yet. If so, we just want this to return an empty dict
output = run_command_output(
f"cat {config_file}", user=properties["slurm-user"], quiet=quiet
)
for line in output.splitlines():
if match := re.search(
rf"^\s*(\S+?)\s*{re.escape(delimiter)}\s*(.*)$", line
):
parameter_name, parameter_value = (
match.group(1),
match.group(2).rstrip(),
)
if parameter_name.lower() in [
"downnodes",
"name",
"nodename",
"nodeset",
"partitionname",
"switchname",
]:
instance_name, subparameters = parameter_value.split(" ", 1)
subparameters_dict = {}
for subparameter_name, subparameter_value in re.findall(
rf" *([^= ]+) *{re.escape(delimiter)} *([^ ]+)", subparameters
):
# Reformat the value if necessary
if is_integer(subparameter_value):
subparameter_value = int(subparameter_value)
elif is_float(subparameter_value):
subparameter_value = float(subparameter_value)
elif subparameter_value == "(null)":
subparameter_value = None
subparameters_dict[subparameter_name] = subparameter_value
if parameter_name not in slurm_dict:
slurm_dict[parameter_name] = {}
slurm_dict[parameter_name][instance_name] = subparameters_dict
else:
# Reformat the value if necessary
if is_integer(parameter_value):
parameter_value = int(parameter_value)
elif is_float(parameter_value):
parameter_value = float(parameter_value)
elif parameter_value == "(null)":
parameter_value = None
slurm_dict[parameter_name] = parameter_value
return slurm_dict
def get_config_parameter(name, default=None, **get_config_kwargs):
"""Obtains the value for a Slurm configuration parameter.
This function accepts the same arguments as get_config.
Args:
name (string): The parameter name.
default (string or None): This value is returned if the parameter
is not found.
Returns:
The value of the specified parameter, or the default if not found.
Example:
>>> get_config_parameter('JobAcctGatherFrequency')
'30'
>>> get_config_parameter('MaxJobCount', default='10000')
'10000'
>>> get_config_parameter('partitionname', default='debug', live=True, source='scontrol')
'debug'
"""
config_dict = get_config(**get_config_kwargs)
# Convert keys to lower case so we can do a case-insensitive search
lower_dict = dict(
(key.casefold(), str(value).casefold()) for key, value in config_dict.items()
)
if name.casefold() in lower_dict:
return lower_dict[name.casefold()]
else:
return default
def config_parameter_includes(name, value, **get_config_kwargs):
"""Checks whether a configuration parameter includes a specific value.
When a parameter may contain a comma-separated list of values, this
function can be used to determine whether a specific value is within
the list.
This function accepts the same arguments as get_config.
Args:
name (string): The parameter name.
value (string): The value you are looking for.
Returns:
True if the specified string value is found within the parameter
value list, False otherwise.
Example:
>>> config_parameter_includes('SlurmdParameters', 'config_overrides')
False
"""
config_dict = get_config(**get_config_kwargs)
# Convert keys to lower case so we can do a case-insensitive search
lower_dict = dict((key.casefold(), value) for key, value in config_dict.items())
if name.casefold() in lower_dict and value.lower() in map(
str.lower, lower_dict[name.casefold()].split(",")
):
return True
else:
return False
def set_config_parameter(
parameter_name,
parameter_value,
source="slurm",
restart=False,
delimiter="=",
):
"""Sets the value of the specified configuration parameter.
This function modifies the specified slurm configuration file and
reconfigures slurm (or restarts slurm if restart=True). A backup
is automatically created and the original configuration is restored
after the test completes.
This function may only be used in auto-config mode.
Args:
parameter_name (string): The parameter name.
parameter_value (string): The parameter value.
Use a value of None to unset a parameter.
source (string): Name of the config file without the .conf prefix.
restart (boolean): If True and slurm is running, slurm will be
restarted rather than reconfigured.
delimiter (string): The delimiter between the parameter name and the value.
Note:
When setting a complex parameter (one which may be repeated and has
its own subparameters, such as with nodes, partitions and gres),
the parameter_value should be a dictionary of dictionaries.
AccountingStorageType is a special case: it also updates
properties["accounting"], and if slurm is running it is stopped
before the change and started after it, so slurmdbd is stopped or
started as needed.
Returns:
None
Example:
>>> set_config_parameter("ClusterName", "cluster1")
>>> set_config_parameter("required", "/tmp/spank_plugin.so", source="plugstack", delimiter=" ")
"""
if not properties["auto-config"]:
require_auto_config("wants to modify parameters")
if source == "dbd":
config = "slurmdbd"
else:
config = source
config_file = f"{properties['slurm-config-dir']}/{config}.conf"
slurmctld_running = is_slurmctld_running(quiet=True)
# Handle AccountingStorageType as a special case: it decides whether
# slurmdbd is part of slurm, so stop slurm with the current value and
# start it again with the new one.
accounting_storage_type = (
config == "slurm" and parameter_name.casefold() == "accountingstoragetype"
)
if accounting_storage_type and slurmctld_running:
stop_slurm(quiet=True)
# Remove all matching parameters and append the new parameter
lines = []
output = run_command_output(
f"cat {config_file}", user=properties["slurm-user"], quiet=True
)
for line in output.splitlines():
if not re.search(rf"(?i)^\s*{parameter_name}\s*{re.escape(delimiter)}", line):
lines.append(line)
if isinstance(parameter_value, dict):
for instance_name in parameter_value:
line = f"{parameter_name}{delimiter}{instance_name}"
for subparameter_name, subparameter_value in parameter_value[
instance_name
].items():
line += f" {subparameter_name}{delimiter}{subparameter_value}"
lines.append(line)
elif parameter_value is not None:
lines.append(f"{parameter_name}{delimiter}{parameter_value}")
input = "\n".join(lines)
run_command(
f"cat > {config_file}",
input=input,
user=properties["slurm-user"],
fatal=True,
quiet=True,
)
if accounting_storage_type:
if str(parameter_value).casefold() == "accounting_storage/slurmdbd":
if not properties["accounting"]:
properties["accounting"] = "enabled"
else:
# Restore the original DB now, the teardown won't do it anymore
if properties["accounting"] == "db-saved":
failures = restore_accounting_database(properties["sql-db-backup"])
if failures:
pytest.fail("; ".join(failures))
properties["accounting"] = None
# Remove clustername state file if we aim to change the cluster name
if parameter_name.lower() == "clustername":
state_save_location = get_config_parameter(
"StateSaveLocation", live=slurmctld_running, quiet=True
)
run_command(
f"rm -f {state_save_location}/clustername",
user=properties["slurm-user"],
quiet=True,
)
# Reconfigure (or restart) slurm controller if it is already running
if slurmctld_running:
if accounting_storage_type:
start_slurm(quiet=True)
elif source != "slurm" or parameter_name.lower() in [
"rebootprogram",
]:
restart_slurm(quiet=True)
elif restart or parameter_name.lower() in [
"authtype",
"controlmach",
"plugindir",
"statesavelocation",
"slurmctldhost",
"slurmctldport",
"slurmdport",
]:
restart_slurmctld(quiet=True)
else:
run_command(
"scontrol reconfigure", user=properties["slurm-user"], quiet=True
)
def add_config_parameter_value(name, value, source="slurm"):
"""Appends a value to configuration parameter list.
When a parameter may contain a comma-separated list of values, this
function can be used to add a value to the list.
This function may only be used in auto-config mode.
Args:
name (string): The parameter name.
value (string): The value to add.
source (string): Name of the config file without the .conf prefix.
Returns:
None
Example:
>>> add_config_parameter_value('SlurmdParameters', 'config_overrides')
"""
if config_parameter_includes(name, value, live=False, quiet=True, source=source):
return
original_value_string = get_config_parameter(
name, live=False, quiet=True, source=source
)
if original_value_string is None:
set_config_parameter(name, value, source=source)
else:
value_list = original_value_string.split(",")
value_list.append(value)
set_config_parameter(name, ",".join(value_list), source=source)
def remove_config_parameter_value(name, value, source="slurm"):
"""Removes a value from a configuration parameter list.
When a parameter may contain a comma-separated list of values, this
function can be used to remove a value from the list.
This function may only be used in auto-config mode.
Args:
name (string): The parameter name.
value (string): The value to remove.
source (string): Name of the config file without the .conf prefix.
Returns:
None
Example:
>>> remove_config_parameter_value('SlurmdParameters', 'config_overrides')
"""
if not config_parameter_includes(
name, value, live=False, quiet=True, source=source
):
return
value_list = get_config_parameter(
name, live=False, quiet=True, source=source
).split(",")
value_list.remove(value.casefold())
if value_list:
set_config_parameter(name, ",".join(value_list), source=source)
else:
set_config_parameter(name, None, source=source)
def is_tool(tool):
"""Returns True if the tool is found in PATH.
Args:
tool (string): The name of the tool to check for in the PATH environment variable.
Returns:
True if the tool is found in PATH, False otherwise.
Example:
>>> is_tool('ls')
True
>>> is_tool('uninstalled-tool-name')
False
"""
from shutil import which
return which(tool) is not None
def get_node_cpu_topology(node=None, require_numa=False):
"""Return the CPU topology a node reports, or skip.
Maps every CPU id the node presents to the socket, core and NUMA node
holding it. This map may include cpuset-restricted CPUs.
Args:
node (string): The node to read or None to read the test host directly.
require_numa (boolean): Skip unless a NUMA node is reported for every
CPU. Otherwise an unreported NUMA node is returned as None.
Returns:
A map from cpu_id to a dict of socket, core, numa_node.
Example:
>>> get_node_cpu_topology('node1')[3]
{'socket': 0, 'core': 1, 'numa_node': 0}
"""
command = "lscpu -p=CPU,CORE,SOCKET,NODE -b"
node_str = "the test host"
if node is None:
require_tool("lscpu")
results = run_command(command, quiet=True, fatal=True)
else:
node_str = f"node {node}"
srun = f"srun --nodelist={node} -N1"
run_command(f"{srun} true", quiet=True, fatal=True)
results = run_command(f"{srun} {command}", quiet=True)
if results["exit_code"] != 0:
pytest.skip(f"This test requires lscpu on {node_str}")
topology = {}
for line in results["stdout"].splitlines():
if not line or line.startswith("#"):
continue
fields = line.split(",")
if len(fields) < 3 or not all(field.isdigit() for field in fields[:3]):
pytest.skip(f"This test requires lscpu on {node_str} to report a CPU")
cpu_id, core, socket = (int(field) for field in fields[:3])
numa_field = fields[3] if len(fields) > 3 else ""
if not numa_field.isdigit():
if require_numa:
pytest.skip(
f"This test requires lscpu on {node_str} to report a NUMA node"
)
numa_node = None
else:
numa_node = int(numa_field)
topology[cpu_id] = {"socket": socket, "core": core, "numa_node": numa_node}
if not topology:
pytest.skip(f"This test requires lscpu on {node_str} to report a CPU")
return topology
def require_tool(tool):
"""Skips if the supplied tool is not found.
Args:
tool (string): The name of the tool to check for in the PATH environment variable.
Returns:
None
Example:
>>> require_tool('ls')
>>> require_tool('uninstalled-tool-name')
"""
if not is_tool(tool):
msg = f"This test requires '{tool}' and it was not found"
pytest.skip(msg, allow_module_level=True)
def require_mpi(mpi_option="pmix", mpi_compiler="mpicc"):
"""Skips if the build lacks the mpi_option plugin or mpi_compiler is missing.
The pmi2 and cray_shasta plugins are always built, so they are never
skipped and a failure to load one fails the test. The pmix plugin is
built only when configure found libpmix (HAVE_PMIX).
Args:
mpi_option (string): The value to use with --mpi when submitting jobs.
mpi_compiler (string): The required compiler in the system.
Returns:
None
"""
require_tool(mpi_compiler)
if mpi_option.startswith("pmix"):
require_build_config(
"HAVE_PMIX", reason=f"Slurm was built without the mpi/{mpi_option} plugin"
)
def require_lmod():
"""
Skips if Lmod (environment modules) is not available.
"""
require_tool("lmod")
def _module(action, *modules):
"""
Run `lmod python <action> [<modules>...]` and apply env changes
to the current Python process's os.environ.
"""
# Lmod's `python` shell emits Python code that mutates os.environ.
output = run_command_output(
f"lmod python {action} {' '.join(modules)}", fatal=True, quiet=True
)
# Applying the module and saving the difference so we can preserve them
# in sudo commands
before = dict(os.environ)
exec(output, {"os": os})
for k in set(before) | set(os.environ):
if before.get(k) != os.environ.get(k):
properties["lmod-touched-vars"].add(k)
def module_load(*modules):
"""
Load Lmod environment module(s) into the current process's env.
Args:
*modules: Module name(s) to load (e.g. "openmpi/5.0.10").
Example:
>>> require_lmod()
>>> module_load("openmpi/5.0.10")
"""
_module("load", *modules)
def module_unload(*modules):
"""
Unload Lmod environment module(s) from the current process's env.
Args:
*modules: Module name(s) to unload.
"""
_module("unload", *modules)
def module_purge():
"""Unload all currently loaded Lmod environment modules."""
_module("purge")
def require_influxdb(influx_client="influx", jobacct_gather="jobacct_gather/cgroup"):
"""Require the influx client available and the right config.
With auto-config the default values would work. Without you may need to setup the acct_gather values in the testsuite.conf file.
Args:
influx_client (string): The name of the influxdb client. Default: influx.
jobacct_gather (string): To use influxdb we need to set some JobAcctGatherType. Default: cgroup.
Returns:
None
"""
require_tool(influx_client)
require_config_parameter("JobAcctGatherType", jobacct_gather)
require_config_parameter("AcctGatherProfileType", "acct_gather_profile/influxdb")
require_config_parameter(
"ProfileInfluxDBHost",
f"{properties['influxdb_host']}:{properties['influxdb_port']}",
source="acct_gather",
)
require_config_parameter(
"ProfileInfluxDBDatabase", properties["influxdb_db"], source="acct_gather"
)
require_config_parameter("ProfileInfluxDBDefault", "ALL", source="acct_gather")
require_config_parameter("ProfileInfluxDBRTPolicy", "autogen", source="acct_gather")
request_influxdb(f"CREATE DATABASE {properties['influxdb_db']}")
properties["influxdb-started"] = True
def request_influxdb(query):
"""Send a query using the influx client. Requires to run require_influxdb first.
Args:
query (string): The query to send,
Returns:
The stdout returned by the influxdb client.
If the clients fails somehow, this function will fatal.
"""
return run_command_output(
f"influx -host {properties['influxdb_host']} -port {properties['influxdb_port']} -database {properties['influxdb_db']} -format column -execute \"{query}\"",
fatal=True,
)
def require_whereami():
"""Compiles the whereami.c program to be used by tests.
This function installs the whereami program. To get the
correct output, TaskPlugin is required in the slurm.conf
file before slurm starts up.
ex: TaskPlugin=task/cgroup,task/affinity
The file will be installed in the testsuite/python/lib/scripts
directory where the whereami.c file is located
Args:
None
Returns:
None
Examples:
>>> atf.require_whereami()
>>> print('\nwhereami is located at', atf.properties['whereami'])
>>> output = atf.run_command(f"srun {atf.properties['whereami']}",
>>> user=atf.properties['slurm-user'])
"""
require_config_parameter("TaskPlugin", "task/cgroup,task/affinity")
# If the file already exists and we don't need to recompile
dest_file = f"{properties['testsuite_scripts_dir']}/whereami"
if os.path.isfile(dest_file):
properties["whereami"] = dest_file
return
source_file = f"{properties['testsuite_scripts_dir']}/whereami.c"
if not os.path.isfile(source_file):
pytest.skip("Could not find whereami.c!", allow_module_level=True)
run_command(
f"gcc {source_file} -o {dest_file}", fatal=True, user=properties["slurm-user"]
)
properties["whereami"] = dest_file
def require_config_file(
filename,
content,
):
"""
Ensures that a configuration file exists with the required content.
In local-config mode, the test is skipped if the required config file
doesn't exists or is not equivalent.
In auto-config mode, save current config file if exists and creates the
required one.
Note that it cannot be used with some base config files like slurm.conf,
use require_config_parameter() for those config files instead.
Supported extensions for local-config are .conf and .yaml.
Args:
filename (string): The filename of the config path (e.g. "resources.yaml").
content (string): The desired content of the config file.
Returns:
None
It will fail or skip the test if required config file cannot be set.
"""
# Some config files need to be modified, cannot be created from scratch
if filename in {
"slurm.conf",
"slurmdbd.conf",
"cgroup.conf",
"testsuite.conf",
}:
pytest.fail(f"Use require_config_param() for file {filename}")
file_path = f"{properties['slurm-config-dir']}/{filename}"
# If not auto-condig:
if not properties["auto-config"]:
supported_ext = {".conf", ".yaml"}
file, ext = os.path.splitext(filename)
match ext:
case ".yaml":
desired_yaml = yaml.safe_load(content)
local_yaml = None
if os.path.exists(file_path):
with open(file_path, "r", encoding="utf-8") as f:
local_yaml = yaml.safe_load(f)
if desired_yaml != local_yaml:
pytest.skip(f"{filename} is not as required by the test")
else:
logging.debug(f"Found compatible required {filename}")
return
case ".conf":
# TODO: Ignore blank or commented lines
with open(file_path, "r", encoding="utf-8") as f:
local_content = f.read()
if content != local_content:
pytest.skip(f"{filename} is not as required by the test")
else:
logging.debug(f"Found compatible required {filename}")
return
case _:
pytest.fail(f"Supported config files are only {supported_ext}")
pytest.fail(
"Internal error: function should already skip, fail or return in local-config"
)
# In auto-config
run_command(
f"cat > {file_path}", input=content, user=properties["slurm-user"], fatal=True
)
def require_config_parameter(
parameter_name,
parameter_value,
operator="==",
source="slurm",
skip_message=None,
delimiter="=",
):
"""Ensures that a configuration parameter has the required value.
In local-config mode, the test is skipped if the required configuration is not set.
In auto-config mode, sets the required configuration value if necessary.
Args:
parameter_name (string): The parameter name.
parameter_value: The target parameter value or a list of them.
If a list of acceptable values is given, the requirement is
considered satisfied when the current value matches ANY of them.
In auto-config mode, when none match, the FIRST listed value is
used as the target — so callers should list their preferred
default first.
If a dict is passed then multiple values are assigned.
None means parameter is required to be unset.
operator (string): By default the observed current value and the
desired values are expected to be equal ("=="), but we can require
just "<=" or ">=" too (usually for numerical values).
If a dict is passed as parameter_value, only "==" is supported.
source (string): Name of the config file without the .conf prefix.
skip_message (string): Message to be displayed if in local-config mode
and parameter not present.
delimiter (string): The delimiter between the parameter name and the value.
Note:
When requiring a complex parameter (one which may be repeated and has
its own subparameters, such as with nodes, partitions and gres),
the parameter_value should be a dictionary of dictionaries. See the
fourth example for multi-line parameters.
Returns:
None
Examples:
>>> require_config_parameter('SelectType', 'select/cons_tres')
>>> require_config_parameter('SlurmdTimeout', 5, '<=')
>>> require_config_parameter('SelectType', ['select/cons_tres', 'select/linear'])
>>> require_config_parameter('Name', {'gpu': {'File': '/dev/tty0'}, 'mps': {'Count': 100}}, source='gres')
>>> require_config_parameter("PartitionName", {"primary": {"Nodes": "ALL"}, "dynamic1": {"Nodes": "ns1"}, "dynamic2": {"Nodes": "ns2"}, "dynamic3": {"Nodes": "ns1,ns2"}})
"""
_ALLOWED_OPERATORS = ("==", "<=", ">=")
if operator not in _ALLOWED_OPERATORS:
pytest.fail(
f"Unsupported operator {operator!r} (allowed: {_ALLOWED_OPERATORS})"
)
# Dict case: params like PartitionName. The outer key is the stanza
# identifier; the inner dict is the stanza's sub-key=value pairs.
if isinstance(parameter_value, dict):
if operator != "==":
pytest.fail(
f"A dict parameter_value only supports operator '==', "
f"not {operator!r}"
)
tmp1_dict = dict()
for k1, v1 in parameter_value.items():
tmp2_dict = dict()
for k2, v2 in v1.items():
if isinstance(v2, str):
tmp2_dict[k2.casefold()] = v2.casefold()
else:
tmp2_dict[k2.casefold()] = v2
tmp1_dict[k1.casefold()] = tmp2_dict
parameter_value = tmp1_dict
observed_value = get_config_parameter(
parameter_name, live=False, source=source, quiet=True, delimiter=delimiter
)
if str(observed_value) == str(parameter_value):
return
if properties["auto-config"]:
set_config_parameter(
parameter_name, parameter_value, source=source, delimiter=delimiter
)
else:
if skip_message is None:
skip_message = f"This test requires the {parameter_name} parameter to be {parameter_value} (but it is {observed_value})"
pytest.skip(skip_message, allow_module_level=True)
return
if isinstance(parameter_value, (list, tuple)):
values = list(parameter_value)
else:
values = [parameter_value]
if not values:
pytest.fail(
f"require_config_parameter({parameter_name!r}, ...) needs at least one value"
)
values = [v.casefold() if isinstance(v, str) else v for v in values]
if operator != "==" and len(values) > 1:
pytest.fail(
f"Using operator {operator!r} with multiple values doesn't make sense, right?"
)
observed_value = get_config_parameter(
parameter_name, live=False, source=source, quiet=True, delimiter=delimiter
)
if not observed_value and None in values:
# already unset, and unset is acceptable
return
condition_satisfied = False
if observed_value:
for v in values:
if v is None:
# None means expected unset, but parameter was set
continue
try:
obs = type(v)(observed_value)
except ValueError:
# The observed value cannot be converted to the expected
# type (e.g. "infinite" for an int parameter), so it cannot
# satisfy this requirement; try the next acceptable value.
continue
if operator == "==" and obs == v:
condition_satisfied = True
break
if operator == "<=" and obs <= v:
condition_satisfied = True
break
if operator == ">=" and obs >= v:
condition_satisfied = True
break
if not condition_satisfied:
target = values[0]
if properties["auto-config"]:
set_config_parameter(
parameter_name, target, source=source, delimiter=delimiter
)
else:
if skip_message is None:
skip_message = f"This test requires the {parameter_name} parameter to be {parameter_value} (but it is {observed_value})"
pytest.skip(skip_message, allow_module_level=True)
def require_config_parameter_includes(name, value, source="slurm"):
"""Ensures that a configuration parameter list contains the required value.
In local-config mode, the test is skipped if the configuration parameter
list does not include the required value.
In auto-config mode, adds the required value to the configuration parameter
list if necessary.
Args:
name (string): The parameter name.
value: The required value, in one of these forms:
* string: A single token that must be present in the comma list.
* (key, val) tuple: A "key=val" pair.
(e.g. ``('sched_interval', 1)`` in ``SchedulerParameters``).
* list: A set of acceptable alternatives (strings or tuples).
The requirement is satisfied if ANY of them is satisfied.
source (string): Name of the config file without the .conf prefix.
Returns:
None
Example:
>>> # Single value:
>>> require_config_parameter_includes('SlurmdParameters', 'config_overrides')
>>>
>>> # key=value: replaces any existing 'sched_interval=<other>' token:
>>> require_config_parameter_includes(
... 'SchedulerParameters', ('sched_interval', 1))
>>>
>>> # Any of several acceptable values; if none present in auto-config,
>>> # adds 'CR_Core_Memory' (the first listed):
>>> require_config_parameter_includes(
... 'SelectTypeParameters',
... ['CR_Core_Memory', 'CR_Memory'],
... )
"""
# A list means "one of these alternatives"; anything else is a single
# requirement (string token, or (key, val) tuple). Wrap the single case
# in a list so we can iterate uniformly below.
values = value if isinstance(value, list) else [value]
if not values:
pytest.fail(
f"require_config_parameter_includes({name!r}, ...) needs at least one value"
)
def _as_token(v):
"""Render a requirement as the token that must appear in the list."""
if isinstance(v, tuple):
if len(v) != 2:
pytest.fail("Only supporting key-value tuples")
return f"{v[0]}={v[1]}"
return str(v)
def _is_satisfied(v):
return config_parameter_includes(
name, _as_token(v), source=source, live=False, quiet=True
)
already_present = any(_is_satisfied(v) for v in values)
if properties["auto-config"]:
if already_present:
return
first = values[0]
if isinstance(first, tuple):
# (key, val): replace any existing "key=<other>" token in place;
# otherwise append. This avoids ambiguous duplicate keys.
key, _ = first
kv_token = _as_token(first)
current = get_config_parameter(name, live=False, quiet=True, source=source)
tokens = current.split(",") if current else []
key_prefix = f"{key.casefold()}="
match_idx = next(
(
i
for i, t in enumerate(tokens)
if t.casefold().startswith(key_prefix)
),
None,
)
if match_idx is not None:
tokens[match_idx] = kv_token
set_config_parameter(name, ",".join(tokens), source=source)
else:
add_config_parameter_value(name, kv_token, source=source)
else:
add_config_parameter_value(name, _as_token(first), source=source)
else:
if not already_present:
pytest.skip(
f"This test requires the {name} parameter to include any of these: {value}",
allow_module_level=True,
)
def require_config_parameter_excludes(name, value, source="slurm"):
"""Ensures that a configuration parameter list does not contain a value.
In local-config mode, the test is skipped if the configuration parameter
includes the specified value. In auto-config mode, removes the specified
value from the configuration parameter list if necessary.
Args:
name (string): The parameter name.
value (string): The value we do not want to be in the list.
source (string): Name of the config file without the .conf prefix.
Returns:
None
Example:
>>> require_config_parameter_excludes('SlurmdParameters', 'config_overrides')
"""
if properties["auto-config"]:
remove_config_parameter_value(name, value, source=source)
else:
if config_parameter_includes(
name, value, source=source, live=False, quiet=True
):
pytest.skip(
f"This test requires the {name} parameter to exclude {value}",
allow_module_level=True,
)
def require_tls():
"""Skips the test unless the cluster is configured to use TLS.
Unlike require_config_parameter('TLSType', ...), this never enables TLS in
auto-config mode: a working TLSType needs certificate infrastructure (a CA,
per-daemon certificates and the TLSParameters pointing at them) that cannot
be synthesized here, so a cluster without it is skipped in both modes.
The config file is consulted first so that a non-TLS cluster is skipped
without having to start Slurm. That read does not follow Include
directives, so a TLSType found only in an included file reads as absent
here; fall back to the live value when the file says nothing and a
controller is available.
Returns:
None
"""
tls_type = get_config_parameter("TLSType", default=None, live=False, quiet=True)
if tls_type is None and is_slurmctld_running(quiet=True):
tls_type = get_config_parameter("TLSType", default=None, quiet=True)
if not tls_type or tls_type.casefold() == "tls/none":
pytest.skip(
"This test requires the TLSType parameter to be set to a TLS "
"plugin such as tls/s2n (run under the -s2n test variant)",
allow_module_level=True,
)
def _require_node_tls_credentials(node_names, quiet=False):
"""Ensures slurmds with the given node names can get their TLS certificates.
Tests do not call this: start_slurmctld() does it for every configured
node, and start_slurmd() for the node it starts.
Without TLS, or for nodes that can already get their token, nothing is
done.
Args:
node_names (list): The node names the slurmds will register as.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
None
"""
if not node_names:
return
# get_config_parameter() casefolds values, and these are paths
config = {
key.casefold(): str(value)
for key, value in get_config(live=False, quiet=True).items()
}
tls_type = config.get("tlstype")
if not tls_type or tls_type.casefold() == "tls/none":
return
if not config.get("certmgrtype") or config.get("certmgrtype") == "None":
# Without certmgr, slurmd loads a static slurmd_cert_file (with %n
# expanded) signed by the CA, which the test environment does not use
slurmd_cert = re.search(
r"(?:^|,)slurmd_cert_file=([^,]+)", config.get("tlsparameters", "")
)
if slurmd_cert:
for node_name in node_names:
if not os.path.exists(slurmd_cert.group(1).replace("%n", node_name)):
pytest.fail(
f"No TLS certificate for node {node_name} and "
"provisioning static slurmd_cert_file certificates is "
"not supported"
)
return
certmgr_params = dict(
param.split("=", 1)
for param in config.get("certmgrparameters", "").split(",")
if "=" in param
)
token_script = certmgr_params.get("get_node_token_script")
validate_script = certmgr_params.get("validate_node_script")
if not token_script or not validate_script:
pytest.fail(
"CertmgrParameters needs get_node_token_script and "
"validate_node_script to provision certmgr node tokens"
)
# The token script is the source of truth: a node it returns a token for
# can get its certificate without any help. One sudo for all the nodes.
def _missing():
return run_command_output(
f"for n in {' '.join(node_names)}; do "
f"{token_script} $n >/dev/null 2>&1 || echo $n; done",
user="root",
quiet=quiet,
).split()
missing = _missing()
if not missing:
return
# Locate the files the scripts use, e.g.:
# TOKEN_PATH=/opt/slurm/etc/$1_token.txt
# NODE_TOKEN_LIST_FILE=/opt/slurm/etc/node_token_list.txt
token_path = re.search(
r"^TOKEN_PATH=(\S*)\$1(\S*)$",
run_command_output(f"cat {token_script}", user="root", quiet=True),
re.MULTILINE,
)
token_list = re.search(
r"^NODE_TOKEN_LIST_FILE=(\S+)$",
run_command_output(f"cat {validate_script}", user="root", quiet=True),
re.MULTILINE,
)
if not token_path or not token_list:
pytest.fail(
f"Unable to provision certmgr node tokens for {missing}: "
f"{token_script} or {validate_script} has an unknown format"
)
# Hex keeps the tokens free of characters special to validate_node_script's
# grep; ownership and 0600 mode are what get_node_token_script requires
logging.debug(f"Provisioning certmgr node tokens for {missing}")
slurm_user = properties["slurm-user"]
commands = []
for node_name in missing:
token = os.urandom(16).hex()
token_file = f"{token_path.group(1)}{node_name}{token_path.group(2)}"
commands.append(
f"printf '%s' '{token}' > {token_file} && chmod 0600 {token_file}"
f" && chown {slurm_user}: {token_file}"
f" && echo '{node_name}: {token}' >> {token_list.group(1)}"
)
run_command(" && ".join(commands), user="root", fatal=True, quiet=quiet)
if still_missing := _missing():
pytest.fail(f"{token_script} still has no token for {still_missing}")
def require_tty(number):
"""Creates a TTY device file if it does not exist.
Args:
number (integer): The number of the TTY device.
Returns:
None
Example:
>>> require_tty(1)
>>> require_tty(2)
"""
tty_file = f"/dev/tty{number}"
if not os.path.exists(tty_file):
run_command(
f"mknod -m 666 {tty_file} c 4 {number}", user="root", fatal=True, quiet=True
)
## Use this to create an entry in gres.conf and create an associated tty
# def require_gres_device(name):
#
# gres_value = get_config_parameter('Name', live=False, source='gres', quiet=True)
# if gres_value is None or name not in gres_value:
# if not properties['auto-config']:
# pytest.skip(f"This test requires a '{name}' gres device to be defined in gres.conf", allow_module_level=True)
# else:
# require_tty(0)
# require_config_parameter('Name', {name: {'File': '/dev/tty0'}}, source='gres')
def require_auto_config(reason=""):
"""Ensures that auto-config mode is being used.
This function skips the test if auto-config mode is not enabled.
Args:
reason (string): Augments the skip reason with a context-specific
explanation for why the auto-config mode is needed by the test.
Returns:
None
Example:
>>> require_auto_config("wants to set the Epilog")
"""
if not properties["auto-config"]:
message = "This test requires auto-config to be enabled"
if reason != "":
message += f" ({reason})"
pytest.skip(message, allow_module_level=True)
def require_accounting(modify=False, enabled=True):
"""Ensures that Slurm accounting is configured (or not).
In local-config mode, the test is skipped if Slurm accounting is not
configured as required. In auto-config mode, configures (or unconfigures)
Slurm accounting if necessary.
Args:
modify (boolean): If True, this indicates to the ATF that the test
will modify the accounting database (e.g. adding accounts, etc).
In local-config mode the test is skipped unless
--allow-slurmdbd-modify is used. In auto-config mode it makes no
difference: the original database is always saved and dropped
before slurmdbd is started, so the module starts with an empty
one, and it is restored when the module completes.
enabled (boolean): If False, ensures that Slurm accounting is NOT
configured. Incompatible with modify=True.
Returns:
None
Example:
>>> require_accounting()
>>> require_accounting(modify=True)
>>> require_accounting(enabled=False)
"""
if not enabled:
if modify:
pytest.fail("require_accounting(): modify=True requires enabled=True")
require_config_parameter(
"AccountingStorageType",
None,
skip_message="This test requires accounting not to be configured",
)
return
if (
modify
and not properties["auto-config"]
and not properties["allow-slurmdbd-modify"]
):
require_auto_config("wants to modify the accounting database")
require_config_parameter(
"AccountingStorageType",
"accounting_storage/slurmdbd",
skip_message="This test requires accounting to be configured",
)
def require_slurmrestd(openapi_plugins, data_parsers):
properties["openapi_plugins"] = openapi_plugins
properties["data_parsers"] = data_parsers
if properties["auto-config"]:
properties["slurmrestd-started"] = True
elif "SLURM_TESTSUITE_SLURMRESTD_URL" in os.environ:
properties["slurmrestd_url"] = os.environ["SLURM_TESTSUITE_SLURMRESTD_URL"]
# Setup auth token
setup_slurmrestd_headers()
# Check version is the expected one
if not is_slurmrestd_running():
pytest.skip(
f"This test needs slurmrestd running in SLURM_TESTSUITE_SLURMRESTD_URL but cannot connect with {os.environ['SLURM_TESTSUITE_SLURMRESTD_URL']}",
allow_module_level=True,
)
else:
pytest.skip(
"This test requires to start slurmrestd or SLURM_TESTSUITE_SLURMRESTD_URL",
allow_module_level=True,
)
def start_slurmrestd():
os.environ["SLURM_JWT"] = "daemon"
port = None
attempts = 0
if "slurm-logs-dir" not in properties:
properties["slurm-logs-dir"] = os.path.dirname(
get_config_parameter("SlurmctldLogFile", live=False, quiet=True)
)
properties["slurmrestd_log"] = open(
f"{properties['slurm-logs-dir']}/slurmrestd.log", "w"
)
if not properties["slurmrestd_log"]:
pytest.fail(
f"Unable to open slurmrestd log: {properties['slurm-logs-dir']}/slurmrestd.log"
)
while not port and attempts < 15:
if "slurmrestd_port" not in properties or attempts > 0:
port = get_open_port()
else:
port = properties["slurmrestd_port"]
attempts += 1
args = [
"slurmrestd",
"-a",
"jwt",
"-s",
properties["openapi_plugins"],
]
if properties["data_parsers"] is not None:
args.extend(["-d", properties["data_parsers"]])
args.append(f"localhost:{port}")
logging.debug(f"Trying to start slurmrestd: {args}")
properties["slurmrestd"] = subprocess.Popen(
args,
stdin=subprocess.DEVNULL,
stdout=properties["slurmrestd_log"],
stderr=properties["slurmrestd_log"],
)
s = None
for i in range(100):
if properties["slurmrestd"].poll():
break
try:
s = socket.create_connection(("localhost", port))
break
except Exception as e:
logging.debug(f"Unable to connect to port {port}: {e}")
time.sleep(1)
if s:
s.close()
break
logging.debug(f"slurmrestd accepting on port {port} but is still running")
properties["slurmrestd"].kill()
properties["slurmrestd"].wait()
port = None
if not port:
pytest.fail(f"Unable start slurmrestd after trying {attempts} different ports")
del os.environ["SLURM_JWT"]
properties["slurmrestd_port"] = port
properties["slurmrestd_url"] = f"http://localhost:{port}/"
# Setup auth token
setup_slurmrestd_headers()
# Check slurmrestd is up
if not is_slurmrestd_running():
pytest.fail("Slurmrestd not responding")
def setup_slurmrestd_headers():
# Create the headers with the token to connect later
token = (
run_command_output("scontrol token lifespan=600", fatal=True)
.replace("SLURM_JWT=", "")
.replace("\n", "")
)
if token == "":
logging.warning("unable to get auth/jwt token")
properties["slurmrestd-headers"] = {
"X-SLURM-USER-NAME": get_user_name(),
"X-SLURM-USER-TOKEN": token,
}
def get_user_name():
"""Returns the username of the current user.
Args:
None
Returns:
The username of the current user.
Example:
>>> get_user_name()
'john_doe'
"""
return pwd.getpwuid(os.getuid()).pw_name
def cancel_jobs(
job_list,
timeout=default_polling_timeout,
poll_interval=0.1,
**run_command_kwargs,
):
"""Cancels a list of jobs and waits for them to complete.
Args:
job_list (list): A list of job ids to cancel. All 0s will be ignored.
timeout (integer): Number of seconds to wait for jobs to be done before
timing out.
poll_interval (float): Number of seconds to wait between job state
polls.
Returns:
True if all jobs were successfully cancelled and completed within
the timeout period, False otherwise.
Example:
>>> cancel_jobs([1234, 5678], timeout=60, fatal=True)
True
>>> cancel_jobs([9876, 5432], timeout=30, fatal=False)
False
"""
# Get the quiet and fatal values but don't forward them
quiet = run_command_kwargs.pop("quiet", None)
fatal = run_command_kwargs.pop("fatal", None)
# Filter list to ignore job_ids being 0
job_list = [i for i in job_list if i != 0]
job_list_string = " ".join(str(i) for i in job_list)
if job_list_string == "":
return True
run_command(
f"scancel {job_list_string}", fatal=fatal, quiet=quiet, **run_command_kwargs
)
for t in timer(timeout=timeout, poll_interval=poll_interval):
jobs = get_jobs(quiet=True, use_json=True, **run_command_kwargs)
jobs_not_in_system, jobs_not_done = [], []
for job_id in job_list:
if job_id not in jobs:
jobs_not_in_system.append(job_id)
continue
job_state = jobs[job_id]["job_state"]
if not job_state:
pytest.fail(f"JobState info not found for job {job_id}: {jobs[job_id]}")
for state in job_state:
if not is_job_state_done(state):
jobs_not_done.append(job_id)
break
if jobs_not_in_system and not quiet:
logging.warning(
f"Cancelled job(s) {jobs_not_in_system} not anymore in the system"
)
if not jobs_not_done:
if not quiet:
logging.debug(f"Jobs {job_list} successfully cancelled")
return True
if not quiet:
logging.debug(f"Cancelled job(s) {jobs_not_done} still not DONE")
else:
if fatal:
pytest.fail("Unable to cancel all jobs")
return False
def cancel_all_jobs(
timeout=default_polling_timeout, poll_interval=0.1, fatal=False, quiet=False
):
"""Cancels all jobs in the system, and waits for them to be cancelled.
Args:
fatal (boolean): If True, a timeout will result in the test failing.
timeout (integer): If timeout number of seconds expires before the
jobs are verified to be cancelled, fail.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
True if all jobs were successfully cancelled and completed within
the timeout period, False otherwise.
Example:
>>> cancel_all_jobs(timeout=60, fatal=True)
True
>>> cancel_all_jobs(timeout=30, fatal=False)
False
"""
jobs = get_jobs(quiet=True, use_json=True, user=properties["slurm-user"])
if not jobs:
logging.debug("No jobs to cancel")
return True
return cancel_jobs(
jobs.keys(),
fatal=fatal,
quiet=quiet,
timeout=timeout,
poll_interval=poll_interval,
user=properties["slurm-user"],
)
def is_integer(value):
try:
int(value)
except ValueError:
return False
else:
return True
def is_float(value):
try:
float(value)
except ValueError:
return False
else:
return True
def get_nodes(live=True, quiet=False, **run_command_kwargs):
"""Returns the node configuration as a dictionary of dictionaries.
If the live argument is not True, the dictionary will contain the literal
configuration values and the DEFAULT node will be separately indexed.
Args:
live (boolean):
If True, the node configuration information is obtained via a query
to the slurmctld daemon (e.g. scontrol show config).
If False, the node configuration information is obtained by
directly parsing the Slurm configuration file (slurm.conf).
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
A dictionary of dictionaries where the first level keys are the node
names and with their values being a dictionary of configuration
parameters for the respective node.
Example:
>>> get_nodes()
{'node1': {'NodeName': 'node1', 'Arch': 'x86_64', 'CoresPerSocket': 1, 'CPUAlloc': 0, 'CPUEfctv': 1, 'CPUTot': 1, 'CPULoad': 91.4, 'AvailableFeatures': None, 'ActiveFeatures': None, 'Gres': None, 'NodeAddr': 'slurm-host', 'NodeHostName': 'slurm-host', 'Port': 6821, 'Version': '24.05.0-0rc1', ...}}
>>> get_nodes(live=False, quiet=True)
{'node1': {'NodeHostname': 'slurm-host', 'Port': 6821, 'NodeName': 'node1'}}
"""
nodes_dict = {}
if live:
# TODO: Remove extra debug info for t22858 instead of fatal
run_command_kwargs.pop("fatal", None)
result = run_command(
"scontrol show nodes --json -F",
fatal=False,
quiet=quiet,
**run_command_kwargs,
)
if result["exit_code"]:
logging.debug(
"Fatal failure of 'scontrol show nodes', probably due 'Unable to contact slurm controller' (t22858)"
)
logging.debug("Getting gcore from slurmctld before fatal.")
gcore("slurmctld")
pytest.fail(
f"Command 'scontrol show nodes -oF' failed with rc={result['exit_code']}: {result['stderr']}"
)
output = result["stdout"]
node_dict_json = json.loads(output)["nodes"]
node_dict = {}
for node in node_dict_json:
# Add the node dictionary to the nodes dictionary
nodes_dict[node["name"]] = node
else:
# Get the config dictionary
config_dict = get_config(live=False, quiet=quiet)
# Convert keys to lower case so we can do a case-insensitive search
lower_config_dict = dict(
(key.lower(), value) for key, value in config_dict.items()
)
# DEFAULT will be included separately
if "nodename" in lower_config_dict:
for node_expression, node_expression_dict in lower_config_dict[
"nodename"
].items():
port_expression = (
node_expression_dict["Port"]
if "Port" in node_expression_dict
else ""
)
# Break up the node expression and port expression into lists
node_list = node_range_to_list(node_expression)
port_list = range_to_list(port_expression)
# Iterate over the nodes in the expression
for node_index in range(len(node_list)):
node_name = node_list[node_index]
# Add the parameters to the temporary node dictionary
node_dict = dict(node_expression_dict)
node_dict["NodeName"] = node_name
if node_index < len(port_list):
node_dict["Port"] = int(port_list[node_index])
# Add the node dictionary to the nodes dictionary
nodes_dict[node_name] = node_dict
return nodes_dict
def get_node_parameter(
node_name, parameter_name, default=None, live=True, **run_command_kwargs
):
"""Obtains the value for a node configuration parameter.
This function accepts the same keyword arguments as run_command
(e.g. user='slurm').
Args:
node_name (string): The node name.
parameter_name (string): The parameter name.
default (string or None): This value is returned if the parameter
is not found.
live (boolean):
If True, the node configuration information is obtained via a query
to the slurmctld daemon (e.g. scontrol show config).
If False, the node configuration information is obtained by
directly parsing the Slurm configuration file (slurm.conf).
Returns:
The value of the specified node parameter, or the default if not found.
Example:
>>> get_node_parameter('node1', 'State')
'IDLE'
>>> get_node_parameter('node2', 'RealMemory', default=4)
1
>>> get_node_parameter('node3', 'Partitions', default='primary', live=False)
'primary'
"""
nodes_dict = get_nodes(live=live, quiet=True, **run_command_kwargs)
if node_name in nodes_dict:
node_dict = nodes_dict[node_name]
else:
pytest.fail(f"Node ({node_name}) was not found in the node configuration")
if parameter_name in node_dict:
return node_dict[parameter_name]
else:
return default
def get_node_detail(node_name, quiet=True):
"""Returns the output of "scontrol show node <node> -d".
The detailed (-d) form is the only one that carries the GRES details and
the drained TRES; a plain "scontrol show node" (and therefore get_nodes()
and get_node_parameter()) omits them because gathering them is expensive.
Args:
node_name (string): The node name.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
The command output as a string.
Example:
>>> get_node_detail('node1')
'NodeName=node1 Arch=x86_64 ... DrainedTRES=gpu(IDX:0)(ecc)\\n...'
"""
return run_command_output(
f"scontrol show node {node_name} -d",
user=properties["slurm-user"],
quiet=quiet,
)
def get_node_tres_drained(node_name, quiet=True):
"""Returns a node's DrainedTRES= value, or None when nothing is drained.
The field is omitted entirely by "scontrol show node -d" when the node has
no drained TRES, so None (rather than an empty string) means "nothing
drained".
Args:
node_name (string): The node name.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
The DrainedTRES value as a string, or None if the node has none.
Example:
>>> get_node_tres_drained('node1')
'gpu(IDX:0)(ecc),cpu(IDX:2)(flaky)'
"""
# A reason can contain spaces, so take the rest of the line.
match = re.search(
r"^\s*DrainedTRES=(.*?)\s*$",
get_node_detail(node_name, quiet=quiet),
re.MULTILINE,
)
return match.group(1) if match else None
def update_tres_drain(
node_name,
tres_name,
selector="",
action="Drain",
reason=None,
fatal=False,
quiet=False,
user=None,
):
"""Drains or resumes a node-local TRES with "scontrol update TresName=".
Args:
node_name (string): The node the TRES lives on.
tres_name (string): The TRES to act on ("gres/gpu", "gres/gpu:a100",
"cpu" or "mem").
selector (string): The TRES to act on, as the scontrol field it is
given with (e.g. "Index=0,2", "Index=ALL", "Count=512",
"Uuid=GPU-..."). An empty string omits the selector, which is
useful to check that it is required.
action (string): "Drain", "Resume" or "SetDrain".
reason (string): Why the TRES are being drained. None omits Reason,
which is required when draining and refused when resuming.
fatal (boolean): If True, the test fails if the command fails.
quiet (boolean): If True, logging is performed at the TRACE log level.
user (string): The user to run scontrol as. Defaults to SlurmUser.
Returns:
The run_command() result dictionary, so the caller can assert on the
exit code of a request it expects to be refused.
Example:
>>> update_tres_drain('node1', 'gres/gpu', 'Index=0', reason='ecc')
>>> update_tres_drain('node1', 'cpu', 'Index=0-1', action='Resume')
"""
command = f"scontrol update TresName={tres_name} NodeName={node_name}"
if selector:
command += f" {selector}"
command += f" Action={action}"
if reason is not None:
command += f" Reason={shlex.quote(reason)}"
return run_command(
command, user=user or properties["slurm-user"], fatal=fatal, quiet=quiet
)
def fake_gpus_conf(*devices, gpu_type="a100"):
"""Builds a fake_gpus.conf body for the given devices.
Used with gres.conf "Autodetect=full". slurmd numbers the devices in
device file order, not in the order of the lines, so index i is devices[i]
only when the ttys are listed in ascending order.
Args:
*devices: One tuple per device. Either ``(tty, uuid)`` or
``(tty, gpu_type, uuid)``, where tty is the /dev/ttyN number that
backs the fake device and uuid is what it reports (use "(null)"
for a device that reports none).
gpu_type (string): The GPU type used for two-element device tuples.
Returns:
The file content as a string.
Example:
>>> fake_gpus_conf((0, "GPU-aaa"), (1, "GPU-bbb"))
'a100|2|(null)|(null)|/dev/tty0|GPU-aaa|nvidia_gpu_env\\na100|...'
>>> fake_gpus_conf((0, "h100", "GPU-ccc"))
'h100|2|(null)|(null)|/dev/tty0|GPU-ccc|nvidia_gpu_env\\n'
"""
content = ""
for device in devices:
if len(device) == 2:
tty, uuid = device
device_type = gpu_type
else:
tty, device_type, uuid = device
content += (
f"{device_type}|2|(null)|(null)|/dev/tty{tty}|{uuid}|nvidia_gpu_env\n"
)
return content
def get_node_field(node_name, field, quiet=True):
"""Returns a field of "scontrol show node -d" as its literal text.
State is best read this way: the JSON node state is a list of exact flag
names, so a substring check such as "INVAL" not in state would always pass
against it.
Args:
node_name (string): The node name.
field (string): The field name, e.g. "State".
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
The field value as a string, or None if the node has no such field.
Example:
>>> get_node_field('node1', 'State')
'IDLE+DEGRADED'
"""
match = re.search(
rf"\b{re.escape(field)}=(\S+)", get_node_detail(node_name, quiet=quiet)
)
return match.group(1) if match else None
def restart_slurmd_with_fake_gpus(node_name, *devices, timeout=60):
"""Rewrites fake_gpus.conf to exactly ``devices`` and restarts only slurmd.
slurmd autodetects the new device set and registers with the running
controller. The node's configured Gres is left alone, so a shorter device
set registers as fewer devices than configured. Fails the test unless the
node then settles neither DOWN nor INVAL.
Args:
node_name (string): The node whose slurmd to restart.
*devices: The devices, as for fake_gpus_conf().
timeout (integer): Seconds to wait for the node to settle.
Returns:
None
Example:
>>> restart_slurmd_with_fake_gpus('node1', (0, 'GPU-aaa'), (2, 'GPU-ccc'))
"""
require_config_file("fake_gpus.conf", fake_gpus_conf(*devices))
start_slurmd(node_name)
for _ in timer(timeout=timeout, fatal=True):
state = get_node_field(node_name, "State")
if state is not None and "DOWN" not in state and "INVAL" not in state:
break
def gpu_gres_conf(node_name, count=2, gpu_type="a100"):
"""Builds a gres.conf body giving one node autodetected GPUs.
The devices are backed by /dev/tty0..N-1, so they line up with the fake
devices of a fake_gpus_conf() written for the same count. Naming the node
keeps the GPUs off any other node in the cluster.
Args:
node_name (string): The node the GPUs belong to.
count (integer): How many GPUs the node has.
gpu_type (string): The GPU type the devices are configured with.
Returns:
The file content as a string.
Example:
>>> gpu_gres_conf("node100", count=1)
'Autodetect=full\\nNodeName=node100 Name=gpu File=/dev/tty0 Type=a100\\n'
"""
content = "Autodetect=full\n"
for tty in range(count):
content += f"NodeName={node_name} Name=gpu File=/dev/tty{tty} Type={gpu_type}\n"
return content
def set_gpu_set(node_name, fake_gpus, count):
"""Rewrites fake_gpus.conf and the node's configured Gres, then restarts.
set_node_parameter() rewrites the NodeName line and restarts Slurm with
clean=False, so the controller restores the node, including any saved
UUID order. gres.conf is
left alone (typically "Autodetect=full"); only the fake device set and the
configured count change, kept equal so config_overrides never marks the
node INVALID.
This function may only be used in auto-config mode.
Args:
node_name (string): The static node whose device set is changing.
fake_gpus (string): The new fake_gpus.conf content (see
fake_gpus_conf()).
count (integer): The number of devices, used as the configured
"gpu:a100:<count>".
Returns:
None
Example:
>>> set_gpu_set('node1', fake_gpus_conf((1, GPU_B), (2, GPU_C)), 2)
"""
require_config_file("fake_gpus.conf", fake_gpus)
set_node_parameter(node_name, "Gres", f"gpu:a100:{count}")
def get_node_events(node_name):
"""Returns a node's accounting events as raw '|'-separated lines.
An open (active) event has an End field of "Unknown"; a closed event has an
End time. Every event is returned, including one opened in the current
second: "start=" makes sacctmgr ask the database for "time_start < now",
which hides an event until the clock ticks past the second it was opened
in, and that is a flake source for anything polling for a fresh event. The
bare "all_time" keyword drops the time predicate entirely instead.
The queried format is nodename|reason|tres|tresreference|end|state|start.
The state and start fields are last so the earlier indexes keep their
meaning.
Args:
node_name (string): The node name.
Returns:
The sacctmgr output as a string.
"""
command = (
f"sacctmgr -n -P list events all_time Event=Node Nodes={node_name} "
"format=nodename,reason,tres,tresreference,end,state,start"
)
return run_command_output(
command,
user=properties["slurm-user"],
fatal=True,
quiet=True,
)
def get_open_node_event(node_name):
"""Returns (reason, tres, tres_reference) of a node's open (active) event.
A node has one event open at a time, whatever took its resources away: a
whole-node event covering a Down or Drain, or a drained TRES event covering
the TRES drained on it. Opening either closes the other, so there is never
more than one to choose between.
Args:
node_name (string): The node name.
Returns:
A (reason, tres, tres_reference) tuple, or None if no event is open.
"""
for parts in _open_node_events(node_name):
return parts[1], parts[2], parts[3]
return None
def _open_node_events(node_name):
"""Yields the split fields of a node's open events, in query order.
Args:
node_name (string): The node name.
"""
for line in (get_node_events(node_name) or "").splitlines():
parts = line.split("|")
if len(parts) < 6 or parts[0] != node_name:
continue
if parts[4].strip() != "Unknown":
continue
yield parts
def get_open_node_event_reason(node_name):
"""Returns the Reason column of a node's open (active) event, or None."""
event = get_open_node_event(node_name)
return event[0] if event else None
def get_open_node_event_tres(node_name):
"""Returns the TRES column of a node's open (active) event, or None."""
event = get_open_node_event(node_name)
return event[1] if event else None
def get_open_node_event_tres_reference(node_name):
"""Returns the TresReference column of a node's open event, or None."""
event = get_open_node_event(node_name)
return event[2] if event else None
def get_open_node_event_count(node_name):
"""Returns how many of a node's accounting events are open (active).
A node has one event open at a time: opening one closes whatever was open
for it. More than one open at once would double count, since the rollup
adds up every event row overlapping the hour.
Args:
node_name (string): The node name.
Returns:
The number of open events (int).
"""
return len(list(_open_node_events(node_name)))
def get_overlapping_node_events(node_name):
"""Returns a node's accounting events that overlap one another in time.
A node has one event open at a time and each new one closes the last, so
none of a node's events should overlap at all. Any that do would double
count, since the rollup adds up every event row overlapping the hour.
Event times are printed as YYYY-MM-DDTHH:MM:SS, which sorts correctly as a
string. An open event ("Unknown" end) extends to infinity.
Args:
node_name (string): The node name.
Returns:
A list of (start, end) tuples that start before another event ended.
Empty when they are disjoint.
"""
intervals = []
for line in (get_node_events(node_name) or "").splitlines():
parts = line.split("|")
if len(parts) < 7 or parts[0] != node_name:
continue
end = parts[4].strip()
intervals.append((parts[6].strip(), "9999" if end == "Unknown" else end))
overlapping = []
prev_end = None
for start, end in sorted(intervals):
if prev_end is not None and start < prev_end:
overlapping.append((start, end))
# Events are sorted by start, so the running bound is the latest end
# seen: a long event followed by a short one still has to be caught.
prev_end = max(prev_end, end) if prev_end is not None else end
return overlapping
def get_dbd_agent_queue_size(quiet=True):
"""Returns how many messages slurmctld still has queued for slurmdbd.
Args:
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
The "DBD Agent queue size" sdiag reports (int).
Example:
>>> get_dbd_agent_queue_size()
0
"""
output = run_command_output("sdiag", quiet=quiet, fatal=True)
match = re.search(r"DBD Agent queue size:\s*(\d+)", output)
if match is None:
pytest.fail(f"sdiag reported no DBD agent queue size:\n{output}")
return int(match.group(1))
def wait_for_dbd_agent_drained(timeout=default_polling_timeout, fatal=False):
"""Waits until slurmctld has nothing left queued for slurmdbd.
Messages such as node events reach the database only when the agent
delivers them. Two empty readings in a row are required, so a queue that
is only momentarily empty does not pass.
Args:
timeout (integer): The number of seconds to wait before timing out.
fatal (boolean): If True, a timeout will cause the test to fail.
Returns:
True if the queue drained before the timeout, False otherwise.
Example:
>>> wait_for_dbd_agent_drained(fatal=True)
True
"""
empty_polls = 0
for _ in timer(timeout=timeout, fatal=fatal):
empty_polls = empty_polls + 1 if get_dbd_agent_queue_size() == 0 else 0
if empty_polls == 2:
return True
return False
def set_node_parameter(node_name, new_parameter_name, new_parameter_value):
"""Sets the value of the specified node configuration parameter.
This function sets a node property for the specified node and restarts
the relevant slurm daemons. A backup is automatically created and the
original configuration is restored after the test completes.
This function may only be used in auto-config mode.
Args:
node_name (string): The node name.
new_parameter_name (string): The parameter name.
new_parameter_value (string): The parameter value.
Use a value of None to unset a node parameter.
Returns:
None
Example:
>>> set_node_parameter('node1', 'Features', 'f1')
"""
if not properties["auto-config"]:
require_auto_config("wants to modify node parameters")
config_file = f"{properties['slurm-config-dir']}/slurm.conf"
# Read the original slurm.conf into a list of lines
output = run_command_output(
f"cat {config_file}", user=properties["slurm-user"], quiet=True
)
original_config_lines = output.splitlines()
new_config_lines = original_config_lines.copy()
# Locate the node among the various NodeName definitions
found_node_name = False
for line_index in range(len(original_config_lines)):
line = original_config_lines[line_index]
words = re.split(r" +", line.strip())
if not words[0]:
continue
if words[0][0] == "#":
continue
parameter_name, parameter_value = words[0].split("=", 1)
if parameter_name.lower() != "nodename":
continue
# We found a NodeName line. Read in the node parameters
node_expression = parameter_value
port_expression = ""
original_node_parameters = collections.OrderedDict()
for word in words[1:]:
parameter_name, parameter_value = word.split("=", 1)
if parameter_name.lower() == "port":
port_expression = parameter_value
else:
original_node_parameters[parameter_name] = parameter_value
node_list = node_range_to_list(node_expression)
port_list = range_to_list(port_expression)
# Determine whether our node is included in this node expression
if node_name in node_list:
# Yes. We found the node
found_node_name = True
node_index = node_list.index(node_name)
# Delete the original node definition
new_config_lines.pop(line_index)
# Add the modified definition for the specified node
modified_node_parameters = original_node_parameters.copy()
if new_parameter_value is None:
if new_parameter_name in modified_node_parameters:
del modified_node_parameters[new_parameter_name]
else:
modified_node_parameters[new_parameter_name] = new_parameter_value
modified_node_line = f"NodeName={node_name}"
if node_index < len(port_list):
modified_node_line += f" Port={port_list[node_index]}"
for parameter_name, parameter_value in modified_node_parameters.items():
modified_node_line += f" {parameter_name}={parameter_value}"
new_config_lines.insert(line_index, modified_node_line)
# If the node was part of an aggregate node definition
node_list.remove(node_name)
if len(node_list):
# Write the remainder of the aggregate using the original attributes
remainder_node_expression = node_list_to_range(node_list)
remainder_node_line = f"NodeName={remainder_node_expression}"
if node_index < len(port_list):
port_list.pop(node_index)
if port_list:
remainder_port_expression = list_to_range(port_list)
remainder_node_line += f" Port={remainder_port_expression}"
for parameter_name, parameter_value in original_node_parameters.items():
remainder_node_line += f" {parameter_name}={parameter_value}"
new_config_lines.insert(line_index, remainder_node_line)
break
if not found_node_name:
pytest.fail(
f"Invalid node specified in set_node_parameter(). Node ({node_name}) does not exist"
)
# Write the config file back out with the modifications
new_config_string = "\n".join(new_config_lines)
run_command(
f"echo '{new_config_string}' > {config_file}",
user=properties["slurm-user"],
fatal=True,
quiet=True,
)
# Restart slurm controller if it is already running
if is_slurmctld_running(quiet=True):
restart_slurm(quiet=True)
def get_reservations(quiet=False, **run_command_kwargs):
"""Returns the reservations as a dictionary of dictionaries.
Args:
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
A dictionary of dictionaries where the first level keys are the
reservation names and with their values being a dictionary of
configuration parameters for the respective reservation.
Example:
>>> get_reservations()
{'resv1': {'ReservationName': 'resv1', 'StartTime': '2024-04-06T00:00:17', 'EndTime': '2024-04-06T00:15:17', 'Duration': '00:15:00', 'Nodes': 'node1', 'NodeCnt': 1, 'CoreCnt': 1, 'Features': None, 'PartitionName': 'primary', 'Flags': 'DAILY,REPLACE_DOWN,PURGE_COMP=00:01:00', 'TRES': 'cpu=1', 'Users': 'atf', 'Groups': None, 'Accounts': None, 'Licenses': None, 'State': 'INACTIVE', 'BurstBuffer': None, 'MaxStartDelay': None}}
"""
resvs_dict = {}
resv_dict = {}
output = run_command_output(
"scontrol show reservations -o", quiet=quiet, **run_command_kwargs
)
for line in output.splitlines():
if line == "":
continue
while match := re.search(r"^ *([^ =]+)=(.*?)(?= +[^ =]+=| *$)", line):
parameter_name, parameter_value = match.group(1), match.group(2)
# Remove the consumed parameter from the line
line = re.sub(r"^ *([^ =]+)=(.*?)(?= +[^ =]+=| *$)", "", line)
# Reformat the value if necessary
if is_integer(parameter_value):
parameter_value = int(parameter_value)
elif is_float(parameter_value):
parameter_value = float(parameter_value)
elif parameter_value == "(null)":
parameter_value = None
# Add it to the temporary resv dictionary
resv_dict[parameter_name] = parameter_value
if "ReservationName" in resv_dict:
# Add the resv dictionary to the resvs dictionary
resvs_dict[resv_dict["ReservationName"]] = resv_dict
# Clear the resv dictionary for use by the next resv
resv_dict = {}
return resvs_dict
def get_reservation_parameter(resv_name, parameter_name, default=None):
"""Obtains the value for a reservation configuration parameter.
Args:
resv_name (string): The reservation name.
parameter_name (string): The parameter name.
default (string or None): This value is returned if the parameter
is not found.
Returns:
The value of the specified reservation parameter, or the default if not found.
Example:
>>> get_reservation_parameter('resv1', 'PartitionName')
'primary'
>>> get_reservation_parameter('resv2', 'EndTime', default='2024-04-06T00:15:17')
'2024-04-06T00:15:17'
"""
resvs_dict = get_reservations()
if resv_name in resvs_dict:
resv_dict = resvs_dict[resv_name]
else:
pytest.fail(f"reservation ({resv_name}) was not found")
if parameter_name in resv_dict:
return resv_dict[parameter_name]
else:
return default
def is_super_user():
"""Checks if the current user is a super user.
Args:
None
Returns:
True if the current user is a super user, False otherwise.
Example:
>>> is_super_user()
False
"""
uid = os.getuid()
if uid == 0:
return True
user = pwd.getpwuid(uid)[0]
if get_config_parameter("SlurmUser") == user:
return True
return False
def require_sudo_rights():
"""Skips the test if the test user does not have unprompted sudo privileges.
Args:
None
Returns:
None
Example:
>>> require_sudo_rights()
"""
if not properties["sudo-rights"]:
pytest.skip(
"This test requires the test user to have unprompted sudo privileges",
allow_module_level=True,
)
def submit_job_sbatch(sbatch_args='--wrap "sleep 60"', **run_command_kwargs):
"""Submits a job using sbatch and returns the job id.
The submitted job will automatically be cancelled when the test ends.
Args:
sbatch_args (string): The arguments to sbatch.
Returns:
The job id.
Note:
The background parameter is ignored and always treated as False, since
sbatch is already non-blocking and the function needs to capture the
job ID from output.
Example:
>>> submit_job_sbatch('--wrap "echo Hello"')
1234
>>> submit_job_sbatch('-J myjob --wrap "echo World"', fatal=True)
5678
"""
# Since sbatch is non-blocking anyway, make sure we're not in the background
run_command_kwargs["background"] = False
output = run_command_output(f"sbatch {sbatch_args}", **run_command_kwargs)
if match := re.search(r"Submitted \S+ job (\d+)", output):
job_id = int(match.group(1))
properties["submitted-jobs"].append(job_id)
return job_id
else:
return 0
# Returns results
def run_job(srun_args, **run_command_kwargs):
"""Runs a job using srun and returns the run_command results dictionary.
If the srun command times out, it will automatically be cancelled when the
test ends.
Args:
srun_args (string): The arguments to srun.
Returns:
The srun run_command results dictionary.
Example:
>>> run_job('-n 4 --output=output.txt ./my_executable', timeout=60)
{'command': 'srun -n 4 --output=output.txt ./my_executable', 'start_time': 1712276708.827, 'duration': 60.0, 'exit_code': 110, 'stdout': '', 'stderr': 'srun: Requested partition configuration not available now\nsrun: job 15 queued and waiting for resources\n'}
>>> run_job('-n 1 --error=error.txt ./my_executable', fatal=True)
{'command': 'srun -n 1 --error=error.txt my_executable', 'start_time': 1712277798.016, 'duration': 0.06, 'exit_code': 0, 'stdout': 'foo bar\n', 'stderr': ''}
"""
results = run_command(f"srun {srun_args}", **run_command_kwargs)
return results
# Return exit code
def run_job_exit(srun_args, **run_command_kwargs):
"""Runs a job using srun and returns the exit code.
If the srun command times out, it will automatically be cancelled when the
test ends.
Args:
srun_args (string): The arguments to srun.
Returns:
The exit code from srun.
Example:
>>> run_job_exit('-n 4 --output=output.txt ./my_executable', timeout=60)
2
>>> run_job_exit('-n 2 --error=error.txt ./my_executable', fatal=True)
0
"""
results = run_job(srun_args, **run_command_kwargs)
return results["exit_code"]
# Return output
def run_job_output(srun_args, **run_command_kwargs):
"""Runs a job using srun and returns the standard output.
If the srun command times out, it will automatically be cancelled when the
test ends.
Args:
srun_args (string): The arguments to srun.
Returns:
The standard output from srun.
Example:
>>> run_job_output('-n 4 ./my_executable', timeout=60)
'stdout of the command'
>>> run_job_output('-n 2 --output=output.txt ./my_executable', fatal=True)
''
"""
results = run_job(srun_args, **run_command_kwargs)
return results["stdout"]
# Return error
def run_job_error(srun_args, **run_command_kwargs):
"""Runs a job using srun and returns the standard error.
If the srun command times out, it will automatically be cancelled when the
test ends.
Args:
srun_args (string): The arguments to srun.
Returns:
The standard error from srun.
Example:
>>> run_job_error('-n 4 --output=output.txt ./my_executable', timeout=60)
'stderr of the command'
>>> run_job_error('-n 200000 ./my_executable', fatal=True) # Will automatically fail the test due to resources
"""
results = run_job(srun_args, **run_command_kwargs)
return results["stderr"]
# Return job id
def submit_job_srun(srun_args, return_process=False, **run_command_kwargs):
"""Runs a job using srun and returns the job id.
This function obtains the job id by adding the -v option to srun and parsing
out the job id. If the srun command times out, it will automatically be
cancelled when the test ends.
Args:
srun_args (string): The arguments to srun.
return_process (boolean): If True, with background=True, also return
the srun process, so the caller can keep reading its stderr after
the job id.
Returns:
The job id (int) or 0 if failed, or the tuple (job id, process) with
return_process.
Example:
>>> submit_job_srun('-n 4 --output=output.txt ./my_executable')
12345
>>> submit_job_srun('-n 2 --output=output.txt ./my_executable', timeout=60)
67890
>>> submit_job_srun('-n 2 --output=output.txt ./my_executable', background=True)
34567
>>> job_id, process = submit_job_srun('./my_executable', background=True, return_process=True)
"""
if return_process and not run_command_kwargs.get("background", False):
pytest.fail("submit_job_srun(): return_process requires background=True")
verbose_arg = "-v"
if run_command_kwargs.get("background", False):
# Use timeout for finding job id
timeout = run_command_kwargs.pop("timeout", 30)
# Extract fatal/xfail to handle at job submission level
fatal = run_command_kwargs.pop("fatal", False)
xfail = run_command_kwargs.pop("xfail", False)
if get_version("bin/srun") < (25, 11):
# Older versions may not print the jobid with -v until job is scheduled
verbose_arg = "--begin=now+1"
logging.warning(
"Using {verbose_arg} to be able to extract jobid when using srun background"
)
results = run_command(f"srun {verbose_arg} {srun_args}", **run_command_kwargs)
if run_command_kwargs.get("background", False):
job_id = None
start_time = time.time()
while time.time() - start_time < timeout:
line = _readline_with_timeout(results["process"].stderr, timeout=5)
if line is None:
# Timeout or EOF
if results["process"].poll() is not None:
break
continue
if match := re.search(r"job(id)?[ =](\d+)", line, re.IGNORECASE):
job_id = int(match.group(2))
break
if results["process"].poll() is not None:
break
else:
if match := re.search(r"job(id)?[ =](\d+)", results["stderr"], re.IGNORECASE):
job_id = int(match.group(2))
else:
job_id = None
if job_id is not None:
properties["submitted-jobs"].append(job_id)
# Handle fatal/xfail for background jobs
if run_command_kwargs.get("background", False):
if fatal:
if xfail and job_id is not None:
pytest.fail(
f'Command "srun -v {srun_args}" was expected to fail but got job id {job_id}'
)
elif not xfail and job_id is None:
pytest.fail(f'Command "srun -v {srun_args}" failed to get job id')
job_id = job_id if job_id else 0
if return_process:
return job_id, results["process"]
return job_id
# Reads in progress (thread and result) by stream, see _readline_with_timeout()
_pending_readlines = {}
def _readline_with_timeout(stream, timeout=default_command_timeout):
"""Read a line from a stream with a timeout.
A read that times out is not abandoned, as it would still be blocked in
readline() and would swallow the next line. It is kept in progress and
the next call for the same stream waits for it, so no line is lost.
Args:
stream: The stream to read from (e.g., process.stderr)
timeout: Maximum seconds to wait for a line
Returns:
The line read from the stream, or None if timeout/EOF
"""
if stream not in _pending_readlines:
result = [None]
def read_line():
try:
result[0] = stream.readline()
except Exception:
pass
thread = threading.Thread(target=read_line)
thread.daemon = True
thread.start()
_pending_readlines[stream] = (thread, result)
thread, result = _pending_readlines[stream]
thread.join(timeout)
if thread.is_alive():
# Timeout occurred
return None
del _pending_readlines[stream]
return result[0]
# Return job id (command should not be interactive/shell)
def submit_job_salloc(salloc_args, **run_command_kwargs):
"""Submits a job using salloc and returns the job id.
The submitted job will automatically be cancelled when the test ends.
Args:
salloc_args (string): The arguments to salloc.
Returns:
The job id (int) or 0 if failed.
Example:
>>> submit_job_salloc('-N 1 -t 60 --output=output.txt ./my_executable')
12345
>>> submit_job_salloc('-N 2 -t 120 --output=output.txt ./my_executable', timeout=60)
67890
>>> submit_job_salloc('-N 2 -t 120 --output=output.txt ./my_executable', background=True)
34567
"""
# If background=True, use timeout for finding job id
if run_command_kwargs.get("background", False):
timeout = run_command_kwargs.pop("timeout", 30)
# Extract fatal/xfail to handle at job submission level
fatal = run_command_kwargs.pop("fatal", False)
xfail = run_command_kwargs.pop("xfail", False)
results = run_command(f"salloc {salloc_args}", **run_command_kwargs)
if run_command_kwargs.get("background", False):
job_id = None
start_time = time.time()
while time.time() - start_time < timeout:
line = _readline_with_timeout(results["process"].stderr, timeout=5)
if line is None:
# Timeout or EOF
if results["process"].poll() is not None:
break
continue
if match := re.search(r"job allocation (\d+)", line):
job_id = int(match.group(1))
break
if results["process"].poll() is not None:
break
else:
if match := re.search(r"Granted job allocation (\d+)", results["stderr"]):
job_id = int(match.group(1))
else:
job_id = None
if job_id is not None:
properties["submitted-jobs"].append(job_id)
# Handle fatal/xfail for background jobs
if run_command_kwargs.get("background", False):
if fatal:
if xfail and job_id is not None:
pytest.fail(
f'Command "salloc {salloc_args}" was expected to fail but got job id {job_id}'
)
elif not xfail and job_id is None:
pytest.fail(f'Command "salloc {salloc_args}" failed to get job id')
return job_id if job_id else 0
# Return job id
def submit_job(command, job_param, job, *, wrap_job=True, **run_command_kwargs):
"""Submits a job using the given command and returns the job id.
Args:
command (string): The command to submit the job (salloc, srun, sbatch).
job_param (string): The arguments to the job.
job (string): The command or job file to be executed.
wrap_job (boolean): If True, the job will be wrapped when the command is 'sbatch'.
Returns:
The job id.
Example:
>>> submit_job('salloc', '-N 1 -t 60', './my_executable', quiet=True)
12345
>>> submit_job('srun', '-N 2 -t 120', './my_executable', quiet=True)
67890
>>> submit_job('sbatch', '-N 1 -t 60', './my_script.sh', wrap_job=False, quiet=True)
23456
"""
# Make sure command is a legal command to run a job
assert command in [
"salloc",
"srun",
"sbatch",
], f"Invalid command '{command}'. Should be salloc, srun, or sbatch."
if command == "salloc":
return submit_job_salloc(f"{job_param} {job}", **run_command_kwargs)
elif command == "srun":
return submit_job_srun(f"{job_param} {job}", **run_command_kwargs)
elif command == "sbatch":
# If the job should be wrapped, do so before submitting
if wrap_job:
job = f"--wrap '{job}'"
return submit_job_sbatch(f"{job_param} {job}", **run_command_kwargs)
def run_job_nodes(srun_args, **run_command_kwargs):
"""Runs a job using srun and returns the allocated node list.
This function obtains the job id by adding the -v option to srun and parsing
out the allocated node list. If the srun command times out, it will
automatically be cancelled when the test ends.
Args:
srun_args (string): The arguments to srun.
Returns:
The allocated node list for the job.
Example:
>>> run_job_nodes('-N 2 hostname', quiet=True)
['node001', 'node002']
>>> run_job_nodes('-N 1 --exclude=node001 hostname', quiet=True)
['node002']
"""
results = run_command(f"srun -v {srun_args}", **run_command_kwargs)
node_list = []
if results["exit_code"] == 0:
if match := re.search(r"jobid \d+: nodes\(\d+\):`([^']+)'", results["stderr"]):
node_list = node_range_to_list(match.group(1))
return node_list
def get_qos(name=None, **run_command_kwargs):
"""Returns the QOSes in the system as a dictionary of dictionaries.
Args:
name: The name of a specific QOS of which to get parameters.
Returns:
A dictionary of dictionaries where the first level keys are the QOSes names
and with their values being a dictionary of configuration parameters for
the respective QOS.
"""
command = "sacctmgr --json show qos"
if id is not None:
command += f" {name}"
output = run_command_output(command, **run_command_kwargs)
qos_dict = {}
qos_list = json.loads(output)["qos"]
for qos in qos_list:
qos_dict[qos["name"]] = qos
return qos_dict
def get_jobs(job_id=None, dbd=False, use_json=False, **run_command_kwargs):
"""Returns the jobs in the system as a dictionary of dictionaries.
Args:
job_id (integer): The id of a specific job of which to get parameters.
dbd (boolean): If True, obtain the jobs from sacct instead of scontrol.
Returns:
A dictionary of dictionaries where the first level keys are the job ids
and with their values being a dictionary of configuration parameters for
the respective job.
Example:
>>> get_jobs()
{38: {'JobId': 38, 'JobName': 'wrap', 'UserId': 'atf(1002)', 'GroupId': 'atf(1002)', ...},
39: {'JobId': 39, 'JobName': 'wrap', 'UserId': 'atf(1002)', 'GroupId': 'atf(1002)', ...}}
>>> get_jobs(job_id='12345', quiet=True)
{12345: {'JobId': '12345', 'JobName': 'foo.sh', 'UserId': 'test(1003)', ...}}
"""
jobs_dict = {}
if dbd:
command = "sacct --json -X"
if job_id is not None:
command += f" -j {job_id}"
output = run_command_output(command, **run_command_kwargs)
jobs_list = json.loads(output)["jobs"]
for job in jobs_list:
jobs_dict[job["job_id"]] = job
else:
# TODO: We should always use --json to properly parse Slurm parameters.
# This is still optional because using it means that the keys
# of the jobs_dict will be different in all existing tests.
if use_json:
command = "scontrol --json -d -a show jobs"
if job_id is not None:
command += f" {job_id}"
output = run_command_output(command, **run_command_kwargs)
try:
jobs_list = json.loads(output)["jobs"]
except json.JSONDecodeError as e:
msg = f"Error parsing scontrol output: {output}\n{e}"
if "fatal" in run_command_kwargs and run_command_kwargs["fatal"]:
pytest.fail(msg)
else:
logging.warning(msg)
return jobs_dict
for job in jobs_list:
jobs_dict[job["job_id"]] = job
return jobs_dict
command = "scontrol -d -o show jobs"
if job_id is not None:
command += f" {job_id}"
# TODO: Remove extra debug info for t22858 instead of forwarding fatal
run_command_kwargs.pop("fatal", None)
result = run_command(command, fatal=False, **run_command_kwargs)
if result["exit_code"]:
logging.debug(
f"Fatal failure of {command}, probably due 'Unable to contact slurm controller' (t22858)"
)
logging.debug("Getting gcore from slurmctld before fatal.")
gcore("slurmctld")
pytest.fail(
f"Command {command} failed with rc={result['exit_code']}: {result['stderr']}"
)
output = result["stdout"]
job_dict = {}
for line in output.splitlines():
if line == "":
continue
original_line = line
while match := re.search(r"^ *([^ =]+)=(.*?)(?= +[^ =]+=| *$)", line):
param_name, param_value = match.group(1), match.group(2)
# Remove the consumed parameter from the line
line = re.sub(r"^ *([^ =]+)=(.*?)(?= +[^ =]+=| *$)", "", line)
# Reformat the value if necessary
if is_integer(param_value):
param_value = int(param_value)
elif is_float(param_value):
param_value = float(param_value)
elif param_value == "(null)":
param_value = None
# Add it to the temporary job dictionary
job_dict[param_name] = param_value
# Add the job dictionary to the jobs dictionary
if "JobId" in job_dict:
jobs_dict[job_dict["JobId"]] = job_dict
# Clear the job dictionary for use by the next job
job_dict = {}
elif job_dict:
# A line carrying parameters but no JobId belongs to no job of
# its own, the way a detail line of the record before it does.
# Keeping it would attribute its parameters to the next job.
logging.debug(f"Ignoring job line without a JobId: {original_line}")
job_dict = {}
return jobs_dict
def get_steps(job_id, **run_command_kwargs):
"""Returns the steps of a given job_id as a dictionary of dictionaries.
Args:
job_id (string): The specific JobId to retrieve information for.
Returns:
A dictionary of dictionaries where the first level keys are the step ids
and with their values being a dictionary of configuration parameters for
the respective step.
Example:
>>> get_steps(44)
{'44.batch': {'StepId': '44.batch', 'UserId': 1002, 'StartTime': '2024-04-05T01:17:53', ...},
'44.0': {'StepId': 44.0, 'UserId': 1002, 'StartTime': '2024-04-05T01:17:54', ...}}
"""
steps_dict = {}
step_dict = {}
command = f"scontrol -d -o show steps {job_id}"
result = run_command(command, **run_command_kwargs)
if result["exit_code"]:
logging.debug("scontrol command failed, no steps returned")
return step_dict
output = result["stdout"]
for line in output.splitlines():
if line == "":
continue
while match := re.search(r"^ *([^ =]+)=(.*?)(?= +[^ =]+=| *$)", line):
param_name, param_value = match.group(1), match.group(2)
# Remove the consumed parameter from the line
line = re.sub(r"^ *([^ =]+)=(.*?)(?= +[^ =]+=| *$)", "", line)
# Reformat the value if necessary
if is_integer(param_value):
param_value = int(param_value)
elif is_float(param_value) and param_name != "StepId":
param_value = float(param_value)
elif param_value == "(null)":
param_value = None
# Add it to the temporary step dictionary
step_dict[param_name] = param_value
# Add the step dictionary to the steps dictionary
if step_dict:
steps_dict[str(step_dict["StepId"])] = step_dict
# Clear the step dictionary for use by the next step
step_dict = {}
return steps_dict
def get_job(job_id, quiet=False):
"""Returns job information for a specific job as a dictionary.
Args:
job_id (integer): The id of the job for which information is requested.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
A dictionary containing parameters for the specified job. If the job id
is not found, an empty dictionary is returned.
Example:
>>> get_job(51)
{'JobId': 51, 'JobName': 'wrap', 'UserId': 'atf(1002)', 'GroupId': 'atf(1002)', ...}
>>> get_job(182, quiet=True)
{'JobId': 182, 'JobName': 'foo.sh', 'UserId': 'atf(1002)', 'GroupId': 'atf(1002)', ...}
"""
jobs_dict = get_jobs(job_id, quiet=quiet)
return jobs_dict[job_id] if job_id in jobs_dict else {}
def get_job_parameter(job_id, parameter_name, default=None, **run_command_kwargs):
"""Returns the value of a specific parameter for a given job.
Args:
job_id (integer): The id of the job for which the parameter value is requested.
parameter_name (string): The name of the parameter whose value is to be obtained.
default (string or None): The value to be returned if the parameter is not found.
Returns:
The value of the specified job parameter, or the default value if the
parameter is not found.
Example:
>>> get_job_parameter(12345, 'UserId')
'atf(1002)'
>>> get_job_parameter(67890, 'Partition', default='normal', quiet=True)
'primary'
"""
jobs_dict = get_jobs(**run_command_kwargs)
if job_id in jobs_dict:
job_dict = jobs_dict[job_id]
else:
pytest.fail(f"Job ({job_id}) was not found in the job configuration")
if parameter_name in job_dict:
return job_dict[parameter_name]
else:
return default
def get_stepmgr_host(job_id, **run_command_kwargs):
"""Returns the host running a stepmgr-enabled job's step manager.
The stepmgr runs in the job's extern slurmstepd on its BatchHost, which
is the host commands such as swait talk to directly.
Args:
job_id (integer): The id of the job whose stepmgr host is requested.
Note:
fatal= is not honored: get_jobs() drops it. This function always
pytest.fail()s when the job has no BatchHost.
Returns:
The hostname of the job's stepmgr.
Example:
>>> get_stepmgr_host(12345)
'node1'
"""
# get_jobs() pops fatal=, so get_job_parameter() cannot fail on a
# missing value; fail explicitly rather than returning None.
host = get_job_parameter(job_id, "BatchHost", **run_command_kwargs)
if not host:
pytest.fail(f"Job ({job_id}) has no BatchHost, so it has no stepmgr")
return host
# swait -v logs this on stderr once the stepmgr has accepted its subscribe
# request. swait(1) documents it under -v, so it is a contract, not an
# incidental log line.
SWAIT_SUBSCRIBED = "subscribed to stepmgr"
# Script exit code meaning "the subscribe sentinel never appeared", kept
# well clear of swait's own 0/1/2/3 so a harness timeout is not read as a
# feature failure.
SWAIT_SUBSCRIBE_TIMEOUT_RC = 90
def swait_await_subscribe(out_file, quote="'"):
"""Returns a shell fragment that blocks until a backgrounded swait -v
has subscribed.
Signalling a step before the subscribe lands makes the stepmgr
fast-return 0 with no output, which reads as a content failure rather
than the timing problem it is. The fragment exits
SWAIT_SUBSCRIBE_TIMEOUT_RC instead, so an exhausted poll names itself.
Args:
out_file (string): File the backgrounded swait -v writes stderr to.
quote (string): Quote character to wrap the sentinel in. Pass '"'
to embed the fragment in a single-quoted --wrap.
Returns:
The shell fragment, as a string.
Example:
>>> swait_await_subscribe("swait.out")
"for _ in $(seq 150); do grep -q 'subscribed to stepmgr' ..."
"""
return (
f"for _ in $(seq 150); do "
f"grep -q {quote}{SWAIT_SUBSCRIBED}{quote} {out_file} 2>/dev/null "
f"&& break; "
f"sleep 0.2; "
f"done; "
f"grep -q {quote}{SWAIT_SUBSCRIBED}{quote} {out_file} 2>/dev/null || "
f"exit {SWAIT_SUBSCRIBE_TIMEOUT_RC}; "
)
def get_pending_async_step(job_id):
"""Returns the StepId of the job's pending --async step, or None.
Args:
job_id (integer): The id of the job to inspect.
Returns:
The StepId (e.g. '123.1'), or None if none appeared in time.
Example:
>>> get_pending_async_step(123)
'123.1'
"""
for _ in timer(timeout=30, poll_interval=0.5, quiet=True):
for sid, info in get_steps(job_id, quiet=True).items():
# Skip .0: before it starts running it can itself read
# PENDING for an instant and shadow the async step we want.
if sid.endswith((".0", ".batch", ".extern", ".interactive")):
continue
if info.get("State") == "PENDING":
return sid
return None
def get_data_parser_number(value):
"""Returns the number carried by a data_parser no-val field.
A no-val field (UINT32_NO_VAL and friends) is a bare int under a
complex-mode data_parser and a {set, infinite, number} object under the
default one, so a test that reads one has to accept both shapes.
Args:
value: The field as parsed from the command's JSON or YAML output.
Returns:
The number the field carries, or None if it is unset.
Example:
>>> get_data_parser_number({'set': True, 'infinite': False, 'number': 7})
7
>>> get_data_parser_number({'set': False, 'infinite': False, 'number': 0})
None
"""
if isinstance(value, dict):
return value["number"] if value.get("set") else None
return value
def get_data_parser_flag(value):
"""Returns the single flag carried by a data_parser flag field.
A flag field renders as a bare string or as a one-element list,
depending on the data_parser in use.
Args:
value: The field as parsed from the command's JSON or YAML output.
Returns:
The flag string.
Example:
>>> get_data_parser_flag(['SUCCESS'])
'SUCCESS'
"""
return value[0] if isinstance(value, list) else value
def get_job_gres_index_blocks(job_id, **run_command_kwargs):
"""Returns the GRES device indexes of a running job, one set per block.
`scontrol show job -d` prints one Nodes= block per group of nodes that
share an identical allocation, with one IDX list per GRES record. This
returns one set per block, holding every IDX list of the block with any
ranges expanded, so a caller can tell a job holding {0,1} on one node and
{2,3} on another from one holding {0,1,2,3} on a single node.
Args:
job_id (integer): The id of the job whose device indexes are wanted.
Returns:
A list of sets of integer device indexes, one per block, in the order
printed.
Example:
>>> get_job_gres_index_blocks(12345)
[{0, 1}]
"""
run_command_kwargs.setdefault("fatal", True)
output = run_command_output(f"scontrol show job -d {job_id}", **run_command_kwargs)
blocks = [
set().union(
*(range_to_list(idx) for idx in re.findall(r"IDX:([\d,\-]+)", line))
)
for line in output.splitlines()
if line.strip().startswith("Nodes=") and "IDX:" in line
]
if not blocks:
pytest.fail(f"Expected a GRES IDX list for job {job_id} in: {output}")
return blocks
def get_job_gres_indexes(job_id, **run_command_kwargs):
"""Returns the set of GRES device indexes allocated to a running job.
This unions every block, so the result is only interpretable for a job on
a single node. Use get_job_gres_index_blocks() for a multi-node job, where
the union cannot tell which devices came from which node.
Args:
job_id (integer): The id of the job whose device indexes are wanted.
Returns:
A set of integer device indexes, unioned over every node and GRES
record of the job.
Example:
>>> get_job_gres_indexes(12345)
{0, 1}
"""
return set().union(*get_job_gres_index_blocks(job_id, **run_command_kwargs))
def get_job_id_from_array_task(array_job_id, array_task_id, fatal=False, quiet=True):
"""Returns the raw job id of a task of a job array.
Args:
array_job_id (integer): The id of the job array.
array_task_id (integer): The id of the task of the job array.
fatal (boolean): If True, fails if the raw job id is not found in the system.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
The raw job id of the given task of a job array, or 0 if not found.
Example:
>>> get_job_id_from_array_task(234, 2)
241
"""
jobs_dict = get_jobs(quiet=quiet)
for job_id, job_values in jobs_dict.items():
# Non-array jobs left in the system by sibling tests carry neither
# key, so use .get() rather than raising on them.
if (
job_values.get("ArrayJobId") == array_job_id
and job_values.get("ArrayTaskId") == array_task_id
):
return job_id
if fatal:
pytest.fail(f"{array_job_id}_{array_task_id} was not found in the system")
return 0
def get_step_parameter(step_id, parameter_name, default=None, quiet=False):
"""Returns the value of a specific parameter for a given step.
Args:
step_id (string): The id of the step for which the parameter value is requested.
parameter_name (string): The name of the parameter whose value is to be obtained.
default (string or None): The value to be returned if the parameter is not found.
quiet (boolean): If True, logging is performed at the TRACE log level.
Returns:
The value of the specified step parameter, or the default value if the
parameter is not found.
Example:
>>> get_step_parameter("45.0", 'NodeList')
'node[2,4]'
>>> get_step_parameter("60.1", 'Partition', default='normal', quiet=True)
'primary'
"""
steps_dict = get_steps(step_id, quiet=quiet)
if step_id not in steps_dict:
logging.debug(f"Step ({step_id}) was not found in the step list")
return default
step_dict = steps_dict[step_id]
if parameter_name in step_dict:
return step_dict[parameter_name]
else:
return default
def _parse_json_tasks(output, hex_fields):
task_data = []
for line in output.split("\n"):
if line.strip(): # Skip empty lines
data = json.loads(line)
for key in hex_fields:
data[key] = int(data[key], 16)
task_data.append(data)
return task_data
def parse_taskget(output):
"""Parse taskget output, return an array of reported task bindings"""
return _parse_json_tasks(output, ["mask"])
def parse_numaget(output):
"""Parse numaget output, return an array of reported task bindings"""
return _parse_json_tasks(output, ["mem_mask", "allowed_mask", "policy_mask"])
def get_licenses(
**run_command_kwargs,
):
"""Returns the licenses/HRES in the system as a dictionary of dictionaries.
Returns:
A dictionary of dictionaries. The first level keys are "license_name"
for licenses defined in slurm.conf or remote licenses defined in the
Slurm database, and "license_name(layer_name)" for HRES (only in Slurm
>= 26.11). Their values are a dictionary of parameters for the
respective license. A test using get_licenses() for HRES should use
atf.require_version((26, 11), component="bin/scontrol"). If duplicate
entries are found (which would only happen for Slurm < 26.11 with
HRES), an empty dictionary is returned.
Example:
>>> get_licenses()
{'lic1': {'LicenseName': 'lic1', 'Total': 30, 'Used': 0, 'Free': 30, 'LayerName': '', ...},
'lic2': {'LicenseName': 'lic2', 'Total': 10, 'Used': 0, 'Free': 10, 'LayerName': '', ...},
'power(leaf1)': {'LicenseName': 'power', 'Total': 2000, 'Used': 2000, 'Free': 0, 'LayerName': 'leaf1', ...},
'power(root)': {'LicenseName': 'power', 'Total': 100000, 'Used': 2000, 'Free': 98000, 'LayerName': 'root', ...}}
>>> get_licenses()["lic1"]
{'LicenseName': 'lic1', 'ConfTotal': 30, 'Total': 30, 'BaseUsage': 0,
'Used': 0, 'Free': 30, 'Remote': False, 'Reserved': 0,
'LastConsumed': 0, 'LastDeficit': 0, 'LastUpdate': 0, 'LayerName': '',
'ParentName': '', 'DisableHRES': False, 'DisableLayer': False,
'Nodes': '', 'Base': []}
>>> get_licenses()["power(leaf1)"]
{'LicenseName': 'power', 'ConfTotal': 2000, 'Total': 2000,
'BaseUsage': 0, 'Used': 2000, 'Free': 0, 'Remote': False,
'Reserved': 0, 'LastConsumed': 0, 'LastDeficit': 0, 'LastUpdate': 0,
'LayerName': 'leaf1', 'ParentName': 'root', 'DisableHRES': False,
'DisableLayer': False, 'Nodes': 'node[1-3]', 'Base': []}
"""
licenses_dict = {}
command = "scontrol --json show licenses"
output = run_command_output(command, **run_command_kwargs)
try:
lics_list = json.loads(output)["licenses"]
except json.JSONDecodeError as e:
msg = f"Error parsing scontrol output: {output}\n{e}"
if run_command_kwargs.get("fatal"):
pytest.fail(msg)
else:
logging.warning(msg)
return licenses_dict
for lic in lics_list:
key = lic["LicenseName"]
# LayerName is only included in Slurm >= 26.11. Before that, the layers
# of a HRES have the same LicenseName and cannot be told apart.
if lic.get("LayerName"):
key += f"({lic['LayerName']})"
if key in licenses_dict:
msg = f"Duplicate license {key}: HRES layers need Slurm >= 26.11"
if run_command_kwargs.get("fatal"):
pytest.fail(msg)
else:
logging.warning(msg)
return {}
licenses_dict[key] = lic
return licenses_dict
def get_log(component, num_lines=0, fatal=False, quiet=True):
"""Returns the log of a Slurm daemon.
The log file is the one configured for the daemon (SlurmctldLogFile,
SlurmdLogFile or LogFile in slurmdbd.conf), read from the config files so
it works even when the daemons are down. slurmrestd and sackd are started
by the testsuite with their output redirected to the log directory of
slurmctld.
Args:
component (string): 'slurmctld', 'slurmdbd', 'slurmrestd', 'sackd',
'slurmd' to get the logs of all the slurmds concatenated, or a
node name to get the log of its slurmd.
num_lines (integer): Number of trailing lines to return (tail), of
each log file. If 0, the whole log is returned.
fatal (boolean): If True, fail the test if the log cannot be read.
quiet (boolean): If True, the command and the log are logged at the
TRACE log level.
Returns:
The requested log content as a string, or an empty string if the log
cannot be read and fatal is False.
Example:
>>> get_log("slurmctld", num_lines=15)
'[2026-10-26T20:12:01.123] debug: sched: Running job scheduler...'
>>> get_log("node2", fatal=True)
'[2026-10-26T20:11:58.456] slurmd version 26.11.0 started...'
>>> get_log("slurmd")
'[2026-10-26T20:11:58.123] slurmd version 26.11.0 started...\\n...'
"""
def _conf_value(name, source):
conf = get_config(live=False, source=source, quiet=True)
return next(
(v for k, v in conf.items() if k.casefold() == name.casefold()), None
)
match component:
case "slurmctld":
log_file = _conf_value("SlurmctldLogFile", "slurm")
case "slurmdbd":
log_file = _conf_value("LogFile", "slurmdbd")
case "slurmrestd" | "sackd":
log_file = _conf_value("SlurmctldLogFile", "slurm")
if log_file:
log_file = f"{os.path.dirname(log_file)}/{component}.log"
case "slurmd":
log_file = _conf_value("SlurmdLogFile", "slurm")
if log_file:
# All the nodes (the shell expands the glob)
log_file = log_file.replace("%n", "*")
case node:
log_file = _conf_value("SlurmdLogFile", "slurm")
if log_file:
log_file = log_file.replace("%n", node)
if not log_file:
message = f"No log file configured for {component}"
if fatal:
pytest.fail(message)
logging.warning(message)
return ""
if num_lines:
command = f"tail -q -n {num_lines} {log_file}"
else:
command = f"cat {log_file}"
return run_command_output(command, user="root", fatal=fatal, quiet=quiet)
def get_log_time(line):
"""Returns the timestamp of a Slurm log line.
Args:
line (string): A line of a Slurm daemon log, as returned by get_log().
Returns:
The timestamp of the line in seconds since the epoch (float), taking
the log time as local time, or None if the line does not start with a
timestamp.
Example:
>>> end = get_log_time("[2026-10-26T21:21:24.680] Saving all slurm state")
>>> start = get_log_time("[2026-10-26T21:19:52.979] Reconfigure signal (SIGHUP) received")
>>> end - start
91.70099997520447
>>> get_log_time(" continuation of a multi-line message")
None
"""
match = re.match(r"\[(\d{4}-\d\d-\d\dT[\d:.]+)\]", line)
if not match:
return None
return datetime.datetime.fromisoformat(match.group(1)).timestamp()
def get_log_delays(component, start_regex, end_regex, min_delay=0):
"""Returns the delays between two events in the log of a Slurm daemon.
Each line matching end_regex is paired with the latest line matching
start_regex before it that was not paired yet. Lines without a timestamp
are ignored.
Args:
component (string): 'slurmctld', 'slurmdbd', 'slurmrestd', 'sackd',
or a node name to get the log of its slurmd. Use a single log, so
not 'slurmd': the concatenated logs of all slurmds are not sorted
by time.
start_regex (string): Regular expression of the line starting the event.
end_regex (string): Regular expression of the line ending the event.
min_delay (float): Only the delays of at least these seconds are
returned.
Returns:
A list of (start_line, end_line, delay) tuples, with the delay in
seconds, in the order of the end lines.
Example:
>>> get_log_delays(
... "slurmctld",
... r"Reconfigure signal|slurmctld terminating",
... r"Saving all slurm state",
... min_delay=30,
... )
[('[2026-10-26T21:19:52.979] Reconfigure signal (SIGHUP) received', '[2026-10-26T21:21:24.680] Saving all slurm state', 91.70099997520447)]
"""
delays = []
start = None
for line in get_log(component).splitlines():
line_time = get_log_time(line)
if line_time is None:
continue
if re.search(end_regex, line):
if start is not None:
delay = line_time - start[1]
if delay >= min_delay:
delays.append((start[0], line, delay))
start = None
elif re.search(start_regex, line):
start = (line, line_time)
return delays
class SetMatch(enum.Enum):
EQUAL = "=="
INTERSECTS = "~="
def wait_for_node_state(
nodename,
desired_node_state,
operator=SetMatch.INTERSECTS,
timeout=default_polling_timeout,
poll_interval=1,
fatal=False,
reverse=False,
):
"""Wait for a specified node state to be reached on one or more nodes.
Polls the node state every poll interval seconds, waiting up to the timeout
for the specified node state to be reached. When given a list of nodes,
the per-node condition must hold for *all* of them; with reverse=True, the
per-node condition is inverted (so a list with reverse=True means "no node
satisfies the condition").
A single query to slurmctld is issued per poll regardless of the number of
nodes being monitored.
Args:
nodename (string or list/set of strings): A node name, or a list of
node names, whose state is being monitored.
desired_node_state (string or list/set of strings): A state name, or a
list of state names, that the node(s) are expected to reach.
operator (SetMatch or string): How desired_node_state is matched
against the actual node state list. Accepts either a SetMatch
member or its underlying string value ("~=" or "=="). Defaults to
~= (SetMatch.INTERSECTS). See SetMatch.
timeout (integer): The number of seconds to wait before timing out.
poll_interval (float): Number of seconds between node state polls.
Defaults to 1s.
fatal (boolean): If True, a timeout will cause the test to fail.
reverse (boolean): If True, wait for the per-node condition to be
false instead of true.
Returns:
Boolean value indicating whether the desired condition was reached on
all the specified node(s) before the timeout.
Example:
>>> wait_for_node_state('node1', 'IDLE', timeout=60, poll_interval=5)
True
>>> wait_for_node_state('node2', 'DOWN', timeout=30, fatal=True)
False
>>> wait_for_node_state(['node1', 'node2', 'node3'], 'IDLE', fatal=True)
True
>>> wait_for_node_state('node1', ['REBOOT_REQUESTED', 'REBOOT_ISSUED'], fatal=True)
True
>>> wait_for_node_state(node_list, 'IDLE', operator=SetMatch.EQUAL, fatal=True)
True
>>> wait_for_node_state(node_list, 'IDLE', operator='==', fatal=True)
True
"""
try:
operator = SetMatch(operator)
except ValueError:
pytest.fail(
f"Unsupported operator {operator!r} "
f"(allowed: {[m.value for m in SetMatch]})"
)
nodes_to_check = {nodename} if isinstance(nodename, str) else set(nodename)
desired_set = (
{desired_node_state}
if isinstance(desired_node_state, str)
else set(desired_node_state)
)
def matches(state):
if operator is SetMatch.EQUAL:
ok = set(state) == desired_set
else: # SetMatch.INTERSECTS
ok = bool(desired_set & set(state))
return ok != reverse
for _ in timer(timeout=timeout, poll_interval=poll_interval, fatal=fatal):
nodes = get_nodes(quiet=True)
if all(matches(nodes[n]["state"]) for n in nodes_to_check):
return True
return False
def wait_for_step(job_id, step_id, state="RUNNING", **repeat_until_kwargs):
"""Wait for the specified step of a job to reach a state.
Continuously polls the step state until it matches or until a timeout
occurs.
Args:
job_id (integer or string): The id of the job. For an array task,
either its raw job id or "<array_job_id>_<task_id>".
step_id (integer or string): The id of the step within the job, or
the name of a special step such as "batch".
state (string): The step state to wait for, or None to wait only for
the step to exist. Defaults to RUNNING. Existence alone is not
enough: a step queued waiting for resources is already listed by
`scontrol show step`, under the StepId it was assigned at
submission, while it is still PENDING.
Returns:
A boolean value indicating whether the specified step reached the
state or not.
Example:
>>> wait_for_step(1234, 0, timeout=60, poll_interval=5, fatal=True)
True
>>> wait_for_step(5678, 1, timeout=30)
False
>>> wait_for_step("5678_3", 0, timeout=30)
True
"""
# The JSON "id" is always the raw "<job_id>.<step_id>", with "+<comp>" for
# a het step, and array tasks carry their ids separately under "array".
array_job_id, _, task_id = str(job_id).partition("_")
if task_id:
# Unless stepmgr is used, scontrol only sends "<array_job_id>.<step_id>"
# to the controller for "<array_job_id>_<task_id>.<step_id>", which only
# matches the task whose raw job id is <array_job_id>, so filter here.
query = array_job_id
def _is_step(step):
return (
step["array"]["job_id"] == int(array_job_id)
and step["array"]["task_id"] == int(task_id)
and step["id"].split("+")[0].split(".", 1)[1] == str(step_id)
)
else:
query = f"{job_id}.{step_id}"
def _is_step(step):
return step["id"].split("+")[0] == query
def _get_steps():
out = run_command_output(f"scontrol --json show step {query}", quiet=True)
try:
steps = json.loads(out)["steps"]
except (json.JSONDecodeError, KeyError):
logging.debug(f"Unable to parse scontrol output: {out}")
return []
# Keep only the fields used here in repeat_until logs.
return [
{
"id": step["id"],
"array": {
"job_id": step["array"]["job_id"],
"task_id": step["array"]["task_id"],
},
"state": step["state"],
}
for step in steps
]
return repeat_until(
_get_steps,
lambda steps: any(
_is_step(step) and (state is None or state in step["state"])
for step in steps
),
**repeat_until_kwargs,
)
def wait_for_step_count_snippet(count, op="-ge", timeout=60):
"""Return Bash lines that wait for the script's job to reach a step count.
Polls `scontrol show step` from inside a batch script until the number of
the job's steps with numeric StepIds (excluding .batch/.extern, which
never match) bears the given test(1) integer relation to count. Aborts
the script on timeout: the gate orders step submissions, so a script
that ran anyway would report the missing ordering as a product failure.
Args:
count (integer): The step count to compare against.
op (string): A test(1) integer relation (e.g. -ge, -eq). Defaults
to -ge: wait until at least count steps are registered.
timeout (integer): Seconds to wait before aborting the script.
Returns:
Bash lines embedding the wait, terminated by a newline.
Example:
>>> make_bash_script("s.sh", wait_for_step_count_snippet(2) + "srun true\\n")
None
"""
pattern = "StepId=$SLURM_JOB_ID[.][0-9]+([[:space:]+]|$)"
reached = (
f'[ "$(scontrol -o show step "$SLURM_JOB_ID" 2>/dev/null | '
f'grep -cE "{pattern}")" {op} {count} ]'
)
return (
f"for _ in $(seq 1 {int(timeout * 2)}); do {reached} && break; "
"sleep 0.5; done\n"
f"{reached} || {{ echo STEP_COUNT_NOT_{op}_{count}; exit 1; }}\n"
)
def wait_for_step_snippet(step_id, timeout=30):
"""Return Bash lines that wait for a step of the script's job to register.
Polls `scontrol show step` from inside a batch script until the step is
listed, under the StepId it was assigned at submission (a step queued
waiting for resources is already listed, while still PENDING). Aborts
the script on timeout: the step's registration is a precondition for
whatever the script does to it next.
Args:
step_id (integer): The id of the step within $SLURM_JOB_ID.
timeout (integer): Seconds to wait before aborting the script.
Returns:
Bash lines embedding the wait, terminated by a newline.
Example:
>>> make_bash_script("s.sh", "srun true &\\n" + wait_for_step_snippet(0))
None
"""
pattern = f"StepId=$SLURM_JOB_ID[.]{step_id}([[:space:]+]|$)"
return (
f"for _ in $(seq 1 {int(timeout * 2)}); do "
f'scontrol -o show step "$SLURM_JOB_ID" 2>/dev/null | '
f'grep -qE "{pattern}" && break; sleep 0.5; done\n'
f'scontrol -o show step "$SLURM_JOB_ID" 2>/dev/null | '
f'grep -qE "{pattern}" || '
f"{{ echo STEP_{step_id}_NOT_REGISTERED; exit 1; }}\n"
)
def hog_allocation_snippet(ready_file, srun_args="--exclusive -n2", timeout=30):
"""Return Bash lines that hog a node's CPUs and block until they are held.
Launches a background srun that touches ready_file once its tasks hold
the node's CPUs, then blocks until that file exists, so a following
srun sees a genuinely busy allocation. Aborts the script on timeout:
the busy allocation is a precondition, so a script that ran anyway
would report the missing contention as a product failure.
Args:
ready_file (string): Path the hog touches once it holds the CPUs.
srun_args (string): Resource arguments for the hog srun. Defaults
to "--exclusive -n2" (both CPUs of a 2-CPU node).
timeout (integer): Seconds to wait for the hog before aborting.
Returns:
Bash lines embedding the hog and the readiness gate, terminated by a
newline.
Example:
>>> make_bash_script("s.sh", hog_allocation_snippet("ready") + "wait\\n")
None
"""
return (
f"srun {srun_args} sh -c 'touch \"{ready_file}\"; exec sleep infinity' &\n"
f"for _ in $(seq 1 {int(timeout * 2)}); do "
f"[ -f '{ready_file}' ] && break; sleep 0.5; done\n"
f"[ -f '{ready_file}' ] || {{ echo HOG_NOT_READY; exit 1; }}\n"
)
def wait_for_step_accounted(job_id, step_id, **repeat_until_kwargs):
"""Wait for specified job step to appear in accounting database (`sacct`).
Continuously polls the database until the step is accounted for or until a
timeout occurs.
Args:
job_id (integer): The id of the job.
step_id (integer): The id of the step within the job.
Returns:
A boolean value indicating whether the specified step is accounted for
in the database or not.
Example:
>>> wait_for_step_accounted(1234, 0, timeout=60, poll_interval=5, fatal=True)
True
>>> wait_for_step_accounted(5678, 1, timeout=30)
False
"""
step_str = f"{job_id}.{step_id}"
return repeat_until(
lambda: run_command_output(f"sacct -j {job_id} -o jobid"),
lambda out: re.search(rf"{step_str}", out) is not None,
**repeat_until_kwargs,
)
def wait_for_job_accounted(job_id, field="Start", value=None, **repeat_until_kwargs):
"""Wait for specified job field to appear in accounting database (`sacct`).
Continuously polls the database until the job field is accounted for or until a
timeout occurs. If a value is specified, it waits until the field has that value.
Args:
job_id (integer): The id of the job.
field (string): The field to query from the job.
value (string): The desired value to wait for.
None means "any value but empty"
Returns:
A boolean value indicating whether the specified job is accounted for
in the database or not.
Example:
>>> wait_for_job_accounted(1234, timeout=60, poll_interval=5, fatal=True)
True
>>> wait_for_job_accounted(5678, "End", timeout=30)
False
"""
if not value:
value = r"."
return repeat_until(
lambda: run_command_output(f"sacct -XPnj {job_id} -o {field}"),
lambda out: re.search(value, out.strip()) is not None,
**repeat_until_kwargs,
)
def is_job_state_done(job_state):
"""Return True if job_state is one of the final JobStates, or False otherwise"""
done_states = [
"NOT_FOUND",
"BOOT_FAIL",
"CANCELLED",
"COMPLETED",
"DEADLINE",
"FAILED",
"NODE_FAIL",
"OUT_OF_MEMORY",
"TIMEOUT",
"PREEMPTED",
]
return job_state in done_states
def wait_for_job_state(
job_id,
desired_job_state,
desired_reason=None,
timeout=default_polling_timeout,
poll_interval=None,
**run_command_kwargs,
):
"""Wait for the specified job to reach the desired state.
Continuously polls the job state until the desired state is reached or until
a timeout occurs.
Some supported job states are aggregate states, which may include multiple
discrete states. Some logic is built-in to fail if a job reaches a state
that makes the desired job state impossible to reach.
Current supported aggregate states:
- DONE
Args:
job_id (integer): The id of the job.
desired_job_state (string): The desired state of the job.
desired_reason (string): Optional reason to also match.
timeout (integer): The number of seconds to poll before timing out.
poll_interval (float): Time (in seconds) between job state polls.
Returns:
Boolean value indicating whether the job reached the desired state.
Example:
>>> wait_for_job_state(1234, 'COMPLETED', timeout=300, poll_interval=10, fatal=True)
True
>>> wait_for_job_state(5678, 'RUNNING', timeout=60)
False
"""
if poll_interval is None:
if timeout <= 5:
poll_interval = 0.1
elif timeout <= 10:
poll_interval = 0.2
else:
poll_interval = 1
# Get the quiet, fatal and xfail values but don't forward them
quiet = run_command_kwargs.pop("quiet", None)
fatal = run_command_kwargs.pop("fatal", None)
xfail = run_command_kwargs.pop("xfail", None)
if quiet:
log_level = logging.TRACE
else:
log_level = logging.DEBUG
# We don't use repeat_until here because we support pseudo-job states and
# we want to allow early return (e.g. for a DONE state if we want RUNNING)
xfail_str = ""
if xfail:
xfail_str = "not "
message = (
f"Waiting for job ({job_id}) to {xfail_str}reach state {desired_job_state}"
)
if desired_reason is not None:
message += f" and reason {desired_reason}"
logging.log(log_level, message)
begin_time = time.time()
while time.time() < begin_time + timeout:
job_state = get_job_parameter(
job_id, "JobState", default="NOT_FOUND", quiet=True, **run_command_kwargs
)
message = f"Job ({job_id}) is in state {job_state}, but we are waiting for {desired_job_state}"
if is_job_state_done(job_state):
if desired_job_state == "DONE" or job_state == desired_job_state:
message = f"Job ({job_id}) is in the {xfail_str}desired state {desired_job_state}"
reason = get_job_parameter(
job_id,
"Reason",
default="NOT_FOUND",
quiet=True,
**run_command_kwargs,
)
if desired_reason is None or reason == desired_reason:
if desired_reason is not None:
message += (
f" with the {xfail_str}desired reason {desired_reason}"
)
if not xfail:
logging.log(log_level, message)
else:
logging.warning(message)
return True
else:
message += (
f", but with reason {reason} and we waited for {desired_reason}"
)
if not xfail:
if fatal:
pytest.fail(message)
else:
logging.warning(message)
else:
logging.log(log_level, message)
return False
elif job_state == desired_job_state:
message = (
f"Job ({job_id}) is in the {xfail_str}desired state {desired_job_state}"
)
reason = get_job_parameter(
job_id, "Reason", default="NOT_FOUND", quiet=True, **run_command_kwargs
)
if desired_reason is None or reason == desired_reason:
if desired_reason is not None:
message += f" with the {xfail_str}desired reason {desired_reason}"
if not xfail:
logging.log(log_level, message)
else:
logging.warning(message)
return True
else:
message += (
f", but with reason {reason} and we waited for {desired_reason}"
)
logging.log(log_level, message)
time.sleep(poll_interval)
message = f"Job ({job_id}) did not reach the {desired_job_state} state"
if desired_reason is not None:
message += f" or the reason {desired_reason}"
message += f" within the {timeout} second(s) timeout"
if not xfail:
if fatal:
pytest.fail(message)
else:
logging.warning(message)
else:
logging.log(log_level, message)
return False
def job_stays_pending(
job_id,
timeout=3,
pending_timeout=default_polling_timeout,
poll_interval=0.2,
):
"""Wait for a job to be pending, then check that it stays pending.
wait_for_job_state(job_id, "PENDING") returns as soon as the job is seen
pending, which every newly submitted job is before the scheduler starts it.
Use this instead to show that something (e.g. a drained TRES) is actually
holding the job back.
Args:
job_id (integer): The id of the job.
timeout (integer): Seconds to watch the job for once the scheduler has
given it a pending reason. The job must not leave PENDING during
this window.
pending_timeout (integer): Seconds to wait for the job to be pending in
the first place, and again for the scheduler to give it a reason.
poll_interval (float): Time (in seconds) between job state polls.
Returns:
True if the job became pending and was still pending for the whole
window, False if it never became pending, the scheduler never gave it
a reason, or it started during the window.
Example:
>>> job_stays_pending(1234)
True
"""
if not wait_for_job_state(job_id, "PENDING", timeout=pending_timeout):
return False
# A new job is pending with Reason=None until the scheduler first looks at
# it, so a window started before then could pass without anything holding
# the job back. Start the window once the scheduler has given a reason.
def _state_and_reason():
job = get_job(job_id, quiet=True)
return job.get("JobState"), job.get("Reason")
if not repeat_until(
_state_and_reason,
lambda state_reason: state_reason[0] != "PENDING"
or state_reason[1] not in (None, "None"),
timeout=pending_timeout,
poll_interval=poll_interval,
):
return False
started = repeat_until(
lambda: get_job_parameter(job_id, "JobState", default="NOT_FOUND", quiet=True),
lambda job_state: job_state != "PENDING",
timeout=timeout,
poll_interval=poll_interval,
)
return not started
def check_steps_delayed(
job_id,
expected_delayed,
first_parallel_step=0,
min_delay_seconds=2,
**repeat_until_kwargs,
):
"""Verify that job steps were delayed waiting for resources.
Reads per-step Start timestamps from sacct after the job ends. A step is
considered "delayed" if its Start is at least min_delay_seconds after the
earliest start within the parallel-batch group. The count of such steps
must equal expected_delayed.
Args:
job_id (integer): The job ID that we're interested in.
expected_delayed (integer): The number of steps that should have been
delayed waiting for resources.
first_parallel_step (integer): The step ID at which the parallel
batch begins. Steps with a lower ID (typically sequential
pre-steps in the job script) are ignored. Defaults to 0, which
considers all steps as part of the parallel batch.
min_delay_seconds (integer): A step is "delayed" if its Start is at
least this many seconds after the earliest step Start of the
considered set. Default 2 — fits between sub-second
parallel-batch jitter and the >=3s gap created by typical test
step sleeps.
Returns:
True if exactly expected_delayed steps were delayed, else False.
"""
# End being recorded implies all step rows have flushed (steps end before their parent job).
if not wait_for_job_accounted(job_id, field="End", **repeat_until_kwargs):
logging.error(f"check_steps_delayed: job {job_id} End not in accounting")
return False
# -n omits header, -P pipe-separates, -j scopes to this job only.
out = run_command_output(f"sacct -nPj {job_id} -o jobid,start", quiet=True)
# Numeric step IDs only — skips ".batch" and ".extern" rows.
step_re = re.compile(rf"^{job_id}\.(\d+)\|(\S+)$")
parsed = []
for line in out.splitlines():
m = step_re.match(line.strip())
if not m:
continue
stepnum = int(m.group(1))
# Skip sequential pre-steps that aren't part of the parallel batch.
if stepnum < first_parallel_step:
continue
ts = m.group(2)
try:
parsed.append(
(stepnum, datetime.datetime.strptime(ts, "%Y-%m-%dT%H:%M:%S"))
)
except ValueError:
logging.error(f"check_steps_delayed: step {stepnum} has bad Start {ts!r}")
return False
if not parsed:
logging.error(
f"check_steps_delayed: no step rows for job {job_id} at or "
f"after step {first_parallel_step}"
)
return False
# Sort by start time; tie-break on step number for stable ordering.
parsed.sort(key=lambda x: (x[1], x[0]))
earliest = parsed[0][1]
threshold = datetime.timedelta(seconds=min_delay_seconds)
delayed_count = sum(1 for _, ts in parsed if ts - earliest >= threshold)
if delayed_count != expected_delayed:
summary = ", ".join(f"{n}@{t.isoformat()}" for n, t in parsed)
logging.error(
f"check_steps_delayed: expected {expected_delayed} delayed "
f"step(s), got {delayed_count} (>= {min_delay_seconds}s after "
f"earliest start). Step starts: {summary}"
)
return False
return True
def create_node(node_dict):
"""Creates a node with the properties described by the supplied dictionary.
This function is currently only used as a helper function within other
library functions (e.g. require_nodes). It modifies the Slurm configuration
file and restarts the relevant Slurm daemons. A backup is automatically
created, and the original configuration is restored after the test
completes. This function may only be used in auto-config mode.
Args:
node_dict (dictionary): A dictionary containing the desired node
properties.
Returns:
None
Example:
>>> create_node({'NodeName': 'new_node', 'CPUs': 4, 'RealMemory': 16384})
"""
if not properties["auto-config"]:
require_auto_config("wants to add a node")
config_file = f"{properties['slurm-config-dir']}/slurm.conf"
# Read the original slurm.conf into a list of lines
output = run_command_output(
f"cat {config_file}", user=properties["slurm-user"], quiet=True
)
original_config_lines = output.splitlines()
new_config_lines = original_config_lines.copy()
# Locate the last NodeName definition
last_node_line_index = 0
for line_index in range(len(original_config_lines)):
line = original_config_lines[line_index]
if re.search(r"(?i)^ *NodeName", line) is not None:
last_node_line_index = line_index
if last_node_line_index == 0:
last_node_line_index = line_index
# Build up the new node line
node_line = ""
if "NodeName" in node_dict:
node_line = f"NodeName={node_dict['NodeName']}"
node_range = node_dict.pop("NodeName")
if "Port" in node_dict:
node_line += f" Port={node_dict['Port']}"
node_dict.pop("Port")
for parameter_name, parameter_value in sorted(node_dict.items()):
node_line += f" {parameter_name}={parameter_value}"
# Add the new node line
new_config_lines.insert(last_node_line_index + 1, node_line)
# Write the config file back out with the modifications
new_config_string = "\n".join(new_config_lines)
run_command(
f"echo '{new_config_string}' > {config_file}",
user=properties["slurm-user"],
fatal=True,
quiet=True,
)
# Create the required node directories
for node_name in node_range_to_list(node_range):
spool_dir = properties["slurm-spool-dir"].replace("%n", node_name)
tmpfs_dir = properties["slurm-tmpfs"].replace("%n", node_name)
properties["nodes"].append(node_name)
run_command(f"sudo mkdir -p {spool_dir}", fatal=True, quiet=True)
run_command(f"sudo mkdir -p {tmpfs_dir}", fatal=True, quiet=True)
# Restart slurm if it is already running
if is_slurmctld_running(quiet=True):
restart_slurm(quiet=True)
def _hardware_node_parameters(requirements_list):
"""Resolve hardware-matching requirements against the test host's topology.
Args:
requirements_list (list of tuples): List of (parameter_name,
parameter_value) tuples, as passed to require_nodes().
Returns:
A 2-tuple (hardware_values, remaining_requirements). hardware_values
is a dict of the topology parameters to apply to any node created to
satisfy require_nodes(). remaining_requirements is requirements_list
with the parameters checked here removed.
"""
if not properties["auto-config"]:
logging.warning(
"Assuming the nodes are configured with the hardware they run on. "
"A test failing right after this may be running on a node whose "
"configuration does not match what 'slurmd -C' reports for it."
)
return ({}, requirements_list)
hardware = get_slurmd_C()
sockets = hardware["Boards"] * hardware["SocketsPerBoard"]
available = {
"CPUs": hardware["CPUs"],
"Cores": sockets * hardware["CoresPerSocket"],
"Sockets": sockets,
"Boards": hardware["Boards"],
"SocketsPerBoard": hardware["SocketsPerBoard"],
"CoresPerSocket": hardware["CoresPerSocket"],
"ThreadsPerCore": hardware["ThreadsPerCore"],
}
write_params = (
"Boards",
"SocketsPerBoard",
"CoresPerSocket",
"ThreadsPerCore",
"CPUs",
)
# Cores is checked against the hardware and then dropped because it is
# derived from the topology rather than written to the node line.
drop_params = write_params + ("Sockets", "Cores")
remaining_requirements = []
for requirement in requirements_list:
parameter_name, parameter_value = requirement[0:2]
if parameter_name in available and available[parameter_name] < int(
parameter_value
):
pytest.skip(
f"This test requires a node with {parameter_value} "
f"{parameter_name} matching its hardware, but the host has "
f"{available[parameter_name]}",
allow_module_level=True,
)
if parameter_name not in drop_params:
remaining_requirements.append(requirement)
hardware_values = {name: hardware[name] for name in write_params}
logging.info(f"Nodes will be created with the detected hardware: {hardware_values}")
return (hardware_values, remaining_requirements)
# requirements_list is a list of (parameter_name, parameter_value) tuples.
# Uses non-live node info because must copy from existing node config line
# We implemented requirements_list as a list of tuples so that this could
# later be extended to include a comparator, etc.
# atf.require_nodes(1, [('CPUs', 4), ('RealMemory', 40)])
# atf.require_nodes(2, [('Gres', 'gpu:1,mps:100')])
def require_nodes(requested_node_count, requirements_list=[], require_hardware=False):
"""Ensure that a requested number of nodes have the required properties.
In local-config mode, the test is skipped if an insufficient number of
nodes possess the required properties.
In auto-config mode, nodes are created as needed with the required
properties.
Args:
requested_node_count (integer): Number of required nodes.
requirements_list (list of tuples): List of (parameter_name,
parameter_value) tuples.
require_hardware (boolean): In auto-config mode, nodes created by this
call are given the CPU topology reported by 'slurmd -C' on the test
host, and existing nodes must be configured to the reported topology
to qualify. The test is skipped when the detected hardware cannot
satisfy the CPU, core or socket requirements. In local-config mode
a warning is logged and the node configuration is assumed to match
the hardware already.
Currently supported node requirement types include:
CPUs
Cores
RealMemory
Gres
Features
Other node requirement types will still be appended to the requirements,
but this could stop slurm from starting.
Returns:
At least requested_node_count node names with the required properties.
Example:
>>> require_nodes(2, [('CPUs', 4), ('RealMemory', 40)])
['node1', 'node2']
>>> require_nodes(2, [('CPUs', 2), ('RealMemory', 30), ('Features', 'gpu,mpi')])
['node1', 'node2', 'node3']
>>> require_nodes(2, [('CPUs', 4), ('Sockets', 1)])
['node4', 'node5']
>>> require_nodes(1, [('CPUs', 4)], require_hardware=True)
['node2', 'node3']
"""
hardware_values = {}
if require_hardware:
hardware_values, requirements_list = _hardware_node_parameters(
requirements_list
)
# Always read from slurm.conf (live=False), so we never use --json.
nodes_dict = get_nodes(live=False, quiet=True)
original_nodes = {}
default_node = {}
# Instantiate original node names from nodes_dict
for node_name in nodes_dict:
# Split out the default node
if node_name == "DEFAULT":
default_node = nodes_dict[node_name]
else:
original_nodes[node_name] = {}
# Populate with any default parameters
if default_node:
for node_name in original_nodes:
for parameter_name, parameter_value in default_node.items():
if parameter_name.lower() != "nodename":
original_nodes[node_name][parameter_name] = parameter_value
# Merge in parameters from nodes_dict
for node_name in original_nodes:
for parameter_name, parameter_value in nodes_dict[node_name].items():
if parameter_name.lower() != "nodename":
original_nodes[node_name][parameter_name] = parameter_value
# Check to see how many qualifying nodes we have
qualifying_node_names = []
node_count = 0
nonqualifying_node_count = 0
first_node_name = ""
node_indices = {}
augmentation_dict = {}
for node_name in sorted(original_nodes):
lower_node_dict = dict(
(key.lower(), value) for key, value in original_nodes[node_name].items()
)
node_count += 1
if node_count == 1:
first_node_name = node_name
# Build up node indices for use when having to create new nodes
match = re.search(r"^(.*?)(\d*)$", node_name)
node_prefix, node_index = match.group(1), match.group(2)
if node_index == "":
node_indices[node_prefix] = node_indices.get(node_prefix, [])
else:
node_indices[node_prefix] = node_indices.get(node_prefix, []) + [
int(node_index)
]
node_qualifies = True
if hardware_values:
boards = int(lower_node_dict.get("boards", 1))
sockets_per_board = int(lower_node_dict.get("socketsperboard", 1))
cores_per_socket = int(lower_node_dict.get("corespersocket", 1))
threads_per_core = int(lower_node_dict.get("threadspercore", 1))
default_cpus = (
boards * sockets_per_board * cores_per_socket * threads_per_core
)
node_hardware = {
"Boards": boards,
"SocketsPerBoard": sockets_per_board,
"CoresPerSocket": cores_per_socket,
"ThreadsPerCore": threads_per_core,
"CPUs": int(lower_node_dict.get("cpus", default_cpus)),
}
if "sockets" in lower_node_dict or any(
node_hardware[parameter_name] != hardware_values[parameter_name]
for parameter_name in hardware_values.keys()
):
if node_qualifies:
node_qualifies = False
nonqualifying_node_count += 1
for requirement_tuple in requirements_list:
parameter_name, parameter_value = requirement_tuple[0:2]
if parameter_name in ["CPUs", "RealMemory"]:
if parameter_name.lower() in lower_node_dict:
if int(lower_node_dict[parameter_name.lower()]) < int(
parameter_value
):
if node_qualifies:
node_qualifies = False
nonqualifying_node_count += 1
if nonqualifying_node_count == 1:
augmentation_dict[parameter_name] = parameter_value
else:
if node_qualifies:
node_qualifies = False
nonqualifying_node_count += 1
if nonqualifying_node_count == 1:
augmentation_dict[parameter_name] = parameter_value
elif parameter_name == "Cores":
boards = int(lower_node_dict.get("boards", 1))
sockets_per_board = int(lower_node_dict.get("socketsperboard", 1))
cores_per_socket = int(lower_node_dict.get("corespersocket", 1))
sockets = boards * sockets_per_board
cores = sockets * cores_per_socket
parameter_value = int(parameter_value)
if cores < parameter_value:
if node_qualifies:
node_qualifies = False
nonqualifying_node_count += 1
if nonqualifying_node_count == 1:
augmentation_dict["CoresPerSocket"] = math.ceil(
parameter_value / sockets
)
elif parameter_name == "Gres":
if parameter_name.lower() in lower_node_dict:
gres_list = parameter_value.split(",")
for gres_value in gres_list:
if match := re.search(r"^(\w+)(?::(\w+))?:(\d+)$", gres_value):
(
required_gres_name,
required_gres_type,
required_gres_value,
) = (
match.group(1),
match.group(2),
match.group(3),
)
else:
pytest.fail(
"Gres requirement must be of the form <name>[:<type>]:<count>"
)
required_gres_key = required_gres_name
if required_gres_type:
required_gres_key += f":{required_gres_type}"
if match := re.search(
rf"{re.escape(required_gres_key)}:(\d+)",
lower_node_dict[parameter_name.lower()],
):
if int(match.group(1)) < int(required_gres_value):
if node_qualifies:
node_qualifies = False
nonqualifying_node_count += 1
if nonqualifying_node_count == 1:
augmentation_dict[parameter_name] = parameter_value
else:
if node_qualifies:
node_qualifies = False
nonqualifying_node_count += 1
if nonqualifying_node_count == 1:
augmentation_dict[parameter_name] = parameter_value
else:
if node_qualifies:
node_qualifies = False
nonqualifying_node_count += 1
if nonqualifying_node_count == 1:
augmentation_dict[parameter_name] = parameter_value
elif parameter_name == "Features":
required_features = set(parameter_value.split(","))
features = lower_node_dict.get("features", [])
features = features[0] if features else ""
node_features = set(features.split(","))
if not required_features.issubset(node_features):
if node_qualifies:
node_qualifies = False
nonqualifying_node_count += 1
if nonqualifying_node_count == 1:
augmentation_dict[parameter_name] = parameter_value
else:
logging.debug(
f"{parameter_name} is not a supported node requirement type."
)
logging.debug(
f"{parameter_name}={parameter_value} will be added anyways!"
)
augmentation_dict[parameter_name] = parameter_value
if node_qualifies:
node_qualifies = False
nonqualifying_node_count += 1
if node_qualifies:
qualifying_node_names.append(node_name)
# Not enough qualifying nodes
if len(qualifying_node_names) < requested_node_count:
# If auto-config, configure what is required
if properties["auto-config"]:
# Create new nodes to meet requirements ignoring default node0
new_node_count = requested_node_count
# If we already have a qualifying node, we will use it as the template
if qualifying_node_names:
template_node_name = qualifying_node_names[0]
template_node = nodes_dict[template_node_name].copy()
# Otherwise we will use the first node as a template and augment it
else:
template_node_name = first_node_name
template_node = nodes_dict[template_node_name].copy()
for parameter_name, parameter_value in augmentation_dict.items():
template_node[parameter_name] = parameter_value
if hardware_values:
# Drop every spelling of the topology keys so each is set once
hardware_keys = {key.lower() for key in hardware_values}
hardware_keys.add("sockets")
# Iterate a snapshot, not the real dict so we can delete
for key in list(template_node):
if key.lower() in hardware_keys:
del template_node[key]
template_node.update(hardware_values)
base_port = int(nodes_dict[template_node_name]["Port"])
# Build up a list of available new indices starting after the template
match = re.search(r"^(.*?)(\d*)$", template_node_name)
template_node_prefix, template_node_index = match.group(1), int(
match.group(2)
)
used_indices = sorted(node_indices[template_node_prefix])
new_indices = []
new_index = template_node_index
for i in range(new_node_count):
new_index += 1
while new_index in used_indices:
new_index += 1
new_indices.append(new_index)
# Create a new aggregate node
# Later, we could consider collapsing the node into the template node if unmodified
new_node_dict = template_node.copy()
if new_node_count == 1:
new_node_dict["NodeName"] = template_node_prefix + str(new_indices[0])
new_node_dict["Port"] = base_port - template_node_index + new_indices[0]
else:
new_node_dict["NodeName"] = (
f"{template_node_prefix}[{list_to_range(new_indices)}]"
)
new_node_dict["Port"] = list_to_range(
list(
map(lambda x: base_port - template_node_index + x, new_indices)
)
)
new_node_names = [
template_node_prefix + str(new_index) for new_index in new_indices
]
create_node(new_node_dict)
return qualifying_node_names + new_node_names
# If local-config, skip
else:
message = f"This test requires {requested_node_count} nodes"
if requirements_list:
message += f" with {requirements_list}"
pytest.skip(message, allow_module_level=True)
return qualifying_node_names
def make_bash_script(script_name, script_contents):
"""Creates an executable Bash script with the specified contents.
Args:
script_name (string): Name of the script to create.
script_contents (string): Contents of the script.
Returns:
None
Example:
>>> make_bash_script("my_script.sh", "echo 'Hello, World!'") # This creates an executable Bash script named "my_script.sh" with the contents "echo 'Hello, World!'"
"""
with open(script_name, "w") as f:
f.write("#!/bin/bash\n")
f.write(script_contents)
run_command(f"chmod 755 {script_name}", user="root", fatal=True, quiet=True)
if properties["test-user-set"]:
run_command(
f"chown {properties['test-user']} {script_name}",
user="root",
fatal=True,
quiet=True,
)
def wait_for_file(file_name, **repeat_until_kwargs):
"""Waits for the specified file to be present.
This function waits up to the specified timeout seconds for the file to be
present, polling every poll_interval seconds. The default timeout and
poll_interval are inherited from repeat_until.
Args:
file_name (string): The file name.
Returns:
True if the file was found, False otherwise.
Example:
>>> wait_for_file("my_file.txt", timeout=30, poll_interval=0.5)
True
"""
logging.debug(f"Waiting for file ({file_name}) to be present")
return repeat_until(
lambda: os.path.isfile(file_name), lambda exists: exists, **repeat_until_kwargs
)
def assert_file_contents(
file_name,
expected,
contains=False,
strip=True,
message=None,
timeout=default_polling_timeout,
poll_interval=None,
quiet=False,
):
"""Waits for a file to exist and its content to match `expected`, or assert False.
Args:
file_name (string): The file name.
expected (string): The expected file content.
contains (bool): If True, `expected` must be a substring of the
file content. If False (default), the content must equal
`expected` exactly.
strip (bool): If True (default), whitespace is stripped from the
file content before comparison.
message (string): Optional custom message on timeout. If None, a
default message including `file_name`, `expected`, and the
last observed content is used.
timeout (float): Maximum time in seconds to wait (default:
default_polling_timeout). Applies both to the existence gate
and to the content wait.
poll_interval (float): Seconds between content checks. Defaults
to timeout / 10.0.
quiet (bool): If True, silences the per-iteration `cat` logs and
the timer's progress/timeout logs. The existence-check log
(from wait_for_file) is DEBUG-level and unaffected.
Example:
>>> assert_file_contents("tasks-0.out", "task-0")
>>> assert_file_contents("job.out", "hello", contains=True)
"""
wait_for_file(file_name, timeout=timeout, poll_interval=poll_interval, fatal=True)
text = ""
for _ in timer(timeout=timeout, poll_interval=poll_interval, quiet=quiet):
text = run_command_output(f"cat {file_name}", quiet=quiet)
if strip:
text = text.strip()
if (expected in text) if contains else (text == expected):
return
assert False, (
message
or f"File {file_name} content should {'contain' if contains else 'be'} {expected!r}; got {text!r}"
)
def dump_accounting_database(dst, drop=False):
"""Dumps the accounting database to a gzipped SQL file.
Works in both auto-config and local-config modes. The dump can be used
later by restore_accounting_database (e.g. by require_accounting(modify=True)
for restore-on-teardown) or as a debugging artifact attached to a failing
test.
Args:
dst (string): Destination file path. Always gzipped; callers should
use a .sql.gz suffix.
drop (boolean): If True, also drop the database after dumping it.
Only allowed in auto-config mode.
Returns:
None
Example:
>>> dump_accounting_database(f"{atf.properties['slurm-logs-dir']}/slurm_acct_db.sql.gz")
"""
if drop and not properties["auto-config"]:
pytest.fail("The accounting database can only be dropped in auto-config")
mysqldump_path = shutil.which("mysqldump")
if mysqldump_path is None:
pytest.fail(
"Unable to dump the accounting database. mysqldump was not found in your path"
)
mysql_path = shutil.which("mysql")
if mysql_path is None:
pytest.fail(
"Unable to dump the accounting database. mysql was not found in your path"
)
slurmdbd_dict = get_config(live=False, source="slurmdbd", quiet=True)
database_host, database_port, database_name, database_user, database_password = (
slurmdbd_dict.get(key)
for key in [
"StorageHost",
"StoragePort",
"StorageLoc",
"StorageUser",
"StoragePass",
]
)
mysql_options = ""
if database_host:
mysql_options += f" -h {database_host}"
if database_port:
mysql_options += f" -P {database_port}"
if database_user:
mysql_options += f" -u {database_user}"
else:
mysql_options += f" -u {properties['slurm-user']}"
if database_password:
mysql_options += f" -p {database_password}"
if not database_name:
database_name = "slurm_acct_db"
mysql_command = f"{mysql_path} {mysql_options} -e \"USE '{database_name}'\""
if run_command_exit(mysql_command, quiet=True) != 0:
logging.debug(f"Slurm accounting database ({database_name}) is not present")
return
if os.path.isfile(dst):
logging.warning(f"Dump file already exists ({dst})")
else:
mysqldump_command = (
f"{mysqldump_path} {mysql_options} {database_name} | gzip -c > {dst}"
)
run_command(
mysqldump_command, fatal=True, quiet=False, timeout=default_sql_cmd_timeout
)
if drop:
run_command(
f'{mysql_path} {mysql_options} -e "drop database {database_name}"',
fatal=True,
quiet=False,
timeout=default_sql_cmd_timeout,
)
def restore_accounting_database(src):
"""Restores the accounting database from a previous dump.
This function may only be used in auto-config mode. The dump file is
removed after a successful restore.
This behaves like a teardown/finally step: it never aborts but returns
a list of failures for the caller to handle.
Args:
src (string): Source dump file path (gzipped, as produced by
dump_accounting_database).
Returns:
list[str]: One message per step that failed (empty if the restore
completed cleanly).
Example:
>>> restore_accounting_database(properties["sql-db-backup"])
[]
"""
# Should be single line failures
failures = []
if not properties["auto-config"]:
return failures
mysql_path = shutil.which("mysql")
if mysql_path is None:
failures.append(
"Unable to restore the accounting database. mysql was not found in your path"
)
return failures
slurmdbd_dict = get_config(live=False, source="slurmdbd", quiet=True)
database_host, database_port, database_name, database_user, database_password = (
slurmdbd_dict.get(key)
for key in [
"StorageHost",
"StoragePort",
"StorageLoc",
"StorageUser",
"StoragePass",
]
)
if not database_name:
database_name = "slurm_acct_db"
base_command = mysql_path
if database_host:
base_command += f" -h {database_host}"
if database_port:
base_command += f" -P {database_port}"
if database_user:
base_command += f" -u {database_user}"
else:
base_command += f" -u {properties['slurm-user']}"
if database_password:
base_command += f" -p {database_password}"
# If DB exists, drop it and try to restore the dump file
mysql_command = f"{base_command} -e \"USE '{database_name}'\""
if run_command_exit(mysql_command, quiet=True) == 0:
result = run_command(
f'{base_command} -e "drop database {database_name}"',
quiet=False,
timeout=default_sql_cmd_timeout,
)
if result["exit_code"] != 0:
failures.append(f"Failed to drop the accounting database ({database_name})")
# If the dump file doesn't exist, it has probably already been
# restored by a previous call to restore_accounting_database
if not os.path.isfile(src):
logging.debug(
f"Slurm accounting database backup ({src}) is not present. It has probably already been restored."
)
return failures
dump_stat = os.stat(src)
if not (dump_stat.st_size == 0 and dump_stat.st_mode & stat.S_ISVTX):
result = run_command(
f'{base_command} -e "create database {database_name}"',
quiet=False,
)
if result["exit_code"] != 0:
failures.append(
f"Failed to create the accounting database ({database_name})"
)
result = run_command(
f"gunzip -c {src} | {base_command} {database_name}",
quiet=False,
timeout=default_sql_cmd_timeout,
)
if result["exit_code"] != 0:
failures.append(f"Failed to restore the accounting database dump ({src})")
# In either case, remove the dump file
result = run_command(f"rm -f {src}", quiet=False)
if result["exit_code"] != 0:
failures.append(f"Failed to remove the accounting database dump ({src})")
return failures
def run_check_test(source_file, build_args=""):
"""Compiles and runs a libcheck test
Args:
source_file (string): The name of the source file, relative to testsuite_check_dir/tests.
build_args (string): Additional string to be appended to the build command.
"""
import xmltodict
check_test = (
f"{module_tmp_path}/{os.path.splitext(os.path.basename(source_file))[0]}"
)
xml_test = check_test + ".xml"
# Add python/data and the scheck helpers to the include path
build_args = " ".join(
[
build_args,
"-lcheck -lm -lsubunit",
f"-I{properties['testsuite_data_dir']}",
f"-I{properties['testsuite_check_dir']}/include",
]
)
compile_against_libslurm(
f"{properties['testsuite_check_dir']}/tests/{source_file}",
check_test,
full=True,
build_args=build_args,
fatal=True,
quiet=True,
)
# Run the libcheck test setting an xml output. Also export `srcdir`
# so the test can locate any sidecar file (e.g. topology.conf)
#
# stdbuf -oL forces line-buffered stdout so printf() output survives
# libcheck's _exit() on assertion failure (default fully-buffered pipe
# would otherwise discard the buffered lines).
src_dir = os.path.dirname(
f"{properties['testsuite_check_dir']}/tests/{source_file}"
)
result = run_command(
f"stdbuf -oL {check_test}",
quiet=True,
env_vars=f"CK_XML_LOG_FILE_NAME={xml_test} srcdir={src_dir}",
)
logging.info(f"{result['stdout']}")
if result["exit_code"]:
logging.error(f"\n{result['stderr']}")
# Parse the xml output
if not os.path.exists(xml_test):
pytest.fail(f"Test results not found: {xml_test}")
with open(xml_test) as f:
xml_data = xmltodict.parse(f.read())
return xml_data["testsuites"]["suite"]
def compile_against_libslurm(
source_file,
dest_file,
build_args="",
full=False,
shared=False,
new_prefixes=False,
**run_command_kwargs,
):
"""Compiles a test program against either libslurm.so or libslurmfull.so.
This function compiles the specified source file against the Slurm library,
either libslurm.so or libslurmfull.so, and creates the target binary file.
Args:
source_file (string): The name of the source file.
dest_file (string): The name of the target binary file.
build_args (string): Additional string to be appended to the build command.
full (boolean): Use libslurmfull.so instead of libslurm.so.
shared (boolean): Produces a shared library (adds the -shared compiler option
and adds a .so suffix to the output file name).
**run_command_kwargs: Auxiliary arguments to be passed to the
run_command function (e.g., quiet, fatal, timeout, etc.).
Returns:
None
Example:
>>> compile_against_libslurm("my_test.c", "my_test", build_args="-Wall -Werror")
"""
slurm_prefix = properties["slurm-prefix"]
slurm_source = properties["slurm-source-dir"]
slurm_build = properties["slurm-build-dir"]
if new_prefixes:
slurm_prefix = properties["new-slurm-prefix"]
slurm_source = properties["new-source-prefix"]
slurm_build = properties["new-build-prefix"]
if full:
slurm_library = "slurmfull"
else:
slurm_library = "slurm"
if os.path.isfile(f"{slurm_prefix}/lib64/slurm/lib{slurm_library}.so"):
lib_dir = "lib64"
else:
lib_dir = "lib"
if full:
lib_path = f"{slurm_prefix}/{lib_dir}/slurm"
else:
lib_path = f"{slurm_prefix}/{lib_dir}"
command = f"gcc {source_file} -g -pthread"
if shared:
command += " -fPIC -shared"
command += f" -o {dest_file}"
command += f" -I{slurm_source} -I{slurm_build} -I{slurm_prefix}/include -Wl,-rpath={lib_path} -L{lib_path} -l{slurm_library} -lresolv"
if build_args != "":
command += f" {build_args}"
run_command(command, **run_command_kwargs)
def get_partitions(**run_command_kwargs):
"""Returns the Slurm partition configuration as a dictionary of dictionaries.
Args:
**run_command_kwargs: Auxiliary arguments to be passed to the
run_command function (e.g., quiet, fatal, timeout, etc.).
Returns:
A dictionary of dictionaries, where the first-level keys are the
partition names, and the values are dictionaries containing the
configuration parameters for the respective partitions.
Example:
>>> get_partitions(quiet=True)
{'partition1': {'PartitionName': 'partition1', 'AllowGroups': 'ALL', 'Defaults': 'YES', ...},
'partition2': {'PartitionName': 'partition2', 'AllowGroups': 'group1,group2', 'Defaults': 'YES', ...}}
"""
partitions_dict = {}
output = run_command_output("scontrol show partition -o", **run_command_kwargs)
partition_dict = {}
for line in output.splitlines():
if line == "":
continue
while match := re.search(r"^ *([^ =]+)=(.*?)(?= +[^ =]+=| *$)", line):
param_name, param_value = match.group(1), match.group(2)
# Remove the consumed parameter from the line
line = re.sub(r"^ *([^ =]+)=(.*?)(?= +[^ =]+=| *$)", "", line)
# Reformat the value if necessary
if is_integer(param_value):
param_value = int(param_value)
elif is_float(param_value):
param_value = float(param_value)
elif param_value == "(null)":
param_value = None
# Add it to the temporary partition dictionary
partition_dict[param_name] = param_value
# Add the partition dictionary to the partitions dictionary
partitions_dict[partition_dict["PartitionName"]] = partition_dict
# Clear the partition dictionary for use by the next partition
partition_dict = {}
return partitions_dict
def get_partition_parameter(partition_name, parameter_name, default=None):
"""Obtains the value for a Slurm partition configuration parameter.
This function retrieves the value of the specified parameter for the given
partition. If the parameter is not present, the default value is returned.
Args:
partition_name (string): The name of the partition.
parameter_name (string): The name of the parameter to retrieve.
default (string or None): The default value to return if the parameter is
not found.
Returns:
The value of the specified partition parameter, or the default if not
found.
Example:
>>> get_partition_parameter('my_partition', 'AllowAccounts')
'ALL'
>>> get_partition_parameter('second_partition', 'DefaultTime', '00:30:00')
'00:30:00'
"""
partitions_dict = get_partitions()
if partition_name in partitions_dict:
partition_dict = partitions_dict[partition_name]
else:
pytest.fail(
f"Partition ({partition_name}) was not found in the partition configuration"
)
if parameter_name in partition_dict:
return partition_dict[parameter_name]
else:
return default
def set_partition_parameter(partition_name, new_parameter_name, new_parameter_value):
"""Sets the value of the specified partition configuration parameter.
This function the specified partition property and reconfigures
the slurm daemons. A backup is automatically created and the
original configuration is restored after the test completes.
This function may only be used in auto-config mode.
Args:
partition_name (string): The partition name.
new_parameter_name (string): The parameter name.
new_parameter_value (string): The parameter value.
Use a value of None to unset a partition parameter.
Example:
>>> set_partition_parameter('partition1', 'MaxTime', 'INFINITE')
"""
if not properties["auto-config"]:
require_auto_config("wants to modify partition parameters")
config_file = f"{properties['slurm-config-dir']}/slurm.conf"
# Read the original slurm.conf into a list of lines
output = run_command_output(
f"cat {config_file}", user=properties["slurm-user"], quiet=True
)
original_config_lines = output.splitlines()
new_config_lines = original_config_lines.copy()
# Locate the partition among the various Partition definitions
found_partition_name = False
for line_index in range(len(original_config_lines)):
line = original_config_lines[line_index]
words = re.split(r" +", line.strip())
if not words[0]:
continue
if words[0][0] == "#":
continue
parameter_name, parameter_value = words[0].split("=", 1)
if parameter_name.lower() != "partitionname":
continue
if parameter_value == partition_name:
# We found a matching PartitionName line
found_partition_name = True
# Read in the partition parameters
original_partition_parameters = collections.OrderedDict()
for word in words[1:]:
parameter_name, parameter_value = word.split("=", 1)
original_partition_parameters[parameter_name] = parameter_value
# Delete the original partition definition
new_config_lines.pop(line_index)
# Add the modified definition for the specified partition
modified_partition_parameters = original_partition_parameters.copy()
if new_parameter_value is None:
if new_parameter_name in modified_partition_parameters:
del modified_partition_parameters[new_parameter_name]
else:
modified_partition_parameters[new_parameter_name] = new_parameter_value
modified_partition_line = f"PartitionName={partition_name}"
for (
parameter_name,
parameter_value,
) in modified_partition_parameters.items():
modified_partition_line += f" {parameter_name}={parameter_value}"
new_config_lines.insert(line_index, modified_partition_line)
break
if not found_partition_name:
pytest.fail(
f"Invalid partition name specified in set_partition_parameter(). Partition {partition_name} does not exist"
)
# Write the config file back out with the modifications
new_config_string = "\n".join(new_config_lines)
run_command(
f"echo '{new_config_string}' > {config_file}",
user=properties["slurm-user"],
fatal=True,
quiet=True,
)
# Reconfigure slurm controller if it is already running
if is_slurmctld_running(quiet=True):
run_command("scontrol reconfigure", user=properties["slurm-user"], quiet=True)
def default_partition():
"""Returns the name of the default Slurm partition.
This function retrieves the Slurm partition configuration and returns the
name of the partition that is marked as the default.
Args:
None
Returns:
The name of the default Slurm partition.
Example:
>>> default_partition()
'my_default_partition'
"""
partitions_dict = get_partitions()
for partition_name in partitions_dict:
if partitions_dict[partition_name]["Default"] == "YES":
return partition_name
def get_run_dir_path():
if "slurm-run-dir" not in properties:
properties["slurm-run-dir"] = get_config_parameter(
"SlurmdPidFile", live=False, quiet=True
)
return os.path.dirname(properties["slurm-run-dir"])
# This is supplied for ease-of-use in test development only.
# Tests should not use this permanently. Use logging.debug() instead.
def log_debug(msg):
logging.debug(msg)
##############################################################################
# ATF module initialization
##############################################################################
# This is a logging filter that adds a new LogRecord traceback attribute
class TraceBackFilter(logging.Filter):
def filter(self, record):
call_stack = []
within_atf_context = False
for frame_summary in (traceback.extract_stack())[-5::-1]:
if within_atf_context:
if "testsuite/python" not in frame_summary.filename:
break
else:
if "testsuite/python" in frame_summary.filename:
within_atf_context = True
else:
continue
function = frame_summary.name
short_filename = frame_summary.filename.rpartition("testsuite/python/")[2]
lineno = frame_summary.lineno
call_stack.append(f"{function}@{short_filename}:{lineno}")
record.traceback = ",".join(call_stack)
return True
# Add a new traceback LogRecord attribute
logging.getLogger().addFilter(TraceBackFilter())
# Add a custom TRACE logging level
# This has to be done early enough to allow pytest --log-level=TRACE to be used
logging.TRACE = logging.NOTSET + 5
logging.addLevelName(logging.TRACE, "TRACE")
def _trace(message, *args, **kwargs):
logging.log(logging.TRACE, message, *args, **kwargs)
logging.trace = _trace
logging.getLogger().trace = _trace
# Add a custom NOTE logging level in between INFO and DEBUG
logging.NOTE = logging.DEBUG + 5
logging.addLevelName(logging.NOTE, "NOTE")
def _note(message, *args, **kwargs):
logging.log(logging.NOTE, message, *args, **kwargs)
logging.note = _note
logging.getLogger().note = _note
# The module-level temporary directory is initialized in conftest.py
module_tmp_path = None
# Instantiate and populate testrun-level properties
properties = {}
# Initialize directory properties
properties["testsuite_base_dir"] = str(pathlib.Path(__file__).resolve().parents[2])
properties["testsuite_python_lib"] = properties["testsuite_base_dir"] + "/python/lib"
properties["slurm-source-dir"] = str(pathlib.Path(__file__).resolve().parents[3])
properties["slurm-build-dir"] = properties["slurm-source-dir"]
properties["slurm-prefix"] = "/usr/local"
properties["testsuite_scripts_dir"] = (
properties["testsuite_base_dir"] + "/python/scripts"
)
properties["testsuite_data_dir"] = properties["testsuite_base_dir"] + "/python/data"
properties["testsuite_check_dir"] = properties["testsuite_base_dir"] + "/python/check"
properties["influxdb_host"] = "localhost"
properties["influxdb_port"] = 8086
properties["influxdb_db"] = "slurm"
# Override directory properties with values from testsuite.conf file
testsuite_config = {}
# The default location for the testsuite.conf file (in SRCDIR/testsuite)
# can be overridden with the SLURM_TESTSUITE_CONF environment variable.
testsuite_config_file = os.getenv(
"SLURM_TESTSUITE_CONF", f"{properties['testsuite_base_dir']}/testsuite.conf"
)
if not os.path.isfile(testsuite_config_file):
pytest.fail(
f"The unified testsuite configuration file (testsuite.conf) was not found. This file can be created from a copy of the sample provided in the source tree at SRCDIR/testsuite/testsuite.conf.sample. By default, this file is expected to be found in SRCDIR/testsuite ({properties['testsuite_base_dir']}). If placed elsewhere, set the SLURM_TESTSUITE_CONF environment variable to the full path of your testsuite.conf file."
)
with open(testsuite_config_file, "r") as f:
for line in f.readlines():
if match := re.search(r"^\s*(\w+)\s*=\s*(.*)$", line):
testsuite_config[match.group(1).lower()] = match.group(2)
if "slurmsourcedir" in testsuite_config:
properties["slurm-source-dir"] = testsuite_config["slurmsourcedir"]
if "slurmbuilddir" in testsuite_config:
properties["slurm-build-dir"] = testsuite_config["slurmbuilddir"]
if "slurminstalldir" in testsuite_config:
properties["slurm-prefix"] = testsuite_config["slurminstalldir"]
if "slurmconfigdir" in testsuite_config:
properties["slurm-config-dir"] = testsuite_config["slurmconfigdir"]
if "influxdb_host" in testsuite_config:
properties["influxdb_host"] = testsuite_config["influxdb_host"]
if "influxdb_port" in testsuite_config:
properties["influxdb_port"] = testsuite_config["influxdb_port"]
if "influxdb_db" in testsuite_config:
properties["influxdb_db"] = testsuite_config["influxdb_db"]
# TODO: The SlurmConfigDirBase should be passed from the testsuite.conf too,
# instead of the hadcoded prefix+etc.orig form here.
properties["slurm-config-orig-dir"] = properties["slurm-prefix"] + "/etc.orig"
# Set derived directory properties
# The environment (e.g. PATH, SLURM_CONF) overrides the configuration.
# If the Slurm clients and daemons are not in the current PATH
# but can be found using the configured SlurmInstallDir, add the
# derived bin and sbin dir to the current PATH.
properties["slurm-bin-dir"] = f"{properties['slurm-prefix']}/bin"
if squeue_path := shutil.which("squeue"):
properties["slurm-bin-dir"] = os.path.dirname(squeue_path)
elif os.access(f"{properties['slurm-bin-dir']}/squeue", os.X_OK):
os.environ["PATH"] += ":" + properties["slurm-bin-dir"]
properties["slurm-sbin-dir"] = f"{properties['slurm-prefix']}/sbin"
if slurmctld_path := shutil.which("slurmctld"):
properties["slurm-sbin-dir"] = os.path.dirname(slurmctld_path)
elif os.access(f"{properties['slurm-sbin-dir']}/slurmctld", os.X_OK):
os.environ["PATH"] += ":" + properties["slurm-sbin-dir"]
properties["slurm-config-dir"] = re.sub(
r"\${prefix}", properties["slurm-prefix"], properties["slurm-config-dir"]
)
if slurm_conf_path := os.getenv("SLURM_CONF"):
properties["slurm-config-dir"] = os.path.dirname(slurm_conf_path)
# Derive the slurm-user value
properties["slurm-user"] = "root"
slurm_config_file = f"{properties['slurm-config-dir']}/slurm.conf"
if not os.path.isfile(slurm_config_file):
pytest.fail(
f"The python testsuite was expecting your slurm.conf to be found in {properties['slurm-config-dir']}. Please create it or use the SLURM_CONF environment variable to indicate its location."
)
if os.access(slurm_config_file, os.R_OK):
with open(slurm_config_file, "r") as f:
for line in f.readlines():
if match := re.search(r"^\s*(?i:SlurmUser)\s*=\s*(.*)$", line):
properties["slurm-user"] = match.group(1)
else:
# slurm.conf is not readable, we will try reading it as root
results = run_command(
f"grep -i SlurmUser {slurm_config_file}", user="root", quiet=True
)
if results["exit_code"] == 0:
pytest.fail(f"Unable to read {slurm_config_file}")
for line in results["stdout"].splitlines():
if match := re.search(r"^\s*(?i:SlurmUser)\s*=\s*(.*)$", line):
properties["slurm-user"] = match.group(1)
properties["submitted-jobs"] = []
properties["lmod-touched-vars"] = set()
if "slurmtestuser" in testsuite_config:
properties["test-user"] = testsuite_config["slurmtestuser"]
properties["test-user-set"] = True
else:
properties["test-user"] = pwd.getpwuid(os.getuid()).pw_name
properties["test-user-set"] = False
properties["test-user-uid"] = pwd.getpwnam(properties["test-user"]).pw_uid
properties["test-user-gid"] = pwd.getpwnam(properties["test-user"]).pw_gid
properties["auto-config"] = False
properties["allow-slurmdbd-modify"] = False
properties["slurmrestd-started"] = False
# Accounting state of the module: None if not configured, "enabled" if
# configured, or "db-saved" if also the original DB was already saved (and
# dropped, in auto-config). Only updated by start_slurmdbd() and
# set_config_parameter("AccountingStorageType"); reset by conftest's
# module_setup().
properties["accounting"] = None
# Instantiate a nodes dictionary. These are populated in require_slurm_running.
nodes = {}
# Check if user has sudo privileges
results = subprocess.run(
"sudo -ln | grep -q '(ALL.*) NOPASSWD: ALL'",
shell=True,
capture_output=True,
text=True,
)
if results.returncode == 0:
properties["sudo-rights"] = True
else:
properties["sudo-rights"] = False