PYSURFEX scheduler documentation

https://coveralls.io/github/metno/pysurfex-scheduler

Python abstraction layer for a scheduling system like Ecflow

See online documentation in https://metno.github.io/pysurfex-scheduler/

Installation of pregenerated packages from pypi (pip)

pip3 install pysurfex-scheduler

User installation:

pip3 install pysurfex-scheduler --user

Installation on debian based Linux system

Install the required pacakges (some might be obsolete if the pip packages contain the needed depedencies):

sudo apt-get update
# Python tools
sudo apt-get install python3-setuptools
# Ecflow
sudo apt-get install ecflow-server ecflow-client python3-ecflow

The following depencies are needed. Install the non-standard ones e.g. with pip or your system installation system.

General dependencies (from pypi)

toml
json; python_version < '3'

For testing:

unittest
nose

Download the source code, then install pysurfex-scheduler by executing the following inside the extracted folder:

sudo pip3 install -e .

or

pip3 install -e . --user

Create documentation

cd docs
# Create html documentation
make html
# Create latex documentation
make latex
# Create a pdf documentation
make latexpdf

Usage

import scheduler

# EcFlow variables parsed in EcFlow job
ecf_name = "%ECF_NAME%"
ecf_pass = "%ECF_PASS%"
ecf_tryno = "%ECF_TRYNO%"
ecf_rid = "%ECF_RID%"
submission_id = "%SUBMISSION_ID%"
task_name = "%TASK%"

task = scheduler.EcflowTask(ecf_name, ecf_tryno, ecf_pass, ecf_rid, submission_id)
env_submit = {}
env_server = {
  "ECF_HOST": "localhost",
  "ECF_PORT": 3141,
  "ECF_PORT_OFFSET": 0
}
joboutdir = "/tmp/job"

sub = scheduler.EcflowSubmitTask(task, env_submit, env_server, joboutdir)
sub.submit()

Running an experiment in EcFlow

See examples in unit tests (test directory)

Classes

class scheduler.SuiteDefinition(suite_name, joboutdir, ecf_files, env_submit, ecf_home=None, ecf_include=None, ecf_out=None, ecf_jobout=None, ecf_job_cmd=None, ecf_status_cmd=None, ecf_kill_cmd=None, pythonpath='', path='')

The definition of the suite.

Parameters:

object (_type_) – _description_

class scheduler.EcflowSuite(name, **kwargs)

EcflowSuite.

Parameters:

EcflowNodeContainer (EcflowNodeContainer) – A child of the EcflowNodeContainer class.

class scheduler.EcflowSuiteTriggers(triggers, **kwargs)

Triggers to an ecflow suite.

class scheduler.EcflowSuiteTrigger(node, mode='complete')

EcFlow Trigger in a suite.

class scheduler.EcflowSuiteVariable(name, value)

A variable in an ecflow suite.

class scheduler.EcflowSuiteFamily(name, parent, **kwargs)

A family in ecflow.

Parameters:

EcflowNodeContainer (_type_) – _description_

class scheduler.EcflowSuiteTask(name, parent, **kwargs)

A task in an ecflow suite/family.

Parameters:

EcflowNode (EcflowNodeContainer) – The node container.

class scheduler.EcflowSubmitTask(task, env_submit, server, joboutdir, stream=None, dbfile=None, interpreter='#!/usr/bin/env python3', ensmbr=None, submit_exceptions=None, coldstart=False, env_file=None)

Submit class for ecflow.

class scheduler.TaskSettings(task, submission_defs, joboutdirs, submit_exceptions=None, interpreter='#!/usr/bin/env python3', complete=False, coldstart=False)

Set the task specific setttings.

class scheduler.SubmitException(msg, task, task_settings)

Submit exception.

class scheduler.KillException(msg, task, task_settings)

Kill exception.

class scheduler.StatusException(msg, task, task_settings)

Status exception.

class scheduler.BackgroundSubmission(task, task_settings, server, db_file=None)

Backgrpund submission.

A subprocess on the host system.

Parameters:

SubmissionBaseClass (_type_) – _description_

class scheduler.BatchSubmission(task, task_settings, server, db_file=None, sub=None, stat=None, kill=None, prefix='#')

A general batch system class for a task using a job scheduler system.

Parameters:

SubmissionBaseClass (_type_) – _description_

class scheduler.PBSSubmission(task, task_settings, server, sub='qsub', stat='qstat -j', kill='qdel', prefix='#PBS', db_file=None)

Job submission on a PBS job scheduler.

Parameters:

BatchSubmission (_type_) – _description_

class scheduler.SlurmSubmission(task, task_settings, server, sub='sbatch', stat='squeue -j', kill='scancel', prefix='#SBATCH', db_file=None)

General slurm submssion class.

Parameters:

BatchSubmission (_type_) – _description_

class scheduler.GridEngineSubmission(task, task_settings, server, db_file=None, sub='qsub', stat='qstat -j', kill='qdel', prefix='#$')

Sun Grid Engine (SGE) job submission.

Parameters:

BatchSubmission (_type_) – _description_

class scheduler.Server

Base server/scheduler class.

class scheduler.EcflowServer(ecf_host, ecf_port, logfile)

Ecflow server.

Parameters:

Server (Server) – Is a child of the base server.

class scheduler.EcflowServerFromFile(ecflow_server_file, logfile)

Construct an ecflow server from a config file.

class scheduler.EcflowLogServer(ecf_loghost, ecf_logport)

Ecflow log server.

class scheduler.EcflowTask(ecf_name, ecf_tryno, ecf_pass, ecf_rid, submission_id=None, ecf_timeout=20)

Ecflow scheduler task.

class scheduler.EcflowClient(server, task)

An ecflow client.

Encapsulate communication with the ecflow server. This will automatically call the child command init()/complete(), for job start/finish. It will also handle exceptions and signals, by calling the abort child command. ONLY one instance of this class, should be used. Otherwise zombies will be created.

Class methods

SuiteDefinition.__init__(suite_name, joboutdir, ecf_files, env_submit, ecf_home=None, ecf_include=None, ecf_out=None, ecf_jobout=None, ecf_job_cmd=None, ecf_status_cmd=None, ecf_kill_cmd=None, pythonpath='', path='')

Construct the definition.

Parameters:

• suite_name (_type_) – _description_

• joboutdir (_type_) – _description_

• ecf_files (_type_) – _description_

• env_submit (_type_) – _description_

• ecf_home (_type_, optional) – _description_. Defaults to None.

• ecf_include (_type_, optional) – _description_. Defaults to None.

• ecf_out (_type_, optional) – _description_. Defaults to None.

• ecf_jobout (_type_, optional) – _description_. Defaults to None.

• ecf_job_cmd (_type_, optional) – _description_. Defaults to None.

• ecf_status_cmd (_type_, optional) – _description_. Defaults to None.

• ecf_kill_cmd (_type_, optional) – _description_. Defaults to None.

• pythonpath (str, optional) – _description_. Defaults to “”.

• path (str, optional) – _description_. Defaults to “”.

Raises:

Exception – _description_

SuiteDefinition.save_as_defs(def_file)

Save definition file.

Parameters:

def_file (str) – Name of definition file

EcflowSuite.__init__(name, **kwargs)

Construct the Ecflow suite.

Parameters:

name (_type_) – _description_

EcflowSuite.save_as_defs(def_file)

Save defintion file.

Parameters:

def_file (str) – Name of the definition file.

EcflowSuiteTriggers.__init__(triggers, **kwargs)

Construct EcflowSuiteTriggers.

Parameters:

triggers (list) – List of EcflowSuiteTrigger objects.

static EcflowSuiteTriggers.create_string(triggers, mode)

Create the trigger string.

Parameters:

• triggers (list) – List of trigger objects

• mode (str) – Concatenation type.

Raises:

• Exception – _description_

• Exception – _description_

Returns:

The trigger string based on trigger objects.

Return type:

str

EcflowSuiteTriggers.add_triggers(triggers, mode='AND')

Add triggers.

Parameters:

• triggers (EcflowSuiteTriggers) – The triggers

• mode (str, optional) – Cat mode. Defaults to “AND”.

EcflowSuiteTrigger.__init__(node, mode='complete')

Create a EcFlow trigger object.

Parameters:

• node (scheduler.EcflowNode) – The node to trigger on

• mode (str) –

EcflowSuiteVariable.__init__(name, value)

Constuct the EcflowSuiteVariable.

Parameters:

• name (str) – Name

• value (str) – Value

EcflowSuiteFamily.__init__(name, parent, **kwargs)

Construct the family.

Parameters:

• name (str) – Name of the family.

• parent (EcflowNodeContainer) – Parent node.

EcflowSuiteTask.__init__(name, parent, **kwargs)

Constuct the EcflowSuiteTask.

Parameters:

• name (str) – Name of task

• parent (EcflowNode) – Parent node.

Raises:

Exception – _description_

EcflowSubmitTask.write_header(file_handler)

Write header to file handler.

Parameters:

file_handler (_type_) – _description_

Returns:

_description_

Return type:

_type_

EcflowSubmitTask.write_trailer(file_handler)

Write trailer in job file.

Parameters:

file_handler (_type_) – _description_

EcflowSubmitTask.write_job()

Write job file.

EcflowSubmitTask.submit()

Sumit task.

TaskSettings.check_exceptions(submit_exceptions)

Possibility to create submission exceptions.

Parameters:

submit_exceptions (_type_) – _description_

TaskSettings.process_settings()

Process the settings.

TaskSettings.parse_submission_defs()

Parse the submssion definitions.

SubmitException.__init__(msg, task, task_settings)

Construct SubmitException.

Parameters:

• msg (str) – Exception message.

• task (scheduler.EcflowTask) – Task.

• task_settings (TaskSettings) – Task settings.

KillException.__init__(msg, task, task_settings)

Construct KillException.

Parameters:

• msg (str) – Exception message.

• task (scheduler.EcflowTask) – Task.

• task_settings (TaskSettings) – Task settings.

StatusException.__init__(msg, task, task_settings)

Construct status exception.

Parameters:

• msg (str) – Exception message.

• task (scheduler.EcflowTask) – task.

• task_settings (TaskSettings) – Task settings

BackgroundSubmission.__init__(task, task_settings, server, db_file=None)

Construct BackgroundSubmission.

Parameters:

• task (scheduler.EcflowTask) – Task.

• task_settings (TaskSettings) – Task settings.

• server (scheduler.Server) – Server.

• db_file (str, optional) – Data base for monitoring. Defaults to None.

BackgroundSubmission.set_submit_cmd()

Set submit command.

BackgroundSubmission.set_jobid()

Set job id.

BackgroundSubmission.get_logfile()

Get the logfile.

BackgroundSubmission.set_kill_cmd()

Set the kill command.

BackgroundSubmission.set_job_status()

Set the job status.

BackgroundSubmission.set_output()

Set output.

BackgroundSubmission.set_job_name()

Set job name.

BatchSubmission.__init__(task, task_settings, server, db_file=None, sub=None, stat=None, kill=None, prefix='#')

Construct the BatchSubmission object.

Parameters:

• task (scheduler.EcflowTask) – Task.

• task_settings (TaskSettings) – Task settings.

• server (scheduler.Server) – Server.

• db_file (str, optional) – Data base for monitoring. Defaults to None.

• sub (str, optional) – Submission command. Defaults to None.

• stat (str, optional) – Status command. Defaults to None.

• kill (str, optional) – Kill command. Defaults to None.

• prefix (str, optional) – Batch prefix. Defaults to “#”.

BatchSubmission.set_submit_cmd()

Set submit command.

Parameters:

remote_cmd (_type_, optional) – _description_. Defaults to None.

BatchSubmission.set_jobid()

Set job id.

BatchSubmission.get_logfile()

Get the logfile.

BatchSubmission.set_kill_cmd()

Set kill command.

BatchSubmission.set_job_status()

Set job status.

BatchSubmission.set_output()

Set output.

BatchSubmission.set_job_name()

Set job name.

PBSSubmission.__init__(task, task_settings, server, sub='qsub', stat='qstat -j', kill='qdel', prefix='#PBS', db_file=None)

Construct the PBS job submission object.

Parameters:

• task (scheduler.EcflowTask) – Task.

• task_settings (TaskSettings) – Task settings.

• server (scheduler.Server) – Server.

• sub (str, optional) – Submission command. Defaults to “qsub”.

• stat (str, optional) – Status command. Defaults to “qstat -j”.

• kill (str, optional) – Kill command. Defaults to “qdel”.

• prefix (str, optional) – PBS prefix. Defaults to “#PBS”.

• db_file (str, optional) – Data base for monitoring. Defaults to None.

PBSSubmission.set_jobid()

Set job id.

PBSSubmission.set_job_name()

Set job name.

SlurmSubmission.__init__(task, task_settings, server, sub='sbatch', stat='squeue -j', kill='scancel', prefix='#SBATCH', db_file=None)

Construct SlurmSubmission.

Parameters:

• task (scheduler.EcflowTask) – Task.

• task_settings (TaskSettings) – Task settings.

• server (scheduler.Server) – Server.

• sub (str, optional) – Submission command. Defaults to “sbatch”.

• stat (str, optional) – Status command. Defaults to “squeue -j”.

• kill (str, optional) – Kill command. Defaults to “scancel”.

• prefix (str, optional) – Slurm prefix. Defaults to “#SBATCH”.

• db_file (_type_, optional) – Data base for monitoring. Defaults to None.

SlurmSubmission.set_output()

Set output.

SlurmSubmission.set_jobid()

Set job id.

SlurmSubmission.set_job_name()

Set job name.

GridEngineSubmission.__init__(task, task_settings, server, db_file=None, sub='qsub', stat='qstat -j', kill='qdel', prefix='#$')

Construct the GridEngineSubmission object.

Parameters:

• task (scheduler.EcflowTask) – Task

• task_settings (TaskSettings) – Task settings

• server (scheduler.Server) – Server.

• db_file (_type_, optional) – Data base for monitoring. Defaults to None.

• sub (str, optional) – Sumission command. Defaults to “qsub”.

• stat (str, optional) – Status command. Defaults to “qstat -j”.

• kill (str, optional) – Kill command. Defaults to “qdel”.

• prefix (str, optional) – SGE prefix. Defaults to “#$”.

GridEngineSubmission.set_output()

Set output.

GridEngineSubmission.set_jobid()

Set job id.

GridEngineSubmission.set_job_name()

Set job name.

Server.__init__()

Construct the server.

abstract Server.start_server()

Start the server.

Raises:

NotImplementedError – Must be implemented by the child server object.

abstract Server.replace(suite_name, def_file)

Create or change the suite definition.

Parameters:

• suite_name (str) – Name of the suite.

• def_file (str) – Name of the definition file.

Raises:

NotImplementedError – Must be implemented by the child server object.

Server.start_suite(suite_name, def_file, begin=True)

Start the suite.

All the servers have these methods implemented and can start the server in a server specific way.

Parameters:

• suite_name (str) – Name of the suite

• def_file (str) – Name of the definition file.

• begin (bool, optional) – If the suite should begin. Defaults to True.

EcflowServer.__init__(ecf_host, ecf_port, logfile)

Construct the EcflowServer.

Parameters:

• ecf_host (str) – Host name of the ecflow server.

• ecf_port (int) – Port to listen to.

• logfile (str) – Logfile for the scheduler.

Raises:

Exception – _description_

EcflowServer.start_server()

Start the server.

EcflowServer.force_complete(task)

Force the task complete.

Parameters:

task (scheduler.EcflowTask) – Task to force complete.

EcflowServer.force_aborted(task)

Force the task aborted.

Parameters:

task (scheduler.EcflowTask) – Task to force aborted.

EcflowServer.update_submission_id(task)

Update the submission id.

Parameters:

task (scheduler.EcflowTask) – Ecflow task.

EcflowServer.replace(suite_name, def_file)

Replace the suite name from def_file.

Parameters:

• suite_name (str) – Suite name.

• def_file (str) – Definition file.

Raises:

Exception – _description_

EcflowServer.update_log(text)

Update the log.

Parameters:

text (str) – Text to log

EcflowServerFromFile.__init__(ecflow_server_file, logfile)

Construct EcflowServer from a file.

Parameters:

• ecflow_server_file (str) – File with server definition

• logfile (str) – Logfile

Raises:

FileNotFoundError – if server file was not found.

EcflowServerFromFile.get_var(var, default=None)

Get variable setting.

Parameters:

• var (str) – Key in settings.

• default (_type_, optional) – _description_. Defaults to None.

Raises:

Exception – _description_

Returns:

_description_

Return type:

_type_

EcflowServerFromFile.save_as_file(fname)

Save the server settings to a file.

Parameters:

fname (str) – File name

EcflowLogServer.__init__(ecf_loghost, ecf_logport)

Constuct the ecflow log server.

Parameters:

• ecf_loghost (str) – Name of the loghost.

• ecf_logport (int) – Port to listen to.

EcflowTask.__init__(ecf_name, ecf_tryno, ecf_pass, ecf_rid, submission_id=None, ecf_timeout=20)

Construct a task running and communicating with ecflow server.

Parameters:

• ecf_name (str) – Full name of ecflow task.

• ecf_tryno (int) – Ecflow task try number

• ecf_pass (str) – Ecflow task password

• ecf_rid (int) – Ecflow runtime ID

• submission_id (str, optional) – Submssion ID from submission. Defaults to None.

• ecf_timeout (int, optional) – _description_. Defaults to 20.

EcflowTask.create_submission_log(joboutdir)

Create the submssion log file name.

Parameters:

joboutdir (str) – Location of ecflow created job files.

Returns:

Name of the submission output file.

Return type:

str

EcflowTask.create_kill_log(joboutdir)

Create the kill log file name.

Parameters:

joboutdir (str) – Location of ecflow created job files.

Returns:

Name of the kill output file.

Return type:

str

EcflowTask.create_status_log(joboutdir)

Create the status log file name.

Parameters:

joboutdir (str) – Location of ecflow created job files.

Returns:

Name of the status output file.

Return type:

str

EcflowTask.create_ecf_job(joboutdir)

Create the ecflow job file name.

Parameters:

joboutdir (str) – Location of ecflow created job files.

Returns:

Name of the ecflow job file.

Return type:

str

EcflowTask.create_ecf_jobout(joboutdir)

Create the ecflow output file name.

Parameters:

joboutdir (str) – Location of ecflow created job files.

Returns:

Name of the ecflow output file.

Return type:

str

EcflowClient.__init__(server, task)

Construct the ecflow client.

Parameters:

• server (EcflowServer) – Ecflow server object.

• task (EcflowTask) – Ecflow task object.

static EcflowClient.at_time()

Generate time stamp.

EcflowClient.signal_handler(signum, extra=None)

Signal handler.

Parameters:

• signum (_type_) – _description_

• extra (_type_, optional) – _description_. Defaults to None.

EcflowClient.__enter__()

Enter the object.

Returns:

_description_

Return type:

_type_

EcflowClient.__exit__(ex_type, value, tback)

Exit method.

Parameters:

• ex_type (_type_) – _description_

• value (_type_) – _description_

• tback (_type_) – _description_

Returns:

_description_

Return type:

_type_

Methods

scheduler.parse_submit_cmd(argv)

Parse the command line input arguments.

scheduler.submit_cmd(**kwargs)

Submit command.

scheduler.parse_kill_cmd(argv)

Parse the command line input arguments.

scheduler.kill_cmd(**kwargs)

Kill cmmand.

scheduler.parse_status_cmd(argv)

Parse the command line input arguments.

scheduler.status_cmd(**kwargs)

Status command.

scheduler.get_submission_object(task, task_settings, server, db_file=None)

Get the submission object constructed from a submit type.

Parameters:

• task (scheduler.EcflowTask) – Task.

• task_settings (TaskSettings) – Task settings

• server (scheduler.Server) – Server.

• db_file (str, optional) – Data base for monitoring.. Defaults to None.

Raises:

NotImplementedError – _description_

Returns:

Return a submission object.

Return type:

SubmissionBaseClass

• ref:

  README

Indices and tables

• Index

• Search Page