<s> from __future__ import division <EOL> import random <EOL> import time <EOL> import sys <EOL> import mesos . interface <EOL> from mesos . interface import mesos_pb2 <EOL> from relay_mesos import log <EOL> from relay_mesos . util import catch <EOL> SCALAR_KEYS = { '<STR_LIT>' : float , '<STR_LIT>' : int , '<STR_LIT>' : int } <EOL> RANGE_KEYS = { '<STR_LIT>' : int } <EOL> SET_KEYS = { '<STR_LIT>' : str } <EOL> class MaxFailuresReached ( Exception ) : <EOL> pass <EOL> def filter_offers ( offers , task_resources ) : <EOL> """<STR_LIT>""" <EOL> available_offers = [ ] <EOL> decline_offers = [ ] <EOL> for offer in offers : <EOL> ntasks = calc_tasks_per_offer ( offer , task_resources ) <EOL> if ntasks == <NUM_LIT:0> : <EOL> decline_offers . append ( offer ) <EOL> continue <EOL> available_offers . append ( ( offer , ntasks ) ) <EOL> return available_offers , decline_offers <EOL> def calc_tasks_per_offer ( offer , task_resources ) : <EOL> """<STR_LIT>""" <EOL> num_tasks = float ( '<STR_LIT>' ) <EOL> for res in offer . resources : <EOL> if res . name not in task_resources : <EOL> continue <EOL> if res . name in SCALAR_KEYS : <EOL> oval = float ( res . scalar . value ) <EOL> reqval = float ( task_resources . get ( res . name ) ) <EOL> if reqval <= oval : <EOL> num_tasks = min ( num_tasks , int ( oval / reqval ) ) <EOL> else : <EOL> num_tasks = <NUM_LIT:0> <EOL> break <EOL> elif res . name in RANGE_KEYS : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> elif res . name in SET_KEYS : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> else : <EOL> raise NotImplementedError ( ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) % res . name ) <EOL> if num_tasks == float ( '<STR_LIT>' ) : <EOL> return <NUM_LIT:0> <EOL> else : <EOL> return num_tasks <EOL> def create_tasks ( MV , available_offers , driver , command , ns ) : <EOL> """<STR_LIT>""" <EOL> n_fulfilled = <NUM_LIT:0> <EOL> for offer , ntasks in available_offers : <EOL> if n_fulfilled >= MV : <EOL> driver . declineOffer ( offer . id ) <EOL> continue <EOL> tasks = [ ] <EOL> for ID in range ( ntasks ) : <EOL> if n_fulfilled >= MV : <EOL> break <EOL> n_fulfilled += <NUM_LIT:1> <EOL> tid = "<STR_LIT>" % ( <EOL> ID , offer . id . value , random . randint ( <NUM_LIT:1> , sys . maxint ) ) <EOL> log . debug ( <EOL> "<STR_LIT>" , extra = dict ( <EOL> offer_host = offer . hostname , task_id = tid , <EOL> mesos_framework_name = ns . mesos_framework_name ) ) <EOL> task = _create_task ( <EOL> tid , offer , command , ns ) <EOL> tasks . append ( task ) <EOL> driver . launchTasks ( offer . id , tasks ) <EOL> return n_fulfilled <EOL> def _create_task_add_task_resources ( task , ns ) : <EOL> task_resources = dict ( ns . mesos_task_resources ) <EOL> seen = set ( ) <EOL> for key in set ( SCALAR_KEYS ) . intersection ( task_resources ) : <EOL> seen . add ( key ) <EOL> resource = task . resources . add ( ) <EOL> resource . name = key <EOL> resource . type = mesos_pb2 . Value . SCALAR <EOL> typecast = SCALAR_KEYS [ key ] <EOL> resource . scalar . value = typecast ( task_resources [ key ] ) <EOL> for key in set ( RANGE_KEYS ) . intersection ( task_resources ) : <EOL> seen . add ( key ) <EOL> resource = task . resources . add ( ) <EOL> resource . name = key <EOL> resource . type = mesos_pb2 . Value . RANGES <EOL> for range_data in task_resources [ key ] : <EOL> inst = resource . ranges . range . add ( ) <EOL> typecast = RANGE_KEYS [ key ] <EOL> inst . begin = typecast ( range_data [ <NUM_LIT:0> ] ) <EOL> inst . end = typecast ( range_data [ <NUM_LIT:1> ] ) <EOL> for key in set ( SET_KEYS ) . intersection ( task_resources ) : <EOL> typecast = SET_KEYS [ key ] <EOL> seen . add ( key ) <EOL> resource = task . resources . add ( ) <EOL> resource = task . resources . add ( ) <EOL> resource . name = key <EOL> resource . type = mesos_pb2 . Value . SET <EOL> for elem in task_resources [ key ] : <EOL> resource . set . item . append ( typecast ( elem ) ) <EOL> unrecognized_keys = set ( task_resources ) . difference ( seen ) <EOL> if unrecognized_keys : <EOL> msg = "<STR_LIT>" <EOL> log . error ( msg , extra = dict ( <EOL> unrecognized_keys = unrecognized_keys , <EOL> mesos_framework_name = ns . mesos_framework_name ) ) <EOL> raise UserWarning ( <EOL> "<STR_LIT>" % ( msg , unrecognized_keys ) ) <EOL> def _create_task ( tid , offer , command , ns ) : <EOL> """<STR_LIT>""" <EOL> task = dict ( <EOL> task_id = mesos_pb2 . TaskID ( value = tid ) , <EOL> slave_id = offer . slave_id , <EOL> command = mesos_pb2 . CommandInfo ( <EOL> value = command , <EOL> uris = [ mesos_pb2 . CommandInfo . URI ( value = uri ) for uri in ns . uris ] , <EOL> environment = mesos_pb2 . Environment ( variables = [ <EOL> mesos_pb2 . Environment . Variable ( name = k , value = v ) <EOL> for k , v in ns . mesos_environment ] ) <EOL> ) <EOL> ) <EOL> if ns . mesos_framework_name : <EOL> task . update ( <EOL> name = "<STR_LIT>" % ( ns . mesos_framework_name , tid ) ) <EOL> else : <EOL> task . update ( name = "<STR_LIT>" % tid ) <EOL> if ns . docker_image : <EOL> volumes = [ <EOL> mesos_pb2 . Volume ( <EOL> host_path = host_path , <EOL> container_path = container_path , <EOL> mode = mesos_pb2 . Volume . Mode . Value ( mode . upper ( ) ) ) <EOL> for host_path , container_path , mode in ns . volumes ] <EOL> task . update ( <EOL> container = mesos_pb2 . ContainerInfo ( <EOL> type = mesos_pb2 . ContainerInfo . DOCKER , <EOL> volumes = volumes , <EOL> docker = mesos_pb2 . ContainerInfo . DockerInfo ( <EOL> image = ns . docker_image , <EOL> force_pull_image = ns . force_pull_image , <EOL> network = mesos_pb2 . ContainerInfo . DockerInfo . Network . Value ( <EOL> ns . docker_network ) , <EOL> parameters = [ <EOL> mesos_pb2 . Parameter ( key = k , value = v ) <EOL> for k , v in ns . docker_parameters . items ( ) ] , <EOL> ) <EOL> ) ) <EOL> task = mesos_pb2 . TaskInfo ( ** task ) <EOL> _create_task_add_task_resources ( task , ns ) <EOL> return task <EOL> class Scheduler ( mesos . interface . Scheduler ) : <EOL> def __init__ ( self , MV , exception_sender , mesos_ready , ns ) : <EOL> self . ns = ns <EOL> self . MV = MV <EOL> self . mesos_ready = mesos_ready <EOL> self . exception_sender = exception_sender <EOL> self . failures = <NUM_LIT:0> <EOL> def registered ( self , driver , frameworkId , masterInfo ) : <EOL> """<STR_LIT>""" <EOL> catch ( self . _registered , self . exception_sender ) ( <EOL> driver , frameworkId , masterInfo ) <EOL> def _registered ( self , driver , frameworkId , masterInfo ) : <EOL> self . mesos_ready . acquire ( ) <EOL> self . mesos_ready . notify ( ) <EOL> self . mesos_ready . release ( ) <EOL> log . info ( <EOL> "<STR_LIT>" , extra = dict ( <EOL> framework_id = frameworkId . value , master_pid = masterInfo . pid , <EOL> master_hostname = masterInfo . hostname , master_id = masterInfo . id , <EOL> master_ip = masterInfo . ip , master_port = masterInfo . port , <EOL> mesos_framework_name = self . ns . mesos_framework_name , <EOL> ) ) <EOL> def reregistered ( self , driver , masterInfo ) : <EOL> log . info ( <EOL> "<STR_LIT>" , extra = dict ( <EOL> master_pid = masterInfo . pid , <EOL> master_hostname = masterInfo . hostname , master_id = masterInfo . id , <EOL> master_ip = masterInfo . ip , master_port = masterInfo . port , <EOL> mesos_framework_name = self . ns . mesos_framework_name , <EOL> ) ) <EOL> def resourceOffers ( self , driver , offers ) : <EOL> catch ( self . _resourceOffers , self . exception_sender ) ( <EOL> driver , offers ) <EOL> def _resourceOffers ( self , driver , offers ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" , extra = dict ( <EOL> num_offers = len ( offers ) , <EOL> mesos_framework_name = self . ns . mesos_framework_name ) ) <EOL> available_offers , decline_offers = filter_offers ( <EOL> offers , dict ( self . ns . mesos_task_resources ) ) <EOL> for offer in decline_offers : <EOL> driver . declineOffer ( offer . id ) <EOL> if not available_offers : <EOL> log . debug ( <EOL> '<STR_LIT>' , <EOL> extra = dict ( mesos_framework_name = self . ns . mesos_framework_name ) ) <EOL> return <EOL> log . debug ( <EOL> '<STR_LIT>' , extra = dict ( <EOL> available_offers = len ( available_offers ) , <EOL> max_runnable_tasks = sum ( x [ <NUM_LIT:1> ] for x in available_offers ) , <EOL> mesos_framework_name = self . ns . mesos_framework_name ) ) <EOL> MV , command = self . _get_and_update_relay ( available_offers ) <EOL> if command is None : <EOL> for offer , _ in available_offers : <EOL> driver . declineOffer ( offer . id ) <EOL> return <EOL> create_tasks ( <EOL> MV = abs ( MV ) , available_offers = available_offers , <EOL> driver = driver , command = command , ns = self . ns <EOL> ) <EOL> driver . reviveOffers ( ) <EOL> def _get_and_update_relay ( self , available_offers ) : <EOL> """<STR_LIT>""" <EOL> command = None <EOL> with self . MV . get_lock ( ) : <EOL> MV , t = self . MV <EOL> if MV == <NUM_LIT:0> : <EOL> log . debug ( <EOL> '<STR_LIT>' , <EOL> extra = dict ( <EOL> mesos_framework_name = self . ns . mesos_framework_name ) ) <EOL> else : <EOL> if MV > <NUM_LIT:0> and self . ns . warmer : <EOL> command = self . ns . warmer <EOL> elif MV < <NUM_LIT:0> and self . ns . cooler : <EOL> command = self . ns . cooler <EOL> if abs ( MV ) < len ( available_offers ) : <EOL> self . MV [ : ] = [ <NUM_LIT:0> , time . time ( ) ] <EOL> else : <EOL> new_MV = MV - ( MV > <NUM_LIT:0> or - <NUM_LIT:1> ) * max ( abs ( MV ) , <EOL> len ( available_offers ) ) <EOL> self . MV [ : ] = [ new_MV , time . time ( ) ] <EOL> return ( MV , command ) <EOL> def statusUpdate ( self , driver , update ) : <EOL> catch ( self . _statusUpdate , self . exception_sender ) ( driver , update ) <EOL> def _statusUpdate ( self , driver , update ) : <EOL> log . debug ( '<STR_LIT>' % str ( update . message ) , extra = dict ( <EOL> task_id = update . task_id . value , task_state = update . state , <EOL> slave_id = update . slave_id . value , timestamp = update . timestamp , <EOL> mesos_framework_name = self . ns . mesos_framework_name ) ) <EOL> if self . ns . max_failures == - <NUM_LIT:1> : <EOL> return <EOL> m = mesos_pb2 <EOL> if update . state in [ m . TASK_FAILED , m . TASK_LOST ] : <EOL> self . failures += <NUM_LIT:1> <EOL> elif update . state in [ m . TASK_FINISHED , m . TASK_STARTING ] : <EOL> self . failures = max ( self . failures - <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> if self . failures >= self . ns . max_failures : <EOL> log . error ( <EOL> "<STR_LIT>" , extra = dict ( <EOL> max_failures = self . failures , <EOL> mesos_framework_name = self . ns . mesos_framework_name ) ) <EOL> driver . stop ( ) <EOL> raise MaxFailuresReached ( self . failures ) <EOL> def frameworkMessage ( self , driver , executorId , slaveId , message ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def offerRescinded ( self , driver , offerId ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( '<STR_LIT>' , extra = dict ( <EOL> offer_id = offerId . value , <EOL> mesos_framework_name = self . ns . mesos_framework_name ) ) </s>
