<s> import fnmatch <EOL> import itertools <EOL> import os <EOL> from dcos import http , util <EOL> from dcos . errors import DCOSException , DCOSHTTPException <EOL> from six . moves import urllib <EOL> logger = util . get_logger ( __name__ ) <EOL> def get_master ( dcos_client = None ) : <EOL> """<STR_LIT>""" <EOL> dcos_client = dcos_client or DCOSClient ( ) <EOL> return Master ( dcos_client . get_master_state ( ) ) <EOL> class DCOSClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> config = util . get_config ( ) <EOL> self . _dcos_url = None <EOL> self . _mesos_master_url = None <EOL> mesos_master_url = config . get ( '<STR_LIT>' ) <EOL> if mesos_master_url is None : <EOL> self . _dcos_url = util . get_config_vals ( [ '<STR_LIT>' ] , config ) [ <NUM_LIT:0> ] <EOL> else : <EOL> self . _mesos_master_url = mesos_master_url <EOL> self . _timeout = config . get ( '<STR_LIT>' ) <EOL> def get_dcos_url ( self , path ) : <EOL> """<STR_LIT>""" <EOL> if self . _dcos_url : <EOL> return urllib . parse . urljoin ( self . _dcos_url , path ) <EOL> else : <EOL> raise util . missing_config_exception ( '<STR_LIT>' ) <EOL> def master_url ( self , path ) : <EOL> """<STR_LIT>""" <EOL> base_url = ( self . _mesos_master_url or <EOL> urllib . parse . urljoin ( self . _dcos_url , '<STR_LIT>' ) ) <EOL> return urllib . parse . urljoin ( base_url , path ) <EOL> def slave_url ( self , slave_id , private_url , path ) : <EOL> """<STR_LIT>""" <EOL> if self . _dcos_url : <EOL> return urllib . parse . urljoin ( self . _dcos_url , <EOL> '<STR_LIT>' . format ( slave_id , path ) ) <EOL> else : <EOL> return urllib . parse . urljoin ( private_url , path ) <EOL> def get_master_state ( self ) : <EOL> """<STR_LIT>""" <EOL> url = self . master_url ( '<STR_LIT>' ) <EOL> return http . get ( url , timeout = self . _timeout ) . json ( ) <EOL> def get_slave_state ( self , slave_id , private_url ) : <EOL> """<STR_LIT>""" <EOL> url = self . slave_url ( slave_id , private_url , '<STR_LIT>' ) <EOL> return http . get ( url , timeout = self . _timeout ) . json ( ) <EOL> def get_state_summary ( self ) : <EOL> """<STR_LIT>""" <EOL> url = self . master_url ( '<STR_LIT>' ) <EOL> return http . get ( url , timeout = self . _timeout ) . json ( ) <EOL> def slave_file_read ( self , slave_id , private_url , path , offset , length ) : <EOL> """<STR_LIT>""" <EOL> url = self . slave_url ( slave_id , <EOL> private_url , <EOL> '<STR_LIT>' ) <EOL> params = { '<STR_LIT:path>' : path , <EOL> '<STR_LIT>' : length , <EOL> '<STR_LIT>' : offset } <EOL> return http . get ( url , params = params , timeout = self . _timeout ) . json ( ) <EOL> def master_file_read ( self , path , length , offset ) : <EOL> """<STR_LIT>""" <EOL> url = self . master_url ( '<STR_LIT>' ) <EOL> params = { '<STR_LIT:path>' : path , <EOL> '<STR_LIT>' : length , <EOL> '<STR_LIT>' : offset } <EOL> return http . get ( url , params = params , timeout = self . _timeout ) . json ( ) <EOL> def shutdown_framework ( self , framework_id ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( '<STR_LIT>' . format ( framework_id ) ) <EOL> data = '<STR_LIT>' . format ( framework_id ) <EOL> url = self . master_url ( '<STR_LIT>' ) <EOL> try : <EOL> http . post ( url , data = data , timeout = self . _timeout ) <EOL> except DCOSHTTPException as e : <EOL> if e . response . status_code == <NUM_LIT> : <EOL> url = self . master_url ( '<STR_LIT>' ) <EOL> http . post ( url , data = data , timeout = self . _timeout ) <EOL> else : <EOL> raise <EOL> def metadata ( self ) : <EOL> """<STR_LIT>""" <EOL> url = self . get_dcos_url ( '<STR_LIT>' ) <EOL> return http . get ( url , timeout = self . _timeout ) . json ( ) <EOL> def browse ( self , slave , path ) : <EOL> """<STR_LIT>""" <EOL> url = self . slave_url ( slave [ '<STR_LIT:id>' ] , <EOL> slave . http_url ( ) , <EOL> '<STR_LIT>' ) <EOL> return http . get ( url , params = { '<STR_LIT:path>' : path } ) . json ( ) <EOL> class MesosDNSClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url = None ) : <EOL> self . url = url or urllib . parse . urljoin ( <EOL> util . get_config_vals ( [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> def _path ( self , path ) : <EOL> """<STR_LIT>""" <EOL> return urllib . parse . urljoin ( self . url , path ) <EOL> def hosts ( self , host ) : <EOL> """<STR_LIT>""" <EOL> url = self . _path ( '<STR_LIT>' . format ( host ) ) <EOL> return http . get ( url , headers = { } ) . json ( ) <EOL> class Master ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , state ) : <EOL> self . _state = state <EOL> self . _frameworks = { } <EOL> self . _slaves = { } <EOL> def state ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _state <EOL> def slave_base_url ( self , slave ) : <EOL> """<STR_LIT>""" <EOL> if self . _mesos_master_url is not None : <EOL> slave_ip = slave [ '<STR_LIT>' ] . split ( '<STR_LIT:@>' ) [ <NUM_LIT:1> ] <EOL> return '<STR_LIT>' . format ( slave_ip ) <EOL> else : <EOL> return urllib . parse . urljoin ( self . _dcos_url , <EOL> '<STR_LIT>' . format ( slave [ '<STR_LIT:id>' ] ) ) <EOL> def slave ( self , fltr ) : <EOL> """<STR_LIT>""" <EOL> slaves = self . slaves ( fltr ) <EOL> if len ( slaves ) == <NUM_LIT:0> : <EOL> raise DCOSException ( '<STR_LIT>' . format ( fltr ) ) <EOL> elif len ( slaves ) > <NUM_LIT:1> : <EOL> matches = [ '<STR_LIT>' . format ( slave [ '<STR_LIT:id>' ] ) for slave in slaves ] <EOL> raise DCOSException ( <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" . format ( '<STR_LIT:\n>' . join ( matches ) ) ) <EOL> else : <EOL> return slaves [ <NUM_LIT:0> ] <EOL> def task ( self , fltr ) : <EOL> """<STR_LIT>""" <EOL> tasks = self . tasks ( fltr ) <EOL> if len ( tasks ) == <NUM_LIT:0> : <EOL> raise DCOSException ( <EOL> '<STR_LIT>' . format ( fltr ) ) <EOL> elif len ( tasks ) > <NUM_LIT:1> : <EOL> msg = [ ( "<STR_LIT>" + <EOL> "<STR_LIT>" ) . format ( fltr ) ] <EOL> msg += [ "<STR_LIT>" . format ( t [ "<STR_LIT:id>" ] ) for t in tasks ] <EOL> raise DCOSException ( '<STR_LIT:\n>' . join ( msg ) ) <EOL> else : <EOL> return tasks [ <NUM_LIT:0> ] <EOL> def framework ( self , framework_id ) : <EOL> """<STR_LIT>""" <EOL> for f in self . _framework_dicts ( True , True ) : <EOL> if f [ '<STR_LIT:id>' ] == framework_id : <EOL> return self . _framework_obj ( f ) <EOL> return None <EOL> def slaves ( self , fltr = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return [ self . _slave_obj ( slave ) <EOL> for slave in self . state ( ) [ '<STR_LIT>' ] <EOL> if fltr in slave [ '<STR_LIT:id>' ] ] <EOL> def tasks ( self , fltr = "<STR_LIT>" , completed = False ) : <EOL> """<STR_LIT>""" <EOL> keys = [ '<STR_LIT>' ] <EOL> if completed : <EOL> keys = [ '<STR_LIT>' ] <EOL> tasks = [ ] <EOL> for framework in self . _framework_dicts ( completed , completed ) : <EOL> for task in _merge ( framework , keys ) : <EOL> if fltr in task [ '<STR_LIT:id>' ] or fnmatch . fnmatchcase ( task [ '<STR_LIT:id>' ] , fltr ) : <EOL> task = self . _framework_obj ( framework ) . task ( task [ '<STR_LIT:id>' ] ) <EOL> tasks . append ( task ) <EOL> return tasks <EOL> def frameworks ( self , inactive = False , completed = False ) : <EOL> """<STR_LIT>""" <EOL> return [ self . _framework_obj ( framework ) <EOL> for framework in self . _framework_dicts ( inactive , completed ) ] <EOL> @ util . duration <EOL> def fetch ( self , path , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> url = urllib . parse . urljoin ( self . _base_url ( ) , path ) <EOL> return http . get ( url , ** kwargs ) <EOL> def _slave_obj ( self , slave ) : <EOL> """<STR_LIT>""" <EOL> if slave [ '<STR_LIT:id>' ] not in self . _slaves : <EOL> self . _slaves [ slave [ '<STR_LIT:id>' ] ] = Slave ( slave , None , self ) <EOL> return self . _slaves [ slave [ '<STR_LIT:id>' ] ] <EOL> def _framework_obj ( self , framework ) : <EOL> """<STR_LIT>""" <EOL> if framework [ '<STR_LIT:id>' ] not in self . _frameworks : <EOL> self . _frameworks [ framework [ '<STR_LIT:id>' ] ] = Framework ( framework , self ) <EOL> return self . _frameworks [ framework [ '<STR_LIT:id>' ] ] <EOL> def _framework_dicts ( self , inactive = False , completed = False ) : <EOL> """<STR_LIT>""" <EOL> if completed : <EOL> for framework in self . state ( ) [ '<STR_LIT>' ] : <EOL> yield framework <EOL> for framework in self . state ( ) [ '<STR_LIT>' ] : <EOL> if inactive or framework [ '<STR_LIT>' ] : <EOL> yield framework <EOL> class Slave ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , short_state , state , master ) : <EOL> self . _short_state = short_state <EOL> self . _state = state <EOL> self . _master = master <EOL> def state ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _state : <EOL> self . _state = DCOSClient ( ) . get_slave_state ( self [ '<STR_LIT:id>' ] , <EOL> self . http_url ( ) ) <EOL> return self . _state <EOL> def http_url ( self ) : <EOL> """<STR_LIT>""" <EOL> parsed_pid = parse_pid ( self [ '<STR_LIT>' ] ) <EOL> return '<STR_LIT>' . format ( parsed_pid [ <NUM_LIT:1> ] , parsed_pid [ <NUM_LIT:2> ] ) <EOL> def _framework_dicts ( self ) : <EOL> """<STR_LIT>""" <EOL> return _merge ( self . state ( ) , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def executor_dicts ( self ) : <EOL> """<STR_LIT>""" <EOL> iters = [ _merge ( framework , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> for framework in self . _framework_dicts ( ) ] <EOL> return itertools . chain ( * iters ) <EOL> def __getitem__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> return self . _short_state [ name ] <EOL> class Framework ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , framework , master ) : <EOL> self . _framework = framework <EOL> self . _master = master <EOL> self . _tasks = { } <EOL> def task ( self , task_id ) : <EOL> """<STR_LIT>""" <EOL> for task in _merge ( self . _framework , [ '<STR_LIT>' , '<STR_LIT>' ] ) : <EOL> if task [ '<STR_LIT:id>' ] == task_id : <EOL> return self . _task_obj ( task ) <EOL> return None <EOL> def _task_obj ( self , task ) : <EOL> """<STR_LIT>""" <EOL> if task [ '<STR_LIT:id>' ] not in self . _tasks : <EOL> self . _tasks [ task [ '<STR_LIT:id>' ] ] = Task ( task , self . _master ) <EOL> return self . _tasks [ task [ '<STR_LIT:id>' ] ] <EOL> def dict ( self ) : <EOL> return self . _framework <EOL> def __getitem__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> return self . _framework [ name ] <EOL> class Task ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , task , master ) : <EOL> self . _task = task <EOL> self . _master = master <EOL> def dict ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _task <EOL> def framework ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _master . framework ( self [ "<STR_LIT>" ] ) <EOL> def slave ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _master . slave ( self [ "<STR_LIT>" ] ) <EOL> def user ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . framework ( ) [ '<STR_LIT:user>' ] <EOL> def executor ( self ) : <EOL> """<STR_LIT>""" <EOL> for executor in self . slave ( ) . executor_dicts ( ) : <EOL> tasks = _merge ( executor , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> if any ( task [ '<STR_LIT:id>' ] == self [ '<STR_LIT:id>' ] for task in tasks ) : <EOL> return executor <EOL> return None <EOL> def directory ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . executor ( ) [ '<STR_LIT>' ] <EOL> def __getitem__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> return self . _task [ name ] <EOL> class MesosFile ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , path , task = None , slave = None , dcos_client = None ) : <EOL> if task and slave : <EOL> raise ValueError ( <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" ) <EOL> if slave : <EOL> self . _slave = slave <EOL> elif task : <EOL> self . _slave = task . slave ( ) <EOL> else : <EOL> self . _slave = None <EOL> self . _task = task <EOL> self . _path = path <EOL> self . _dcos_client = dcos_client or DCOSClient ( ) <EOL> self . _cursor = <NUM_LIT:0> <EOL> def size ( self ) : <EOL> """<STR_LIT>""" <EOL> params = self . _params ( <NUM_LIT:0> , offset = - <NUM_LIT:1> ) <EOL> return self . _fetch ( params ) [ "<STR_LIT>" ] <EOL> def seek ( self , offset , whence = os . SEEK_SET ) : <EOL> """<STR_LIT>""" <EOL> if whence == os . SEEK_SET : <EOL> self . _cursor = <NUM_LIT:0> + offset <EOL> elif whence == os . SEEK_CUR : <EOL> self . _cursor += offset <EOL> elif whence == os . SEEK_END : <EOL> self . _cursor = self . size ( ) + offset <EOL> else : <EOL> raise ValueError ( <EOL> "<STR_LIT>" . format ( whence ) ) <EOL> def tell ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _cursor <EOL> def read ( self , length = None ) : <EOL> """<STR_LIT>""" <EOL> data = '<STR_LIT>' <EOL> while length is None or length - len ( data ) > <NUM_LIT:0> : <EOL> chunk_length = - <NUM_LIT:1> if length is None else length - len ( data ) <EOL> chunk = self . _fetch_chunk ( chunk_length ) <EOL> if chunk == '<STR_LIT>' : <EOL> break <EOL> data += chunk <EOL> return data <EOL> def _host_path ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _task : <EOL> directory = self . _task . directory ( ) <EOL> if directory [ - <NUM_LIT:1> ] == '<STR_LIT:/>' : <EOL> return directory + self . _path <EOL> else : <EOL> return directory + '<STR_LIT:/>' + self . _path <EOL> else : <EOL> return self . _path <EOL> def _params ( self , length , offset = None ) : <EOL> """<STR_LIT>""" <EOL> if offset is None : <EOL> offset = self . _cursor <EOL> return { <EOL> '<STR_LIT:path>' : self . _host_path ( ) , <EOL> '<STR_LIT>' : offset , <EOL> '<STR_LIT>' : length <EOL> } <EOL> def _fetch_chunk ( self , length , offset = None ) : <EOL> """<STR_LIT>""" <EOL> if offset is not None : <EOL> self . seek ( offset , os . SEEK_SET ) <EOL> params = self . _params ( length ) <EOL> data = self . _fetch ( params ) [ "<STR_LIT:data>" ] <EOL> self . seek ( len ( data ) , os . SEEK_CUR ) <EOL> return data <EOL> def _fetch ( self , params ) : <EOL> """<STR_LIT>""" <EOL> if self . _slave : <EOL> return self . _dcos_client . slave_file_read ( self . _slave [ '<STR_LIT:id>' ] , <EOL> self . _slave . http_url ( ) , <EOL> ** params ) <EOL> else : <EOL> return self . _dcos_client . master_file_read ( ** params ) <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _task : <EOL> return "<STR_LIT>" . format ( self . _task [ '<STR_LIT:id>' ] , self . _path ) <EOL> elif self . _slave : <EOL> return "<STR_LIT>" . format ( self . _slave [ '<STR_LIT:id>' ] , self . _path ) <EOL> else : <EOL> return "<STR_LIT>" . format ( self . _path ) <EOL> def parse_pid ( pid ) : <EOL> """<STR_LIT>""" <EOL> id_ , second = pid . split ( '<STR_LIT:@>' ) <EOL> ip , port = second . split ( '<STR_LIT::>' ) <EOL> return id_ , ip , port <EOL> def _merge ( d , keys ) : <EOL> """<STR_LIT>""" <EOL> return itertools . chain ( * [ d [ k ] for k in keys ] ) </s>
