mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-28 02:58:18 +08:00
Completed port to SQLalchemy 0.5. Also, fixed couple of bugs, cleaned up cloud provider code, tested w/ EC2.
This commit is contained in:
@@ -109,15 +109,15 @@ class EC2CloudProvider( object ):
|
||||
return
|
||||
try:
|
||||
if uci_state==uci_states.NEW:
|
||||
self.createUCI( uci_wrapper )
|
||||
self.create_uci( uci_wrapper )
|
||||
elif uci_state==uci_states.DELETING:
|
||||
self.deleteUCI( uci_wrapper )
|
||||
self.delete_uci( uci_wrapper )
|
||||
elif uci_state==uci_states.SUBMITTED:
|
||||
self.startUCI( uci_wrapper )
|
||||
self.start_uci( uci_wrapper )
|
||||
elif uci_state==uci_states.SHUTTING_DOWN:
|
||||
self.stopUCI( uci_wrapper )
|
||||
self.stop_uci( uci_wrapper )
|
||||
elif uci_state==uci_states.SNAPSHOT:
|
||||
self.snapshotUCI( uci_wrapper )
|
||||
self.snapshot_uci( uci_wrapper )
|
||||
except:
|
||||
log.exception( "Uncaught exception executing cloud request." )
|
||||
cnt += 1
|
||||
@@ -223,7 +223,7 @@ class EC2CloudProvider( object ):
|
||||
uci_wrapper.set_error( err+". Contact site administrator to ensure needed machine image is registered.", True )
|
||||
return None
|
||||
|
||||
def createUCI( self, uci_wrapper ):
|
||||
def create_uci( self, uci_wrapper ):
|
||||
"""
|
||||
Creates User Configured Instance (UCI). Essentially, creates storage volume on cloud provider
|
||||
and registers relevant information in Galaxy database.
|
||||
@@ -276,7 +276,7 @@ class EC2CloudProvider( object ):
|
||||
uci_wrapper.set_store_status( vol.id, uci_states.ERROR )
|
||||
uci_wrapper.set_error( err, True )
|
||||
|
||||
def deleteUCI( self, uci_wrapper ):
|
||||
def delete_uci( self, uci_wrapper ):
|
||||
"""
|
||||
Deletes UCI. NOTE that this implies deletion of any and all data associated
|
||||
with this UCI from the cloud. All data will be deleted.
|
||||
@@ -315,7 +315,7 @@ class EC2CloudProvider( object ):
|
||||
log.error( err )
|
||||
uci_wrapper.set_error( err, True )
|
||||
|
||||
def snapshotUCI( self, uci_wrapper ):
|
||||
def snapshot_uci( self, uci_wrapper ):
|
||||
"""
|
||||
Creates snapshot of all storage volumes associated with this UCI.
|
||||
"""
|
||||
@@ -346,11 +346,11 @@ class EC2CloudProvider( object ):
|
||||
|
||||
uci_wrapper.change_state( uci_state=uci_states.AVAILABLE )
|
||||
|
||||
def addStorageToUCI( self, name ):
|
||||
def add_storage_to_uci( self, name ):
|
||||
""" Adds more storage to specified UCI
|
||||
TODO"""
|
||||
|
||||
def dummyStartUCI( self, uci_wrapper ):
|
||||
def dummy_start_uci( self, uci_wrapper ):
|
||||
|
||||
uci = uci_wrapper.get_uci()
|
||||
log.debug( "Would be starting instance '%s'" % uci.name )
|
||||
@@ -359,7 +359,7 @@ class EC2CloudProvider( object ):
|
||||
# time.sleep(20)
|
||||
# log.debug( "Woke up! (%s)" % uci.name )
|
||||
|
||||
def startUCI( self, uci_wrapper ):
|
||||
def start_uci( self, uci_wrapper ):
|
||||
"""
|
||||
Starts instance(s) of given UCI on the cloud.
|
||||
"""
|
||||
@@ -456,7 +456,7 @@ class EC2CloudProvider( object ):
|
||||
else:
|
||||
log.error( "UCI '%s' is in 'error' state, starting instance was aborted." % uci_wrapper.get_name() )
|
||||
|
||||
def stopUCI( self, uci_wrapper):
|
||||
def stop_uci( self, uci_wrapper):
|
||||
"""
|
||||
Stops all of cloud instances associated with given UCI.
|
||||
"""
|
||||
@@ -467,7 +467,7 @@ class EC2CloudProvider( object ):
|
||||
# Process list of instances and remove any references to empty instance id's
|
||||
for i in il:
|
||||
if i is None:
|
||||
l.remove( i )
|
||||
il.remove( i )
|
||||
log.debug( 'List of instances being terminated: %s' % il )
|
||||
rl = conn.get_all_instances( il ) # Reservation list associated with given instances
|
||||
|
||||
@@ -554,7 +554,7 @@ class EC2CloudProvider( object ):
|
||||
for inst in instances:
|
||||
if self.type == inst.uci.credentials.provider.type:
|
||||
log.debug( "[%s] Running general status update on instance '%s'" % ( inst.uci.credentials.provider.type, inst.instance_id ) )
|
||||
self.updateInstance( inst )
|
||||
self.update_instance( inst )
|
||||
|
||||
# Update storage volume(s)
|
||||
stores = self.sa_session.query( model.CloudStore ) \
|
||||
@@ -565,7 +565,7 @@ class EC2CloudProvider( object ):
|
||||
for store in stores:
|
||||
if self.type == store.uci.credentials.provider.type: # and store.volume_id != None:
|
||||
log.debug( "[%s] Running general status update on store with local database ID: '%s'" % ( store.uci.credentials.provider.type, store.id ) )
|
||||
self.updateStore( store )
|
||||
self.update_store( store )
|
||||
# else:
|
||||
# log.error( "[%s] There exists an entry for UCI (%s) storage volume without an ID. Storage volume might have been created with "
|
||||
# "cloud provider though. Manual check is recommended." % ( store.uci.credentials.provider.type, store.uci.name ) )
|
||||
@@ -579,7 +579,7 @@ class EC2CloudProvider( object ):
|
||||
|
||||
# Update pending snapshots or delete ones marked for deletion
|
||||
snapshots = self.sa_session.query( model.CloudSnapshot ) \
|
||||
.filter_by( status=snapshot_status.PENDING, status=snapshot_status.DELETE ) \
|
||||
.filter( or_( model.CloudSnapshot.table.c.status == snapshot_status.PENDING, model.CloudSnapshot.table.c.status == snapshot_status.DELETE ) ) \
|
||||
.all()
|
||||
for snapshot in snapshots:
|
||||
if self.type == snapshot.uci.credentials.provider.type and snapshot.status == snapshot_status.PENDING:
|
||||
@@ -603,9 +603,9 @@ class EC2CloudProvider( object ):
|
||||
td = datetime.utcnow() - z_inst.update_time
|
||||
if td.seconds > 180: # if instance has been in SUBMITTED state for more than 3 minutes
|
||||
log.debug( "[%s] Running zombie repair update on instance with DB id '%s'" % ( z_inst.uci.credentials.provider.type, z_inst.id ) )
|
||||
self.processZombie( z_inst )
|
||||
self.process_zombie( z_inst )
|
||||
|
||||
def updateInstance( self, inst ):
|
||||
def update_instance( self, inst ):
|
||||
|
||||
# Get credentials associated wit this instance
|
||||
uci_id = inst.uci_id
|
||||
@@ -678,7 +678,7 @@ class EC2CloudProvider( object ):
|
||||
self.sa_session.flush()
|
||||
return None
|
||||
|
||||
def updateStore( self, store ):
|
||||
def update_store( self, store ):
|
||||
# Get credentials associated wit this store
|
||||
uci_id = store.uci_id
|
||||
uci = self.sa_session.query( model.UCI ).get( uci_id )
|
||||
@@ -748,7 +748,7 @@ class EC2CloudProvider( object ):
|
||||
self.sa_session.add( store )
|
||||
self.sa_session.flush()
|
||||
|
||||
def updateSnapshot( self, snapshot ):
|
||||
def update_snapshot( self, snapshot ):
|
||||
# Get credentials associated wit this store
|
||||
uci_id = snapshot.uci_id
|
||||
uci = self.sa_session.query( model.UCI ).get( uci_id )
|
||||
@@ -839,7 +839,7 @@ class EC2CloudProvider( object ):
|
||||
self.sa_session.add( snapshot )
|
||||
self.sa_session.flush()
|
||||
|
||||
def processZombie( self, inst ):
|
||||
def process_zombie( self, inst ):
|
||||
"""
|
||||
Attempt at discovering if starting an instance was successful but local database was not updated
|
||||
accordingly or if something else failed and instance was never started. Currently, no automatic
|
||||
|
||||
@@ -110,16 +110,16 @@ class EucalyptusCloudProvider( object ):
|
||||
return
|
||||
try:
|
||||
if uci_state==uci_states.NEW:
|
||||
self.createUCI( uci_wrapper )
|
||||
self.create_uci( uci_wrapper )
|
||||
elif uci_state==uci_states.DELETING:
|
||||
self.deleteUCI( uci_wrapper )
|
||||
self.delete_uci( uci_wrapper )
|
||||
elif uci_state==uci_states.SUBMITTED:
|
||||
self.startUCI( uci_wrapper )
|
||||
#self.dummyStartUCI( uci_wrapper )
|
||||
self.start_uci( uci_wrapper )
|
||||
#self.dummy_start_uci( uci_wrapper )
|
||||
elif uci_state==uci_states.SHUTTING_DOWN:
|
||||
self.stopUCI( uci_wrapper )
|
||||
self.stop_uci( uci_wrapper )
|
||||
elif uci_state==uci_states.SNAPSHOT:
|
||||
self.snapshotUCI( uci_wrapper )
|
||||
self.snapshot_uci( uci_wrapper )
|
||||
except:
|
||||
log.exception( "Uncaught exception executing cloud request." )
|
||||
cnt += 1
|
||||
@@ -230,7 +230,7 @@ class EucalyptusCloudProvider( object ):
|
||||
uci_wrapper.set_error( err+". Contact site administrator to ensure needed machine image is registered.", True )
|
||||
return None
|
||||
|
||||
def createUCI( self, uci_wrapper ):
|
||||
def create_uci( self, uci_wrapper ):
|
||||
"""
|
||||
Create User Configured Instance (UCI) - i.e., create storage volume on cloud provider
|
||||
and register relevant information in local Galaxy database.
|
||||
@@ -276,7 +276,7 @@ class EucalyptusCloudProvider( object ):
|
||||
uci_wrapper.set_store_status( vol.id, uci_states.ERROR )
|
||||
uci_wrapper.set_error( err, True )
|
||||
|
||||
def deleteUCI( self, uci_wrapper ):
|
||||
def delete_uci( self, uci_wrapper ):
|
||||
"""
|
||||
Delete UCI - i.e., delete all storage volumes associated with this UCI.
|
||||
NOTE that this implies deletion of any and all data associated
|
||||
@@ -318,7 +318,7 @@ class EucalyptusCloudProvider( object ):
|
||||
log.error( err )
|
||||
uci_wrapper.set_error( err, True )
|
||||
|
||||
def snapshotUCI( self, uci_wrapper ):
|
||||
def snapshot_uci( self, uci_wrapper ):
|
||||
"""
|
||||
Initiate creation of a snapshot by cloud provider for all storage volumes
|
||||
associated with this UCI.
|
||||
@@ -361,10 +361,10 @@ class EucalyptusCloudProvider( object ):
|
||||
# Feel free to resent state of this instance and use it normally.", True )
|
||||
|
||||
|
||||
def addStorageToUCI( self, uci_wrapper ):
|
||||
def add_storage_to_uci( self, uci_wrapper ):
|
||||
""" Adds more storage to specified UCI """
|
||||
|
||||
def dummyStartUCI( self, uci_wrapper ):
|
||||
def dummy_start_uci( self, uci_wrapper ):
|
||||
|
||||
uci = uci_wrapper.get_uci()
|
||||
log.debug( "Would be starting instance '%s'" % uci.name )
|
||||
@@ -374,7 +374,7 @@ class EucalyptusCloudProvider( object ):
|
||||
time.sleep(10)
|
||||
log.debug( "Woke up! (%s)" % uci.name )
|
||||
|
||||
def startUCI( self, uci_wrapper ):
|
||||
def start_uci( self, uci_wrapper ):
|
||||
"""
|
||||
Start instance(s) of given UCI on the cloud.
|
||||
"""
|
||||
@@ -443,7 +443,7 @@ class EucalyptusCloudProvider( object ):
|
||||
else:
|
||||
log.error( "UCI '%s' is in 'error' state, starting instance was aborted." % uci_wrapper.get_name() )
|
||||
|
||||
def stopUCI( self, uci_wrapper):
|
||||
def stop_uci( self, uci_wrapper):
|
||||
"""
|
||||
Stop all cloud instances associated with given UCI.
|
||||
"""
|
||||
@@ -454,7 +454,7 @@ class EucalyptusCloudProvider( object ):
|
||||
# Process list of instances and remove any references to empty instance id's
|
||||
for i in il:
|
||||
if i is None:
|
||||
l.remove( i )
|
||||
il.remove( i )
|
||||
log.debug( 'List of instances being terminated: %s' % il )
|
||||
rl = conn.get_all_instances( il ) # Reservation list associated with given instances
|
||||
|
||||
@@ -543,7 +543,7 @@ class EucalyptusCloudProvider( object ):
|
||||
for inst in instances:
|
||||
if self.type == inst.uci.credentials.provider.type:
|
||||
log.debug( "[%s] Running general status update on instance '%s'" % ( inst.uci.credentials.provider.type, inst.instance_id ) )
|
||||
self.updateInstance( inst )
|
||||
self.update_instance( inst )
|
||||
|
||||
# Update storage volume(s)
|
||||
stores = self.sa_session.query( model.CloudStore ) \
|
||||
@@ -554,11 +554,11 @@ class EucalyptusCloudProvider( object ):
|
||||
for store in stores:
|
||||
if self.type == store.uci.credentials.provider.type: # and store.volume_id != None:
|
||||
log.debug( "[%s] Running general status update on store with local database ID: '%s'" % ( store.uci.credentials.provider.type, store.id ) )
|
||||
self.updateStore( store )
|
||||
self.update_store( store )
|
||||
|
||||
# Update pending snapshots or delete ones marked for deletion
|
||||
snapshots = self.sa_session.query( model.CloudSnapshot ) \
|
||||
.filter_by( status=snapshot_status.PENDING, status=snapshot_status.DELETE ) \
|
||||
.filter( or_( model.CloudSnapshot.table.c.status == snapshot_status.PENDING, model.CloudSnapshot.table.c.status == snapshot_status.DELETE ) ) \
|
||||
.all()
|
||||
for snapshot in snapshots:
|
||||
if self.type == snapshot.uci.credentials.provider.type and snapshot.status == snapshot_status.PENDING:
|
||||
@@ -583,9 +583,9 @@ class EucalyptusCloudProvider( object ):
|
||||
# log.debug( "z_inst.id: %s, time delta is %s sec" % ( z_inst.id, td.seconds ) )
|
||||
if td.seconds > 180: # if instance has been in SUBMITTED state for more than 3 minutes
|
||||
log.debug( "[%s](td=%s) Running zombie repair update on instance with DB id '%s'" % ( z_inst.uci.credentials.provider.type, td.seconds, z_inst.id ) )
|
||||
self.processZombie( z_inst )
|
||||
self.process_zombie( z_inst )
|
||||
|
||||
def updateInstance( self, inst ):
|
||||
def update_instance( self, inst ):
|
||||
"""
|
||||
Update information in local database for given instance as it is obtained from cloud provider.
|
||||
Along with updating information about given instance, information about the UCI controlling
|
||||
@@ -662,7 +662,7 @@ class EucalyptusCloudProvider( object ):
|
||||
self.sa_session.flush()
|
||||
return None
|
||||
|
||||
def updateStore( self, store ):
|
||||
def update_store( self, store ):
|
||||
"""
|
||||
Update information in local database for given storage volume as it is obtained from cloud provider.
|
||||
Along with updating information about given storage volume, information about the UCI controlling
|
||||
@@ -756,7 +756,7 @@ class EucalyptusCloudProvider( object ):
|
||||
self.sa_session.add( store )
|
||||
self.sa_session.flush()
|
||||
|
||||
def updateSnapshot( self, snapshot ):
|
||||
def update_snapshot( self, snapshot ):
|
||||
"""
|
||||
Update information in local database for given snapshot as it is obtained from cloud provider.
|
||||
Along with updating information about given snapshot, information about the UCI controlling
|
||||
@@ -855,7 +855,7 @@ class EucalyptusCloudProvider( object ):
|
||||
self.sa_session.add( snapshot )
|
||||
self.sa_session.flush()
|
||||
|
||||
def processZombie( self, inst ):
|
||||
def process_zombie( self, inst ):
|
||||
"""
|
||||
Attempt at discovering if starting a cloud instance was successful but local database was not updated
|
||||
accordingly or if something else failed and instance was never started. Currently, no automatic
|
||||
|
||||
@@ -515,7 +515,7 @@ class CloudController( BaseController ):
|
||||
error['provider_error'] = "You must select cloud provider type for this machine image."
|
||||
elif image_id=='' or len( image_id ) > 255:
|
||||
error['id_error'] = "Image ID must be between 1 and 255 characters long."
|
||||
elif trans.sa_session.query( model.CloudUserCredentials ) \
|
||||
elif trans.sa_session.query( model.CloudImage ) \
|
||||
.filter_by( deleted=False ) \
|
||||
.filter( model.CloudImage.table.c.image_id == image_id ) \
|
||||
.first():
|
||||
@@ -558,7 +558,7 @@ class CloudController( BaseController ):
|
||||
@web.expose
|
||||
@web.require_login( "use Galaxy cloud" )
|
||||
def list_machine_images( self, trans ):
|
||||
images = trans.sa_session.query( model.CloudImage ).filter( model.CloudImage.table.c.deleted != True ).all()
|
||||
images = trans.sa_session.query( model.CloudImage ).filter_by( deleted=False ).all()
|
||||
return trans.fill_template( '/cloud/list_images.mako', images=images )
|
||||
|
||||
@web.expose
|
||||
@@ -1028,15 +1028,19 @@ class CloudController( BaseController ):
|
||||
@web.json
|
||||
def json_update( self, trans ):
|
||||
user = trans.get_user()
|
||||
UCIs = trans.sa_session.query( model.UCI ).filter_by( user=user ).filter( model.UCI.table.c.deleted != True ).all()
|
||||
UCIs = trans.sa_session.query( model.UCI ).filter_by( user=user, deleted=False ).all()
|
||||
insd = {} # instance name-state dict
|
||||
for uci in UCIs:
|
||||
dict = {}
|
||||
dict['id'] = uci.id
|
||||
dict['state'] = uci.state
|
||||
if uci.error != None:
|
||||
dict['error'] = str( uci.error )
|
||||
else:
|
||||
dict['error'] = None
|
||||
if uci.launch_time != None:
|
||||
dict['launch_time'] = str(uci.launch_time)
|
||||
dict['time_ago'] = str(date.distance_of_time_in_words(uci.launch_time, date.datetime.utcnow() ) )
|
||||
dict['launch_time'] = str( uci.launch_time )
|
||||
dict['time_ago'] = str( date.distance_of_time_in_words( uci.launch_time, date.datetime.utcnow() ) )
|
||||
else:
|
||||
dict['launch_time'] = None
|
||||
dict['time_ago'] = None
|
||||
|
||||
@@ -36,10 +36,11 @@ function trim19(str){
|
||||
old_state = $(elem + "-state").text();
|
||||
prev_old_state = trim19( $(elem + "-state-p").text() );
|
||||
new_state = data[i].state;
|
||||
error_msg = data[i].error;
|
||||
//console.log( "old_state[%d] = %s", i, old_state );
|
||||
//console.log( "prev_old_state[%d] = %s", i, prev_old_state );
|
||||
//console.log( "new_state[%d] = %s", i, new_state );
|
||||
//console.log( trim19(prev_old_state) );
|
||||
//console.log( "error_msg[%d] = %s", i, error_msg );
|
||||
if ( ( old_state=='pending' && new_state=='running' ) || ( old_state=='shutting-down' && new_state=='available' ) || \
|
||||
( old_state=='running' && new_state=='available' ) || ( old_state=='running' && new_state=='error' ) || \
|
||||
( old_state=='pending' && new_state=='available' ) || ( old_state=='submitted' && new_state=='available' ) || \
|
||||
@@ -47,21 +48,11 @@ function trim19(str){
|
||||
var url = "${h.url_for( controller='cloud', action='list')}";
|
||||
location.replace( url );
|
||||
}
|
||||
else if ( ( old_state=='running' && new_state=='error' ) || ( old_state=='pending' && new_state=='error' ) || \
|
||||
( old_state=='submitted' && new_state=='error' ) || ( old_state=='submittedUCI' && new_state=='error' ) || \
|
||||
( old_state=='shutting-down' && new_state=='error' ) || ( prev_old_state.match('newUCI') && new_state=='error' ) || \
|
||||
( prev_old_state.match('new') && new_state=='error' ) || ( prev_old_state.match('deletingUCI') && new_state=='error' ) ) {
|
||||
// TODO: Following clause causes constant page refresh for an exception thrown as a result of instance not starting correctly - need alternative method!
|
||||
//( prev_old_state.match('available') && new_state=='error' ) || ( prev_old_state.match('deleting') && new_state=='error' ) \
|
||||
|
||||
else if ( ( ( old_state != 'error' && old_state != '' ) && new_state == 'error' ) || ( !prev_old_state.match('error') && new_state == 'error' ) ) {
|
||||
var url = "${h.url_for( controller='cloud', action='list')}";
|
||||
location.replace( url );
|
||||
}
|
||||
|
||||
if ( prev_old_state.match('deletingUCI') || prev_old_state.match('deleting') ) {
|
||||
setTimeout("update_state()", 3000);
|
||||
}
|
||||
|
||||
if ( new_state=='shutting-down' || new_state=='shutting-downUCI' ) {
|
||||
$(elem + "-link").text( "" );
|
||||
}
|
||||
@@ -90,15 +81,10 @@ function trim19(str){
|
||||
|
||||
// Update 'state' and 'time alive' fields
|
||||
$(elem + "-state").text( data[i].state );
|
||||
if ( ( prev_old_state.match('newUCI') && new_state=='new' ) || ( prev_old_state.match('newUCI') && new_state=='available' ) || \
|
||||
( prev_old_state.match('newUCI') && new_state=='creating' ) || ( prev_old_state.match('new') && new_state=='creating' ) || \
|
||||
( prev_old_state.match('new') && new_state=='available' ) || \
|
||||
( prev_old_state.match('deletingUCI') && new_state=='deleted' ) || ( prev_old_state.match('deleting') && new_state=='deleted' ) || \
|
||||
( prev_old_state.match('available') && new_state=='error' ) || ( prev_old_state.match('deleting') && new_state=='error' ) ) {
|
||||
// TODO: on state change from available->error and deleting->error page should be refreshed but that causes problems with
|
||||
// constant refreshings depending on what error message is so at least do it here...
|
||||
if ( new_state != 'error' ) { // Because 'error' state is handled as a JS link, don't include it in update
|
||||
$(elem + "-state-p").text( data[i].state );
|
||||
}
|
||||
|
||||
if (data[i].launch_time) {
|
||||
$(elem + "-launch_time").text( data[i].launch_time.substring(0, 16 ) + " UTC (" + data[i].time_ago + ")" );
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user