galaxy.jobs.runners.util package

This module and its submodules contains utilities for running external processes and interfacing with job managers. This module should contain functionality shared between Galaxy and the Pulsar.

galaxy.jobs.runners.util.kill_pid(pid: int, use_psutil: bool = True)[source]

Subpackages

Submodules

galaxy.jobs.runners.util.env module

galaxy.jobs.runners.util.env.env_to_statement(env)[source]

Return the abstraction description of an environment variable definition into a statement for shell script.

>>> env_to_statement(dict(name='X', value='Y'))
'X="Y"; export X'
>>> env_to_statement(dict(name='X', value='Y', raw=True))
'X=Y; export X'
>>> env_to_statement(dict(name='X', value='"A","B","C"'))
'X="\\"A\\",\\"B\\",\\"C\\""; export X'
>>> env_to_statement(dict(file="Y"))
'. "Y"'
>>> env_to_statement(dict(file="'RAW $FILE'", raw=True))
". 'RAW $FILE'"
>>> # Source file takes precedence
>>> env_to_statement(dict(name='X', value='"A","B","C"', file="S"))
'. "S"'
>>> env_to_statement(dict(execute="module load java/1.5.1"))
'module load java/1.5.1'

galaxy.jobs.runners.util.external module

galaxy.jobs.runners.util.external.parse_external_id(output, type=None)[source]

Attempt to parse the output of job submission commands for an external id.__doc__

>>> parse_external_id("12345.pbsmanager")
'12345.pbsmanager'
>>> parse_external_id('Submitted batch job 185')
'185'
>>> parse_external_id('Submitted batch job 185', type='torque')
'Submitted batch job 185'
>>> parse_external_id('submitted to cluster 125.')
'125'
>>> parse_external_id('submitted to cluster 125.', type='slurm')
>>>

galaxy.jobs.runners.util.kill module

galaxy.jobs.runners.util.kill.kill_pid(pid: int, use_psutil: bool = True)[source]

galaxy.jobs.runners.util.process_groups module

galaxy.jobs.runners.util.process_groups.check_pg(pgid)[source]: Check whether processes in process group pgid are still alive.

galaxy.jobs.runners.util.process_groups.kill_pg(pgid)[source]: Kill all processes in process group pgid.

galaxy.jobs.runners.util.pykube_util module

Interface layer for pykube library shared between Galaxy and Pulsar.

galaxy.jobs.runners.util.pykube_util.ensure_pykube()[source]

galaxy.jobs.runners.util.pykube_util.find_service_object_by_name(pykube_api, service_name, namespace=None)[source]

galaxy.jobs.runners.util.pykube_util.find_ingress_object_by_name(pykube_api, ingress_name, namespace=None)[source]

galaxy.jobs.runners.util.pykube_util.find_job_object_by_name(pykube_api, job_name, namespace=None)[source]

galaxy.jobs.runners.util.pykube_util.find_pod_object_by_name(pykube_api, job_name, namespace=None)[source]

galaxy.jobs.runners.util.pykube_util.galaxy_instance_id(params)[source]

Parse and validate the id of the Galaxy instance from supplied dict.

This will be added to Jobs and Pods names, so it needs to be DNS friendly, this means: The Internet standards (Requests for Comments) for protocols mandate that component hostname labels may contain only the ASCII letters ‘a’ through ‘z’ (in a case-insensitive manner), the digits ‘0’ through ‘9’, and the minus sign (‘-‘).

It looks for the value set on params[‘k8s_galaxy_instance_id’], which might or not be set. The idea behind this is to allow the Galaxy instance to trust (or not) existing k8s Jobs and Pods that match the setup of a Job that is being recovered or restarted after a downtime/reboot.

galaxy.jobs.runners.util.pykube_util.is_pod_unschedulable(pykube_api, pod, namespace=None)[source]

galaxy.jobs.runners.util.pykube_util.job_object_dict(params, job_prefix, spec)[source]

galaxy.jobs.runners.util.pykube_util.service_object_dict(params, service_name, spec)[source]

galaxy.jobs.runners.util.pykube_util.ingress_object_dict(params, ingress_name, spec)[source]

galaxy.jobs.runners.util.pykube_util.produce_k8s_job_prefix(app_prefix=None, instance_id=None)[source]

galaxy.jobs.runners.util.pykube_util.pull_policy(params)[source]

galaxy.jobs.runners.util.pykube_util.pykube_client_from_dict(params)[source]

galaxy.jobs.runners.util.pykube_util.delete_job(job, cleanup='always')[source]

galaxy.jobs.runners.util.pykube_util.delete_service(service, cleanup='always', job_failed=False)[source]

galaxy.jobs.runners.util.pykube_util.delete_ingress(ingress, cleanup='always', job_failed=False)[source]

galaxy.jobs.runners.util.pykube_util.get_volume_mounts_for_job(job_wrapper, data_claim=None, working_claim=None)[source]

galaxy.jobs.runners.util.pykube_util.parse_pvc_param_line(pvc_param)[source]

Parameters:

pvc_param (str) – the pvc mount param in the format pvc-name/subpath/desired:/mountpath/desired[:r]

Return type:

dict

Returns:

a dict like:

{"name": "pvc-name",
 "subPath": "subpath/desired",
 "mountPath": "/mountpath/desired",
 "readOnly": False}

galaxy.jobs.runners.util.sudo module

galaxy.jobs.runners.util.sudo.sudo_popen(*args, **kwargs)[source]: Helper method for building and executing Popen command. This is potentially sensetive code so should probably be centralized.