diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py index e8df4c16da2..86b01d4cd19 100644 --- a/lib/galaxy/app.py +++ b/lib/galaxy/app.py @@ -76,6 +76,10 @@ class UniverseApplication( object ): from galaxy.util import memdump if memdump.Memdump: self.memdump = memdump.Memdump() + # Transfer manager client + if self.config.get_bool( 'enable_beta_job_managers', False ): + from jobs import transfer_manager + self.transfer_manager = transfer_manager.TransferManager( self ) # Start the job queue self.job_manager = jobs.JobManager( self ) # FIXME: These are exposed directly for backward compatibility @@ -83,10 +87,6 @@ class UniverseApplication( object ): self.job_stop_queue = self.job_manager.job_stop_queue # Initialize the external service types self.external_service_types = external_service_types.ExternalServiceTypesCollection( self.config.external_service_type_config_file, self.config.external_service_type_path, self ) - # Transfer manager client - if self.config.get_bool( 'enable_deferred_job_queue', False ): - from jobs import transfer_manager - self.transfer_manager = transfer_manager.TransferManager( self ) def shutdown( self ): self.job_manager.shutdown() diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py index c3c44b8a670..fadd73bfc47 100644 --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -80,6 +80,8 @@ class Configuration( object ): self.external_service_type_path = resolve_path( kwargs.get( 'external_service_type_path', 'external_service_types' ), self.root ) # Tasked job runner. self.use_tasked_jobs = string_as_bool( kwargs.get( 'use_tasked_jobs', False ) ) + # The transfer manager and deferred job queue + self.enable_beta_job_managers = string_as_bool( kwargs.get( 'enable_beta_job_managers', 'False' ) ) self.local_task_queue_workers = int(kwargs.get("local_task_queue_workers", 2)) self.default_cluster_job_runner = kwargs.get( 'default_cluster_job_runner', 'local:///' ) self.pbs_application_server = kwargs.get('pbs_application_server', "" ) @@ -104,7 +106,6 @@ class Configuration( object ): self.ftp_upload_site = kwargs.get( 'ftp_upload_site', None ) self.allow_library_path_paste = kwargs.get( 'allow_library_path_paste', False ) self.disable_library_comptypes = kwargs.get( 'disable_library_comptypes', '' ).lower().split( ',' ) - self.transfer_manager_port = int( kwargs.get( 'transfer_manager_port', 8163 ) ) # Location for dependencies if 'tool_dependency_dir' in kwargs: self.tool_dependency_dir = resolve_path( kwargs.get( "tool_dependency_dir" ), self.root ) diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index 3ba0f353bd8..098eeeacc67 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -42,7 +42,7 @@ class JobManager( object ): # Queues for starting and stopping jobs self.job_queue = JobQueue( app, self.dispatcher ) self.job_stop_queue = JobStopQueue( app, self.dispatcher ) - if self.app.config.get_bool( 'enable_deferred_job_queue', False ): + if self.app.config.enable_beta_job_managers: from galaxy.jobs.deferred import DeferredJobQueue self.deferred_job_queue = DeferredJobQueue( app ) else: diff --git a/lib/galaxy/jobs/deferred/__init__.py b/lib/galaxy/jobs/deferred/__init__.py index 7886672ca38..8ff113a119e 100644 --- a/lib/galaxy/jobs/deferred/__init__.py +++ b/lib/galaxy/jobs/deferred/__init__.py @@ -15,7 +15,7 @@ class DeferredJobQueue( object ): INVALID = 'invalid' ) def __init__( self, app ): self.app = app - self.sa_session = app.model.context + self.sa_session = app.model.context.current self.queue = Queue() self.plugins = {} self._load_plugins() @@ -63,7 +63,9 @@ class DeferredJobQueue( object ): if 'check_interval' in dir( self.plugins[job.plugin] ): job.check_interval = self.plugins[job.plugin].check_interval log.info( 'Recovered deferred job (id: %s) at startup' % job.id ) - self.waiting_jobs.append( job ) + # Pass the job ID as opposed to the job, since the monitor thread + # needs to load it in its own threadlocal scoped session. + self.waiting_jobs.append( job.id ) def __monitor( self ): while self.running: @@ -89,19 +91,30 @@ class DeferredJobQueue( object ): self.waiting_jobs.append( job ) new_waiting = [] for job in self.waiting_jobs: + try: + # Recovered jobs are passed in by ID + assert type( job ) is int + job = self.sa_session.query( model.DeferredJob ).get( job ) + except: + pass if job.is_check_time: try: job_state = self.plugins[job.plugin].check_job( job ) - except: - raise # TODO: fail + except Exception, e: + self.__fail_job( job ) + log.error( 'Set deferred job %s to error because of an exception in check_job(): %s' % ( job.id, str( e ) ) ) + continue if job_state == self.job_states.READY: try: self.plugins[job.plugin].run_job( job ) - except: - raise # TODO: fail + except Exception, e: + self.__fail_job( job ) + log.error( 'Set deferred job %s to error because of an exception in run_job(): %s' % ( job.id, str( e ) ) ) + continue elif job_state == self.job_states.INVALID: - # TODO: fail + self.__fail_job( job ) log.error( 'Unable to run deferred job (id: %s): Plugin "%s" marked it as invalid' % ( job.id, job.plugin ) ) + continue else: new_waiting.append( job ) job.last_check = 'now' @@ -113,7 +126,6 @@ class DeferredJobQueue( object ): if job.plugin not in self.plugins: log.error( 'Invalid deferred job plugin: %s' ) % job.plugin job.state = model.DeferredJob.states.ERROR - job.info = 'Invalid deferred job plugin: %s' % job.plugin self.sa_session.add( job ) self.sa_session.flush() return False @@ -122,6 +134,11 @@ class DeferredJobQueue( object ): def __check_if_ready_to_run( self, job ): return self.plugins[job.plugin].check_job( job ) + def __fail_job( self, job ): + job.state = model.DeferredJob.states.ERROR + self.sa_session.add( job ) + self.sa_session.flush() + def shutdown( self ): self.running = False self.sleeper.wake() diff --git a/lib/galaxy/jobs/transfer_manager.py b/lib/galaxy/jobs/transfer_manager.py index cd214478cd0..b80ce1062f1 100644 --- a/lib/galaxy/jobs/transfer_manager.py +++ b/lib/galaxy/jobs/transfer_manager.py @@ -1,8 +1,8 @@ """ -Client interface to the Galaxy Transfer Manager, which is a standalone, -lightweight on-demand daemon. +Manage transfers from arbitrary URLs to temporary files. Socket interface for +IPC with multiple process configurations. """ -import subprocess, socket, logging +import os, subprocess, socket, logging, threading from galaxy import eggs from galaxy.util import listify, json @@ -10,12 +10,19 @@ from galaxy.util import listify, json log = logging.getLogger( __name__ ) class TransferManager( object ): + """ + Manage simple data transfers from URLs to temporary locations. + """ def __init__( self, app ): self.app = app self.sa_session = app.model.context.current - self.tm_port = app.config.transfer_manager_port - self.tm_command = 'python transfer_manager.py' - self.tm_transfer_job_id_flag = '-i' + self.command = 'python %s' % os.path.abspath( os.path.join( os.getcwd(), 'scripts', 'transfer.py' ) ) + if app.config.get_bool( 'enable_job_recovery', True ): + # Only one Galaxy server process should be able to recover jobs! (otherwise you'll have nasty race conditions) + self.running = True + self.sleeper = Sleeper() + self.restarter = threading.Thread( target=self.__restarter ) + self.restarter.start() def new( self, path=None, **kwd ): if 'url' not in kwd: raise Exception( 'Missing required parameter "url".' ) @@ -32,48 +39,117 @@ class TransferManager( object ): transfer_manager will either daemonize or return after submitting to a running daemon, so it should be fairly quick to return. """ - spaced_flag = ' %s ' % self.tm_transfer_job_id_flag - cmd = '%s %s %s' % ( self.tm_command, self.tm_transfer_job_id_flag, spaced_flag.join( [ str( tj.id ) for tj in transfer_jobs ] ) ) - log.debug( 'Initiating Transfer Job(s): %s' % ', '.join( [ str( tj.id ) for tj in transfer_jobs ] ) ) - p = subprocess.Popen( cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT ) - p.wait() - return p.stdout.read() - def status( self, transfer_jobs ): + transfer_jobs = listify( transfer_jobs ) + printable_tj_ids = ', '.join( [ str( tj.id ) for tj in transfer_jobs ] ) + log.debug( 'Initiating transfer job(s): %s' % printable_tj_ids ) + # Set all jobs running before spawning, or else updating the state may + # clobber a state change performed by the worker. + [ tj.__setattr__( 'state', tj.states.RUNNING ) for tj in transfer_jobs ] + self.sa_session.add_all( transfer_jobs ) + self.sa_session.flush() + for tj in transfer_jobs: + # The transfer script should daemonize fairly quickly - if this is + # not the case, this process will need to be moved to a + # non-blocking method. + cmd = '%s %s' % ( self.command, tj.id ) + log.debug( 'Transfer command is: %s' % cmd ) + p = subprocess.Popen( cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT ) + p.wait() + output = p.stdout.read( 32768 ) + if p.returncode != 0: + log.error( 'Spawning transfer job failed: %s: %s' % ( tj.id, output ) ) + tj.state = tj.states.ERROR + tj.info = 'Spawning transfer job failed: %s' % output.splitlines()[-1] + self.sa_session.add( tj ) + self.sa_session.flush() + def get_state( self, transfer_jobs, via_socket=False ): transfer_jobs = listify( transfer_jobs ) rval = [] - sock = socket.socket( socket.AF_INET, socket.SOCK_STREAM ) - sock.settimeout( 10 ) - try: - sock.connect( ( 'localhost', self.tm_port ) ) - except Exception, e: - log.warning( 'sock.connect for status update of Transfer Jobs %s failed (this is okay if all jobs have finished): %s' % ( ', '.join( [ str( tj.id ) for tj in transfer_jobs ] ), str( e ) ) ) - [ self.sa_session.refresh( tj ) for tj in transfer_jobs ] - new_jobs = filter( lambda x: x.state == self.app.model.TransferJob.states.NEW, transfer_jobs ) - #terminal_jobs = filter( lambda x: x.state in [ self.app.model.TransferJob.states.DONE, \ - # self.app.model.TransferJob.states.ERROR ], transfer_jobs ) - if new_jobs: - # This could be a bad idea if the transfer manager daemon is misbehaving. - output = self.run( new_jobs ) - for tj in transfer_jobs: - if tj.state == tj.states.DONE: - log.debug( 'Transfer Job %s is complete' % tj.id ) - rval.append( dict( transfer_job_id=tj.id, state=tj.state ) ) - if len( rval ) == 1: - return rval[0] - return rval - sock.send( json.to_json_string( dict( state_transfer_job_ids=[ t.id for t in transfer_jobs ] ) ) + '\n' ) - resp = sock.recv( 8192 ) - for line in resp.splitlines(): - status = json.from_json_string( line ) - # TODO: need a bunch for this - if status['state'] == 'unknown': - transfer_job = [ tj for tj in transfer_jobs if tj.id == int( status['transfer_job_id'] ) ][0] - self.sa_session.refresh( tj ) - rval.append( dict( transfer_job_id=tj.id, state=tj.state ) ) + for tj in transfer_jobs: + if via_socket and tj.state not in tj.terminal_states and tj.socket: + try: + request = json.jsonrpc_request( method='get_state', id=True ) + sock = socket.socket( socket.AF_INET, socket.SOCK_STREAM ) + sock.settimeout( 5 ) + sock.connect( ( 'localhost', tj.socket ) ) + sock.send( json.to_json_string( request ) ) + response = sock.recv( 8192 ) + valid, response = json.validate_jsonrpc_response( response, id=request['id'] ) + if not valid: + # No valid response received, make some pseudo-json-rpc + raise Exception( dict( code=128, message='Did not receive valid response from transfer daemon for state' ) ) + if 'error' in response: + # Response was valid but Request resulted in an error + raise Exception( error ) + else: + # Request was valid + response['result']['transfer_job_id'] = tj.id + rval.append( response['result'] ) + except Exception, e: + # State checking via the transfer daemon failed, just + # return the state from the database instead. Callers can + # look for the 'error' member of the response to see why + # the check failed. + self.sa_session.refresh( tj ) + error = e.args + if type( error ) != dict: + error = dict( code=256, message='Error connecting to transfer daemon', data=str( e ) ) + rval.append( dict( transfer_job_id=tj.id, state=tj.state, error=error ) ) else: - if status['state'] == 'progress' and 'percent' in status: - log.debug( 'Transfer Job %s is %s complete' % ( status['transfer_job_id'], status['percent'] ) ) - rval.append( status ) + rval.append( dict( transfer_job_id=tj.id, state=tj.state ) ) + for tj_state in rval: + if tj_state['state'] in self.app.model.TransferJob.terminal_states: + log.debug( 'Transfer job %s is in terminal state: %s' % ( tj_state['transfer_job_id'], tj_state['state'] ) ) + elif tj_state['state'] == self.app.model.TransferJob.states.PROGRESS and 'percent' in tj_state: + log.debug( 'Transfer job %s is %s%% complete' % ( tj_state['transfer_job_id'], tj_state['percent'] ) ) if len( rval ) == 1: return rval[0] return rval + def __restarter( self ): + log.info( 'Transfer job restarter starting up...' ) + while self.running: + dead = [] + self.sa_session.expunge_all() # our session is threadlocal so this is safe. + for tj in self.sa_session.query( self.app.model.TransferJob ) \ + .filter( self.app.model.TransferJob.state == self.app.model.TransferJob.states.RUNNING ): + if not tj.pid: + continue + # This will only succeed if the process exists and is owned by the + # user running Galaxy (unless that user is root, in which case it + # can be owned by anyone - but you're not running Galaxy as root, + # right?). This is not guaranteed proof that the transfer is alive + # since another process may have assumed the original process' PID. + # But that will only cause the transfer to not restart until that + # process dies, which hopefully won't be too long from now... If + # it becomes a problem, try to talk to the socket a few times and + # restart the transfer if socket communication fails repeatedly. + try: + os.kill( tj.pid, 0 ) + except: + self.sa_session.refresh( tj ) + if tj.state == tj.states.RUNNING: + log.error( 'Transfer job %s is marked as running but pid %s appears to be dead.' % ( tj.id, tj.pid ) ) + dead.append( tj ) + if dead: + self.run( dead ) + self.sleeper.sleep( 30 ) + log.info( 'Transfer job restarter shutting down...' ) + def shutdown( self ): + self.running = False + self.sleeper.wake() + +class Sleeper( object ): + """ + Provides a 'sleep' method that sleeps for a number of seconds *unless* + the notify method is called (from a different thread). + """ + def __init__( self ): + self.condition = threading.Condition() + def sleep( self, seconds ): + self.condition.acquire() + self.condition.wait( seconds ) + self.condition.release() + def wake( self ): + self.condition.acquire() + self.condition.notify() + self.condition.release() diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 0062b868e05..6d7dea72002 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -2173,14 +2173,22 @@ class VisualizationUserShareAssociation( object ): self.user = None class TransferJob( object ): + # These states are used both by the transfer manager's IPC and the object + # state in the database. Not all states are used by both. states = Bunch( NEW = 'new', + UNKNOWN = 'unknown', + PROGRESS = 'progress', RUNNING = 'running', ERROR = 'error', DONE = 'done' ) - def __init__( self, state=None, path=None, info=None, params=None ): + terminal_states = [ states.ERROR, + states.DONE ] + def __init__( self, state=None, path=None, info=None, pid=None, socket=None, params=None ): self.state = state self.path = path self.info = info + self.pid = pid + self.socket = socket self.params = params class Tag ( object ): diff --git a/lib/galaxy/model/mapping.py b/lib/galaxy/model/mapping.py index cc99207ed16..c66f72a5eaa 100644 --- a/lib/galaxy/model/mapping.py +++ b/lib/galaxy/model/mapping.py @@ -445,6 +445,8 @@ TransferJob.table = Table( "transfer_job", metadata, Column( "state", String( 64 ), index=True ), Column( "path", String( 1024 ) ), Column( "info", TEXT ), + Column( "pid", Integer ), + Column( "socket", Integer ), Column( "params", JSONType ) ) Event.table = Table( "event", metadata, diff --git a/lib/galaxy/model/migrate/versions/0072_add_pid_and_socket_columns_to_transfer_job_table.py b/lib/galaxy/model/migrate/versions/0072_add_pid_and_socket_columns_to_transfer_job_table.py new file mode 100644 index 00000000000..f291858214a --- /dev/null +++ b/lib/galaxy/model/migrate/versions/0072_add_pid_and_socket_columns_to_transfer_job_table.py @@ -0,0 +1,39 @@ +""" +Migration script to add 'pid' and 'socket' columns to the transfer_job table. +""" + +from sqlalchemy import * +from sqlalchemy.orm import * +from migrate import * +from migrate.changeset import * + +import logging +log = logging.getLogger( __name__ ) + +metadata = MetaData( migrate_engine ) +db_session = scoped_session( sessionmaker( bind=migrate_engine, autoflush=False, autocommit=True ) ) + +def upgrade(): + print __doc__ + metadata.reflect() + try: + TransferJob_table = Table( "transfer_job", metadata, autoload=True ) + c = Column( "pid", Integer ) + c.create( TransferJob_table ) + assert c is TransferJob_table.c.pid + c = Column( "socket", Integer ) + c.create( TransferJob_table ) + assert c is TransferJob_table.c.socket + except Exception, e: + print "Adding columns to transfer_job table failed: %s" % str( e ) + log.debug( "Adding columns to transfer_job table failed: %s" % str( e ) ) + +def downgrade(): + metadata.reflect() + try: + TransferJob_table = Table( "transfer_job", metadata, autoload=True ) + TransferJob_table.c.pid.drop() + TransferJob_table.c.socket.drop() + except Exception, e: + print "Dropping columns from transfer_job table failed: %s" % str( e ) + log.debug( "Dropping columns from transfer_job table failed: %s" % str( e ) ) diff --git a/lib/galaxy/util/json.py b/lib/galaxy/util/json.py index 8e69713bc19..5219a44f52a 100644 --- a/lib/galaxy/util/json.py +++ b/lib/galaxy/util/json.py @@ -1,5 +1,8 @@ -__all__ = [ "to_json_string", "from_json_string", "json_fix"] +__all__ = [ "to_json_string", "from_json_string", "json_fix", "validate_jsonrpc_request", "validate_jsonrpc_response", "jsonrpc_request", "jsonrpc_response" ] + +import random, string, logging +import socket import pkg_resources pkg_resources.require( "simplejson" ) @@ -9,6 +12,8 @@ import simplejson to_json_string = simplejson.dumps from_json_string = simplejson.loads +log = logging.getLogger( __name__ ) + def json_fix( val ): if isinstance( val, list ): return [ json_fix( v ) for v in val ] @@ -17,4 +22,90 @@ def json_fix( val ): elif isinstance( val, unicode ): return val.encode( "utf8" ) else: - return val \ No newline at end of file + return val + +# Methods for handling JSON-RPC + +def validate_jsonrpc_request( request, regular_methods, notification_methods ): + try: + request = from_json_string( request ) + except Exception, e: + return False, request, jsonrpc_response( id = None, error = dict( code = -32700, message = 'Parse error', data = str( e ) ) ) + try: + assert 'jsonrpc' in request, \ + 'This server requires JSON-RPC 2.0 and no "jsonrpc" member was sent with the Request object as per the JSON-RPC 2.0 Specification.' + assert request['jsonrpc'] == '2.0', \ + 'Requested JSON-RPC version "%s" != required version "2.0".' % request['jsonrpc'] + assert 'method' in request, 'No "method" member was sent with the Request object' + except AssertionError, e: + return False, request, jsonrpc_response( request = request, error = dict( code = -32600, message = 'Invalid Request', data = str( e ) ) ) + try: + assert request['method'] in ( regular_methods + notification_methods ) + except AssertionError, e: + return False, request, jsonrpc_response( request = request, + error = dict( code = -32601, + message = 'Method not found', + data = 'Valid methods are: %s' % ', '.join( regular_methods + notification_methods ) ) ) + try: + if request['method'] in regular_methods: + assert 'id' in request, 'No "id" member was sent with the Request object and the requested method "%s" is not a notification method' % request['method'] + except AssertionError, e: + return False, request, jsonrpc_response( request = request, error = dict( code = -32600, message = 'Invalid Request', data = str( e ) ) ) + return True, request, None + +def validate_jsonrpc_response( response, id=None ): + try: + response = from_json_string( response ) + except Exception, e: + log.error( 'Response was not valid JSON: %s' % str( e ) ) + log.debug( 'Response was: %s' % response ) + return False, response + try: + assert 'jsonrpc' in response, \ + 'This server requires JSON-RPC 2.0 and no "jsonrpc" member was sent with the Response object as per the JSON-RPC 2.0 Specification.' + assert ( 'result' in response or 'error' in response ), \ + 'Neither of "result" or "error" members were sent with the Response object.' + if 'error' in response: + assert int( response['error']['code'] ), \ + 'The "code" member of the "error" object in the Response is missing or not an integer.' + assert 'message' in response, \ + 'The "message" member of the "error" object in the Response is missing.' + except Exception, e: + log.error( 'Response was not valid JSON-RPC: %s' % str( e ) ) + log.debug( 'Response was: %s' % response ) + return False, response + if id is not None: + try: + assert 'id' in response and response['id'] == id + except Exception, e: + log.error( 'The response id "%s" does not match the request id "%s"' % ( response['id'], id ) ) + return False, response + return True, response + +def jsonrpc_request( method, params=None, id=None, jsonrpc='2.0' ): + if method is None: + log.error( 'jsonrpc_request(): "method" parameter cannot be None' ) + return None + request = dict( jsonrpc = jsonrpc, method = method ) + if params: + request['params'] = params + if id is not None and id is True: + request['id'] = ''.join( [ random.choice( string.hexdigits ) for i in range( 16 ) ] ) + elif id is not None: + request['id'] = id + return request + +def jsonrpc_response( request=None, id=None, result=None, error=None, jsonrpc='2.0' ): + if result: + rval = dict( jsonrpc = jsonrpc, result = result ) + elif error: + rval = dict( jsonrpc = jsonrpc, error = error ) + else: + msg = 'jsonrpc_response() called with out a "result" or "error" parameter' + log.error( msg ) + rval = dict( jsonrpc = jsonrpc, error = msg ) + if id is not None: + rval['id'] = id + elif request is not None and 'id' in request: + rval['id'] = request['id'] + return rval diff --git a/scripts/transfer.py b/scripts/transfer.py new file mode 100644 index 00000000000..8a0b177f2b4 --- /dev/null +++ b/scripts/transfer.py @@ -0,0 +1,258 @@ +#!/usr/bin/env python +""" +Downloads files to temp locations. This script is invoked by the Transfer +Manager (galaxy.jobs.transfer_manager) and should not normally be invoked by +hand. +""" + +import os, sys, optparse, ConfigParser, socket, SocketServer, threading, logging, random + +import urllib2, tempfile + +import time + +log = logging.getLogger( __name__ ) +log.setLevel( logging.INFO ) +handler = logging.StreamHandler( sys.stdout ) +log.addHandler( handler ) + +galaxy_root = os.path.abspath( os.path.join( os.path.dirname( __file__ ), '..' ) ) +sys.path.insert( 0, os.path.abspath( os.path.join( galaxy_root, 'lib' ) ) ) + +from galaxy import eggs +eggs.require( "SQLAlchemy >= 0.4" ) + +from sqlalchemy import * +from sqlalchemy.orm import * +from galaxy.model.mapping import load_egg_for_url + +import galaxy.model +from galaxy.util import json, bunch + +eggs.require( 'python_daemon' ) +from daemon import DaemonContext + +debug = False +slow = False + +class ArgHandler( object ): + """ + Collect command line flags. + """ + def __init__( self ): + self.parser = optparse.OptionParser() + self.parser.add_option( '-c', '--config', dest='config', help='Path to Galaxy config file (universe_wsgi.ini)', + default=os.path.abspath( os.path.join( galaxy_root, 'universe_wsgi.ini' ) ) ) + self.parser.add_option( '-d', '--debug', action='store_true', dest='debug', help="Debug (don't detach)" ) + self.parser.add_option( '-s', '--slow', action='store_true', dest='slow', help="Transfer slowly (for debugging)" ) + self.opts = None + def parse( self ): + self.opts, args = self.parser.parse_args() + if len( args ) != 1: + log.error( 'usage: transfer.py [options] ' ) + sys.exit( 1 ) + try: + self.transfer_job_id = int( args[0] ) + except TypeError: + log.error( 'The provided transfer job ID is not an integer: %s' % args[0] ) + sys.exit( 1 ) + if self.opts.debug: + global debug + debug = True + log.setLevel( logging.DEBUG ) + if self.opts.slow: + global slow + slow = True + +class GalaxyApp( object ): + """ + A shell Galaxy App to provide access to the Galaxy configuration and + model/database. + """ + def __init__( self, config_file ): + self.config = ConfigParser.ConfigParser( dict( database_file = 'database/universe.sqlite', + file_path = 'database/files', + transfer_worker_port_range = '12275-12675', + transfer_worker_log = None ) ) + self.config.read( config_file ) + self.model = bunch.Bunch() + self.connect_database() + def connect_database( self ): + # Avoid loading the entire model since doing so is exceptionally slow + default_dburl = 'sqlite:///%s?isolation_level=IMMEDIATE' % self.config.get( 'app:main', 'database_file' ) + try: + dburl = self.config.get( 'app:main', 'database_connection' ) + except ConfigParser.NoOptionError: + dburl = default_dburl + load_egg_for_url( dburl ) + engine = create_engine( dburl ) + metadata = MetaData( engine ) + self.sa_session = scoped_session( sessionmaker( bind=engine, autoflush=False, autocommit=True ) ) + self.model.TransferJob = galaxy.model.TransferJob + self.model.TransferJob.table = Table( "transfer_job", metadata, autoload=True ) + def get_transfer_job( self, id ): + return self.sa_session.query( self.model.TransferJob ).get( int( id ) ) + +class ListenerServer( SocketServer.ThreadingTCPServer ): + """ + The listener will accept state requests and new transfers for as long as + the manager is running. + """ + def __init__( self, port_range, RequestHandlerClass, app, transfer_job, state_result ): + self.state_result = state_result + # Try random ports until a free one is found + while True: + random_port = random.choice( port_range ) + try: + SocketServer.ThreadingTCPServer.__init__( self, ( 'localhost', random_port ), RequestHandlerClass ) + log.info( 'Listening on port %s' % random_port ) + break + except Exception, e: + log.warning( 'Tried binding port %s: %s' % ( random_port, str( e ) ) ) + transfer_job.socket = random_port + app.sa_session.add( transfer_job ) + app.sa_session.flush() + +class ListenerRequestHandler( SocketServer.BaseRequestHandler ): + """ + Handle state or transfer requests received on the socket. + """ + def handle( self ): + request = self.request.recv( 8192 ) + response = {} + valid, request, response = json.validate_jsonrpc_request( request, ( 'get_state', ), () ) + if valid: + self.request.send( json.to_json_string( json.jsonrpc_response( request=request, result=self.server.state_result.result ) ) ) + else: + error_msg = 'Unable to serve request: %s' % response['error']['message'] + if 'data' in response['error']: + error_msg += ': %s' % response['error']['data'] + log.error( error_msg ) + log.debug( 'Original request was: %s' % request ) + +class StateResult( object ): + """ + A mutable container for the 'result' portion of JSON-RPC responses to state requests. + """ + def __init__( self, result=None ): + self.result = result + +def transfer( app, transfer_job_id ): + + transfer_job = app.get_transfer_job( transfer_job_id ) + if transfer_job is None: + log.error( 'Invalid transfer job ID: %s' % transfer_job_id ) + return False + + port_range = app.config.get( 'app:main', 'transfer_worker_port_range' ) + try: + port_range = [ int( p ) for p in port_range.split( '-' ) ] + except Exception, e: + log.error( 'Invalid port range set in transfer_worker_port_range: %s: %s' % ( port_range, str( e ) ) ) + return False + + protocol = transfer_job.params['url'].split( '://' )[0] + if protocol not in ( 'http', 'https' ): + log.error( 'Unsupported protocol: %s' % protocol ) + return False + + state_result = StateResult( result = dict( state = transfer_job.states.RUNNING, info='Transfer process starting up.' ) ) + + listener_server = ListenerServer( range( port_range[0], port_range[1] + 1 ), ListenerRequestHandler, app, transfer_job, state_result ) + + # daemonize here (if desired) + if not debug: + daemon_context = DaemonContext( files_preserve=[ listener_server.fileno() ], working_directory=os.getcwd() ) + daemon_context.open() + + # If this fails, it'll never be detected. Hopefully it won't fail since it succeeded once. + app.connect_database() # daemon closed the database fd + transfer_job = app.get_transfer_job( transfer_job_id ) + + listener_thread = threading.Thread( target=listener_server.serve_forever ) + listener_thread.setDaemon( True ) + listener_thread.start() + + # Store this process' pid so unhandled deaths can be handled by the restarter + transfer_job.pid = os.getpid() + app.sa_session.add( transfer_job ) + app.sa_session.flush() + + terminal_state = None + if protocol in ( 'http', 'https' ): + for state in http_transfer( transfer_job ): + state_result.result = state + if state['state'] in transfer_job.terminal_states: + terminal_state = state + if terminal_state is not None: + transfer_job.state = terminal_state['state'] + for name in ( 'info', 'path' ): + if name in terminal_state: + transfer_job.__setattr__( name, terminal_state[name] ) + else: + transfer_job.state = transfer_job.states.ERROR + transfer_job.info = 'Unknown error encountered by transfer worker.' + app.sa_session.add( transfer_job ) + app.sa_session.flush() + return True + +def http_transfer( transfer_job ): + """ + "Plugin" for handling http(s) transfers. + """ + url = transfer_job.params['url'] + try: + f = urllib2.urlopen( url ) + except urllib2.URLError, e: + yield dict( state = transfer_job.states.ERROR, info = 'Unable to open URL: %s' % str( e ) ) + return + size = f.info().getheader( 'Content-Length' ) + if size is not None: + size = int( size ) + chunksize = 1024 * 1024 + if slow: + chunksize = 1024 + read = 0 + last = 0 + try: + fh, fn = tempfile.mkstemp() + except Exception, e: + yield dict( state = transfer_job.states.ERROR, info = 'Unable to create temporary file for transfer: %s' % str( e ) ) + return + log.debug( 'Writing %s to %s, size is %s' % ( url, fn, size or 'unknown' ) ) + try: + while True: + chunk = f.read( chunksize ) + if not chunk: + break + os.write( fh, chunk ) + read += chunksize + if size is not None and read < size: + percent = int( float( read ) / size * 100 ) + if percent != last: + yield dict( state = transfer_job.states.PROGRESS, read = read, percent = '%s' % percent ) + last = percent + elif size is None: + yield dict( state = transfer_job.states.PROGRESS, read = read ) + if slow: + time.sleep( 1 ) + os.close( fh ) + yield dict( state = transfer_job.states.DONE, path = fn ) + except Exception, e: + yield dict( state = transfer_job.states.ERROR, info = 'Error during file transfer: %s' % str( e ) ) + return + return + +if __name__ == '__main__': + + arg_handler = ArgHandler() + arg_handler.parse() + app = GalaxyApp( arg_handler.opts.config ) + + log.debug( 'Initiating transfer' ) + if transfer( app, arg_handler.transfer_job_id ): + log.debug( 'Finished' ) + else: + log.error( 'Error in transfer process' ) + sys.exit( 1 ) + sys.exit( 0 ) diff --git a/transfer_manager.py b/transfer_manager.py deleted file mode 100644 index 5787e5f07e2..00000000000 --- a/transfer_manager.py +++ /dev/null @@ -1,330 +0,0 @@ -#!/usr/bin/env python -''' -Downloads files to temp locations. -''' - -import os, sys, optparse, ConfigParser, socket, SocketServer, errno, Queue, threading, subprocess - -import urllib2, tempfile - -import time - -sys.path.insert( 0, os.path.abspath( 'lib' ) ) -from galaxy import eggs -import galaxy.model.mapping -from galaxy.util import json, bunch - -eggs.require( 'python_daemon' ) -from daemon import DaemonContext - -class ArgHandler( object ): - """ - Collect command line flags. - """ - def __init__( self ): - self.parser = optparse.OptionParser() - self.parser.add_option( '-c', '--config', dest='config', help='Path to Galaxy config file (universe_wsgi.ini)', default='universe_wsgi.ini' ) - self.parser.add_option( '-i', '--transfer-job-id', action='append', dest='transfer_job_ids', help='Initiate management of the specified TransferJob id' ) - self.parser.add_option( '--do', dest='initiate_transfer_job_id', help='Used by this script when it calls itself to actually initiate the download' ) - self.parser.add_option( '-s', '--state-transfer-job-id', action='append', dest='state_transfer_job_ids', help='Report the state of the specified TransferJob id' ) - self.parser.add_option( '-d', '--debug', action='store_true', dest='debug', help="Debug (don't detach)" ) - self.opts = None - def parse( self ): - self.opts, args = self.parser.parse_args() - -class GalaxyApp( object ): - """ - A shell Galaxy App to provide access to the Galaxy configuration and - model/database. - """ - def __init__( self, config_file='universe_wsgi.ini' ): - self.config = ConfigParser.ConfigParser( dict( database_file = 'database/universe.sqlite', - file_path = 'database/files', - transfer_manager_port = '8163', - transfer_manager_log = 'transfer_manager.log' ) ) - self.config.read( config_file ) - self.model = None - @property - def sa_session( self ): - if not self.model: - default_dburl = 'sqlite:///%s?isolation_level=IMMEDIATE' % self.config.get( 'app:main', 'database_file' ) - try: - dburl = self.config.get( 'app:main', 'database_connection' ) - except ConfigParser.NoOptionError: - dburl = default_dburl - self.model = galaxy.model.mapping.init( self.config.get( 'app:main', 'file_path' ), dburl, create_tables = False ) - return self.model.context.current - def get_transfer_job( self, id ): - return self.sa_session.query( self.model.TransferJob ).get( int( id ) ) - -class ListenerServer( SocketServer.ThreadingTCPServer ): - """ - The listener will accept state requests and new transfers for as long as - the manager is running. - """ - def __init__( self, server_address, RequestHandlerClass, transfer_manager ): - SocketServer.ThreadingTCPServer.__init__( self, server_address, RequestHandlerClass ) - self.transfer_manager = transfer_manager - self.app = transfer_manager.app - -class ListenerRequestHandler( SocketServer.BaseRequestHandler ): - """ - Handle state or transfer requests received on the socket. - """ - def handle( self ): - if not self.server.transfer_manager.accepting: - # TODO: does the submitter handle this condition? i'm sure it doesn't... - self.request.send( 'Manager shutting down\n' ) - return - data = '' - while len( data ) < 8 * 1024 * 1024: - # read data up to 8MB (overkill, but be safe) - chunk = self.request.recv( 1024 ) - if not chunk: - break - data += chunk - if '\n' in data: - break - else: - self.request.send( 'Message too large\n' ) - return - data = json.from_json_string( data ) - if 'transfer_job_ids' in data: - # Get all of the TransferJob objects and stick them on the queue. - print 'Adding transfer job ids to transfer queue: %s' % data['transfer_job_ids'] - [ self.server.transfer_manager.transfer_queue.put( self.server.app.get_transfer_job( transfer_job_id ) ) for transfer_job_id in data['transfer_job_ids'] ] - self.request.send( "Added jobs to transfer queue: %s\n" % ', '.join( data['transfer_job_ids'] ) ) - elif 'state_transfer_job_ids' in data: - print 'Servicing state request for transfer job ids: %s' % data['state_transfer_job_ids'] - for state_transfer_job_id in data['state_transfer_job_ids']: - state = self.server.transfer_manager.get_state( int( state_transfer_job_id ) ) - state['transfer_job_id'] = state_transfer_job_id - print 'State of transfer job id %s is: %s' % ( state_transfer_job_id, state ) - self.request.send( json.to_json_string( state ) ) - -class Transfer( object ): - """ - Instantiated for each transfer to track the progress of the transfer via - commmunication with a subprocess. - """ - states = bunch.Bunch( NEW = 'new', - UNKNOWN = 'unknown', - STARTED = 'started', - PROGRESS = 'progress', - DONE = 'done', - ERROR = 'error' ) - def __init__( self, transfer_job ): - self.transfer_job = transfer_job - self.state = dict( state = self.states.NEW ) - self.done = False - def run( self ): - cmd = '%s -u %s --do %s' % ( sys.executable, os.path.abspath( __file__ ), self.transfer_job.id ) - print cmd - self.p = subprocess.Popen( cmd, bufsize=0, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT ) - line = self.p.stdout.readline() - while line: - try: - self.state = json.from_json_string( line ) - assert 'state' in self.state - except: - print 'Received unknown state from transfer (transfer job id: %s): %s' % ( self.transfer_job.id, line ) - self.state = dict( state = self.states.UNKNOWN, info=line ) - line = self.p.stdout.readline() - self.p.wait() - self.done = True - -class TransferManager( object ): - """ - Manage the queue of transfers, handle the setup of new transfers and - completion of finished transfers. - """ - def __init__( self, app, transfer_job_ids, debug=False ): - self.app = app - self.daemon_context = None - self.sa_session = app.sa_session - self.transfer_job_ids = transfer_job_ids - self.debug = debug - self.port = int( app.config.get( 'app:main', 'transfer_manager_port' ) ) - self.log_file_name = app.config.get( 'app:main', 'transfer_manager_log' ) - self.transfer_queue = Queue.Queue() - self.accepting = True - self.watchlist = [] - self.listen_or_submit() - def listen_or_submit( self ): - """ - The transfer manager is invoked the same way every time, via the - command line, but if a transfer manager is already running, the new - manager will simply submit the request to the old manager and - terminate. - """ - try: - self.listener_server = ListenerServer( ( 'localhost', self.port ), ListenerRequestHandler, self ) - except socket.error, e: - if e[0] == errno.EADDRINUSE: - self.submit() - sys.exit() - - # Daemonize - if not self.debug: - log_file = open( self.log_file_name, 'a+' ) - self.daemon_context = DaemonContext( files_preserve=[ self.listener_server.fileno() ], - working_directory=os.getcwd(), - stdout=log_file, - stderr=log_file ) - self.daemon_context.open() - - # Listen for more stuff on the socket - self.listener = threading.Thread( target=self.listener_server.serve_forever ) - self.listener.start() - - # Put all the URLs into the queue, additional URLs received before this - # instance of the manager terminates will also be handled. - print 'Started up with transfer job ids: %s' % self.transfer_job_ids - [ self.transfer_queue.put( app.get_transfer_job( tj_id ) ) for tj_id in self.transfer_job_ids ] - - while True: - # TODO: max transfer limit - try: - transfer_job = self.transfer_queue.get_nowait() - print 'Fetching transfer job:', transfer_job.id, 'URL is:', transfer_job.params['url'] - t = Transfer( transfer_job ) - transfer_job.state = app.model.TransferJob.states.RUNNING - self.sa_session.add( transfer_job ) - self.sa_session.flush() - tt = threading.Thread( target=t.run ) - tt.start() - self.watchlist.append( t ) - except Queue.Empty: - # TODO: this will shutdown the queue as soon the queue is - # empty, which is almost immediately. instead, we need to - # accept until all transfers are finished, then stop accepting, - # but finish out the queue in case one was added in the instant - # between empty and no longer accepting. - # (check the queue one more time after accepting = False) - if not self.watchlist: - self.acecpting = False - break - new_watchlist = [] - for transfer in self.watchlist: - # TODO: handle failure - if transfer.done: - if transfer.state['state'] == transfer.states.DONE: - transfer.transfer_job.state = app.model.TransferJob.states.DONE - transfer.transfer_job.path = transfer.state['path'] - print 'Transfer of job %s ended successfully, output path is: %s' % ( transfer.transfer_job.id, transfer.transfer_job.path ) - elif transfer.state['state'] == transfer.states.ERROR: - transfer.transfer_job.info = transfer.state.get( 'info', None ) - transfer.transfer_job.state = app.model.TransferJob.states.ERROR - print 'Transfer of job %s ended in error: %s' % ( transfer.transfer_job.id, transfer.transfer_job.info ) - else: - print 'Unknown state received for transfer job %s: %s' % ( transfer.transfer_job.id, transfer.state['state'] ) - transfer.transfer_job.info = 'Unknown error encountered in transfer manager' - transfer.transfer_job.state = app.model.TransferJob.states.ERROR - self.sa_session.add( transfer.transfer_job ) - self.sa_session.flush() - else: - new_watchlist.append( transfer ) - time.sleep( 1 ) - self.watchlist = new_watchlist - #except Queue.Empty: - # self.accepting = False - self.listener_server.shutdown() - - def get_state( self, transfer_job_id ): - rval = {} - for transfer in self.watchlist: - if transfer.transfer_job.id == transfer_job_id: - rval = transfer.state - break - else: - rval['state'] = Transfer.states.UNKNOWN # should be DONE? - return rval - - def submit( self ): - # TODO: can fail if shutdown occurs between failure to bind and submission - # Needs error handling. - # This may not work at all right now. - print "Submitting jobs to running transfer manager: %s" % ', '.join( self.transfer_job_ids ) - sock = socket.socket( socket.AF_INET, socket.SOCK_STREAM ) - sock.connect( ( 'localhost', self.port ) ) - sock.send( json.to_json_string( dict( transfer_job_ids=self.transfer_job_ids ) ) + '\n' ) - print sock.recv( 8192 ), - sock.close() - sys.exit() - -def do_transfer( transfer_job ): - """ - Actually initiate a transfer (used when the transfer manager calls itself - with the '--do' flag). - """ - protocol = transfer_job.params['url'].split( '://' )[0] - if protocol in ( 'http', 'https' ): - return do_http_transfer( transfer_job ) - else: - print >>sys.stderr, 'ERROR: Unsupported protocol: %s' % protocol - sys.exit( 1 ) - -def do_http_transfer( transfer_job ): - """ - "Plugin" for handling http(s) transfers. - """ - url = transfer_job.params['url'] - try: - f = urllib2.urlopen( url ) - except urllib2.URLError, e: - print json.to_json_string( dict( state=Transfer.states.ERROR, - info=str( e ) ) ) - return - size = f.info().getheader( 'Content-Length' ) - if size is not None: - size = int( size ) - chunksize = 1024 * 1024 - read = 0 - last = 0 - fh, fn = tempfile.mkstemp() - while True: - chunk = f.read( chunksize ) - if not chunk: - break - os.write( fh, chunk ) - if read == 0 and size is None: - print json.to_json_string( dict( state=Transfer.states.STARTED, - size=None ) ) #+ '\n' - elif read == 0: - print json.to_json_string( dict( state=Transfer.states.STARTED, - size=size ) ) #+ '\n' - read += chunksize - if size is not None and read < size: - percent = int( float( read ) / size * 100 ) - if percent != last: - print json.to_json_string( dict( state=Transfer.states.PROGRESS, - read=read, - percent='%s%%' % percent ) ) #+ '\n' - last = percent - elif size is None: - print json.to_json_string( dict( state=Transfer.states.PROGRESS, - read=read ) ) - os.close( fh ) - print json.to_json_string( dict( state=Transfer.states.DONE, path=fn ) ) #+ '\n' - -def request_state( state_transfer_job_ids, port ): - sock = socket.socket( socket.AF_INET, socket.SOCK_STREAM ) - sock.connect( ( 'localhost', port ) ) - sock.send( json.to_json_string( dict( state_transfer_job_ids=state_transfer_job_ids ) ) + '\n' ) - print sock.recv( 1024 ), - sock.close() - sys.exit() - -if __name__ == '__main__': - arg_handler = ArgHandler() - arg_handler.parse() - app = GalaxyApp( config_file=arg_handler.opts.config ) - if arg_handler.opts.initiate_transfer_job_id is not None: - do_transfer( app.get_transfer_job( arg_handler.opts.initiate_transfer_job_id ) ) - elif arg_handler.opts.state_transfer_job_ids: - request_state( arg_handler.opts.state_transfer_job_ids, int( app.config.get( 'app:main', 'transfer_manager_port' ) ) ) - elif arg_handler.opts.transfer_job_ids: - transfer_manager = TransferManager( app, arg_handler.opts.transfer_job_ids, arg_handler.opts.debug ) - else: - arg_handler.parser.print_usage( sys.stderr ) - sys.exit( 1 )