diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py index ffa6433fa68..f979d10424c 100644 --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -123,6 +123,10 @@ class Configuration( object ): self.enable_cloud_execution = string_as_bool( kwargs.get( 'enable_cloud_execution', 'True' ) ) else: self.enable_cloud_execution = string_as_bool( kwargs.get( 'enable_cloud_execution', 'False' ) ) + # Galaxy messaging (AMQP) configuration options + self.amqp = {} + for k, v in global_conf_parser.items("galaxy_amqp"): + self.amqp[k] = v def get( self, key, default ): return self.config_dict.get( key, default ) def get_bool( self, key, default ): diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index ae207176acb..a35cd86879e 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -18,6 +18,7 @@ from galaxy.web.form_builder import * import logging log = logging.getLogger( __name__ ) from sqlalchemy.orm import object_session +import pexpect datatypes_registry = galaxy.datatypes.registry.Registry() #Default Value Required for unit tests @@ -1455,7 +1456,9 @@ class RequestType( object ): class Sample( object ): transfer_status = Bunch( NOT_STARTED = 'Not started', - IN_PROGRESS = 'In progress', + IN_QUEUE = 'In queue', + TRANSFERRING = 'Transferring dataset', + ADD_TO_LIBRARY = 'Adding to data library', COMPLETE = 'Complete', ERROR = 'Error') def __init__(self, name=None, desc=None, request=None, form_values=None, @@ -1474,22 +1477,33 @@ class Sample( object ): return None def untransferred_dataset_files(self): count = 0 - for df, status in self.dataset_files: - if status == self.transfer_status.NOT_STARTED: + for df in self.dataset_files: + if df['status'] == self.transfer_status.NOT_STARTED: count = count + 1 return count def inprogress_dataset_files(self): count = 0 - for df, status in self.dataset_files: - if status == self.transfer_status.IN_PROGRESS: + for df in self.dataset_files: + if df['status'] not in [self.transfer_status.NOT_STARTED, self.transfer_status.COMPLETE]: count = count + 1 return count def transferred_dataset_files(self): count = 0 - for df, status in self.dataset_files: - if status == self.transfer_status.COMPLETE: + for df in self.dataset_files: + if df['status'] == self.transfer_status.COMPLETE: count = count + 1 return count + def dataset_size(self, filepath): + def print_ticks(d): + pass + datatx_info = self.request.type.datatx_info + cmd = 'ssh %s@%s "du -sh %s"' % ( datatx_info['username'], + datatx_info['host'], + filepath) + output = pexpect.run(cmd, events={'.ssword:*': datatx_info['password']+'\r\n', + pexpect.TIMEOUT:print_ticks}, + timeout=10) + return output.split('\t')[0] class SampleState( object ): def __init__(self, name=None, desc=None, request_type=None): diff --git a/lib/galaxy/web/controllers/requests_admin.py b/lib/galaxy/web/controllers/requests_admin.py index 0789d33801b..9919b909e96 100644 --- a/lib/galaxy/web/controllers/requests_admin.py +++ b/lib/galaxy/web/controllers/requests_admin.py @@ -12,6 +12,7 @@ from sqlalchemy.sql.expression import func, and_ from sqlalchemy.sql import select import pexpect import ConfigParser, threading, time +from amqplib import client_0_8 as amqp log = logging.getLogger( __name__ ) @@ -64,7 +65,6 @@ class RequestsGrid( grids.Grid ): .filter( self.event_class.table.c.id.in_(select(columns=[func.max(self.event_class.table.c.id)], from_obj=self.event_class.table, group_by=self.event_class.table.c.request_id))) - #print column_filter, q return q def get_accepted_filters( self ): """ Returns a list of accepted filters for this column. """ @@ -1509,8 +1509,11 @@ class RequestsAdmin( BaseController ): params = util.Params( kwd ) message = util.restore_text( params.get( 'message', '' ) ) status = params.get( 'status', 'done' ) - folder_path = util.restore_text( params.get( 'folder_path', '' ) ) + folder_path = util.restore_text( params.get( 'folder_path', + sample.request.type.datatx_info['data_dir'] ) ) files_list = util.listify( params.get( 'files_list', '' ) ) + if params.get( 'start_transfer_button', False ) == 'True': + return self.__start_datatx(trans, sample) if not folder_path: return trans.fill_template( '/admin/requests/get_data.mako', sample=sample, files=[], @@ -1544,32 +1547,43 @@ class RequestsAdmin( BaseController ): dataset_files=sample.dataset_files, folder_path=folder_path ) elif params.get( 'remove_dataset_button', False ): + # get the filenames from the remote host + files = self.__get_files(trans, sample, folder_path) dataset_index = int(params.get( 'dataset_index', 0 )) del sample.dataset_files[dataset_index] trans.sa_session.add( sample ) trans.sa_session.flush() return trans.fill_template( '/admin/requests/get_data.mako', - sample=sample, - dataset_files=sample.dataset_files) - elif params.get( 'start_transfer_button', False ): + sample=sample, files=files, + dataset_files=sample.dataset_files, + folder_path=folder_path) + elif params.get( 'select_files_button', False ): folder_files = [] if len(files_list): for f in files_list: + filepath = os.path.join(folder_path, f) if f[-1] == os.sep: # the selected item is a folder so transfer all the # folder contents - self.__get_files_in_dir(trans, sample, os.path.join(folder_path, f)) + self.__get_files_in_dir(trans, sample, filepath) else: - sample.dataset_files.append([os.path.join(folder_path, f), - sample.transfer_status.NOT_STARTED]) + sample.dataset_files.append(dict(filepath=filepath, + status=sample.transfer_status.NOT_STARTED, + name=filepath.split('/')[-1], + error_msg='', + size=sample.dataset_size(filepath))) trans.sa_session.add( sample ) trans.sa_session.flush() - return self.__start_datatx(trans, sample) return trans.response.send_redirect( web.url_for( controller='requests_admin', action='show_datatx_page', sample_id=trans.security.encode_id(sample.id), folder_path=folder_path)) + return trans.response.send_redirect( web.url_for( controller='requests_admin', + action='show_datatx_page', + sample_id=trans.security.encode_id(sample.id), + folder_path=folder_path)) + def __setup_datatx_user(self, trans, library, folder): ''' This method sets up the datatx user: @@ -1620,7 +1634,62 @@ class RequestsAdmin( BaseController ): trans.sa_session.add( dp ) trans.sa_session.flush() return datatx_user - + + def __send_message(self, trans, datatx_info, sample): + ''' + This method creates the xml message and sends it to the rabbitmq server + ''' + # first create the xml message based on the following template + xml = \ + ''' + %(DATA_HOST)s + %(DATA_USER)s + %(DATA_PASSWORD)s + %(SAMPLE_ID)s + %(LIBRARY_ID)s + %(FOLDER_ID)s + %(DATASETS)s + ''' + dataset_xml = \ + ''' + %(INDEX)s + %(NAME)s + %(FILE)s + ''' + datasets = '' + for index, dataset in enumerate(sample.dataset_files): + if dataset['status'] == sample.transfer_status.NOT_STARTED: + datasets = datasets + dataset_xml % dict(INDEX=str(index), + NAME=dataset['name'], + FILE=dataset['filepath']) + sample.dataset_files[index]['status'] = sample.transfer_status.IN_QUEUE + + trans.sa_session.add( sample ) + trans.sa_session.flush() + data = xml % dict(DATA_HOST=datatx_info['host'], + DATA_USER=datatx_info['username'], + DATA_PASSWORD=datatx_info['password'], + SAMPLE_ID=str(sample.id), + LIBRARY_ID=str(sample.library.id), + FOLDER_ID=str(sample.folder.id), + DATASETS=datasets) + # now send this message + conn = amqp.Connection(host=trans.app.config.amqp['host']+":"+trans.app.config.amqp['port'], + userid=trans.app.config.amqp['userid'], + password=trans.app.config.amqp['password'], + virtual_host=trans.app.config.amqp['virtual_host'], + insist=False) + chan = conn.channel() + msg = amqp.Message(data, + content_type='text/plain', + application_headers={'msg_type': 'data_transfer'}) + msg.properties["delivery_mode"] = 2 + chan.basic_publish(msg, + exchange=trans.app.config.amqp['exchange'], + routing_key=trans.app.config.amqp['routing_key']) + chan.close() + conn.close() + def __start_datatx(self, trans, sample): # data transfer user datatx_user = self.__setup_datatx_user(trans, sample.library, sample.folder) @@ -1635,6 +1704,11 @@ class RequestsAdmin( BaseController ): sample_id=trans.security.encode_id(sample.id), status='error', message=message)) + self.__send_message(trans, datatx_info, sample) + return trans.response.send_redirect( web.url_for( controller='requests_admin', + action='show_datatx_page', + sample_id=trans.security.encode_id(sample.id), + folder_path=datatx_info['data_dir'])) error_message = '' transfer_script = "scripts/galaxy_messaging/server/data_transfer.py" for index, dataset in enumerate(sample.dataset_files): @@ -1670,6 +1744,33 @@ class RequestsAdmin( BaseController ): action='show_datatx_page', sample_id=trans.security.encode_id(sample.id), folder_path=os.path.dirname(dfile))) + + @web.expose + @web.require_admin + def dataset_details( self, trans, **kwd ): + try: + sample = trans.sa_session.query( trans.app.model.Sample ).get( trans.security.decode_id(kwd['sample_id']) ) + except: + return trans.response.send_redirect( web.url_for( controller='requests_admin', + action='list', + status='error', + message="Invalid sample ID" ) ) + params = util.Params( kwd ) + message = util.restore_text( params.get( 'message', '' ) ) + status = params.get( 'status', 'done' ) + dataset_index = int( params.get( 'dataset_index', '' ) ) + if params.get('save', '') == 'Save': + sample.dataset_files[dataset_index]['name'] = util.restore_text( params.get( 'name', + sample.dataset_files[dataset_index]['name'] ) ) + trans.sa_session.add( sample ) + trans.sa_session.flush() + status = 'done' + message = 'Saved the changes made to the dataset.' + return trans.fill_template( '/admin/requests/dataset.mako', + sample=sample, + dataset_index=dataset_index, + message=message, + status=status) ## #### Request Type Stuff ################################################### ## diff --git a/lib/galaxy/web/framework/__init__.py b/lib/galaxy/web/framework/__init__.py index df9ca0358b8..5107eb9cfb7 100644 --- a/lib/galaxy/web/framework/__init__.py +++ b/lib/galaxy/web/framework/__init__.py @@ -32,6 +32,7 @@ pkg_resources.require( "SQLAlchemy >= 0.4" ) from sqlalchemy import and_ pkg_resources.require( "pexpect" ) +pkg_resources.require( "amqplib" ) import logging log = logging.getLogger( __name__ ) diff --git a/run_galaxy_listener.sh b/run_galaxy_listener.sh index 4493f8e0066..a8e272080da 100644 --- a/run_galaxy_listener.sh +++ b/run_galaxy_listener.sh @@ -1,4 +1,4 @@ #!/bin/sh cd `dirname $0` -python scripts/galaxy_messaging/server/amqp_consumer.py universe_wsgi.ini >> galaxy_listener.log 2>&1 \ No newline at end of file +python scripts/galaxy_messaging/server/amqp_consumer.py universe_wsgi.ini 2>&1 \ No newline at end of file diff --git a/scripts/galaxy_messaging/client/amqp_publisher.py b/scripts/galaxy_messaging/client/amqp_publisher.py index ae22fb64c6f..05074053019 100644 --- a/scripts/galaxy_messaging/client/amqp_publisher.py +++ b/scripts/galaxy_messaging/client/amqp_publisher.py @@ -35,7 +35,9 @@ def handle_scan(states, amqp_config, barcode): virtual_host=amqp_config['virtual_host'], insist=False) chan = conn.channel() - msg = amqp.Message(data) + msg = amqp.Message(data, + content_type='text/plain', + application_headers={'msg_type': 'sample_state_update'}) msg.properties["delivery_mode"] = 2 chan.basic_publish(msg, exchange=amqp_config['exchange'], diff --git a/scripts/galaxy_messaging/server/amqp_consumer.py b/scripts/galaxy_messaging/server/amqp_consumer.py index 7de500ff31b..8db4ca0a7bb 100644 --- a/scripts/galaxy_messaging/server/amqp_consumer.py +++ b/scripts/galaxy_messaging/server/amqp_consumer.py @@ -13,6 +13,7 @@ import ConfigParser import sys, os import optparse import xml.dom.minidom +import subprocess from galaxydb_interface import GalaxyDbInterface assert sys.version_info[:2] >= ( 2, 4 ) @@ -27,8 +28,13 @@ pkg_resources.require( "amqplib" ) from amqplib import client_0_8 as amqp import logging -logging.basicConfig(level=logging.DEBUG) -log = logging.getLogger( 'GalaxyAMQP' ) +log = logging.getLogger("GalaxyAMQP") +log.setLevel(logging.DEBUG) +fh = logging.FileHandler("galaxy_listener.log") +fh.setLevel(logging.DEBUG) +formatter = logging.Formatter("%(asctime)s - %(name)s - %(message)s") +fh.setFormatter(formatter) +log.addHandler(fh) global dbconnstr @@ -43,19 +49,47 @@ def get_value(dom, tag_name): rc = rc + node.data return rc +def get_value_index(dom, tag_name, index): + ''' + This method extracts the tag value from the xml message + ''' + try: + nodelist = dom.getElementsByTagName(tag_name)[index].childNodes + except: + return None + rc = "" + for node in nodelist: + if node.nodeType == node.TEXT_NODE: + rc = rc + node.data + return rc + def recv_callback(msg): - dom = xml.dom.minidom.parseString(msg.body) - barcode = get_value(dom, 'barcode') - state = get_value(dom, 'state') - log.debug('Barcode: '+barcode) - log.debug('State: '+state) - # update the galaxy db - galaxy = GalaxyDbInterface(dbconnstr) - sample_id = galaxy.get_sample_id(field_name='bar_code', value=barcode) - if sample_id == -1: - log.debug('Invalid barcode.') - return - galaxy.change_state(sample_id, state) + # check the meesage type. + msg_type = msg.properties['application_headers'].get('msg_type') + log.debug('\nMESSAGE RECVD: '+str(msg_type)) + if msg_type == 'data_transfer': + log.debug('DATA TRANSFER') + # fork a new process to transfer datasets + transfer_script = "scripts/galaxy_messaging/server/data_transfer.py" + cmd = ( "python", + transfer_script, + msg.body ) + pid = subprocess.Popen(cmd).pid + log.debug('Started process (%i): %s' % (pid, str(cmd))) + elif msg_type == 'sample_state_update': + log.debug('SAMPLE STATE UPDATE') + dom = xml.dom.minidom.parseString(msg.body) + barcode = get_value(dom, 'barcode') + state = get_value(dom, 'state') + log.debug('Barcode: '+barcode) + log.debug('State: '+state) + # update the galaxy db + galaxy = GalaxyDbInterface(dbconnstr) + sample_id = galaxy.get_sample_id(field_name='bar_code', value=barcode) + if sample_id == -1: + log.debug('Invalid barcode.') + return + galaxy.change_state(sample_id, state) def main(): if len(sys.argv) < 2: @@ -66,8 +100,8 @@ def main(): global dbconnstr dbconnstr = config.get("app:main", "database_connection") amqp_config = {} - for option in config.options("galaxy:amqp"): - amqp_config[option] = config.get("galaxy:amqp", option) + for option in config.options("galaxy_amqp"): + amqp_config[option] = config.get("galaxy_amqp", option) log.debug(str(amqp_config)) conn = amqp.Connection(host=amqp_config['host']+":"+amqp_config['port'], userid=amqp_config['userid'], diff --git a/scripts/galaxy_messaging/server/data_transfer.py b/scripts/galaxy_messaging/server/data_transfer.py index 9603ce9428e..7422f282364 100755 --- a/scripts/galaxy_messaging/server/data_transfer.py +++ b/scripts/galaxy_messaging/server/data_transfer.py @@ -8,28 +8,36 @@ transfer process using the user interface. Usage: -python data_transfer.py - - - - - - - +python data_transfer.py + + """ import ConfigParser import sys, os, time, traceback import optparse import urllib,urllib2, cookielib, shutil import logging, time +import xml.dom.minidom + +sp = sys.path[0] + from galaxydb_interface import GalaxyDbInterface +assert sys.version_info[:2] >= ( 2, 4 ) +new_path = [ sp ] +new_path.extend( sys.path ) +sys.path = new_path + +from galaxyweb_interface import GalaxyWebInterface + assert sys.version_info[:2] >= ( 2, 4 ) new_path = [ os.path.join( os.getcwd(), "lib" ) ] new_path.extend( sys.path[1:] ) # remove scripts/ from the path sys.path = new_path + from galaxy.util.json import from_json_string, to_json_string +from galaxy.model import Sample from galaxy import eggs import pkg_resources pkg_resources.require( "pexpect" ) @@ -38,28 +46,39 @@ import pexpect pkg_resources.require( "simplejson" ) import simplejson -logging.basicConfig(filename=sys.stderr, level=logging.DEBUG, - format="%(asctime)s [%(levelname)s] %(message)s") - -class DataTransferException(Exception): - def __init__(self, value): - self.msg = value - def __str__(self): - return repr(self.msg) +log = logging.getLogger("datatx_"+str(os.getpid())) +log.setLevel(logging.DEBUG) +fh = logging.FileHandler("data_transfer.log") +fh.setLevel(logging.DEBUG) +formatter = logging.Formatter("%(asctime)s - %(name)s - %(message)s") +fh.setFormatter(formatter) +log.addHandler(fh) class DataTransfer(object): - def __init__(self, host, username, password, remote_file, sample_id, - dataset_index, library_id, folder_id): - self.host = host - self.username = username - self.password = password - self.remote_file = remote_file - self.sample_id = sample_id - self.dataset_index = dataset_index - self.library_id = library_id - self.folder_id = folder_id + def __init__(self, msg): + log.info(msg) + self.dom = xml.dom.minidom.parseString(msg) + self.host = self.get_value(self.dom, 'data_host') + self.username = self.get_value(self.dom, 'data_user') + self.password = self.get_value(self.dom, 'data_password') + self.sample_id = self.get_value(self.dom, 'sample_id') + self.library_id = self.get_value(self.dom, 'library_id') + self.folder_id = self.get_value(self.dom, 'folder_id') + self.dataset_files = [] + count=0 + while True: + index = self.get_value_index(self.dom, 'index', count) + file = self.get_value_index(self.dom, 'file', count) + name = self.get_value_index(self.dom, 'name', count) + if file: + self.dataset_files.append(dict(name=name, + index=int(index), + file=file)) + else: + break + count=count+1 try: # Retrieve the upload user login information from the config file config = ConfigParser.ConfigParser() @@ -75,11 +94,13 @@ class DataTransfer(object): os.mkdir(self.server_dir) if not os.path.exists(self.server_dir): raise Exception + # connect to db + self.galaxydb = GalaxyDbInterface(self.database_connection) except: - logging.error(traceback.format_exc()) - logging.error('FATAL ERROR') + log.error(traceback.format_exc()) + log.error('FATAL ERROR') if self.database_connection: - self.update_status('Error') + self.error_and_exit('Error') sys.exit(1) def start(self): @@ -88,13 +109,13 @@ class DataTransfer(object): to the data library & finally updates the data transfer status in the db ''' # datatx - self.transfer_file() + self.transfer_files() # add the dataset to the given library self.add_to_library() # update the data transfer status in the db - self.update_status('Complete') + self.update_status(Sample.transfer_status.COMPLETE) # cleanup - self.cleanup() + #self.cleanup() sys.exit(0) def cleanup(self): @@ -114,34 +135,39 @@ class DataTransfer(object): This method is called any exception is raised. This prints the traceback and terminates this script ''' - logging.error(traceback.format_exc()) - logging.error('FATAL ERROR.'+msg) - self.update_status('Error.'+msg) + log.error(traceback.format_exc()) + log.error('FATAL ERROR.'+msg) + self.update_status('Error.', 'All', msg) sys.exit(1) - def transfer_file(self): + def transfer_files(self): ''' This method executes a scp process using pexpect library to transfer the dataset file from the remote sequencer to the Galaxy server ''' def print_ticks(d): pass - try: - cmd = "scp %s@%s:%s %s" % ( self.username, - self.host, - self.remote_file, - self.server_dir) - logging.debug(cmd) - output = pexpect.run(cmd, events={'.ssword:*': self.password+'\r\n', - pexpect.TIMEOUT:print_ticks}, - timeout=10) - logging.debug(output) - if not os.path.exists(os.path.join(self.server_dir, os.path.basename(self.remote_file))): - raise DataTransferException('Could not find the local file after transfer (%s)' % os.path.join(self.server_dir, os.path.basename(self.remote_file))) - except DataTransferException, (e): - self.error_and_exit(e.msg) - except: - self.error_and_exit() + for i, df in enumerate(self.dataset_files): + self.update_status(Sample.transfer_status.TRANSFERRING, df['index']) + try: + cmd = "scp %s@%s:%s %s/%s" % ( self.username, + self.host, + df['file'], + self.server_dir, + df['name']) + log.debug(cmd) + output = pexpect.run(cmd, events={'.ssword:*': self.password+'\r\n', + pexpect.TIMEOUT:print_ticks}, + timeout=10) + log.debug(output) + path = os.path.join(self.server_dir, os.path.basename(df['file'])) + if not os.path.exists(path): + msg = 'Could not find the local file after transfer (%s)' % path + log.error(msg) + raise Exception(msg) + except Exception, e: + msg = traceback.format_exc() + self.update_status('Error', df['index'], msg) def add_to_library(self): @@ -149,73 +175,72 @@ class DataTransfer(object): This method adds the dataset file to the target data library & folder by opening the corresponding url in Galaxy server running. ''' - try: - logging.debug('Adding %s to library...' % os.path.basename(self.remote_file)) - # create url - base_url = "http://%s:%s" % (self.server_host, self.server_port) - # login - url = "%s/user/login?email=%s&password=%s" % (base_url, self.datatx_email, self.datatx_password) - cj = cookielib.CookieJar() - opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj)) - f = opener.open(url) - if f.read().find("ogged in as "+self.datatx_email) == -1: - # if the user doesnt exist, create the user - url = "%s/user/create?email=%s&username=%s&password=%s&confirm=%s&create_user_button=Submit" % ( base_url, self.datatx_email, self.datatx_email, self.datatx_password, self.datatx_password ) - f = opener.open(url) - if f.read().find("ogged in as "+self.datatx_email) == -1: - raise DataTransferException("The "+self.datatx_email+" user could not login to Galaxy") - # after login, add dataset to the library - params = urllib.urlencode(dict( cntrller='library_admin', - tool_id='upload1', - tool_state='None', - library_id=self.library_id, - folder_id=self.folder_id, - upload_option='upload_directory', - file_type='auto', - server_dir=os.path.basename(self.server_dir), - dbkey='', - runtool_btn='Upload to library')) - #url = "http://localhost:8080/library_common/upload_library_dataset?cntrller=library_admin&tool_id=upload1&tool_state=None&library_id=adb5f5c93f827949&folder_id=adb5f5c93f827949&upload_option=upload_directory&file_type=auto&server_dir=003&dbkey=%3F&message=&runtool_btn=Upload+to+library" - #url = base_url+"/library_common/upload_library_dataset?library_id=adb5f5c93f827949&tool_id=upload1&file_type=auto&server_dir=datatx_22858&dbkey=%3F&upload_option=upload_directory&folder_id=529fd61ab1c6cc36&cntrller=library_admin&tool_state=None&runtool_btn=Upload+to+library" - url = base_url+"/library_common/upload_library_dataset" - logging.debug(url) - logging.debug(params) - f = opener.open(url, params) - if f.read().find("Data Library") == -1: - raise DataTransferException("Dataset could not be uploaded to the data library") - # finally logout - f = opener.open(base_url+'/user/logout') - if f.read().find("You have been logged out.") == -1: - raise DataTransferException("The "+self.datatx_email+" user could not logout of Galaxy") - except DataTransferException, (e): - self.error_and_exit(e.msg) - except: - self.error_and_exit() + self.update_status(Sample.transfer_status.ADD_TO_LIBRARY) + galaxyweb = GalaxyWebInterface(self.server_host, self.server_port, + self.datatx_email, self.datatx_password) + galaxyweb.add_to_library(self.server_dir, self.library_id, self.folder_id) + galaxyweb.logout() - def update_status(self, status): + def update_status(self, status, dataset_index='All', msg=''): ''' Update the data transfer status for this dataset in the database ''' try: - galaxy = GalaxyDbInterface(self.database_connection) - df = from_json_string(galaxy.get_sample_dataset_files(self.sample_id)) - logging.debug(df) - df[self.dataset_index][1] = status - galaxy.set_sample_dataset_files(self.sample_id, to_json_string(df)) - logging.debug("######################\n"+str(from_json_string(galaxy.get_sample_dataset_files(self.sample_id))[self.dataset_index])) + log.debug('Setting status "%s" for sample "%s"' % ( status, str(dataset_index) ) ) + df = from_json_string(self.galaxydb.get_sample_dataset_files(self.sample_id)) + if dataset_index == 'All': + for dataset in self.dataset_files: + df[dataset['index']]['status'] = status + if status == 'Error': + df[dataset['index']]['error_msg'] = msg + else: + df[dataset['index']]['error_msg'] = '' + + else: + df[dataset_index]['status'] = status + if status == 'Error': + df[dataset_index]['error_msg'] = msg + else: + df[dataset_index]['error_msg'] = '' + + self.galaxydb.set_sample_dataset_files(self.sample_id, to_json_string(df)) + log.debug('done.') except: - logging.error(traceback.format_exc()) - logging.error('FATAL ERROR') + log.error(traceback.format_exc()) + log.error('FATAL ERROR') sys.exit(1) + + def get_value(self, dom, tag_name): + ''' + This method extracts the tag value from the xml message + ''' + nodelist = dom.getElementsByTagName(tag_name)[0].childNodes + rc = "" + for node in nodelist: + if node.nodeType == node.TEXT_NODE: + rc = rc + node.data + return rc + + def get_value_index(self, dom, tag_name, index): + ''' + This method extracts the tag value from the xml message + ''' + try: + nodelist = dom.getElementsByTagName(tag_name)[index].childNodes + except: + return None + rc = "" + for node in nodelist: + if node.nodeType == node.TEXT_NODE: + rc = rc + node.data + return rc if __name__ == '__main__': - logging.info('STARTING %i %s' % (os.getpid(), str(sys.argv))) - logging.info('daemonized %i' % os.getpid()) + log.info('STARTING %i %s' % (os.getpid(), str(sys.argv))) # # Start the daemon - # - dt = DataTransfer(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4], - int(sys.argv[5]), int(sys.argv[6]), sys.argv[7], sys.argv[8]) + # + dt = DataTransfer(sys.argv[1]) dt.start() sys.exit(0) diff --git a/scripts/galaxy_messaging/server/galaxydb_interface.py b/scripts/galaxy_messaging/server/galaxydb_interface.py index 70ed9820816..3f0107ac458 100644 --- a/scripts/galaxy_messaging/server/galaxydb_interface.py +++ b/scripts/galaxy_messaging/server/galaxydb_interface.py @@ -20,8 +20,8 @@ pkg_resources.require( "SQLAlchemy >= 0.4" ) from sqlalchemy import * from sqlalchemy.orm import sessionmaker -logging.basicConfig(level=logging.DEBUG) -log = logging.getLogger( 'GalaxyDbInterface' ) +#logging.basicConfig(level=logging.DEBUG) +#log = logging.getLogger( 'GalaxyDbInterface' ) class GalaxyDbInterface(object): @@ -53,9 +53,8 @@ class GalaxyDbInterface(object): x = result.fetchone() if x: sample_id = x[0] - log.debug('Sample ID: %i' % sample_id) + #log.debug('Sample ID: %i' % sample_id) return sample_id - log.warning('This sample %s %s does not belong to any sample in the database.' % (field_name, value)) return -1 def current_state(self, sample_id): @@ -74,16 +73,16 @@ class GalaxyDbInterface(object): subsubquery = select(columns=[self.sample_table.c.request_id], whereclause=self.sample_table.c.id==sample_id) self.request_id = subsubquery.execute().fetchall()[0][0] - log.debug('REQUESTID: %i' % self.request_id) + #log.debug('REQUESTID: %i' % self.request_id) subquery = select(columns=[self.request_table.c.request_type_id], whereclause=self.request_table.c.id==self.request_id) request_type_id = subquery.execute().fetchall()[0][0] - log.debug('REQUESTTYPEID: %i' % request_type_id) + #log.debug('REQUESTTYPEID: %i' % request_type_id) query = select(columns=[self.state_table.c.id, self.state_table.c.name], whereclause=self.state_table.c.request_type_id==request_type_id, order_by=self.state_table.c.id.asc()) states = query.execute().fetchall() - log.debug('POSSIBLESTATES: '+ str(states)) + #log.debug('POSSIBLESTATES: '+ str(states)) return states def change_state(self, sample_id, new_state=None): @@ -100,7 +99,7 @@ class GalaxyDbInterface(object): new_state_id = state_id if new_state_id == -1: return - log.debug('Updating sample_id %i state to %s' % (sample_id, new_state)) + #log.debug('Updating sample_id %i state to %s' % (sample_id, new_state)) i = self.event_table.insert() i.execute(update_time=datetime.utcnow(), create_time=datetime.utcnow(), @@ -120,7 +119,7 @@ class GalaxyDbInterface(object): break if request_complete: request_state = 'Complete' - log.debug('Updating request_id %i state to "%s"' % (self.request_id, request_state)) + #log.debug('Updating request_id %i state to "%s"' % (self.request_id, request_state)) i = self.request_event_table.insert() i.execute(update_time=datetime.utcnow(), create_time=datetime.utcnow(), diff --git a/scripts/galaxy_messaging/server/galaxyweb_interface.py b/scripts/galaxy_messaging/server/galaxyweb_interface.py new file mode 100644 index 00000000000..1b4a3795fc0 --- /dev/null +++ b/scripts/galaxy_messaging/server/galaxyweb_interface.py @@ -0,0 +1,132 @@ +import ConfigParser +import sys, os +import serial +import array +import time +import optparse,array +import shutil, traceback +import urllib,urllib2, cookielib + +assert sys.version_info[:2] >= ( 2, 4 ) +new_path = [ os.path.join( os.getcwd(), "lib" ) ] +new_path.extend( sys.path[1:] ) # remove scripts/ from the path +sys.path = new_path + +from galaxy import eggs +import pkg_resources + +import pkg_resources +pkg_resources.require( "pycrypto" ) + +from Crypto.Cipher import Blowfish +from Crypto.Util.randpool import RandomPool +from Crypto.Util import number + + +class GalaxyWebInterface(object): + def __init__(self, server_host, server_port, datatx_email, datatx_password): + self.server_host = server_host#config.get("main", "server_host") + self.server_port = server_port#config.get("main", "server_port") + self.datatx_email = datatx_email#config.get("main", "datatx_email") + self.datatx_password = datatx_password#config.get("main", "datatx_password") + try: + # create url + self.base_url = "http://%s:%s" % (self.server_host, self.server_port) + # login + url = "%s/user/login?email=%s&password=%s&login_button=Login" % (self.base_url, self.datatx_email, self.datatx_password) + cj = cookielib.CookieJar() + self.opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj)) + #print url + f = self.opener.open(url) + if f.read().find("ogged in as "+self.datatx_email) == -1: + # if the user doesnt exist, create the user + url = "%s/user/create?email=%s&username=%s&password=%s&confirm=%s&create_user_button=Submit" % ( self.base_url, self.datatx_email, self.datatx_email, self.datatx_password, self.datatx_password ) + f = self.opener.open(url) + if f.read().find("ogged in as "+self.datatx_email) == -1: + raise "The "+self.datatx_email+" user could not login to Galaxy" + except: + print traceback.format_exc() + sys.exit(1) + + def add_to_library(self, server_dir, library_id, folder_id, dbkey=''): + ''' + This method adds the dataset file to the target data library & folder + by opening the corresponding url in Galaxy server running. + ''' + try: + params = urllib.urlencode(dict( cntrller='library_admin', + tool_id='upload1', + tool_state='None', + library_id=self.encode_id(library_id), + folder_id=self.encode_id(folder_id), + upload_option='upload_directory', + file_type='auto', + server_dir=os.path.basename(server_dir), + dbkey=dbkey, + show_dataset_id='True', + runtool_btn='Upload to library')) + #url = "http://localhost:8080/library_common/upload_library_dataset?cntrller=library_admin&tool_id=upload1&tool_state=None&library_id=adb5f5c93f827949&folder_id=adb5f5c93f827949&upload_option=upload_directory&file_type=auto&server_dir=003&dbkey=%3F&message=&runtool_btn=Upload+to+library" + #url = base_url+"/library_common/upload_library_dataset?library_id=adb5f5c93f827949&tool_id=upload1&file_type=auto&server_dir=datatx_22858&dbkey=%3F&upload_option=upload_directory&folder_id=529fd61ab1c6cc36&cntrller=library_admin&tool_state=None&runtool_btn=Upload+to+library" + url = self.base_url+"/library_common/upload_library_dataset" + #print url + #print params + f = self.opener.open(url, params) + if f.read().find("Data Library") == -1: + raise "Dataset could not be uploaded to the data library" + except: + print traceback.format_exc() + sys.exit(1) + + def import_to_history(self, ldda_id, library_id, folder_id): + try: + params = urllib.urlencode(dict( cntrller='library_admin', + show_deleted='False', + library_id=self.encode_id(library_id), + folder_id=self.encode_id(folder_id), + ldda_ids=self.encode_id(ldda_id), + do_action='import_to_history', + use_panels='False')) + #url = "http://lion.bx.psu.edu:8080/library_common/act_on_multiple_datasets?library_id=adb5f5c93f827949&show_deleted=False&ldda_ids=adb5f5c93f827949&cntrller=library_admin&do_action=import_to_history&use_panels=False" + #url = base_url+"/library_common/upload_library_dataset?library_id=adb5f5c93f827949&tool_id=upload1&file_type=auto&server_dir=datatx_22858&dbkey=%3F&upload_option=upload_directory&folder_id=529fd61ab1c6cc36&cntrller=library_admin&tool_state=None&runtool_btn=Upload+to+library" + url = self.base_url+"/library_common/act_on_multiple_datasets" + #print url + #print params + f = self.opener.open(url, params) + x = f.read() + if x.find("1 dataset(s) have been imported into your history.") == -1: + #print x + raise Exception("Dataset could not be imported into history") + except: + print traceback.format_exc() + sys.exit(1) + + + def run_workflow(self, workflow_id, hid, workflow_step): + input = str(workflow_step)+'|input' + try: + params = urllib.urlencode({'id':self.encode_id(workflow_id), + 'run_workflow': 'Run workflow', + input: hid}) + url = self.base_url+"/workflow/run" + #print url+'?'+params + f = self.opener.open(url, params) +# if f.read().find("1 dataset(s) have been imported into your history.") == -1: +# raise Exception("Error in running the workflow") + except: + print traceback.format_exc() + sys.exit(1) + + + def logout(self): + # finally logout + f = self.opener.open(self.base_url+'/user/logout') + + def encode_id(self, obj_id ): + id_secret = 'changethisinproductiontoo' + id_cipher = Blowfish.new( id_secret ) + # Convert to string + s = str( obj_id ) + # Pad to a multiple of 8 with leading "!" + s = ( "!" * ( 8 - len(s) % 8 ) ) + s + # Encrypt + return id_cipher.encrypt( s ).encode( 'hex' ) diff --git a/templates/admin/requests/dataset.mako b/templates/admin/requests/dataset.mako new file mode 100644 index 00000000000..08a53f6788a --- /dev/null +++ b/templates/admin/requests/dataset.mako @@ -0,0 +1,71 @@ +<%inherit file="/base.mako"/> +<%namespace file="/message.mako" import="render_msg" /> + + +%if message: + ${render_msg( message, status )} +%endif + +
+
+ + + +
+
Dataset details
+
+
+ <% + dataset = sample.dataset_files[dataset_index] + %> +
+ +
+ %if dataset['status'] in [sample.transfer_status.IN_QUEUE, sample.transfer_status.NOT_STARTED]: + + %else: + ${dataset['name']} + %endif + +
+
+
+
+ +
+ ${dataset['filepath']} + ## +
+
+
+
+ +
+ ${dataset.get('size', 'Unknown')} +
+
+
+
+ +
+ ${dataset['status']} +
+ %if dataset['status'] == sample.transfer_status.ERROR: + ${dataset['error_msg']} + %endif +
+
+
+ %if dataset['status'] in [sample.transfer_status.IN_QUEUE, sample.transfer_status.NOT_STARTED]: +
+ +
+ %endif +
+
+
\ No newline at end of file diff --git a/templates/admin/requests/get_data.mako b/templates/admin/requests/get_data.mako index 234e76622ec..9064bd3d8df 100644 --- a/templates/admin/requests/get_data.mako +++ b/templates/admin/requests/get_data.mako @@ -53,29 +53,44 @@ $(document).ready(function(){
%if len(dataset_files): ##
+
+

Sample Dataset(s)

+ %if sample.untransferred_dataset_files():
-

Sample Dataset(s)

-
- - - - - - - - - - %for dataset_index, dataset_file in enumerate(dataset_files): - ${sample_dataset_files( dataset_index, dataset_file[0], dataset_file[1] )} - %endfor - -
Dataset FileTransfer Status
-
-
+ +
+ %endif +
+ + + + + + + + + + %for dataset_index, dataset_file in enumerate(dataset_files): + ${sample_dataset_files( dataset_index, dataset_file['name'], dataset_file['status'] )} + %endfor + +
Dataset FileTransfer Status
+
+
+ ## ## + + +

%endif + ##
@@ -102,24 +117,24 @@ $(document).ready(function(){ navigate away from this page. Once the transfer is complete the dataset(s) will show up on this page.
- +
-<%def name="sample_dataset_files( dataset_index, dataset_file, status )"> +<%def name="sample_dataset_files( dataset_index, dataset_name, status )"> -## -
${dataset_file.split('/')[-1]}
-
- ${dataset_file} -
+ +##
${dataset_file.split('/')[-1]}
+##
+## ${dataset_file} +##
- %if status == sample.transfer_status.IN_PROGRESS: + %if status not in [sample.transfer_status.NOT_STARTED, sample.transfer_status.COMPLETE]: ${status} %else: ${status} diff --git a/universe_wsgi.ini.sample b/universe_wsgi.ini.sample index 6bc006d5de3..c3c1365b189 100644 --- a/universe_wsgi.ini.sample +++ b/universe_wsgi.ini.sample @@ -287,7 +287,7 @@ upload1 = local:/// # to be set up with a user account and other parameters listed below. The 'host' # and 'port' fields should point to where the RabbitMQ server is running. -#[galaxy:amqp] +[galaxy_amqp] #host = 127.0.0.1 #port = 5672 #userid = galaxy