mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Delete godocker.py
This commit is contained in:
Varun Shankar S
@@ -1,504 +0,0 @@
|
||||
import logging
|
||||
import json
|
||||
import time
|
||||
import inspect
|
||||
import requests
|
||||
from datetime import datetime
|
||||
|
||||
from galaxy import model
|
||||
from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner
|
||||
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
__all__ = ['GodockerJobRunner']
|
||||
|
||||
|
||||
class Godocker(object):
|
||||
"""
|
||||
API parameters
|
||||
"""
|
||||
def __init__(self, server, login, apikey, noCert):
|
||||
self.token = None
|
||||
self.server = server
|
||||
self.login = login
|
||||
self.apikey = apikey
|
||||
self.noCert = noCert
|
||||
|
||||
def setToken(self, token):
|
||||
self.token = token
|
||||
|
||||
def http_post_request(self, query, data, header):
|
||||
""" post request with query """
|
||||
|
||||
# remove warnings if using --no-certificate
|
||||
requests.packages.urllib3.disable_warnings()
|
||||
verify_ssl = not self.noCert
|
||||
try:
|
||||
url = self.server + query
|
||||
res = requests.post(url, data, headers=header, verify=verify_ssl)
|
||||
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
log.debug('A Connection error occurred:', e)
|
||||
return False
|
||||
|
||||
except requests.exceptions.HTTPError as e:
|
||||
log.debug('A HTTP error occurred:', e)
|
||||
return False
|
||||
|
||||
return self.test_status_code(res)
|
||||
|
||||
def http_get_request(self, query, header):
|
||||
""" get request with query, server and header required """
|
||||
|
||||
# remove warnings if using --no-certificate
|
||||
requests.packages.urllib3.disable_warnings()
|
||||
verify_ssl = not self.noCert
|
||||
try:
|
||||
url = self.server + query
|
||||
res = requests.get(url, headers=header, verify=verify_ssl)
|
||||
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
log.debug('A Connection error occurred:', e)
|
||||
return False
|
||||
|
||||
except requests.exceptions.HTTPError as e:
|
||||
log.debug('A HTTP error occurred:', e)
|
||||
return False
|
||||
|
||||
return self.test_status_code(res)
|
||||
|
||||
def http_delete_request(self, query, header):
|
||||
""" delete request with query, server and header required """
|
||||
|
||||
# remove warnings if using --no-certificate
|
||||
requests.packages.urllib3.disable_warnings()
|
||||
verify_ssl = not self.noCert
|
||||
try:
|
||||
url = self.server + query
|
||||
res = requests.delete(url, headers=header, verify=verify_ssl)
|
||||
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
log.debug('A Connection error occurred:', e)
|
||||
return False
|
||||
|
||||
except requests.exceptions.HTTPError as e:
|
||||
log.debug('A HTTP error occurred:', e)
|
||||
return False
|
||||
|
||||
return self.test_status_code(res)
|
||||
|
||||
def http_put_request(self, query, data, header):
|
||||
""" put request with query """
|
||||
|
||||
# remove warnings if using --no-certificate
|
||||
requests.packages.urllib3.disable_warnings()
|
||||
verify_ssl = not self.noCert
|
||||
try:
|
||||
url = self.server + query
|
||||
res = requests.put(url, data, headers=header, verify=verify_ssl)
|
||||
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
log.debug('A Connection error occurred:', e)
|
||||
return False
|
||||
|
||||
except requests.exceptions.HTTPError as e:
|
||||
log.debug('A HTTP error occurred:', e)
|
||||
return False
|
||||
|
||||
return self.test_status_code(res)
|
||||
|
||||
def test_status_code(self, httpresult):
|
||||
""" exit if status code is 401 or 403 or 404 or 200"""
|
||||
if httpresult.status_code == 401:
|
||||
log.debug('Unauthorized : this server could not verify that you are authorized to access the document you requested.')
|
||||
|
||||
elif httpresult.status_code == 403:
|
||||
log.debug('Forbidden : Access was denied to this resource. Not authorized to access this resource.')
|
||||
|
||||
elif httpresult.status_code == 404:
|
||||
log.debug('Not Found : The resource could not be found.')
|
||||
|
||||
elif httpresult.status_code == 200:
|
||||
return httpresult
|
||||
|
||||
return False
|
||||
|
||||
|
||||
class GodockerJobRunner(AsynchronousJobRunner):
|
||||
"""
|
||||
Job runner backed by a finite pool of worker threads. FIFO scheduling
|
||||
"""
|
||||
runner_name = "GodockerJobRunner"
|
||||
|
||||
def __init__(self, app, nworkers, **kwargs):
|
||||
""" 1: Get runner_param_specs from job_conf.xml
|
||||
2: Initialise job runner parent object
|
||||
3: Login to godocker and store the token
|
||||
4: Start the worker and monitor threads
|
||||
"""
|
||||
log.debug("Loading app %s", app)
|
||||
runner_param_specs = dict(godocker_master=dict(map=str), user=dict(map=str), key=dict(map=str), godocker_project=dict(map=str))
|
||||
if 'runner_param_specs' not in kwargs:
|
||||
kwargs['runner_param_specs'] = dict()
|
||||
kwargs['runner_param_specs'].update(runner_param_specs)
|
||||
|
||||
# Start the job runner parent object
|
||||
super(GodockerJobRunner, self).__init__(app, nworkers, **kwargs)
|
||||
|
||||
# godocker API login call
|
||||
self.auth = self.login(self.runner_params["key"], self.runner_params["user"], self.runner_params["godocker_master"])
|
||||
|
||||
if not self.auth:
|
||||
log.debug("Authentication failure!! Runner cannot be started")
|
||||
else:
|
||||
""" Following methods starts threads.
|
||||
These methods invoke threading.Thread(name,target)
|
||||
which inturn invokes methods monitor() and run_next().
|
||||
"""
|
||||
self._init_monitor_thread()
|
||||
self._init_worker_threads()
|
||||
|
||||
def queue_job(self, job_wrapper):
|
||||
""" Create job script and submit it to godocker """
|
||||
if not self.prepare_job(job_wrapper, include_metadata=False, include_work_dir_outputs=True, modify_command_for_container=False):
|
||||
return
|
||||
|
||||
job_destination = job_wrapper.job_destination
|
||||
""" Submit job to godocker """
|
||||
job_id = self.post_task(job_wrapper)
|
||||
if not job_id:
|
||||
log.debug("Job creation faliure!! No Response from GoDocker")
|
||||
job_wrapper.fail("Not submitted")
|
||||
else:
|
||||
log.debug("Starting queue_job for job " + job_id)
|
||||
# Create an object of AsynchronousJobState and add it to the monitor queue.
|
||||
ajs = AsynchronousJobState(files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper, job_id=job_id, job_destination=job_destination)
|
||||
self.monitor_queue.put(ajs)
|
||||
return None
|
||||
|
||||
def check_watched_item(self, job_state):
|
||||
""" Get the job current status from GoDocker
|
||||
using job_id and update the status in galaxy.
|
||||
If the job execution is successful, call
|
||||
mark_as_finished() and return 'None' to galaxy.
|
||||
else if the job failed, call mark_as_failed()
|
||||
and return 'None' to galaxy.
|
||||
else if the job is running or in pending state, simply
|
||||
return the 'AsynchronousJobState object' (job_state).
|
||||
"""
|
||||
''' This function is called by check_watched_items() where
|
||||
param job_state is an object of AsynchronousJobState.
|
||||
Expected return type of this function is None or
|
||||
AsynchronousJobState object with updated running status.
|
||||
'''
|
||||
""" Get task from GoDocker """
|
||||
job_status_god = self.get_task(job_state.job_id)
|
||||
log.debug("Job ID: " + str(job_state.job_id) + " Job Status: " + str(job_status_god['status']['primary']))
|
||||
|
||||
if job_status_god['status']['primary'] == "over":
|
||||
job_state.running = False
|
||||
job_state.job_wrapper.change_state(model.Job.states.OK)
|
||||
if self.create_log_file(job_state, job_status_god):
|
||||
self.mark_as_finished(job_state)
|
||||
else:
|
||||
self.mark_as_failed(job_state)
|
||||
'''The function mark_as_finished() executes:
|
||||
self.work_queue.put((self.finish_job, job_state))
|
||||
*self.finish_job ->
|
||||
job_state.job_wrapper.finish( stdout, stderr, exit_code )
|
||||
job_state.job_wrapper.reclaim_ownership()
|
||||
job_state.cleanup()
|
||||
*self.work_queue.put( method , arg ) ->
|
||||
The run_next() method starts execution on starting worker threads.
|
||||
This run_next() method executes method(arg)
|
||||
by using self.work_queue.get()
|
||||
*Possible outcomes of finish_job(job_state) ->
|
||||
job_state.job_wrapper.finish( stdout, stderr, exit_code )
|
||||
job_state.job_wrapper.fail( "Unable to finish job", exception=True)
|
||||
*Similar workflow is done for mark_as_failed() method.
|
||||
'''
|
||||
return None
|
||||
|
||||
elif job_status_god['status']['primary'] == "running":
|
||||
job_state.running = True
|
||||
job_state.job_wrapper.change_state(model.Job.states.RUNNING)
|
||||
return job_state
|
||||
|
||||
elif job_status_god['status']['primary'] == "pending":
|
||||
return job_state
|
||||
|
||||
elif job_status_god['status']['exitcode'] not in [None, 0]:
|
||||
job_state.running = False
|
||||
job_state.job_wrapper.change_state(model.Job.states.ERROR)
|
||||
self.create_log_file(job_state, job_status_god)
|
||||
self.mark_as_failed(job_state)
|
||||
return None
|
||||
|
||||
else:
|
||||
job_state.running = False
|
||||
self.create_log_file(job_state, job_status_god)
|
||||
self.mark_as_failed(job_state)
|
||||
return None
|
||||
|
||||
def stop_job(self, job):
|
||||
""" Attempts to delete a dispatched executing Job in GoDocker """
|
||||
'''This function is called by fail_job()
|
||||
where param job = self.sa_session.query( self.app.model.Job ).get( job_state.job_wrapper.job_id )
|
||||
No Return data expected
|
||||
'''
|
||||
log.debug("STOP JOB EXECUTION OF JOB ID: " + str(job.id))
|
||||
# Get task status from GoDocker.
|
||||
job_status_god = self.get_task_status(job.id)
|
||||
if job_status_god['status']['primary'] != "over":
|
||||
# Initiate a delete call,if the job is running in GoDocker.
|
||||
self.delete_task(job.id)
|
||||
return None
|
||||
|
||||
def recover(self, job, job_wrapper):
|
||||
""" Recovers jobs stuck in the queued/running state when Galaxy started """
|
||||
""" This method is called by galaxy at the time of startup.
|
||||
Jobs in Running & Queued status in galaxy are put in the monitor_queue by creating an AsynchronousJobState object
|
||||
"""
|
||||
job_id = job_wrapper.job_id
|
||||
ajs = AsynchronousJobState(files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper)
|
||||
ajs.job_id = str(job_id)
|
||||
ajs.job_destination = job_wrapper.job_destination
|
||||
job_wrapper.command_line = job.command_line
|
||||
ajs.job_wrapper = job_wrapper
|
||||
if job.state == model.Job.states.RUNNING:
|
||||
log.debug("(%s/%s) is still in running state, adding to the god queue" % (job.id, job.get_job_runner_external_id()))
|
||||
ajs.old_state = 'R'
|
||||
ajs.running = True
|
||||
self.monitor_queue.put(ajs)
|
||||
|
||||
elif job.state == model.Job.states.QUEUED:
|
||||
log.debug("(%s/%s) is still in god queued state, adding to the god queue" % (job.id, job.get_job_runner_external_id()))
|
||||
ajs.old_state = 'Q'
|
||||
ajs.running = False
|
||||
self.monitor_queue.put(ajs)
|
||||
|
||||
# Helper functions
|
||||
|
||||
def create_log_file(self, job_state, job_status_god):
|
||||
""" Create log files in galaxy, namely error_file, output_file, exit_code_file
|
||||
Return true, if all the file creations are successful
|
||||
"""
|
||||
path = None
|
||||
for vol in job_status_god['container']['volumes']:
|
||||
if vol['name'] == "go-docker":
|
||||
path = str(vol['path'])
|
||||
if path:
|
||||
god_output_file = path + "/god.log"
|
||||
god_error_file = path + "/god.err"
|
||||
try:
|
||||
# Read from GoDocker output_file and write it into galaxy output_file.
|
||||
f = open(god_output_file, "r")
|
||||
out_log = f.read()
|
||||
log_file = open(job_state.output_file, "w")
|
||||
log_file.write(out_log)
|
||||
log_file.close()
|
||||
f.close()
|
||||
# Read from GoDocker error_file and write it into galaxy error_file.
|
||||
f = open(god_error_file, "r")
|
||||
out_log = f.read()
|
||||
log_file = open(job_state.error_file, "w")
|
||||
log_file.write(out_log)
|
||||
log_file.close()
|
||||
f.close()
|
||||
# Read from GoDocker exit_code and write it into galaxy exit_code_file.
|
||||
out_log = str(job_status_god['status']['exitcode'])
|
||||
log_file = open(job_state.exit_code_file, "w")
|
||||
log_file.write(out_log)
|
||||
log_file.close()
|
||||
f.close()
|
||||
log.debug("CREATE OUTPUT FILE: " + str(job_state.output_file))
|
||||
log.debug("CREATE ERROR FILE: " + str(job_state.error_file))
|
||||
log.debug("CREATE EXIT CODE FILE: " + str(job_state.exit_code_file))
|
||||
except IOError as e:
|
||||
log.error('Could not access task log file %s' % str(e))
|
||||
log.debug("IO Error occurred when accessing the files!!")
|
||||
return False
|
||||
return True
|
||||
|
||||
# GoDocker API helper functions
|
||||
|
||||
def login(self, apikey, login, server, noCert=False):
|
||||
""" Login to GoDocker and return the token
|
||||
Create Login model schema of GoDocker and call the http_post_request method.
|
||||
"""
|
||||
log.debug("LOGIN TASK TO BE EXECUTED \n")
|
||||
log.debug("GODOCKER LOGIN: " + str(login))
|
||||
data = json.dumps({'user': login, 'apikey': apikey})
|
||||
# Create object of Godocker class
|
||||
g_auth = Godocker(server, login, apikey, noCert)
|
||||
auth = g_auth.http_post_request("/api/1.0/authenticate", data, {'Content-type': 'application/json', 'Accept': 'application/json'})
|
||||
if not auth:
|
||||
log.debug("Authentication Error!!")
|
||||
else:
|
||||
log.debug("Authentication Successful!!")
|
||||
token = auth.json()['token']
|
||||
g_auth.setToken(token)
|
||||
# Return the object of Godocker class
|
||||
return g_auth
|
||||
|
||||
def post_task(self, job_wrapper):
|
||||
""" Sumbit job to GoDocker and return jobid
|
||||
Create Job model schema of GoDocker and call the http_post_request method.
|
||||
"""
|
||||
# Get the params from <destination> tag in job_conf by using job_destination.params[param]
|
||||
if self.auth.token:
|
||||
job_destination = job_wrapper.job_destination
|
||||
try:
|
||||
docker_cpu = int(job_destination.params["docker_cpu"])
|
||||
except:
|
||||
docker_cpu = 1
|
||||
try:
|
||||
docker_ram = int(job_destination.params["docker_memory"])
|
||||
except:
|
||||
docker_ram = 1
|
||||
try:
|
||||
docker_image = self._find_container(job_wrapper).container_id
|
||||
log.debug("DOCKER IMAGE: \n")
|
||||
log.debug(docker_image)
|
||||
except:
|
||||
log.debug("Error: Docker_image not specified in Job config and Tool config!!")
|
||||
return False
|
||||
|
||||
volumes = []
|
||||
labels = []
|
||||
tags_tab = ['galaxy', job_wrapper.tool.id]
|
||||
tasks_depends = []
|
||||
name = job_wrapper.tool.name
|
||||
description = "galaxy job"
|
||||
array = None
|
||||
project = None
|
||||
try:
|
||||
log.debug(self.runner_params["godocker_project"])
|
||||
project = str(self.runner_params["godocker_project"])
|
||||
except KeyError:
|
||||
log.debug("godocker_project not defined, using defaults")
|
||||
try:
|
||||
log.debug(job_destination.params["godocker_volumes"])
|
||||
volume = job_destination.params["godocker_volumes"]
|
||||
volume = volume.split(",")
|
||||
for i in volume:
|
||||
temp = dict({"name": i})
|
||||
volumes.append(temp)
|
||||
except:
|
||||
log.debug("godocker_volumes not set. Getting default volume!!")
|
||||
|
||||
dt = datetime.now()
|
||||
# Enable galaxy venv in the docker containers
|
||||
try:
|
||||
if(job_destination.params["virtualenv"] == "true"):
|
||||
log.debug("Virtual environment is set!!!")
|
||||
GALAXY_VENV_TEMPLATE = """GALAXY_VIRTUAL_ENV="%s"; if [ "$GALAXY_VIRTUAL_ENV" != "None" -a -z "$VIRTUAL_ENV" -a -f "$GALAXY_VIRTUAL_ENV/bin/activate" ]; then . "$GALAXY_VIRTUAL_ENV/bin/activate"; fi;"""
|
||||
venv = GALAXY_VENV_TEMPLATE % job_wrapper.galaxy_virtual_env
|
||||
command = "#!/bin/bash\n" + "cd " + job_wrapper.working_directory + "\n" + venv + "\n" + job_wrapper.runner_command_line
|
||||
else:
|
||||
command = "#!/bin/bash\n" + "cd " + job_wrapper.working_directory + "\n" + job_wrapper.runner_command_line
|
||||
except:
|
||||
command = "#!/bin/bash\n" + "cd " + job_wrapper.working_directory + "\n" + job_wrapper.runner_command_line
|
||||
|
||||
# GoDocker Job model schema
|
||||
job = {
|
||||
'date': time.mktime(dt.timetuple()),
|
||||
'meta': {
|
||||
'name': name,
|
||||
'description': description,
|
||||
'tags': tags_tab
|
||||
},
|
||||
'requirements': {
|
||||
'cpu': docker_cpu,
|
||||
'ram': docker_ram,
|
||||
'array': {'values': array},
|
||||
'label': labels,
|
||||
'tasks': tasks_depends,
|
||||
'tmpstorage': None
|
||||
},
|
||||
'container': {
|
||||
'image': str(docker_image),
|
||||
'volumes': volumes,
|
||||
'network': True,
|
||||
'id': None,
|
||||
'meta': None,
|
||||
'stats': None,
|
||||
'ports': [],
|
||||
'root': False
|
||||
},
|
||||
'command': {
|
||||
'interactive': False,
|
||||
'cmd': command,
|
||||
},
|
||||
'status': {
|
||||
'primary': None,
|
||||
'secondary': None
|
||||
}
|
||||
}
|
||||
if project is not None:
|
||||
job['user'] = {"project": project}
|
||||
|
||||
log.debug("\n JOB POST TASK TO BE EXECUTED \n")
|
||||
result = self.auth.http_post_request(
|
||||
"/api/1.0/task", json.dumps(job),
|
||||
{'Authorization': 'Bearer ' + self.auth.token, 'Content-type': 'application/json', 'Accept': 'application/json'}
|
||||
)
|
||||
log.debug("Response from godocker: " + str(result.json()['msg']) + " ID: " + str(result.json()['id']))
|
||||
# Return job_id
|
||||
return str(result.json()['id'])
|
||||
|
||||
def get_task(self, job_id):
|
||||
""" Get job details from GoDocker and return the job.
|
||||
Pass job_id to the http_get_request method.
|
||||
"""
|
||||
job = False
|
||||
if self.auth.token:
|
||||
result = self.auth.http_get_request("/api/1.0/task/" + str(job_id), {'Authorization': 'Bearer ' + self.auth.token})
|
||||
job = result.json()
|
||||
# Return the job
|
||||
return job
|
||||
|
||||
def task_suspend(self, job_id):
|
||||
""" Suspend actively running job in galaxy.
|
||||
Pass job_id to the http_get_request method.
|
||||
"""
|
||||
job = False
|
||||
if self.auth.token:
|
||||
result = self.auth.http_get_request("/api/1.0/task/" + str(job_id) + "/suspend", {'Authorization': 'Bearer ' + self.auth.token})
|
||||
job = result.json()
|
||||
# Return the job
|
||||
return job
|
||||
|
||||
def get_task_status(self, job_id):
|
||||
""" Get job status from GoDocker and return the status of job.
|
||||
Pass job_id to http_get_request method.
|
||||
"""
|
||||
job = False
|
||||
if self.auth.token:
|
||||
result = self.auth.http_get_request("/api/1.0/task/" + str(job_id) + "/status", {'Authorization': 'Bearer ' + self.auth.token})
|
||||
job = result.json()
|
||||
# Return task status
|
||||
return job
|
||||
|
||||
def delete_task(self, job_id):
|
||||
""" Delete a suspended task in GoDocker.
|
||||
Pass job_id to http_delete_request method.
|
||||
"""
|
||||
job = False
|
||||
if self.auth.token:
|
||||
result = self.auth.http_delete_request("/api/1.0/task/" + str(job_id), {'Authorization': 'Bearer ' + self.auth.token})
|
||||
job = result.json()
|
||||
# Return the job
|
||||
return job
|
||||
|
||||
def get_structure(self, obj):
|
||||
""" Get data and class structure of any object.
|
||||
Meant for Debugging purpose.
|
||||
"""
|
||||
log.debug("\n STRUCTURE \n")
|
||||
memb = inspect.getmembers(obj)
|
||||
for i in memb:
|
||||
log.debug(i)
|
||||
log.debug("\n END OF STRUCTURE \n")
|
||||
return
|
||||
Reference in New Issue
Block a user