redun.executors package

Submodules

redun.executors.alias module

class redun.executors.alias.AliasExecutor(name: str, scheduler: Optional[Scheduler] = None, config: Optional[SectionProxy] = None, target: Optional[Executor] = None)

Bases: Executor

A simple executor that lazily defers to another one by name. This can be useful when tasks are defined with differing executors, allowing them to be configured separately, but sometimes need to share the underlying executor; this can be important to enforce per-executor resource limits.

Unfortunately, this implementation isn’t able to check validity of the link at init time.

start() → None

stop() → None

submit(job: Job) → None

Execute the provided job.

Implementations must provide results back to the scheduler by either calling done_job or reject_job.

submit_script(job: Job) → None

Execute the provided script job.

Implementations must provide results back to the scheduler by either calling done_job or reject_job.

redun.executors.aws_batch module

class redun.executors.aws_batch.AWSBatchExecutor(name: str, scheduler: Optional[Scheduler] = None, config: Optional[SectionProxy] = None)

Bases: Executor

A redun Executor for running jobs on AWS Batch.

gather_inflight_jobs() → None

get_array_child_jobs(job_id: str, statuses: list[str] = ['SUBMITTED', 'PENDING', 'RUNNABLE', 'STARTING', 'RUNNING']) → list[dict[str, Any]]

get_jobs(statuses: Optional[list[str]] = None) → Iterator[Mapping[str, Any]]

Returns AWS Batch Job statuses from the AWS API.

kill_jobs(job_ids: Iterable[str], reason: str = 'Terminated by user') → Iterator[dict]

Kill AWS Batch Jobs.

scratch_root() → str

set_scheduler(scheduler: Scheduler) → None

stop() → None

Stop Executor and monitoring thread.

submit(job: Job) → None

Submit Job to executor.

submit_script(job: Job) → None

Submit Job for script task to executor.

exception redun.executors.aws_batch.AWSBatchJobTimeoutError

Bases: Exception

Custom exception to raise when AWS Batch Jobs are killed due to timeout.

redun.executors.aws_batch.aws_describe_jobs(job_ids: list[str], chunk_size: int = 100, aws_region: str = 'us-west-2') → Iterator[Mapping[str, Any]]

Returns AWS Batch Job descriptions from the AWS API.

redun.executors.aws_batch.batch_submit(batch_job_args: dict, queue: str, image: str, job_def_name: Optional[str] = None, job_def_suffix: str = '-jd', job_name: str = 'batch-job', array_size: int = 0, memory: int = 4, vcpus: int = 1, gpus: int = 0, num_nodes: Optional[int] = None, shared_memory: Optional[int] = None, retries: int = 1, role: Optional[str] = None, job_def_extra: Optional[dict] = None, aws_region: str = 'us-west-2', privileged: bool = False, autocreate_job_def: bool = True, timeout: Optional[int] = None, batch_tags: Optional[dict[str, str]] = None, propagate_tags: bool = True, share_id: Optional[str] = None, scheduling_priority_override: Optional[str] = None) → SubmitJobResponseTypeDef

Actually perform job submission to AWS batch. Create or retrieve the job definition, then use it to submit the job.

batch_job_argsdict

These are passed as kwargs to the submit_job API. Generally, it must configure the commands to be run by setting node or container overrides. This function will modify the provided dictionary, at a minimum applying overrides for the resource requirements stated by other arguments to this function.

redun.executors.aws_batch.create_job_override_command(command: list[str], command_worker: Optional[list[str]] = None, num_nodes: Optional[int] = None) → dict[str, Any]

Format the command into the form needed for the AWS Batch submit_job API.

command: A list of tokens comprising the command command_worker: List of command tokens for worker nodes, if present. num_nodes: If None, this is a single-node job. If not None, a multi-node job.

Returns a dictionary to be passed to submit_job as kwargs.

redun.executors.aws_batch.ecr_push(repo_name: str, registry: Optional[str] = None) → None

redun.executors.aws_batch.equiv_job_def(job_def1: Mapping[str, Any], job_def2: Mapping[str, Any]) → bool

Returns True if two job definition are equivalent.

Limit equality to the keys of job_def1.

redun.executors.aws_batch.format_log_stream_event(event: Mapping[str, Any]) → str

Format a logStream event as a line.

redun.executors.aws_batch.get_batch_job_name(prefix: str, job_hash: str, array: bool = False) → str

Return a AWS Batch Job name by either job or job hash.

redun.executors.aws_batch.get_batch_job_options(job_options: dict) → dict

Returns AWS Batch-specific job options from general job options.

redun.executors.aws_batch.get_default_registry() → str

Returns the default ECR registry.

redun.executors.aws_batch.get_hash_from_job_name(job_name: str) → Optional[str]

Returns the job/task eval_hash that corresponds with a particular job name on Batch.

redun.executors.aws_batch.get_job_def_revision(job_def_name: str) → int

Returns the job definition revision from a job definition name.

redun.executors.aws_batch.get_job_definition(job_def_name: str, aws_region: str = 'us-west-2', batch_client: Optional[BatchClient] = None) → dict

Returns a job definition with the supplied name or empty dict if no matches are found..

The returned job definition will be the most recent, active revision if there are more than one returned from the API.

redun.executors.aws_batch.get_job_details(image: str, command: Optional[list[str]] = None, memory: int = 4, vcpus: int = 1, num_nodes: Optional[int] = None, shared_memory: Optional[int] = None, role: Optional[str] = None, aws_region: str = 'us-west-2', privileged: bool = False) → dict

Returns a JSON that can be used for creating a job definition.

redun.executors.aws_batch.get_job_log_stream(job: Optional[Mapping[str, Any]], aws_region: str) → Optional[str]

Extract the log stream from a JobDetail status dictionary. For non-multi-node jobs, (i.e., single node and array jobs), this is simply a field in the detail dictionary. But for multi-node jobs, this requires another query to get the log stream for the main node.

redun.executors.aws_batch.get_or_create_job_definition(job_def_name: str, image: str, command: Optional[list[str]] = None, memory: int = 4, vcpus: int = 1, num_nodes: Optional[int] = None, shared_memory: Optional[int] = None, role: Optional[str] = None, aws_region: str = 'us-west-2', privileged: bool = False, job_def_extra: Optional[dict] = None) → Mapping[str, Any]

Returns a job definition with the specified requirements. Although the resource requirements provided are used when creating a job, they are specifically excluded from creating new job definitions.

Either an existing active job definition is used or a new one is created.

num_nodes - if present, create a multi-node batch job.

redun.executors.aws_batch.get_or_create_repo(repo_name: str) → dict

Get or create an ECR repository.

redun.executors.aws_batch.is_array_job_name(job_name: str) → bool

redun.executors.aws_batch.iter_batch_job_log_lines(job_id: str, aws_region: str = 'us-west-2', log_group_name: str = '/aws/batch/job', reverse: bool = False, required: bool = True) → Iterator[str]

Iterate through the log lines of an AWS Batch job.

redun.executors.aws_batch.iter_batch_job_logs(job_id: str, aws_region: str = 'us-west-2', log_group_name: str = '/aws/batch/job', limit: Optional[int] = None, reverse: bool = False, required: bool = True) → Iterator[Mapping[str, Any]]

Iterate through the log events of an AWS Batch job.

redun.executors.aws_batch.iter_batch_job_status(job_ids: list[str], pending_truncate: int = 10, aws_region: str = 'us-west-2') → Iterator[Mapping[str, Any]]

Yields AWS Batch jobs statuses.

If pending_truncate is used (> 0) then rely on AWS Batch’s behavior of running jobs approximately in order. This allows us to truncate the polling of jobs once we see a sufficient number of pending jobs.

Parameters

• job_ids (List[str]) – Batch job ids that should be in order of submission.

• pending_truncate (int) – After seeing pending_truncate number of pending jobs, assume the rest are pending. Use a negative int to disable this optimization.

• aws_region (str) – AWS region that jobs are running in.

redun.executors.aws_batch.make_job_def_name(image_name: str, job_def_suffix: str = '-jd') → str

Autogenerate a job definition name from an image name.

redun.executors.aws_batch.parse_job_error(s3_scratch_prefix: str, job: Job, batch_job_metadata: Optional[Mapping[str, Any]] = None) → tuple[Exception, redun.scheduler.Traceback]

Parse task error from s3 scratch path.

redun.executors.aws_batch.parse_job_logs(batch_job_id: str, max_lines: int = 1000, required: bool = True, aws_region: str = 'us-west-2') → Iterator[str]

Iterates through most recent CloudWatch logs of an AWS Batch Job.

redun.executors.aws_batch.parse_nullable_json(text: Optional[str]) → Any

redun.executors.aws_batch.submit_command(image: str, queue: str, s3_scratch_prefix: str, job: Job, command: str, job_options: dict = {}, aws_region: str = 'us-west-2') → SubmitJobResponseTypeDef

Submit a shell command to AWS Batch.

redun.executors.aws_batch.submit_task(image: str, queue: str, s3_scratch_prefix: str, job: Job, a_task: Task, args: tuple = (), kwargs: dict[str, Any] = {}, job_options: dict = {}, array_uuid: Optional[str] = None, array_size: int = 0, code_file: Optional[File] = None, aws_region: str = 'us-west-2') → SubmitJobResponseTypeDef

Submit a redun Task to AWS Batch.

redun.executors.aws_glue module

exception redun.executors.aws_glue.AWSGlueError

Bases: Exception

class redun.executors.aws_glue.AWSGlueExecutor(name: str, scheduler: Optional[Scheduler] = None, config=None)

Bases: Executor

A redun Executor for running jobs on AWS Glue.

gather_inflight_jobs() → None

get_jobs(statuses: Optional[list[str]] = None) → Iterator[Mapping[str, Any]]

Gets all job runs with given status.

get_or_create_job_definition() → None

Get or create the default Glue job.

scratch_root() → str

stop() → None

submit(job: Job) → None

Submit job to executor.

submit_pending_job(job: Job) → str | None

Returns true if job submission was successful

exception redun.executors.aws_glue.AWSGlueJobStoppedError

Bases: Exception

exception redun.executors.aws_glue.AWSGlueJobTimeoutError

Bases: Exception

Custom exception to raise when AWS Glue jobs are killed due to timeout.

redun.executors.aws_glue.get_default_glue_service_role(account_num: Optional[str] = None, aws_region: str = 'us-west-2') → str

Returns the default Glue service role for the current account.

redun.executors.aws_glue.get_glue_oneshot_scratch_file(s3_scratch_prefix: str, code_hash: str) → str

Returns s3 scratch path for a code package tar file.

redun.executors.aws_glue.get_job_insight_traceback(job_id: str, log_group_name: str, aws_region: str = 'us-west-2', max_results=200) → list[str]

Gets the traceback from AWS’ Glue Job insights.

redun.executors.aws_glue.get_or_create_glue_job_definition(script_location: str, role: str, glue_version: str, worker_type: str, temp_dir: str, extra_py_files: str, spark_history_dir: str, enable_metrics: bool = False, additional_python_modules: list[str] = [], aws_region: str = 'us-west-2') → str

Gets or creates an AWS Glue Job.

Parameters

• script_location (str) – S3 path of script that runs Spark job.

• role (str) – ARN of IAM role to associate with Glue job.

• glue_version (str) – Glue version to use for the job.

• worker_type (str) – Type of worker type for the job (e.g. ‘G.1X’, ‘G.2X’, etc.). See: https://docs.aws.amazon.com/en_us/glue/latest/dg/worker-types.html

• temp_dir (str) – S3 path of scratch directory associated with job data and code.

• extra_py_files (str) – Comma separated S3 file paths for additional Python code needed for job.

• spark_history_dir (str) – S3 path where Spark event logs are stored.

• enable_metrics (bool) – Whether to enable observability and profiling via Cloudwatch.

• additional_python_modules (List[str]) – Python modules to be installed with pip before job start.

• aws_region (str) – AWS region in which to run the Glue job.

Returns

Unique Glue job definition name.

Return type

str

redun.executors.aws_glue.get_redun_lib_files() → Iterator[str]

Iterates through the files of the redun library.

redun.executors.aws_glue.get_redun_lib_scratch_file(s3_scratch_prefix: str, lib_hash: str) → str

Returns s3 scratch path for a code package tar file.

redun.executors.aws_glue.get_spark_history_dir(s3_scratch_prefix: str) → str

Returns s3 scratch path for Spark UI monitoring files.

redun.executors.aws_glue.glue_describe_jobs(job_ids: list[str], glue_job_name: str, aws_region: str = 'us-west-2') → Iterator[Mapping[str, Any]]

redun.executors.aws_glue.package_redun_lib(s3_scratch_prefix: str) → File

Package redun lib to S3.

redun.executors.aws_glue.parse_job_error(s3_scratch_prefix: str, job: Job, glue_job_metadata: Optional[Mapping[str, Any]] = None) → tuple[Exception, redun.scheduler.Traceback]

Parse job error from s3 scratch path.

redun.executors.aws_glue.submit_glue_job(job: Job, a_task: Task, s3_scratch_prefix: str, glue_job_name: str, redun_zip_location: str, code_file: File, job_options: dict = {}, aws_region: str = 'us-west-2') → StartJobRunResponseTypeDef

Submits a redun task to AWS glue.

Parameters

• job (Job) – The redun.scheduler.Job that is in the process of resolving the Task a_task.

• a_task (Task) – The redun.task.Task to be run by the job.

• s3_scratch_prefix (str) – Prefix for S3 url where job files, including additional Python and data files, will reside.

• glue_job_name (str) – Name of Glue job definition to run.

• redun_zip_location (str) – S3 path of zipped redun source code.

• code_file (File) – Code to be executed as a redun workflow. Path within redun.file.File should be an S3 url.

• job_options (Dict[str, Any]) –

  Option map to configure job initialization and execution.

  Required fields

  worker_type: str timeout: int workers: int

  Optional fields

  additional_libs: List[str] extra_py_files: List[str] extra_files: List[str]

• aws_region (str) – AWS region in which to run the Glue job.

Returns

Response contains JobRunId field.

Return type

Dict[str, Any]

Raises

ValueError – If worker_type is not a valid predefined Glue worker type string.

redun.executors.aws_utils module

class redun.executors.aws_utils.JobStatus(all, pending, inflight, success, failure, stopped, timeout)

Bases: NamedTuple

all: list[str]

Alias for field number 0

failure: list[str]

Alias for field number 4

inflight: list[str]

Alias for field number 2

pending: list[str]

Alias for field number 1

stopped: list[str]

Alias for field number 5

success: list[str]

Alias for field number 3

timeout: list[str]

Alias for field number 6

redun.executors.aws_utils.copy_to_s3(file_path: str, s3_scratch_dir: str) → str

Copies a file to the S3 scratch directory if it is not already on S3. Returns the path to the file on S3.

redun.executors.aws_utils.get_aws_client(service: Literal['batch'], aws_region: str = 'us-west-2') → BatchClient

redun.executors.aws_utils.get_aws_client(service: Literal['sts'], aws_region: str = 'us-west-2') → STSClient

redun.executors.aws_utils.get_aws_client(service: Literal['logs'], aws_region: str = 'us-west-2') → CloudWatchLogsClient

redun.executors.aws_utils.get_aws_client(service: Literal['glue'], aws_region: str = 'us-west-2') → GlueClient

redun.executors.aws_utils.get_aws_client(service: Literal['lambda'], aws_region: str = 'us-west-2') → LambdaClient

redun.executors.aws_utils.get_aws_client(service: Literal['secretsmanager'], aws_region: str = 'us-west-2') → SecretsManagerClient

redun.executors.aws_utils.get_aws_client(service: str, aws_region: str = 'us-west-2') → Any

Get an AWS Client with caching.

redun.executors.aws_utils.get_aws_env_vars() → dict[str, str]

Determines the current AWS credentials.

redun.executors.aws_utils.get_aws_user(aws_region: str = 'us-west-2') → str

Returns the current AWS user.

redun.executors.aws_utils.get_default_region() → str

Returns the default AWS region.

redun.executors.aws_utils.get_simple_aws_user(aws_region: str = 'us-west-2') → str

Returns the current AWS user simplified.

By full AWS identify has the format:

arn:aws:sts::ACCOUNT_ID:USER_NAME arn:aws:sts::ACCOUNT_ID:assumed-role/ROLE_NAME/USER_NAME

This function will return USER_NAME as a simplification.

redun.executors.aws_utils.is_ec2_instance() → bool

Returns True if this process is running on an EC2 instance.

We use the presence of a link-local address as a sign we are on an EC2 instance.

redun.executors.aws_utils.iter_log_stream(log_group_name: str, log_stream: str, limit: Optional[int] = None, reverse: bool = False, required: bool = True, aws_region: str = 'us-west-2') → Iterator[Mapping[str, Any]]

Iterate through the events of logStream.

redun.executors.aws_utils.set_boto_config(config: dict[str, Any]) → None

Set the boto3 config. Resets the cache of clients as a side effect, in case any were made that did not use this config.

redun.executors.base module

class redun.executors.base.Executor(name: str, scheduler: Optional[Scheduler] = None, config=None)

Bases: object

Note that most Executors should Register themselves with the method register_executor, below.

log(*messages: Any, **kwargs) → None

Display log message through Scheduler.

scratch_root() → str

set_scheduler(scheduler: Scheduler) → None

start() → None

stop() → None

submit(job: Job) → None

Execute the provided job.

Implementations must provide results back to the scheduler by either calling done_job or reject_job.

submit_script(job: Job) → None

Execute the provided script job.

Implementations must provide results back to the scheduler by either calling done_job or reject_job.

supports_async() → bool

Returns True if Executor supports async tasks.

exception redun.executors.base.ExecutorError

Bases: Exception

redun.executors.base.get_executor_class(executor_name: str, required: bool = True, class_path: Optional[str] = None) → Optional[type[redun.executors.base.Executor]]

Get an Executor by name from the executor registry.

Parameters

• executor_name (str) – Name of executor class to retrieve.

• required (bool) – If True, raises error if executor is not registered. If False, None is returned for unknown executor name.

• class_path (str, optional) – Fully-qualified class name (e.g. ‘mypackage.executors.MyExecutor’). Used as a fallback to lazily register the executor if it is not already in the registry.

redun.executors.base.get_executor_from_config(executors_config: dict, executor_name: str) → Executor

Create and return the executor by name. Raise an error if it is not present.

redun.executors.base.get_executors_from_config(executors_config: dict) → Iterator[Executor]

Instantiate executors defined in an executors config section.

redun.executors.base.load_task_module(module_name: str, task_name: str) → None

Helper method that Executors may call to load a task’s module code.

Presently, the main benefit is this method provides better error handling.

Parameters

• module_name – Task module name.

• task_name – Task name

redun.executors.base.register_executor(executor_name: str, executor_class_name: Optional[str] = None) → Callable

Register an Executor, either by decorating a class or passing a fully-specified class name. The executor will be available to the scheduler under the provided name.

Note that registered classes are responsible for ensuring their modules get loaded, so this decorator is actually run. For example, this can be done by adding them to the redun module __init__.py.

Usage: >>> @register_executor(“my_executor”) … class MyExecutor(Executor): … … or >>> register_executor(“my_executor”, “my_module.MyExecutor”)

redun.executors.code_packaging module

redun.executors.code_packaging.create_tar(tar_path: str, file_paths: Iterable[str], arcname_prefix: Optional[str] = None) → File

Create a tar file from local file paths. :param arcname_prefix: prefix to add to each file path in the tar file.

redun.executors.code_packaging.create_zip(zip_path: str, base_path: str, file_paths: Iterable[str], arcname_prefix: Optional[str] = None) → File

Create a zip file from local file paths. :param arcname_prefix: prefix to add to each file path in the tar file.

redun.executors.code_packaging.extract_tar(tar_file: File, dest_dir: str = '.') → None

Extract a tar file to local paths.

redun.executors.code_packaging.find_code_files(basedir: str = '.', includes: Optional[list[str]] = None, excludes: Optional[list[str]] = None) → Iterable[str]

Find all the workflow code files consistent with the include/exclude patterns.

redun.executors.code_packaging.package_code(scratch_prefix: str, code_package: dict = {}, use_zip: bool = False, basename: Optional[str] = None, arcname_prefix: Optional[str] = None) → File

Package code to scratch directory. :param basename: If provided, uses this string as the basename instead of the

calculated tarball hash.

Parameters

arcname_prefix – Optional suffix to append to tarball basename.

redun.executors.code_packaging.parse_code_package_config(config) → dict | bool

Parse the code package options from a AWSBatchExecutor config.

redun.executors.command module

redun.executors.command.get_oneshot_command(scratch_prefix: str, job: Job, a_task: Task, args: tuple = (), kwargs: dict[str, Any] = {}, job_options: dict = {}, code_file: Optional[File] = None, array_uuid: Optional[str] = None, input_path: Optional[str] = None, output_path: Optional[str] = None, error_path: Optional[str] = None) → list[str]

Returns a redun oneshot command for a Job.

redun.executors.command.get_script_task_command(scratch_prefix: str, job: Job, command: str, exit_command: str = '', as_mount: bool = False, suppress_output: bool = False) → list[str]

Returns a shell script to run a script task.

redun.executors.docker module

exception redun.executors.docker.DockerError

Bases: Exception

class redun.executors.docker.DockerExecutor(name: str, scheduler: Optional[Scheduler] = None, config: Optional[SectionProxy] = None)

Bases: Executor

A redun Executor for running jobs on local Docker containers.

scratch_root() → str

set_scheduler(scheduler: Scheduler) → None

stop() → None

Stop Executor and monitoring thread.

submit(job: Job) → None

Submit Job to executor.

submit_script(job: Job) → None

Submit Job for script task to executor.

redun.executors.docker.get_docker_executor_config(config: SectionProxy) → SectionProxy

Returns a config for DockerExecutor.

redun.executors.docker.get_docker_job_options(job_options: dict, scratch_path: str) → dict

Returns Docker-specific job options from general job options.

Adds the scratch_path as a volume mount.

redun.executors.docker.iter_job_status(scratch_prefix: str, job_id2job: dict[str, redun.scheduler.Job]) → Iterator[dict]

Returns local Docker jobs grouped by their status.

redun.executors.docker.run_docker(command: list[str], image: str, volumes: Iterable[tuple[str, str]] = [], interactive: bool = True, cleanup: bool = False, memory: int = 4, vcpus: int = 1, gpus: int = 0, shared_memory: Optional[int] = None, include_aws_env: bool = False) → str

Run a Docker container locally.

Parameters

• command (List[str]) – A shell command to run within the docker container (e.g. [“ls” “-la”]).

• image (str) – A Docker image.

• volumes (Iterable[Tuple[str, srt]]) – A list of (‘host’, ‘container’) path pairs for volume mounting.

• interactive (bool) – If True, the Docker container is run in interactive mode.

• cleanup (bool) – If True, remove the container after execution.

• memory (int) – Number of GB of memory to reserve for the container.

• vcpus (int) – Number of CPUs to reserve for the container.

• gpus (int) – Number of GPUs to reserve for the container.

• shared_memory (Optional[int]) – Number of GB of shared memory to reserve for the container.

• include_aws_env (bool) – If True, forward AWS environment variables to the container.

redun.executors.docker.submit_command(image: str, scratch_prefix: str, job: Job, command: str, job_options: dict = {}, include_aws_env: bool = False) → dict

Submit a shell command to Docker.

redun.executors.docker.submit_task(image: str, scratch_prefix: str, job: Job, a_task: Task, args: tuple = (), kwargs: dict[str, Any] = {}, job_options: dict = {}, code_file: Optional[File] = None, include_aws_env: bool = False) → dict[str, Any]

Submit a redun Task to Docker.

redun.executors.gcp_batch module

class redun.executors.gcp_batch.GCPBatchExecutor(name: str, scheduler: Optional[Scheduler] = None, config: Optional[SectionProxy] = None)

Bases: Executor

A redun Executor for running jobs on GCP Batch.

gather_inflight_jobs() → None

set_scheduler(scheduler: Scheduler) → None

stop() → None

Stop Executor and monitoring thread.

submit(job: Job) → None

Submit RedunJob to executor.

submit_script(job: Job) → None

Submit RedunJob for script task to executor.

redun.executors.gcp_utils module

class redun.executors.gcp_utils.MinCPUPlatform(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

EPYC_MILAN = 'AMD Milan'

EPYC_ROME = 'AMD Rome'

XEON_BROADWELL = 'Intel Broadwell'

XEON_CASCADE_LAKE = 'Intel Cascade Lake'

XEON_HASWELL = 'Intel Haswell'

XEON_ICE_LAKE = 'Intel Ice Lake'

XEON_SANDY_BRIDGE = 'Intel Sandy Bridge'

XEON_SKYLAKE = 'Intel Skylake'

redun.executors.gcp_utils.batch_submit(client: google.cloud.batch_v1.services.batch_service.client.BatchServiceClient | google.cloud.batch_v1.services.batch_service.async_client.BatchServiceAsyncClient, job_name: str, project: str, region: str, machine_type: str, vcpus: int, memory: int, retries: int, priority: int, max_duration: Optional[str] = None, task_count: int = 1, mount_buckets: list[str] = [], boot_disk_size: Optional[int] = None, min_cpu_platform: Optional[MinCPUPlatform] = None, accelerators: list[tuple[str, int]] = [], image: Optional[str] = None, script: str = 'exit 0', commands: list[str] = ['exit 0'], service_account_email: str = '', labels: dict[str, str] = {}, **kwargs) → Job

redun.executors.gcp_utils.format_job_name(project_id: str, region: str, job_name: str) → str

Create a job name from provided details

Parameters

• project_id – project ID or project number of the Cloud project you want to use.

• region – name of the region hosts the job.

• job_name – fully qualified name of the job you want to retrieve information about.

Returns

A formatted job name.

redun.executors.gcp_utils.format_task_group_name(project_id: str, region: str, job_name: str, group_name: str) → str

Create a task group name from provided details

Parameters

• project_id – project ID or project number of the Cloud project you want to use.

• region – name of the region hosting the jobs.

• job_name – name of the job which tasks you want to list.

• group_name – name of the group of tasks. Usually it’s group0.

Returns

A formatted task group name.

redun.executors.gcp_utils.format_task_name(project_id: str, region: str, job_name: str, group_name: str, task_number: int) → str

Create a task name from provided details

Parameters

• project_id – project ID or project number of the Cloud project you want to use.

• region – name of the region hosts the job.

• job_name – the name of the job you want to retrieve information about.

• group_name – the name of the group that owns the task you want to check. Usually it’s group0.

• task_number – number of the task you want to look up.

Returns

A formatted task name.

redun.executors.gcp_utils.gb_to_mib(gb)

redun.executors.gcp_utils.get_compute_machine_type(client: MachineTypesClient, project: str, region: str, machine_type: str) → MachineType

Retrieve information about a GCP MachineType

Args:

project_id: project ID or project number of the Cloud project you want to use. zone: name of the zone for the machine type. machine_type: the machine type to get details about.

Returns:

A MachineType which provides information about a machines available vCPUs and memory.

redun.executors.gcp_utils.get_gcp_batch_client(sync: bool = True) → google.cloud.batch_v1.services.batch_service.client.BatchServiceClient | google.cloud.batch_v1.services.batch_service.async_client.BatchServiceAsyncClient

redun.executors.gcp_utils.get_gcp_compute_client() → MachineTypesClient

redun.executors.gcp_utils.get_job(client: BatchServiceClient, job_name: str) → Job

Retrieve information about a Batch Job.

Parameters

job_name – fully qualified name of the job you want to retrieve information about.

Returns

A Job object representing the specified job.

redun.executors.gcp_utils.get_task(client: BatchServiceClient, task_name: str) → Task

Retrieve information about a Task.

Parameters

task_name – fully qualified name of the task you want to look up.

Returns

A Task object representing the specified task.

redun.executors.gcp_utils.list_jobs(client: BatchServiceClient, project_id: str, region: str) → Iterable[Job]

Get a list of all jobs defined in given region.

Parameters

• project_id – project ID or project number of the Cloud project you want to use.

• region – name of the region hosting the jobs.

Returns

An iterable collection of Job object.

redun.executors.gcp_utils.list_tasks(client: BatchServiceClient, group_name: str) → Iterable[Task]

Get a list of all jobs defined in given region.

Parameters

group_name – fully qualified name of the group of tasks you want to look up

Returns

An iterable collection of Task objects.

redun.executors.k8s module

exception redun.executors.k8s.K8SError

Bases: Exception

K8S-specific exception raised when a k8s job fails.

class redun.executors.k8s.K8SExecutor(name: str, scheduler: Optional[Scheduler] = None, config=None)

Bases: Executor

Implementation of K8SExecutor. This class adapts redun jobs to the k8s API. It uses k8s jobs https://kubernetes.io/docs/concepts/workloads/controllers/job/ to encapsulate the submissions. For workflows with many tasks, it groups multiple tasks into a single k8s job (https://kubernetes.io/docs/tasks/job/indexed-parallel-processing-static/). to avoid overloading the k8s scheduler. and it lacks a debug mode (use minikube or another k8s tool for debug).

Here are the details of the implementation: - The implementation is very similar to AWSBatchExecutor, with some minor differences to adapt to the k8s API. The primary difference is that array jobs are handled differently - There is no debug mode; if you want to debug K8SExecutor-based workflows, use minikube https://minikube.sigs.k8s.io/ to test your workflows locally. Minikube is powerful enough to run many workflows on a single local machine. - K8SExecutor submits redun tasks (shell scripts and code) as K8S Jobs which execute the tasks in K8S Pods (containers). K8S Jobs wrap pods with additional functionality, such as retry and parallel processing. - Like the AWSBatchExecutor, this implementation can batch up multiple tasks into a single Job. This is more efficient for the k8s scheduler, if you are running more than about 100 tasks. - There are a few situations where this implementation can fail to return useful log messages from k8s jobs and pods.

gather_inflight_jobs() → None

Collect existing k8s jobs and match them up to redun jobs

get_jobs() → Iterator[V1Job]

Iterates active k8s jobs.

stop() → None

Stop Executor and monitoring thread.

submit(job: Job) → None

Submit Job to executor.

submit_script(job: Job) → None

Submit Job for script task to executor.

exception redun.executors.k8s.K8STimeoutExceeded

Bases: Exception

Custom exception to raise when K8S jobs are killed due to timeout.

redun.executors.k8s.get_hash_from_job_name(job_name: str) → Optional[str]

Returns the job/task eval_hash that corresponds with a particular job name on K8S.

redun.executors.k8s.get_k8s_job_name(prefix: str, job_hash: str, array: bool = False) → str

Return a K8S Job name by either job or job hash.

redun.executors.k8s.get_k8s_job_options(job_options: dict) → dict

Returns K8S-specific job options from general job options.

redun.executors.k8s.get_k8s_job_pods(core_api: CoreV1Api, job_name: str) → Iterator[V1Pod]

Iterates the pods for a k8s job.

redun.executors.k8s.get_pod_logs(k8s_client: K8SClient, pod: V1Pod, max_lines: Optional[int] = None) → list[str]

Returns the logs of a K8S pod.

redun.executors.k8s.is_array_job_name(job_name: str) → bool

Returns true if the job name looks like an array job

redun.executors.k8s.k8s_describe_jobs(k8s_client: K8SClient, job_names: list[str], namespace: str) → list[kubernetes.client.models.v1_job.V1Job]

Returns K8S Job descriptions.

redun.executors.k8s.k8s_list_jobs(k8s_client: K8SClient) → Iterator[V1Job]

Returns active K8S Jobs.

redun.executors.k8s.k8s_paginate(request_func: Callable) → Iterator

Yields items from a paginated k8s API endpoint.

redun.executors.k8s.k8s_submit(k8s_client: K8SClient, command: list[str], image: str, namespace: str, job_name: str = 'k8s-job', array_size: int = 0, memory: int = 4, vcpus: int = 1, gpus: int = 0, ephemeral_storage: Optional[str] = None, timeout: Optional[int] = None, k8s_labels: Optional[dict[str, str]] = None, retries: int = 1, service_account_name: str = 'default', annotations: Optional[dict[str, str]] = None, secret_name: Optional[str] = None, node_affinity: Optional[dict[str, Any]] = None, env: Optional[dict[str, str]] = None, volumes: Optional[list[dict[str, Any]]] = None, volume_mounts: Optional[list[dict[str, Any]]] = None) → V1Job

Prepares and submits a k8s job to the API server

redun.executors.k8s.parse_pod_logs(k8s_client: K8SClient, pod: V1Pod, max_lines: int = 1000) → Iterator[str]

Iterates through most recent logs of an K8S Job.

redun.executors.k8s.submit_command(k8s_client: K8SClient, image: str, namespace: str, scratch_prefix: str, job: Job, command: str, job_options: dict = {}, secret_name: Optional[str] = None) → V1Job

Submit a shell command to K8S

redun.executors.k8s.submit_task(k8s_client: K8SClient, image: str, namespace: str, scratch_prefix: str, job: Job, a_task: Task, args: tuple = (), kwargs: dict[str, Any] = {}, job_options: dict = {}, array_uuid: Optional[str] = None, array_size: int = 0, code_file: Optional[File] = None, secret_name: Optional[str] = None) → V1Job

Submit a redun Task to K8S.

redun.executors.k8s_utils module

class redun.executors.k8s_utils.K8SClient

Bases: object

This class manages multiple k8s clients and their config loading.

property batch: BatchV1Api

Returns an API client supporting k8s batch API.

https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/BatchV1Api.md

property core: CoreV1Api

Returns an API client support k8s core API.

https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md

property custom_objects: CustomObjectsApi

Returns an API client support k8s custom_objects API.

https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CustomObjectsApi.md

load_config() → None

Loads k8s config if not loaded.

property rbac: RbacAuthorizationV1Api

Returns an API client support k8s RbacAuthorizationV1 API.

https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/RbacAuthorizationV1Api.md

version() → tuple[int, int]

Returns an API client support k8s version API.

https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/VersionApi.md

redun.executors.k8s_utils.annotate_service_account(k8s_client: K8SClient, namespace: str, sa_name: str, annotations_to_append: dict) → None

Annotate a service account with additional annotations

Parameters

• namespace – Namespace wherein service account resides.

• sa_name – Service account name.

• annotations_to_append – Dict of annotations to append.

redun.executors.k8s_utils.create_job(k8s_client: K8SClient, job: V1Job, namespace: str) → V1Job

Creates an actual k8s job.

redun.executors.k8s_utils.create_job_object(name: str = 'redun-job', image: str = 'bash', command: list[str] = ['false'], resources: Optional[V1ResourceRequirements] = None, timeout: Optional[int] = None, labels: Optional[dict] = None, uid: Optional[str] = None, retries: int = 1, service_account_name: Optional[str] = 'default', annotations: Optional[dict[str, str]] = None, secret_name: Optional[str] = None, node_affinity: Optional[dict[str, Any]] = None, env: Optional[dict[str, str]] = None, volumes: Optional[list[dict[str, Any]]] = None, volume_mounts: Optional[list[dict[str, Any]]] = None) → V1Job

Creates a job object for redun job. https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Job.md Also creates necessary sub-objects

redun.executors.k8s_utils.create_k8s_secret(k8s_client: K8SClient, secret_name: str, namespace: str, secret_data: dict, secret_type: str = 'Opaque') → Any

Creates a k8s Secret from a dict of key-value pairs.

redun.executors.k8s_utils.create_namespace(k8s_client: K8SClient, namespace: str) → None

Create a k8s namespace

redun.executors.k8s_utils.create_resources(requests: Optional[dict] = None, limits: Optional[dict] = None) → V1ResourceRequirements

Creates resource limits for k8s pods.

https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1ResourceRequirements.md

redun.executors.k8s_utils.delete_job(k8s_client: K8SClient, name: str, namespace: str) → Any

Deletes an existing k8s job.

redun.executors.k8s_utils.delete_k8s_secret(k8s_client: K8SClient, secret_name: str, namespace: str) → None

Deletes a k8s Secret. If Secret doesn’t exist, it does not throw an error.

redun.executors.k8s_utils.delete_namespace(k8s_client: K8SClient, namespace: str) → None

Delete a k8s namespace

redun.executors.launch module

redun.executors.launch.launch_script(config: Config, script_command: list[str] | str, executor: Optional[Executor] = None, executor_name: Optional[str] = None, task_options: Optional[dict] = None, execution_id: Optional[str] = None) → None

Submit the provided script command to the executor, then exit.

Use a local scheduler with the default config. This means we won’t record the entry point, but we have no intention of being around long enough to record the results, so there’s not much point.

WARNING: This won’t actually work on all executor types, such as the local ones. To work, the executor needs to be “fire and forget” for submit_script.

Parameters

• config (Config) – Config object containing executor information.

• script_command (Union[List[str], str]) – The script command to run on the executor.

• executor (Optional[Executor]) – Optional Executor on which to run the script. Must be set if executor_name is not set.

• executor_name (Optional[str]) – Name of the executor in the config. Must be set if executor is not set.

• task_options (Optional[dict]) – Task options to pass to the script task.

• execution_id (Optional[str]) – If provided, use this execution id. This is only relevant for tagging.

Return type

None

redun.executors.local module

class redun.executors.local.LocalExecutor(name: str, scheduler: Optional[Scheduler] = None, config=None, mode: str = 'thread')

Bases: Executor

A redun Executor for running jobs locally using a thread or process pool.

DEFAULT_START_METHOD = 'forkserver'

MODES = ['thread', 'process']

START_METHODS = ['fork', 'spawn', 'forkserver']

scratch_root() → str

stop() → None

Stop Executor pools.

submit(job: Job) → None

Execute the provided job.

Implementations must provide results back to the scheduler by either calling done_job or reject_job.

submit_script(job: Job) → None

Execute the provided script job.

Implementations must provide results back to the scheduler by either calling done_job or reject_job.

supports_async() → bool

Returns True if Executor supports async tasks.

redun.executors.local.exec_script_task(mode: str, module_name: str, task_fullname: str, args: tuple, kwargs: dict) → bytes

Execute a script task from the task registry.

redun.executors.local.exec_task(mode: str, module_name: str, task_fullname: str, args: tuple, kwargs: dict) → Any

Execute a task in the new process.

redun.executors.local.get_current_job() → tuple[Scheduler, Job]

Returns the current scheduler and job for this context.

redun.executors.local.set_current_job(scheduler: Scheduler, job: Job) → None

Set the current scheduler and job for this context.

redun.executors.scratch module

Utilities for working with executor scratch directories.

class redun.executors.scratch.ArrayJobScratchFiles(input_file: str, output_file: str, error_file: str, eval_file: Optional[str] = None)

Bases: object

error_file: str

eval_file: Optional[str] = None

input_file: str

output_file: str

exception redun.executors.scratch.ExceptionNotFoundError

Bases: ScratchError

Error when serialized exception is not found in scratch directory.

exception redun.executors.scratch.ScratchError

Bases: Exception

Error when reading data from scratch directory.

redun.executors.scratch.get_array_scratch_file(scratch_prefix: str, job_array_id: str, filename: str) → str

Returns an scratch path for a file related to an array of jobs.

redun.executors.scratch.get_code_scratch_file(scratch_prefix: str, tar_hash: str, use_zip: bool = False) → str

Returns scratch path for a code package tar file.

redun.executors.scratch.get_execution_scratch_file(scratch_prefix: str, execution_id: str, filename: str) → str

Returns an scratch path for a sending data to and from a remote execution.

redun.executors.scratch.get_job_scratch_dir(scratch_prefix: str, job: Job) → str

Returns scratch directory for a redun Job.

redun.executors.scratch.get_job_scratch_file(scratch_prefix: str, job: Job, filename: str) → str

Returns scratch path for a file related to a redun Job.

redun.executors.scratch.parse_job_error(scratch_prefix: str, job: Job) → tuple[Exception, redun.scheduler.Traceback]

Returns job error from scratch directory.

redun.executors.scratch.parse_job_result(scratch_prefix: str, job: Job, is_valid_value: Optional[Callable[[Any], bool]] = None) → tuple[Any, bool]

Returns job output from scratch directory.

Returns a tuple of (result, exists).

redun.executors.scratch.write_array_job_scratch_files(jobs: list[redun.scheduler.Job], scratch_prefix: str, array_id: str, include_eval_hash: bool = True) → ArrayJobScratchFiles

Write the scratch files for an Array-style Job.

Module contents