<s> import boto <EOL> import os <EOL> import sys <EOL> import tempfile <EOL> import time <EOL> import subprocess <EOL> import shutil <EOL> import random <EOL> from optparse import OptionParser <EOL> def parse_args ( force_action = True ) : <EOL> parser = OptionParser ( usage = "<STR_LIT>" + <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" + <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:float>" , default = <NUM_LIT:1> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:1> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:1> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:0> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:1> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:0> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT:-c>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:1> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:100> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:float>" , default = <NUM_LIT:2> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = int , default = <NUM_LIT:2> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:1> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:100> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:string>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , type = "<STR_LIT:float>" , default = None , <EOL> help = "<STR_LIT>" + <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:4> , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , type = "<STR_LIT:string>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:8> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:1> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , type = "<STR_LIT:int>" , default = <NUM_LIT:5> , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) , <EOL> parser . add_option ( "<STR_LIT>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> ( opts , args ) = parser . parse_args ( ) <EOL> if len ( args ) < <NUM_LIT:2> and force_action : <EOL> parser . print_help ( ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> if os . getenv ( '<STR_LIT>' ) == None : <EOL> print >> sys . stderr , ( "<STR_LIT>" + <EOL> "<STR_LIT>" ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> if os . getenv ( '<STR_LIT>' ) == None : <EOL> print >> sys . stderr , ( "<STR_LIT>" + <EOL> "<STR_LIT>" ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> return ( opts , args ) <EOL> def get_or_make_group ( conn , name ) : <EOL> groups = conn . get_all_security_groups ( ) <EOL> group = [ g for g in groups if g . name == name ] <EOL> if len ( group ) > <NUM_LIT:0> : <EOL> return group [ <NUM_LIT:0> ] <EOL> else : <EOL> print "<STR_LIT>" + name <EOL> return conn . create_security_group ( name , "<STR_LIT>" ) <EOL> def scp ( host , opts , local_file , dest_file ) : <EOL> subprocess . check_call ( <EOL> "<STR_LIT>" % <EOL> ( opts . identity_file , local_file , host , dest_file ) , shell = True ) <EOL> def scp_from ( host , opts , dest_file , local_file ) : <EOL> subprocess . check_call ( <EOL> "<STR_LIT>" % <EOL> ( opts . identity_file , host , dest_file , local_file ) , shell = True ) <EOL> def rsync_from_all ( hosts , opts , dest_pattern , local_dir , errors = <NUM_LIT:0> ) : <EOL> commands = [ ] <EOL> for host in hosts : <EOL> cmd = "<STR_LIT>" % ( <EOL> opts . identity_file , host , dest_pattern , local_dir ) <EOL> commands . append ( cmd ) <EOL> parallel_commands ( commands , errors ) <EOL> def parallel_commands ( commands , tolerable_failures ) : <EOL> processes = { } <EOL> failures = [ ] <EOL> for c in commands : <EOL> p = subprocess . Popen ( c , shell = True , stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , stdin = subprocess . PIPE ) <EOL> processes [ p ] = c <EOL> for p in processes . keys ( ) : <EOL> ( stdout , stderr ) = p . communicate ( ) <EOL> if p . poll ( ) != <NUM_LIT:0> : <EOL> failures . append ( ( stdout , stderr , processes [ p ] ) ) <EOL> print stdout <EOL> if len ( failures ) > tolerable_failures : <EOL> out = "<STR_LIT>" <EOL> for ( stdout , stderr , cmd ) in failures : <EOL> out = out + "<STR_LIT>" % ( cmd , stdout , stderr ) <EOL> raise Exception ( out ) <EOL> def ssh ( host , opts , command ) : <EOL> subprocess . check_call ( <EOL> "<STR_LIT>" % <EOL> ( opts . identity_file , host , command ) , shell = True ) <EOL> def ssh_all ( hosts , opts , command ) : <EOL> commands = [ ] <EOL> for host in hosts : <EOL> cmd = "<STR_LIT>" % ( opts . identity_file , host , command ) <EOL> commands . append ( cmd ) <EOL> parallel_commands ( commands , <NUM_LIT:0> ) <EOL> def launch_cluster ( conn , opts , cluster_name ) : <EOL> backend_group = get_or_make_group ( conn , "<STR_LIT>" % cluster_name ) <EOL> frontend_group = get_or_make_group ( conn , "<STR_LIT>" % cluster_name ) <EOL> groups = [ backend_group , frontend_group ] <EOL> for group in groups : <EOL> if group . rules == [ ] : <EOL> for group2 in groups : <EOL> group . authorize ( src_group = group2 ) <EOL> group . authorize ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> , '<STR_LIT>' ) <EOL> print "<STR_LIT>" <EOL> try : <EOL> image = conn . get_all_images ( image_ids = [ opts . ami ] ) [ <NUM_LIT:0> ] <EOL> except : <EOL> print >> sys . stderr , "<STR_LIT>" + opts . ami <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> if opts . spot_price != None : <EOL> launch_group_name = "<STR_LIT>" % cluster_name <EOL> req_ids = [ ] <EOL> if opts . frontends > <NUM_LIT:0> : <EOL> print ( "<STR_LIT>" % <EOL> ( opts . frontends , opts . spot_price ) ) <EOL> frontend_reqs = conn . request_spot_instances ( <EOL> price = opts . spot_price , <EOL> image_id = opts . ami , <EOL> launch_group = launch_group_name , <EOL> placement = opts . zone , <EOL> count = opts . frontends , <EOL> key_name = opts . key_pair , <EOL> security_groups = [ frontend_group ] , <EOL> instance_type = opts . instance_type ) <EOL> req_ids += [ req . id for req in frontend_reqs ] <EOL> if opts . backends > <NUM_LIT:0> : <EOL> print ( "<STR_LIT>" % <EOL> ( opts . backends , opts . spot_price ) ) <EOL> backend_reqs = conn . request_spot_instances ( <EOL> price = opts . spot_price , <EOL> image_id = opts . ami , <EOL> launch_group = launch_group_name , <EOL> placement = opts . zone , <EOL> count = opts . backends , <EOL> key_name = opts . key_pair , <EOL> security_groups = [ backend_group ] , <EOL> instance_type = opts . instance_type ) <EOL> req_ids += [ req . id for req in backend_reqs ] <EOL> print "<STR_LIT>" <EOL> instances_requested = opts . frontends + opts . backends <EOL> try : <EOL> while True : <EOL> time . sleep ( <NUM_LIT:10> ) <EOL> reqs = conn . get_all_spot_instance_requests ( ) <EOL> active_instance_ids = [ r . instance_id for r in reqs <EOL> if r . id in req_ids and r . state == "<STR_LIT>" ] <EOL> if len ( active_instance_ids ) == instances_requested : <EOL> print "<STR_LIT>" % ( opts . frontends , opts . backends ) <EOL> break <EOL> else : <EOL> print ( "<STR_LIT>" % <EOL> ( len ( active_instance_ids ) , instances_requested ) ) <EOL> except : <EOL> print "<STR_LIT>" <EOL> conn . cancel_spot_instance_requests ( req_ids ) <EOL> ( frontends , backends ) = find_existing_cluster ( conn , opts , cluster_name ) <EOL> running = len ( frontends ) + len ( backends ) <EOL> if running : <EOL> print >> sys . stderr , ( "<STR_LIT>" % running ) <EOL> sys . exit ( <NUM_LIT:0> ) <EOL> else : <EOL> print "<STR_LIT>" <EOL> frontend_res = image . run ( key_name = opts . key_pair , <EOL> security_groups = [ frontend_group ] , <EOL> instance_type = opts . instance_type , <EOL> placement = opts . zone , <EOL> min_count = opts . frontends , <EOL> max_count = opts . frontends ) <EOL> backend_res = image . run ( key_name = opts . key_pair , <EOL> security_groups = [ backend_group ] , <EOL> instance_type = opts . instance_type , <EOL> placement = opts . zone , <EOL> min_count = opts . backends , <EOL> max_count = opts . backends ) <EOL> print "<STR_LIT>" % ( <EOL> opts . frontends , opts . backends ) <EOL> def wait_for_instances ( instances ) : <EOL> while True : <EOL> for i in instances : <EOL> i . update ( ) <EOL> if len ( [ i for i in instances if i . state == '<STR_LIT>' ] ) > <NUM_LIT:0> : <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> else : <EOL> return <EOL> def is_active ( instance ) : <EOL> return ( instance . state in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def find_existing_cluster ( conn , opts , cluster_name ) : <EOL> print "<STR_LIT>" <EOL> reservations = conn . get_all_instances ( ) <EOL> frontend_nodes = [ ] <EOL> backend_nodes = [ ] <EOL> for res in reservations : <EOL> active = [ i for i in res . instances if is_active ( i ) ] <EOL> if len ( active ) > <NUM_LIT:0> : <EOL> group_names = [ g . name for g in res . groups ] <EOL> if group_names == [ "<STR_LIT>" % cluster_name ] : <EOL> frontend_nodes += res . instances <EOL> elif group_names == [ "<STR_LIT>" % cluster_name ] : <EOL> backend_nodes += res . instances <EOL> if frontend_nodes != [ ] and backend_nodes != [ ] : <EOL> print ( "<STR_LIT>" % <EOL> ( len ( frontend_nodes ) , len ( backend_nodes ) ) ) <EOL> print "<STR_LIT>" <EOL> frontend_nodes = filter ( lambda k : k . public_dns_name != "<STR_LIT>" , frontend_nodes ) <EOL> for fe in frontend_nodes : <EOL> print fe . public_dns_name <EOL> print "<STR_LIT>" <EOL> backend_nodes = filter ( lambda k : k . public_dns_name != "<STR_LIT>" , backend_nodes ) <EOL> for be in backend_nodes : <EOL> print be . public_dns_name <EOL> return ( frontend_nodes , backend_nodes ) <EOL> else : <EOL> print "<STR_LIT>" % ( <EOL> cluster_name , frontend_nodes , backend_nodes ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> """<STR_LIT>""" <EOL> def generate_deploy_files ( frontends , backends , opts , warmup_job_arrival_s , warmup_s , post_warmup_s , <EOL> nm_task_scheduler , users ) : <EOL> tmp_dir = tempfile . mkdtemp ( ) <EOL> template_vars = { <EOL> "<STR_LIT>" : "<STR_LIT:\n>" . join ( [ "<STR_LIT:%s>" % i . public_dns_name for i in frontends ] ) , "<STR_LIT>" : "<STR_LIT:\n>" . join ( [ "<STR_LIT>" % i . private_ip_address for i in frontends ] ) , "<STR_LIT>" : "<STR_LIT:U+002C>" . join ( [ "<STR_LIT>" % i . public_dns_name for i in backends ] ) , <EOL> "<STR_LIT>" : frontends [ <NUM_LIT:0> ] . public_dns_name , <EOL> "<STR_LIT>" : "<STR_LIT:\n>" . join ( [ "<STR_LIT:%s>" % i . public_dns_name for i in backends ] ) , <EOL> "<STR_LIT>" : "<STR_LIT:U+002C>" . join ( [ "<STR_LIT:%s>" % i . public_dns_name for i in backends ] ) , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . arrival_rate , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . branch , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . spark_branch , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . benchmark_iterations , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . benchmark_id , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . tasks_per_job , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . num_preferred_nodes , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . sample_ratio , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . sample_ratio_constrained , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % warmup_job_arrival_s , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % warmup_s , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % post_warmup_s , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % nm_task_scheduler , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . num_partitions , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % ( opts . reduce_tasks ) , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . inter_query_delay , <EOL> "<STR_LIT>" : users , <EOL> "<STR_LIT>" : opts . frontend_type , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . cpus , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . total_cores , <EOL> "<STR_LIT>" : "<STR_LIT:%s>" % opts . spark_backend_mem <EOL> } <EOL> for dirpath , dirnames , filenames in os . walk ( "<STR_LIT>" ) : <EOL> rel_dir_path = dirpath . replace ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if rel_dir_path . startswith ( os . sep ) : <EOL> rel_dir_path = rel_dir_path [ <NUM_LIT:1> : ] <EOL> if rel_dir_path != "<STR_LIT>" : <EOL> os . mkdir ( os . path . join ( tmp_dir , rel_dir_path ) ) <EOL> for filename in filenames : <EOL> if filename [ <NUM_LIT:0> ] not in '<STR_LIT>' and filename [ - <NUM_LIT:1> ] != '<STR_LIT>' : <EOL> local_file = os . path . join ( tmp_dir , rel_dir_path , filename ) <EOL> with open ( os . path . join ( dirpath , filename ) ) as src : <EOL> with open ( local_file , "<STR_LIT:w>" ) as dest : <EOL> text = src . read ( ) <EOL> for key in template_vars : <EOL> text = text . replace ( "<STR_LIT>" + key + "<STR_LIT>" , template_vars [ key ] ) <EOL> dest . write ( text ) <EOL> dest . close ( ) <EOL> return tmp_dir <EOL> def redeploy_sparrow ( machines , frontends , backends , opts , warmup_job_arrival_s = <NUM_LIT:0> , warmup_s = <NUM_LIT:0> , <EOL> post_warmup_s = <NUM_LIT:0> , nm_task_scheduler = "<STR_LIT>" , <EOL> users = "<STR_LIT>" ) : <EOL> tmp_dir = generate_deploy_files ( frontends , backends , opts , warmup_job_arrival_s , warmup_s , <EOL> post_warmup_s , nm_task_scheduler , users ) <EOL> for machine in machines : <EOL> print "<STR_LIT>" % machine . public_dns_name <EOL> command = ( ( "<STR_LIT>" + <EOL> "<STR_LIT>" ) % ( opts . identity_file , tmp_dir , machine . public_dns_name ) ) <EOL> subprocess . check_call ( command , shell = True ) <EOL> def deploy_cluster ( frontends , backends , opts , warmup_job_arrival_s = <NUM_LIT:0> , warmup_s = <NUM_LIT:0> , <EOL> post_warmup_s = <NUM_LIT:0> , nm_task_scheduler = "<STR_LIT>" , <EOL> users = "<STR_LIT>" ) : <EOL> tmp_dir = generate_deploy_files ( frontends , backends , opts , warmup_job_arrival_s , warmup_s , <EOL> post_warmup_s , nm_task_scheduler , users ) <EOL> driver_machine = frontends [ <NUM_LIT:0> ] . public_dns_name <EOL> print "<STR_LIT>" % driver_machine <EOL> command = ( ( "<STR_LIT>" + <EOL> "<STR_LIT>" ) % ( opts . identity_file , tmp_dir , driver_machine ) ) <EOL> subprocess . check_call ( command , shell = True ) <EOL> shutil . rmtree ( tmp_dir ) <EOL> print "<STR_LIT>" % opts . identity_file <EOL> ssh ( driver_machine , opts , '<STR_LIT>' ) <EOL> scp ( driver_machine , opts , opts . identity_file , '<STR_LIT>' ) <EOL> print "<STR_LIT>" <EOL> ssh ( driver_machine , opts , "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> print "<STR_LIT>" <EOL> ssh ( driver_machine , opts , "<STR_LIT>" ) <EOL> def start_sparrow_throughput ( frontends , backends , opts ) : <EOL> start_sparrow ( frontends , backends , opts ) <EOL> start_spark_shark ( frontends , backends , opts ) <EOL> ssh_all ( [ fe . public_dns_name for fe in frontends ] , opts , "<STR_LIT>" ) <EOL> def start_sparrow ( frontends , backends , opts ) : <EOL> all_machines = [ ] <EOL> for fe in frontends : <EOL> all_machines . append ( fe . public_dns_name ) <EOL> for be in backends : <EOL> all_machines . append ( be . public_dns_name ) <EOL> print "<STR_LIT>" <EOL> ssh_all ( all_machines , opts , "<STR_LIT>" ) <EOL> def stop_sparrow ( frontends , backends , opts ) : <EOL> all_machines = [ ] <EOL> for fe in frontends : <EOL> all_machines . append ( fe . public_dns_name ) <EOL> for be in backends : <EOL> all_machines . append ( be . public_dns_name ) <EOL> print "<STR_LIT>" <EOL> ssh_all ( all_machines , opts , "<STR_LIT>" ) <EOL> def start_mesos ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh ( frontends [ <NUM_LIT:0> ] . public_dns_name , opts , "<STR_LIT>" ) <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ be . public_dns_name for be in backends ] , <EOL> opts , "<STR_LIT>" ) <EOL> def stop_mesos ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ be . public_dns_name for be in backends ] , <EOL> opts , "<STR_LIT>" ) <EOL> print "<STR_LIT>" <EOL> ssh ( frontends [ <NUM_LIT:0> ] . public_dns_name , opts , "<STR_LIT>" ) <EOL> """<STR_LIT>""" <EOL> def start_spark_shark ( frontends , backends , opts ) : <EOL> if opts . scheduler != "<STR_LIT>" : <EOL> print "<STR_LIT>" <EOL> return <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ be . public_dns_name for be in backends ] , opts , <EOL> "<STR_LIT>" ) <EOL> def stop_spark ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ fe . public_dns_name for fe in frontends ] , opts , <EOL> "<STR_LIT>" ) <EOL> time . sleep ( opts . kill_delay ) <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ be . public_dns_name for be in backends ] , opts , <EOL> "<STR_LIT>" ) <EOL> def start_hdfs ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh ( frontends [ <NUM_LIT:0> ] . public_dns_name , opts , <EOL> '<STR_LIT>' ) <EOL> def stop_hdfs ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh ( frontends [ <NUM_LIT:0> ] . public_dns_name , opts , <EOL> '<STR_LIT>' ) <EOL> def start_proto ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ be . public_dns_name for be in backends ] , opts , <EOL> "<STR_LIT>" ) <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ fe . public_dns_name for fe in frontends ] , opts , <EOL> "<STR_LIT>" ) <EOL> def stop_proto ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ fe . public_dns_name for fe in frontends ] , opts , <EOL> "<STR_LIT>" ) <EOL> time . sleep ( opts . kill_delay ) <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ be . public_dns_name for be in backends ] , opts , <EOL> "<STR_LIT>" ) <EOL> def create_database ( frontends , opts ) : <EOL> scale_factor = opts . scale_factor <EOL> print "<STR_LIT>" + scale_factor <EOL> ssh ( frontends [ <NUM_LIT:0> ] . public_dns_name , opts , <EOL> "<STR_LIT>" % scale_factor ) <EOL> def create_tpch_tables ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh ( frontends [ <NUM_LIT:0> ] . public_dns_name , opts , <EOL> "<STR_LIT>" ) <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ fe . public_dns_name for fe in frontends [ <NUM_LIT:1> : ] ] , opts , <EOL> "<STR_LIT>" ) <EOL> def start_shark_tpch ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ fe . public_dns_name for fe in frontends ] , opts , <EOL> "<STR_LIT>" ) <EOL> def collect_logs ( frontends , backends , opts ) : <EOL> print "<STR_LIT>" <EOL> ssh_all ( [ fe . public_dns_name for fe in frontends ] , opts , <EOL> "<STR_LIT>" ) <EOL> ssh_all ( [ be . public_dns_name for be in backends ] , opts , <EOL> "<STR_LIT>" ) <EOL> print "<STR_LIT>" <EOL> rsync_from_all ( [ fe . public_dns_name for fe in frontends ] , opts , <EOL> "<STR_LIT>" , opts . log_dir , len ( frontends ) ) <EOL> rsync_from_all ( [ be . public_dns_name for be in backends ] , opts , <EOL> "<STR_LIT>" , opts . log_dir , len ( backends ) ) <EOL> ssh_all ( [ fe . public_dns_name for fe in frontends ] , opts , <EOL> "<STR_LIT>" ) <EOL> ssh_all ( [ be . public_dns_name for be in backends ] , opts , <EOL> "<STR_LIT>" ) <EOL> def destroy_cluster ( frontends , backends , opts ) : <EOL> response = raw_input ( "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" ) <EOL> if response == "<STR_LIT:y>" : <EOL> print "<STR_LIT>" <EOL> for fe in frontends : <EOL> fe . terminate ( ) <EOL> print "<STR_LIT>" <EOL> for be in backends : <EOL> be . terminate ( ) <EOL> def execute_command ( frontends , backends , opts , cmd ) : <EOL> ssh_all ( [ fe . public_dns_name for fe in frontends ] , opts , cmd ) <EOL> ssh_all ( [ be . public_dns_name for be in backends ] , opts , cmd ) <EOL> def login_frontend ( frontends , backends , opts ) : <EOL> node = frontends [ <NUM_LIT:0> ] . public_dns_name <EOL> print "<STR_LIT>" + node <EOL> subprocess . check_call ( "<STR_LIT>" % <EOL> ( opts . identity_file , node ) , shell = True ) <EOL> def login_backend ( frontends , backends , opts ) : <EOL> node = backends [ <NUM_LIT:0> ] . public_dns_name <EOL> print "<STR_LIT>" + node <EOL> subprocess . check_call ( "<STR_LIT>" % <EOL> ( opts . identity_file , node ) , shell = True ) <EOL> def main ( ) : <EOL> ( opts , args ) = parse_args ( ) <EOL> conn = boto . connect_ec2 ( ) <EOL> action = args [ <NUM_LIT:0> ] <EOL> cluster = args [ <NUM_LIT:1> ] <EOL> if action == "<STR_LIT>" : <EOL> launch_cluster ( conn , opts , cluster ) <EOL> return <EOL> if action == "<STR_LIT>" and len ( args ) < <NUM_LIT:3> : <EOL> print "<STR_LIT>" <EOL> ( frontends , backends ) = find_existing_cluster ( conn , opts , cluster ) <EOL> print "<STR_LIT>" <EOL> wait_for_instances ( frontends ) <EOL> wait_for_instances ( backends ) <EOL> print "<STR_LIT>" % opts . wait <EOL> time . sleep ( opts . wait ) <EOL> print "<STR_LIT>" % action <EOL> if action == "<STR_LIT>" : <EOL> cmd = "<STR_LIT:U+0020>" . join ( args [ <NUM_LIT:2> : ] ) <EOL> execute_command ( frontends , backends , opts , cmd ) <EOL> elif action == "<STR_LIT>" : <EOL> deploy_cluster ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> start_sparrow ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> stop_sparrow ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> start_sparrow_throughput ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> start_mesos ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> stop_mesos ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> start_spark_shark ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> stop_spark ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> stop_spark ( frontends , backends , opts ) <EOL> stop_sparrow ( frontends , backends , opts ) <EOL> start_sparrow ( frontends , backends , opts ) <EOL> start_spark_shark ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> start_proto ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> stop_proto ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> start_hdfs ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> stop_hdfs ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> collect_logs ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> destroy_cluster ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> login_frontend ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> login_backend ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> create_database ( frontends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> create_tpch_tables ( frontends , backends , opts ) <EOL> elif action == "<STR_LIT>" : <EOL> start_shark_tpch ( frontends , backends , opts ) <EOL> else : <EOL> print "<STR_LIT>" % action <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> main ( ) </s>
