#NEWLINE# def run_job( self , job_execution ) : #NEWLINE# #INDENT# ctx = context . ctx ( ) #NEWLINE# job = conductor . job_get ( ctx , job_execution . job_id ) #NEWLINE# data_source_urls = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# ( additional_sources , updated_job_configs ) = job_utils . resolve_data_source_references ( job_execution . job_configs , job_execution . id , data_source_urls , self . cluster ) #NEWLINE# job_execution = conductor . job_execution_update ( ctx , job_execution , { #NEWLINE# #INDENT# 'data_source_urls' : job_utils . to_url_dict ( data_source_urls ) , #NEWLINE# #UNINDENT# } ) #NEWLINE# data_source_urls = job_utils . to_url_dict ( data_source_urls , runtime = True ) #NEWLINE# master = plugin_utils . get_instance ( self . cluster , 'nimbus' ) #NEWLINE# wf_dir = job_utils . create_workflow_dir ( master , '/tmp/storm-edp' , job , job_execution . id , '700' ) #NEWLINE# paths = self . _upload_job_files ( master , wf_dir , job , updated_job_configs ) #NEWLINE# paths = [ os . path . basename ( p ) for p in paths ] #NEWLINE# app_jar = paths . pop ( 0 ) #NEWLINE# job_class = updated_job_configs [ 'configs' ] [ 'edp.java.main_class' ] #NEWLINE# topology_name = self . _generate_topology_name ( job . name ) #NEWLINE# host = master . hostname ( ) #NEWLINE# args = updated_job_configs . get ( 'args' , [ ] ) #NEWLINE# args = ' ' . join ( [ arg for arg in args ] ) #NEWLINE# if args : #NEWLINE# #INDENT# args = ( ' ' + args ) #NEWLINE# #UNINDENT# cmd = ( '%(storm_jar)s -c nimbus.host=%(host)s %(job_jar)s %(main_class)s %(topology_name)s%(args)s' % { #NEWLINE# #INDENT# 'storm_jar' : '/usr/local/storm/bin/storm jar' , #NEWLINE# 'main_class' : job_class , #NEWLINE# 'job_jar' : app_jar , #NEWLINE# 'host' : host , #NEWLINE# 'topology_name' : topology_name , #NEWLINE# 'args' : args , #NEWLINE# #UNINDENT# } ) #NEWLINE# job_execution = conductor . job_execution_get ( ctx , job_execution . id ) #NEWLINE# if ( job_execution . info [ 'status' ] == edp . JOB_STATUS_TOBEKILLED ) : #NEWLINE# #INDENT# return ( None , edp . JOB_STATUS_KILLED , None ) #NEWLINE# #UNINDENT# with remote . get_remote ( master ) as r : #NEWLINE# #INDENT# launch = os . path . join ( wf_dir , 'launch_command' ) #NEWLINE# r . write_file_to ( launch , self . _job_script ( ) ) #NEWLINE# r . execute_command ( ( 'chmod +x %s' % launch ) ) #NEWLINE# ( ret , stdout ) = r . execute_command ( ( 'cd %s; ./launch_command %s > /dev/null 2>&1 & echo $!' % ( wf_dir , cmd ) ) ) #NEWLINE# #UNINDENT# if ( ret == 0 ) : #NEWLINE# #INDENT# return ( ( ( topology_name + '@' ) + master . id ) , edp . JOB_STATUS_RUNNING , { #NEWLINE# #INDENT# 'storm-path' : wf_dir , #NEWLINE# #UNINDENT# } ) #NEWLINE# #UNINDENT# raise e . EDPError ( ( _ ( 'Storm job execution failed. Exit status = %(status)s, stdout = %(stdout)s' ) % { #NEWLINE# #INDENT# 'status' : ret , #NEWLINE# 'stdout' : stdout , #NEWLINE# #UNINDENT# } ) )