#NEWLINE# def start( self ) : #NEWLINE# #INDENT# self . task_queue = Queue ( ) #NEWLINE# self . result_queue = Queue ( ) #NEWLINE# framework = mesos_pb2 . FrameworkInfo ( ) #NEWLINE# framework . user = '' #NEWLINE# if ( not configuration . get ( 'mesos' , 'MASTER' ) ) : #NEWLINE# #INDENT# logging . error ( 'Expecting mesos master URL for mesos executor' ) #NEWLINE# raise AirflowException ( 'mesos.master not provided for mesos executor' ) #NEWLINE# #UNINDENT# master = configuration . get ( 'mesos' , 'MASTER' ) #NEWLINE# framework . name = get_framework_name ( ) #NEWLINE# if ( not configuration . get ( 'mesos' , 'TASK_CPU' ) ) : #NEWLINE# #INDENT# task_cpu = 1 #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# task_cpu = configuration . getint ( 'mesos' , 'TASK_CPU' ) #NEWLINE# #UNINDENT# if ( not configuration . get ( 'mesos' , 'TASK_MEMORY' ) ) : #NEWLINE# #INDENT# task_memory = 256 #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# task_memory = configuration . getint ( 'mesos' , 'TASK_MEMORY' ) #NEWLINE# #UNINDENT# if configuration . getboolean ( 'mesos' , 'CHECKPOINT' ) : #NEWLINE# #INDENT# framework . checkpoint = True #NEWLINE# if configuration . get ( 'mesos' , 'FAILOVER_TIMEOUT' ) : #NEWLINE# #INDENT# from airflow.models import Connection #NEWLINE# conn_id = ( FRAMEWORK_CONNID_PREFIX + framework . name ) #NEWLINE# session = Session ( ) #NEWLINE# connection = session . query ( Connection ) . filter_by ( conn_id = conn_id ) . first ( ) #NEWLINE# if ( connection is not None ) : #NEWLINE# #INDENT# framework . id . value = connection . extra #NEWLINE# #UNINDENT# framework . failover_timeout = configuration . getint ( 'mesos' , 'FAILOVER_TIMEOUT' ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# framework . checkpoint = False #NEWLINE# #UNINDENT# logging . info ( 'MesosFramework master : %s, name : %s, cpu : %s, mem : %s, checkpoint : %s' , master , framework . name , str ( task_cpu ) , str ( task_memory ) , str ( framework . checkpoint ) ) #NEWLINE# implicit_acknowledgements = 1 #NEWLINE# if configuration . getboolean ( 'mesos' , 'AUTHENTICATE' ) : #NEWLINE# #INDENT# if ( not configuration . get ( 'mesos' , 'DEFAULT_PRINCIPAL' ) ) : #NEWLINE# #INDENT# logging . error ( 'Expecting authentication principal in the environment' ) #NEWLINE# raise AirflowException ( 'mesos.default_principal not provided in authenticated mode' ) #NEWLINE# #UNINDENT# if ( not configuration . get ( 'mesos' , 'DEFAULT_SECRET' ) ) : #NEWLINE# #INDENT# logging . error ( 'Expecting authentication secret in the environment' ) #NEWLINE# raise AirflowException ( 'mesos.default_secret not provided in authenticated mode' ) #NEWLINE# #UNINDENT# credential = mesos_pb2 . Credential ( ) #NEWLINE# credential . principal = configuration . get ( 'mesos' , 'DEFAULT_PRINCIPAL' ) #NEWLINE# credential . secret = configuration . get ( 'mesos' , 'DEFAULT_SECRET' ) #NEWLINE# framework . principal = credential . principal #NEWLINE# driver = mesos . native . MesosSchedulerDriver ( AirflowMesosScheduler ( self . task_queue , self . result_queue , task_cpu , task_memory ) , framework , master , implicit_acknowledgements , credential ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# framework . principal = 'Airflow' #NEWLINE# driver = mesos . native . MesosSchedulerDriver ( AirflowMesosScheduler ( self . task_queue , self . result_queue , task_cpu , task_memory ) , framework , master , implicit_acknowledgements ) #NEWLINE# #UNINDENT# self . mesos_driver = driver #NEWLINE# self . mesos_driver . start ( )