Job execution and submission#

Modules managing the execution and submission of jobs to computing clusters. This is mostly used for small productions during the validation or verification phase.

Inspect logs and output for errors and warnings.

job_execution.log_inspector.check_plain_logs(log_files, file_test)[source]#

Check plain log file(s) for wanted and forbidden patterns.

Log file can be plain or gzipped.

Parameters:

log_filesList, str: Path to the log file.
file_testdict: Dictionary with the test configuration.

Returns:

bool: True if the logs are correct.

job_execution.log_inspector.check_tar_logs(tar_file, file_test)[source]#

Check log files in tar package for wanted and forbidden patterns.

Parameters:

tar_filestr: Path to the tar file.
file_testdict: Dictionary with the test configuration.

Returns:

bool: True if the logs are correct.

job_execution.log_inspector.collect_final_lines(file, n_lines)[source]#

Collect final lines.

Parameters:

file: str or Path: File to collect the lines from.
n_lines: int: Number of lines to be collected.

Returns:

str: Final lines collected.

job_execution.log_inspector.get_log_excerpt(log_file, n_last_lines=30)[source]#

Get an excerpt from a log file, namely the n_last_lines of the file.

Parameters:

log_file: str or Path: Log file to get the excerpt from.
n_last_lines: int: Number of last lines of the file to get.

Returns:

str: Excerpt from log file with header/footer

job_execution.log_inspector.inspect(log_text)[source]#

Inspect log text for errors and run-time warnings.

Ignore any lines containing “INFO::” (to avoid false positives like “INFO:: Job error stream “).

Parameters:

log_text: str or list of str: Text of the log to inspect.

Returns:

bool: True if no errors or warnings are found, False otherwise.

job_execution.log_inspector.is_safe_tar_member(member_name)[source]#

Validate that a tar member path is safe and doesn’t contain path traversal sequences.

Parameters:

member_namestr: The path of the tar member to validate.

Returns:

bool: True if the path is safe, False otherwise.

Run one process while recording its directly attributable resource use.

job_execution.process_accounting.build_accounting_command(command, output_path, role, run_id, model_version=None, log_file=None, input_file=None)[source]#

Build a command that records resource use for one executable.

The returned command preserves the executable’s stdin unless input_file is supplied. This is important for sim_telarray consumers of an eventIO stream.

Parameters:

commandlist[str | pathlib.Path]: Executable and arguments to run.
output_pathstr or pathlib.Path: JSON resource-record output path.
rolestr: Process role, such as "corsika" or "sim_telarray".
run_idint or str: Production run identifier.
model_versionstr, optional: Simulation-model version associated with this process.
log_filestr or pathlib.Path, optional: Gzip-compressed stdout log file. Stderr is written to a sibling .err.gz file, replacing the .log.gz suffix.
input_filestr or pathlib.Path, optional: File connected to the child’s stdin.

Returns:

list[str]: Command invoking this module with the supplied process command.

job_execution.process_accounting.main()[source]#: Run the requested command and return its exit status.

job_execution.process_accounting.run_and_record(command, output_path, role, run_id, model_version=None, log_file=None, input_file=None, env=None)[source]#

Run a process and write a JSON resource record for its direct PID.

On Linux, user CPU time, system CPU time, and peak resident memory are sampled from /proc/<pid>. Other platforms still produce a record with wall-clock duration and clearly identify unavailable measurements.

Parameters:

commandlist[str | pathlib.Path]: Executable and arguments to run.
output_pathstr or pathlib.Path: JSON resource-record output path.
rolestr: Process role.
run_idint or str: Production run identifier.
model_versionstr, optional: Simulation-model version associated with this process.
log_filestr or pathlib.Path, optional: Gzip-compressed stdout log file. Stderr is written to a sibling .err.gz file, replacing the .log.gz suffix.
input_filestr or pathlib.Path, optional: File connected to the child’s stdin.
envdict, optional: Environment overrides for the child process.

Returns:

int: Child-process return code.

Public execution facade used by long-running applications.

exception job_execution.execution.JobExecutionError[source]#: Raised when a command job fails.

job_execution.execution.collect_submission(submission, *, backend=None)[source]#

Collect finished scheduler jobs without waiting for active jobs.

Parameters:

submissionSubmissionHandle: Persisted scheduler submission to inspect.
backendobject or None: Execution backend; constructed from the submission when omitted.

Returns:

list[JobResult] or None: Completed results, or None while jobs remain in the scheduler queue.

job_execution.execution.execute_jobs(job_specs, options=None)[source]#: Execute ordered jobs through the selected backend.

job_execution.execution.load_submission(path)[source]#: Load a submission handle from a JSON manifest.

job_execution.execution.map_ordered(function, items, *, backend='local', max_workers=None, backend_config=None, initializer=None, initargs=(), mp_start_method='fork', work_dir=None, context=None)[source]#: Apply function to items and return values in input order.

job_execution.execution.options_from_args(args, *, max_workers=None, work_dir=None)[source]#: Build execution options from an application argument dictionary.

job_execution.execution.run_command(command, *, stdin=None, stdout=None, stderr=None, env=None, runtime_environment=None, configuration=None, application_log=None, check=True, capture_output=True)[source]#

Run one command synchronously through the execution worker.

Parameters:

commandstr or list[str] or Path: Command (a string or script path runs through the shell) or argument list.
stdinfile object or str or Path, optional: Standard input: an open file, or a file path that is opened for reading.
stdout, stderrstr or Path, optional: Files receiving standard output and standard error.
envdict, optional: Environment variables added to the current environment.
runtime_environmentstr or list[str], optional: Command prefix running the command in a container.
configurationdict, optional: Application configuration appended as command-line arguments.
application_logstr or Path, optional: Application log; an excerpt is added to the error message on failure.
checkbool, optional: Raise JobExecutionError on a non-zero return code.
capture_outputbool, optional: Capture standard output and error that are not written to files; otherwise they are inherited.

Returns:

JobResult: Return code, wall-clock runtime and the captured output (output) and error output (error_output).

Raises:

JobExecutionError: If the command fails and check is True; the message contains the return code, the captured standard error and excerpts from the log files.

job_execution.execution.submit_jobs(job_specs, options=None)[source]#

Submit independent jobs, detaching when the backend supports it.

Scheduler-backed implementations return without waiting. Backends that do not advertise submit-only support complete synchronously.

job_execution.execution.wait_for_submission(submission, *, options=None, backend=None)[source]#: Wait for a previously submitted job set and validate its outputs.

Backend-neutral job and execution data structures.

class job_execution.job.CommandOptions(stdin=None, stdout=None, stderr=None, env=None, runtime_environment=None, configuration=None, application_log=None, check=True, capture_output=False)[source]#: Streams, environment and checks of a command job.

class job_execution.job.ExecutionOptions(backend='local', max_workers=None, work_dir=None, backend_config=<factory>, initializer=None, initargs=())[source]#: Options shared by execution backends.

class job_execution.job.JobResult(job_id, index, value=None, status='completed', return_code=0, stdout=None, stderr=None, error=None, runtime=0.0, output=None, error_output=None)[source]#: Result and diagnostic information for one job.

class job_execution.job.JobSpec(job_id, index, function=None, item=None, command=None, options=None, initializer=None, initargs=(), context_spec=None, metadata=<factory>, resources=<factory>, mount_paths=(), output_paths=())[source]#: Describe one independent unit of work.

class job_execution.job.SubmissionHandle(backend, work_dir, job_ids, scheduler_id=None, process_ids=<factory>, metadata=<factory>)[source]#

Persistable description of a submitted execution.

as_dict()[source]#: Return a JSON-serializable manifest representation.

classmethod from_dict(payload)[source]#: Restore a submission handle from a manifest mapping.

Protocol and exceptions shared by execution backends.

exception job_execution.backends.base.BackendConfigurationError[source]#: Raised when backend configuration is invalid or unavailable.

exception job_execution.backends.base.BackendError[source]#: Base class for backend failures.

exception job_execution.backends.base.BackendExecutionError[source]#: Raised when one or more submitted jobs fail.

exception job_execution.backends.base.BackendSubmissionError[source]#: Raised when a backend cannot submit a job set.

class job_execution.backends.base.ExecutionBackend(*args, **kwargs)[source]#

Protocol implemented by all job-execution backends.

cancel(submission)[source]#: Cancel an active submission.

collect(submission)[source]#: Collect results after a submission has finished.

is_finished(submission)[source]#: Return whether a submission has completed.

submit(job_specs, options)[source]#: Submit jobs and return a handle.

wait(submission)[source]#: Wait for completion and return ordered results.

Local process execution backend.

class job_execution.backends.local.LocalBackend[source]#

Execute job specifications with local worker processes.

cancel(submission)[source]#: Cancel pending local jobs.

collect(submission)[source]#: Return direct results after local execution.

is_finished(submission)[source]#: Return whether direct local results are already available.

submit(job_specs, options)[source]#: Submit jobs to a local process pool.

wait(submission)[source]#: Collect local results in input order.

job_execution.backends.local.determine_max_workers(max_workers=None, default_fraction=0.6)[source]#: Determine the local worker count.

HTCondor execution backend using the Python bindings.

class job_execution.backends.htcondor.HTCondorBackend[source]#

Submit one HTCondor process for every job specification.

cancel(submission)[source]#: Remove all processes in an active cluster.

collect(submission)[source]#

Read terminal job results without polling or waiting.

Parameters:

submissionSubmissionHandle: Submission already confirmed finished with is_finished.

Returns:

list[JobResult]: Validated results in submission order.

Raises:

BackendExecutionError: A worker failed or a required output is missing.

is_finished(submission)[source]#

Confirm every submitted process has left the queue and reached a terminal state.

Parameters:

submissionSubmissionHandle: Submission whose scheduler state is checked without waiting.

Returns:

bool: False when jobs remain in the scheduler queue; True when history confirms completion.

Raises:

BackendExecutionError: Scheduler state cannot be determined for every process.

submit(job_specs, options)[source]#: Create and submit a shared-filesystem HTCondor cluster.

wait(submission)[source]#: Wait for all HTCondor processes and load their result files.

Registry for execution backend implementations.

job_execution.backends.registry.available_backends()[source]#: Return registered backend names.

job_execution.backends.registry.get_backend(name)[source]#: Instantiate a registered backend.

Execute one serialized job specification for a remote backend.

job_execution.worker.execute_command(job_spec)[source]#

Run a command job with its configured streams and environment.

Returns:

tuple: Return code, wall-clock runtime in seconds, and the captured standard output and standard error (None unless captured).

job_execution.worker.execute_job_spec(job_spec)[source]#: Execute one Python or command job specification.

job_execution.worker.main()[source]#: Run the worker CLI.

job_execution.worker.run(run_directory, job_id, log_file=None)[source]#: Run one serialized job and write its result or failure record.

job_execution.worker.validate_job_id(job_id)[source]#: Validate a job ID used in worker-owned file names.

job_execution.worker.write_job_payload(job_spec, path)[source]#: Serialize a versioned worker payload.