signature:
def scp_package ( package_file , destinations , cl_args ) :
    '''scp and extract package'''

implementation:
pids = [ ]
for dest in destinations :
    if is_self ( dest ) :
        continue
    Log . info ( "Server: %s" % dest )
    file_path = "/tmp/heron.tar.gz"
    dest_file_path = "%s:%s" % ( dest , file_path )
    remote_cmd = "rm -rf ~/.heron && mkdir ~/.heron " "&& tar -xzvf %s -C ~/.heron --strip-components 1" % ( file_path )
    cmd = '%s && %s' % ( scp_cmd ( package_file , dest_file_path , cl_args ) , ssh_remote_execute ( remote_cmd , dest , cl_args ) )
    Log . debug ( cmd )
    pid = subprocess . Popen ( cmd , shell = True , stdout = subprocess . PIPE , stderr = subprocess . PIPE )
    pids . append ( { "pid" : pid , "dest" : dest } )
errors = [ ]
for entry in pids :
    pid = entry [ "pid" ]
    return_code = pid . wait ( )
    output = pid . communicate ( )
    Log . debug ( "return code: %s output: %s" % ( return_code , output ) )
    if return_code != 0 :
        errors . append ( "Failed to scp package to %s with error:\n%s" % ( entry [ "dest" ] , output [ 1 ] ) )
if errors :
    for error in errors :
        Log . error ( error )
    sys . exit ( - 1 )
Log . info ( "Done distributing packages" )


signature:
def make_tarfile ( output_filename , source_dir ) :
    '''Tar a directory'''

implementation:
with tarfile . open ( output_filename , "w:gz" ) as tar :
    tar . add ( source_dir , arcname = os . path . basename ( source_dir ) )


signature:
def start_master_nodes ( masters , cl_args ) :
    '''Start master nodes'''

implementation:
pids = [ ]
for master in masters :
    Log . info ( "Starting master on %s" % master )
    cmd = "%s agent -config %s >> /tmp/nomad_server_log 2>&1 &" % ( get_nomad_path ( cl_args ) , get_nomad_master_config_file ( cl_args ) )
    if not is_self ( master ) :
        cmd = ssh_remote_execute ( cmd , master , cl_args )
    Log . debug ( cmd )
    pid = subprocess . Popen ( cmd , shell = True , stdout = subprocess . PIPE , stderr = subprocess . PIPE )
    pids . append ( { "pid" : pid , "dest" : master } )
errors = [ ]
for entry in pids :
    pid = entry [ "pid" ]
    return_code = pid . wait ( )
    output = pid . communicate ( )
    Log . debug ( "return code: %s output: %s" % ( return_code , output ) )
    if return_code != 0 :
        errors . append ( "Failed to start master on %s with error:\n%s" % ( entry [ "dest" ] , output [ 1 ] ) )
if errors :
    for error in errors :
        Log . error ( error )
    sys . exit ( - 1 )
Log . info ( "Done starting masters" )


signature:
def start_slave_nodes ( slaves , cl_args ) :
    '''Star slave nodes'''

implementation:
pids = [ ]
for slave in slaves :
    Log . info ( "Starting slave on %s" % slave )
    cmd = "%s agent -config %s >> /tmp/nomad_client.log 2>&1 &" % ( get_nomad_path ( cl_args ) , get_nomad_slave_config_file ( cl_args ) )
    if not is_self ( slave ) :
        cmd = ssh_remote_execute ( cmd , slave , cl_args )
    Log . debug ( cmd )
    pid = subprocess . Popen ( cmd , shell = True , stdout = subprocess . PIPE , stderr = subprocess . PIPE )
    pids . append ( { "pid" : pid , "dest" : slave } )
errors = [ ]
for entry in pids :
    pid = entry [ "pid" ]
    return_code = pid . wait ( )
    output = pid . communicate ( )
    Log . debug ( "return code: %s output: %s" % ( return_code , output ) )
    if return_code != 0 :
        errors . append ( "Failed to start slave on %s with error:\n%s" % ( entry [ "dest" ] , output [ 1 ] ) )
if errors :
    for error in errors :
        Log . error ( error )
    sys . exit ( - 1 )
Log . info ( "Done starting slaves" )


signature:
def read_and_parse_roles ( cl_args ) :
    '''read config files to get roles'''

implementation:
roles = dict ( )
with open ( get_inventory_file ( cl_args ) , 'r' ) as stream :
    try :
        roles = yaml . load ( stream )
    except yaml . YAMLError as exc :
        Log . error ( "Error parsing inventory file: %s" % exc )
        sys . exit ( - 1 )
if Role . ZOOKEEPERS not in roles or not roles [ Role . ZOOKEEPERS ] :
    Log . error ( "Zookeeper servers node defined!" )
    sys . exit ( - 1 )
if Role . CLUSTER not in roles or not roles [ Role . CLUSTER ] :
    Log . error ( "Heron cluster nodes defined!" )
    sys . exit ( - 1 )
# Set roles
roles [ Role . MASTERS ] = set ( [ roles [ Role . CLUSTER ] [ 0 ] ] )
roles [ Role . SLAVES ] = set ( roles [ Role . CLUSTER ] )
roles [ Role . ZOOKEEPERS ] = set ( roles [ Role . ZOOKEEPERS ] )
roles [ Role . CLUSTER ] = set ( roles [ Role . CLUSTER ] )
return roles


signature:
def read_file ( file_path ) :
    '''read file'''

implementation:
lines = [ ]
with open ( file_path , "r" ) as tf :
    lines = [ line . strip ( "\n" ) for line in tf . readlines ( ) if not line . startswith ( "#" ) ]
    # filter empty lines
    lines = [ line for line in lines if line ]
return lines


signature:
def call_editor ( file_path ) :
    '''call editor'''

implementation:
EDITOR = os . environ . get ( 'EDITOR' , 'vim' )
with open ( file_path , 'r+' ) as tf :
    call ( [ EDITOR , tf . name ] )


signature:
def get_remote_home ( host , cl_args ) :
    '''get home directory of remote host'''

implementation:
cmd = "echo ~"
if not is_self ( host ) :
    cmd = ssh_remote_execute ( cmd , host , cl_args )
pid = subprocess . Popen ( cmd , shell = True , stdout = subprocess . PIPE , stderr = subprocess . PIPE )
return_code = pid . wait ( )
output = pid . communicate ( )
if return_code != 0 :
    Log . error ( "Failed to get home path for remote host %s with output:\n%s" % ( host , output ) )
    sys . exit ( - 1 )
return output [ 0 ] . strip ( "\n" )


signature:
def get_hostname ( ip_addr , cl_args ) :
    '''get host name of remote host'''

implementation:
if is_self ( ip_addr ) :
    return get_self_hostname ( )
cmd = "hostname"
ssh_cmd = ssh_remote_execute ( cmd , ip_addr , cl_args )
pid = subprocess . Popen ( ssh_cmd , shell = True , stdout = subprocess . PIPE , stderr = subprocess . PIPE )
return_code = pid . wait ( )
output = pid . communicate ( )
if return_code != 0 :
    Log . error ( "Failed to get hostname for remote host %s with output:\n%s" % ( ip_addr , output ) )
    sys . exit ( - 1 )
return output [ 0 ] . strip ( "\n" )


signature:
def is_self ( addr ) :
    '''check if this host is this addr'''

implementation:
ips = [ ]
for i in netifaces . interfaces ( ) :
    entry = netifaces . ifaddresses ( i )
    if netifaces . AF_INET in entry :
        for ipv4 in entry [ netifaces . AF_INET ] :
            if "addr" in ipv4 :
                ips . append ( ipv4 [ "addr" ] )
return addr in ips or addr == get_self_hostname ( )


signature:
def log ( self , message , level = None ) :
    """Log message , optionally providing a logging level
 It is compatible with StreamParse API .
 : type message : str
 : param message : the log message to send
 : type level : str
 : param level : the logging level ,
 one of : trace ( = debug ) , debug , info , warn or error ( default : info )"""

implementation:
if level is None :
    _log_level = logging . INFO
else :
    if level == "trace" or level == "debug" :
        _log_level = logging . DEBUG
    elif level == "info" :
        _log_level = logging . INFO
    elif level == "warn" :
        _log_level = logging . WARNING
    elif level == "error" :
        _log_level = logging . ERROR
    else :
        raise ValueError ( "%s is not supported as logging level" % str ( level ) )
self . logger . log ( _log_level , message )


signature:
def load_py_instance ( self , is_spout ) :
    """Loads user defined component ( spout / bolt )"""

implementation:
try :
    if is_spout :
        spout_proto = self . pplan_helper . get_my_spout ( )
        py_classpath = spout_proto . comp . class_name
        self . logger . info ( "Loading Spout from: %s" , py_classpath )
    else :
        bolt_proto = self . pplan_helper . get_my_bolt ( )
        py_classpath = bolt_proto . comp . class_name
        self . logger . info ( "Loading Bolt from: %s" , py_classpath )
    pex_loader . load_pex ( self . pplan_helper . topology_pex_abs_path )
    spbl_class = pex_loader . import_and_get_class ( self . pplan_helper . topology_pex_abs_path , py_classpath )
except Exception as e :
    spbl = "spout" if is_spout else "bolt"
    self . logger . error ( traceback . format_exc ( ) )
    raise RuntimeError ( "Error when loading a %s from pex: %s" % ( spbl , str ( e ) ) )
return spbl_class


signature:
def get ( self ) :
    """get method"""

implementation:
# Get all the values for parameter " cluster " .
clusters = self . get_arguments ( constants . PARAM_CLUSTER )
# Get all the values for parameter " environ " .
environs = self . get_arguments ( constants . PARAM_ENVIRON )
# Get role
role = self . get_argument_role ( )
ret = { }
topologies = self . tracker . topologies
for topology in topologies :
    cluster = topology . cluster
    environ = topology . environ
    execution_state = topology . execution_state
    if not cluster or not execution_state or not environ :
        continue
    topo_role = execution_state . role
    if not topo_role :
        continue
    # This cluster is not asked for .
    # Note that " if not clusters " , then
    # we show for all the clusters .
    if clusters and cluster not in clusters :
        continue
    # This environ is not asked for .
    # Note that " if not environs " , then
    # we show for all the environs .
    if environs and environ not in environs :
        continue
    # This role is not asked for .
    # Note that " if not role " , then
    # we show for all the roles .
    if role and role != topo_role :
        continue
    if cluster not in ret :
        ret [ cluster ] = { }
    if topo_role not in ret [ cluster ] :
        ret [ cluster ] [ topo_role ] = { }
    if environ not in ret [ cluster ] [ topo_role ] :
        ret [ cluster ] [ topo_role ] [ environ ] = [ ]
    ret [ cluster ] [ topo_role ] [ environ ] . append ( topology . name )
self . write_success_response ( ret )


signature:
def dereference_symlinks ( src ) :
    """Resolve all symbolic references that ` src ` points to . Note that this
 is different than ` os . path . realpath ` as path components leading up to
 the final location may still be symbolic links ."""

implementation:
while os . path . islink ( src ) :
    src = os . path . join ( os . path . dirname ( src ) , os . readlink ( src ) )
return src


implementation:
clusters = [ statemgr . name for statemgr in self . tracker . state_managers ]
self . write_success_response ( clusters )


signature:
def get_cluster_role_env_topologies ( cluster , role , env ) :
    '''Get the list of topologies given a cluster submitted by a given role under a given environment
 : param cluster :
 : param role :
 : param env :
 : return :'''

implementation: return _get_topologies ( cluster , role = role , env = env )


signature:
def get_execution_state ( cluster , environ , topology , role = None ) :
    '''Get the execution state of a topology in a cluster
 : param cluster :
 : param environ :
 : param topology :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( EXECUTION_STATE_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_logical_plan ( cluster , environ , topology , role = None ) :
    '''Get the logical plan state of a topology in a cluster
 : param cluster :
 : param environ :
 : param topology :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( LOGICALPLAN_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_comps ( cluster , environ , topology , role = None ) :
    '''Get the list of component names for the topology from Heron Nest
 : param cluster :
 : param environ :
 : param topology :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( LOGICALPLAN_URL_FMT ) , params )
lplan = yield fetch_url_as_json ( request_url )
comps = lplan [ 'spouts' ] . keys ( ) + lplan [ 'bolts' ] . keys ( )
raise tornado . gen . Return ( comps )


signature:
def get_instances ( cluster , environ , topology , role = None ) :
    '''Get the list of instances for the topology from Heron Nest
 : param cluster :
 : param environ :
 : param topology :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( PHYSICALPLAN_URL_FMT ) , params )
pplan = yield fetch_url_as_json ( request_url )
instances = pplan [ 'instances' ] . keys ( )
raise tornado . gen . Return ( instances )


signature:
def get_physical_plan ( cluster , environ , topology , role = None ) :
    '''Get the physical plan state of a topology in a cluster from tracker
 : param cluster :
 : param environ :
 : param topology :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( PHYSICALPLAN_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_scheduler_location ( cluster , environ , topology , role = None ) :
    '''Get the scheduler location of a topology in a cluster from tracker
 : param cluster :
 : param environ :
 : param topology :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( SCHEDULER_LOCATION_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_component_exceptionsummary ( cluster , environ , topology , component , role = None ) :
    '''Get summary of exception for a component
 : param cluster :
 : param environ :
 : param topology :
 : param component :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , component = component )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( EXCEPTION_SUMMARY_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_component_exceptions ( cluster , environ , topology , component , role = None ) :
    '''Get exceptions for ' component ' for ' topology '
 : param cluster :
 : param environ :
 : param topology :
 : param component :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , component = component )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( EXCEPTIONS_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_comp_instance_metrics ( cluster , environ , topology , component , metrics , instances , time_range , role = None ) :
    '''Get the metrics for some instances of a topology from tracker
 : param cluster :
 : param environ :
 : param topology :
 : param component :
 : param metrics : dict of display name to cuckoo name
 : param instances :
 : param time _ range : 2 - tuple consisting of start and end of range
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , component = component )
if role is not None :
    params [ 'role' ] = role
# form the fetch url
request_url = tornado . httputil . url_concat ( create_url ( METRICS_URL_FMT ) , params )
# convert a single instance to a list , if needed
all_instances = instances if isinstance ( instances , list ) else [ instances ]
# append each metric to the url
for _ , metric_name in metrics . items ( ) :
    request_url = tornado . httputil . url_concat ( request_url , dict ( metricname = metric_name [ 0 ] ) )
# append each instance to the url
for i in all_instances :
    request_url = tornado . httputil . url_concat ( request_url , dict ( instance = i ) )
# append the time interval to the url
request_url = tornado . httputil . url_concat ( request_url , dict ( interval = time_range [ 1 ] ) )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_comp_metrics ( cluster , environ , topology , component , instances , metricnames , time_range , role = None ) :
    '''Get the metrics for all the instances of a topology from Heron Nest
 : param cluster :
 : param environ :
 : param topology :
 : param component :
 : param instances :
 : param metricnames : dict of display name to cuckoo name
 : param time _ range : 2 - tuple consisting of start and end of range
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , component = component )
if role is not None :
    params [ 'role' ] = role
# form the url
request_url = tornado . httputil . url_concat ( create_url ( METRICS_URL_FMT ) , params )
# append each metric to the url
for metric_name in metricnames :
    request_url = tornado . httputil . url_concat ( request_url , dict ( metricname = metric_name ) )
# append each instance to the url
for instance in instances :
    request_url = tornado . httputil . url_concat ( request_url , dict ( instance = instance ) )
# append the time interval to the url
request_url = tornado . httputil . url_concat ( request_url , dict ( interval = time_range [ 1 ] ) )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_metrics ( cluster , environment , topology , timerange , query , role = None ) :
    '''Get the metrics for a topology from tracker
 : param cluster :
 : param environment :
 : param topology :
 : param timerange :
 : param query :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environment , topology = topology , starttime = timerange [ 0 ] , endtime = timerange [ 1 ] , query = query )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( METRICS_QUERY_URL_FMT ) , params )
logging . info ( "get_metrics %s" , request_url )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_comp_metrics_timeline ( cluster , environ , topology , component , instances , metricnames , time_range , role = None ) :
    '''Get the minute - by - minute metrics for all instances of a topology from tracker
 : param cluster :
 : param environ :
 : param topology :
 : param component :
 : param instances :
 : param metricnames : dict of display name to cuckoo name
 : param time _ range : 2 - tuple consisting of start and end of range
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , component = component )
if role is not None :
    params [ 'role' ] = role
# form the url
request_url = tornado . httputil . url_concat ( create_url ( METRICS_TIMELINE_URL_FMT ) , params )
if role is not None :
    request_url = tornado . httputil . url_concat ( request_url , dict ( role = role ) )
# append each metric to the url
for metric_name in metricnames :
    request_url = tornado . httputil . url_concat ( request_url , dict ( metricname = metric_name ) )
# append each instance to the url
for instance in instances :
    request_url = tornado . httputil . url_concat ( request_url , dict ( instance = instance ) )
# append the time interval to the url
request_url = tornado . httputil . url_concat ( request_url , dict ( starttime = time_range [ 0 ] , endtime = time_range [ 1 ] ) )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_topology_info ( cluster , environ , topology , role = None ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( INFO_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_instance_pid ( cluster , environ , topology , instance , role = None ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param instance :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , instance = instance )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( PID_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_instance_jstack ( cluster , environ , topology , instance , role = None ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param instance :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , instance = instance )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( JSTACK_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_instance_mem_histogram ( cluster , environ , topology , instance , role = None ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param instance :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , instance = instance )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( HISTOGRAM_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def run_instance_jmap ( cluster , environ , topology , instance , role = None ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param instance :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , instance = instance )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( JMAP_URL_FMT ) , params )
if role is not None :
    request_url = tornado . httputil . url_concat ( request_url , dict ( role = role ) )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_container_file_download_url ( cluster , environ , topology , container , path , role = None ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param container :
 : param path :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , container = container , path = path )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( FILE_DOWNLOAD_URL_FMT ) , params )
if role is not None :
    request_url = tornado . httputil . url_concat ( request_url , dict ( role = role ) )
return request_url


signature:
def get_container_file_data ( cluster , environ , topology , container , path , offset , length , role = None ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param container :
 : param path :
 : param offset :
 : param length :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , container = container , path = path , offset = offset , length = length )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( FILE_DATA_URL_FMT ) , params )
if role is not None :
    request_url = tornado . httputil . url_concat ( request_url , dict ( role = role ) )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def get_filestats ( cluster , environ , topology , container , path , role = None ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param container :
 : param path :
 : param role :
 : return :'''

implementation:
params = dict ( cluster = cluster , environ = environ , topology = topology , container = container , path = path )
if role is not None :
    params [ 'role' ] = role
request_url = tornado . httputil . url_concat ( create_url ( FILESTATS_URL_FMT ) , params )
raise tornado . gen . Return ( ( yield fetch_url_as_json ( request_url ) ) )


signature:
def fetch ( self , cluster , metric , topology , component , instance , timerange , environ = None ) :
    ''': param cluster :
 : param metric :
 : param topology :
 : param component :
 : param instance :
 : param timerange :
 : param environ :
 : return :'''

implementation:
components = [ component ] if component != "*" else ( yield get_comps ( cluster , environ , topology ) )
futures = [ ]
for comp in components :
    query = self . get_query ( metric , comp , instance )
    future = get_metrics ( cluster , environ , topology , timerange , query )
    futures . append ( future )
results = yield futures
timelines = [ ]
for result in results :
    timelines . extend ( result [ "timeline" ] )
result = self . get_metric_response ( timerange , timelines , False )
raise tornado . gen . Return ( result )


signature:
def fetch_max ( self , cluster , metric , topology , component , instance , timerange , environ = None ) :
    ''': param cluster :
 : param metric :
 : param topology :
 : param component :
 : param instance :
 : param timerange :
 : param environ :
 : return :'''

implementation:
components = [ component ] if component != "*" else ( yield get_comps ( cluster , environ , topology ) )
result = { }
futures = [ ]
for comp in components :
    query = self . get_query ( metric , comp , instance )
    max_query = "MAX(%s)" % query
    future = get_metrics ( cluster , environ , topology , timerange , max_query )
    futures . append ( future )
results = yield futures
data = self . compute_max ( results )
result = self . get_metric_response ( timerange , data , True )
raise tornado . gen . Return ( result )


signature:
def fetch_backpressure ( self , cluster , metric , topology , component , instance , timerange , is_max , environ = None ) :
    ''': param cluster :
 : param metric :
 : param topology :
 : param component :
 : param instance :
 : param timerange :
 : param isMax :
 : param environ :
 : return :'''

implementation:
instances = yield get_instances ( cluster , environ , topology )
if component != "*" :
    filtered_inst = [ instance for instance in instances if instance . split ( "_" ) [ 2 ] == component ]
else :
    filtered_inst = instances
futures_dict = { }
for inst in filtered_inst :
    query = queries . get ( metric ) . format ( inst )
    futures_dict [ inst ] = get_metrics ( cluster , environ , topology , timerange , query )
res = yield futures_dict
if not is_max :
    timelines = [ ]
    for key in res :
        result = res [ key ]
        # Replacing stream manager instance name with component instance name
        if len ( result [ "timeline" ] ) > 0 :
            result [ "timeline" ] [ 0 ] [ "instance" ] = key
        timelines . extend ( result [ "timeline" ] )
    result = self . get_metric_response ( timerange , timelines , is_max )
else :
    data = self . compute_max ( res . values ( ) )
    result = self . get_metric_response ( timerange , data , is_max )
raise tornado . gen . Return ( result )


signature:
def compute_max ( self , multi_ts ) :
    ''': param multi _ ts :
 : return :'''

implementation:
if len ( multi_ts ) > 0 and len ( multi_ts [ 0 ] [ "timeline" ] ) > 0 :
    keys = multi_ts [ 0 ] [ "timeline" ] [ 0 ] [ "data" ] . keys ( )
    timelines = ( [ res [ "timeline" ] [ 0 ] [ "data" ] [ key ] for key in keys ] for res in multi_ts )
    values = ( max ( v ) for v in zip ( * timelines ) )
    return dict ( zip ( keys , values ) )
return { }


signature:
def get_metric_response ( self , timerange , data , isMax ) :
    ''': param timerange :
 : param data :
 : param isMax :
 : return :'''

implementation:
if isMax :
    return dict ( status = "success" , starttime = timerange [ 0 ] , endtime = timerange [ 1 ] , result = dict ( timeline = [ dict ( data = data ) ] ) )
return dict ( status = "success" , starttime = timerange [ 0 ] , endtime = timerange [ 1 ] , result = dict ( timeline = data ) )


signature:
def get_query ( self , metric , component , instance ) :
    ''': param metric :
 : param component :
 : param instance :
 : return :'''

implementation:
q = queries . get ( metric )
return q . format ( component , instance )


signature:
def to_table ( result ) :
    '''normalize raw result to table'''

implementation:
max_count = 20
table , count = [ ] , 0
for role , envs_topos in result . items ( ) :
    for env , topos in envs_topos . items ( ) :
        for topo in topos :
            count += 1
            if count > max_count :
                continue
            else :
                table . append ( [ role , env , topo ] )
header = [ 'role' , 'env' , 'topology' ]
rest_count = 0 if count <= max_count else count - max_count
return table , header , rest_count


signature:
def show_cluster ( cl_args , cluster ) :
    '''print topologies information to stdout'''

implementation:
try :
    result = tracker_access . get_cluster_topologies ( cluster )
    if not result :
        Log . error ( 'No topologies in cluster \'%s\'' % cluster )
        return False
    result = result [ cluster ]
except Exception :
    Log . error ( "Fail to connect to tracker: \'%s\'" , cl_args [ "tracker_url" ] )
    return False
table , header , rest_count = to_table ( result )
print ( 'Topologies running in cluster \'%s\'' % cluster )
if rest_count :
    print ( '  with %d more...' % rest_count )
print ( tabulate ( table , headers = header ) )
return True


signature:
def show_cluster_role ( cl_args , cluster , role ) :
    '''print topologies information to stdout'''

implementation:
try :
    result = tracker_access . get_cluster_role_topologies ( cluster , role )
    if not result :
        Log . error ( 'Unknown cluster/role \'%s\'' % '/' . join ( [ cluster , role ] ) )
        return False
    result = result [ cluster ]
except Exception :
    Log . error ( "Fail to connect to tracker: \'%s\'" , cl_args [ "tracker_url" ] )
    return False
table , header , rest_count = to_table ( result )
print ( 'Topologies running in cluster \'%s\' submitted by \'%s\':' % ( cluster , role ) )
if rest_count :
    print ( '  with %d more...' % rest_count )
print ( tabulate ( table , headers = header ) )
return True


signature:
def show_cluster_role_env ( cl_args , cluster , role , env ) :
    '''print topologies information to stdout'''

implementation:
try :
    result = tracker_access . get_cluster_role_env_topologies ( cluster , role , env )
    if not result :
        Log . error ( 'Unknown cluster/role/env \'%s\'' % '/' . join ( [ cluster , role , env ] ) )
        return False
    result = result [ cluster ]
except Exception :
    Log . error ( "Fail to connect to tracker: \'%s\'" , cl_args [ "tracker_url" ] )
    return False
table , header , rest_count = to_table ( result )
print ( 'Topologies running in cluster \'%s\', submitted by \'%s\', and\
 under environment \'%s\':' % ( cluster , role , env ) )
if rest_count :
    print ( '  with %d more...' % rest_count )
print ( tabulate ( table , headers = header ) )
return True


signature:
def run ( command , parser , cl_args , unknown_args ) :
    """run command"""

implementation:
location = cl_args [ 'cluster/[role]/[env]' ] . split ( '/' )
if len ( location ) == 1 :
    return show_cluster ( cl_args , * location )
elif len ( location ) == 2 :
    return show_cluster_role ( cl_args , * location )
elif len ( location ) == 3 :
    return show_cluster_role_env ( cl_args , * location )
else :
    Log . error ( 'Invalid topologies selection' )
    return False


signature:
def heron_class ( class_name , lib_jars , extra_jars = None , args = None , java_defines = None ) :
    '''Execute a heron class given the args and the jars needed for class path
 : param class _ name :
 : param lib _ jars :
 : param extra _ jars :
 : param args :
 : param java _ defines :
 : return :'''

implementation:
# default optional params to empty list if not provided
if extra_jars is None :
    extra_jars = [ ]
if args is None :
    args = [ ]
if java_defines is None :
    java_defines = [ ]
# Format all java - D options that need to be passed while running
# the class locally .
java_opts = [ '-D' + opt for opt in java_defines ]
# Construct the command line for the sub process to run
# Because of the way Python execute works ,
# the java opts must be passed as part of the list
all_args = [ config . get_java_path ( ) , "-client" , "-Xmx1g" ] + java_opts + [ "-cp" , config . get_classpath ( extra_jars + lib_jars ) ]
all_args += [ class_name ] + list ( args )
# set heron _ config environment variable
heron_env = os . environ . copy ( )
heron_env [ 'HERON_OPTIONS' ] = opts . get_heron_config ( )
# print the verbose message
Log . debug ( "Invoking class using command: ``%s''" , ' ' . join ( all_args ) )
Log . debug ( "Heron options: {%s}" , str ( heron_env [ "HERON_OPTIONS" ] ) )
# invoke the command with subprocess and print error message , if any
process = subprocess . Popen ( all_args , env = heron_env , stdout = subprocess . PIPE , stderr = subprocess . PIPE , bufsize = 1 )
# stdout message has the information Java program sends back
# stderr message has extra information , such as debugging message
return ProcessResult ( process )


signature:
def heron_tar ( class_name , topology_tar , arguments , tmpdir_root , java_defines ) :
    ''': param class _ name :
 : param topology _ tar :
 : param arguments :
 : param tmpdir _ root :
 : param java _ defines :
 : return :'''

implementation:
# Extract tar to a tmp folder .
tmpdir = tempfile . mkdtemp ( dir = tmpdir_root , prefix = 'tmp' )
with contextlib . closing ( tarfile . open ( topology_tar ) ) as tar :
    tar . extractall ( path = tmpdir )
# A tar generated by pants has all dependency jars under libs /
# in addition to the topology jar at top level . Pants keeps
# filename for jar and tar the same except for extension .
topology_jar = os . path . basename ( topology_tar ) . replace ( ".tar.gz" , "" ) . replace ( ".tar" , "" ) + ".jar"
extra_jars = [ os . path . join ( tmpdir , topology_jar ) , os . path . join ( tmpdir , "*" ) , os . path . join ( tmpdir , "libs/*" ) ]
lib_jars = config . get_heron_libs ( jars . topology_jars ( ) )
# Now execute the class
return heron_class ( class_name , lib_jars , extra_jars , arguments , java_defines )


signature:
def get ( self , cluster , environ , topology , comp_name ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param comp _ name :
 : return :'''

implementation:
start_time = time . time ( )
comp_names = [ ]
if comp_name == "All" :
    lplan = yield access . get_logical_plan ( cluster , environ , topology )
    if not lplan :
        self . write ( dict ( ) )
        return
    if not 'spouts' in lplan or not 'bolts' in lplan :
        self . write ( dict ( ) )
        return
    comp_names = lplan [ 'spouts' ] . keys ( )
    comp_names . extend ( lplan [ 'bolts' ] . keys ( ) )
else :
    comp_names = [ comp_name ]
exception_infos = dict ( )
for comp_name in comp_names :
    exception_infos [ comp_name ] = yield access . get_component_exceptionsummary ( cluster , environ , topology , comp_name )
# Combine exceptions from multiple component
aggregate_exceptions = dict ( )
for comp_name , exception_logs in exception_infos . items ( ) :
    for exception_log in exception_logs :
        class_name = exception_log [ 'class_name' ]
        if class_name != '' :
            if not class_name in aggregate_exceptions :
                aggregate_exceptions [ class_name ] = 0
            aggregate_exceptions [ class_name ] += int ( exception_log [ 'count' ] )
# Put the exception value in a table
aggregate_exceptions_table = [ ]
for key in aggregate_exceptions :
    aggregate_exceptions_table . append ( [ key , str ( aggregate_exceptions [ key ] ) ] )
result = dict ( status = "success" , executiontime = time . time ( ) - start_time , result = aggregate_exceptions_table )
self . write ( result )


signature:
def get ( self ) :
    ''': return :'''

implementation:
# get all the topologies from heron nest
topologies = yield access . get_topologies_states ( )
result = dict ( )
# now convert some of the fields to be displayable
for cluster , cluster_value in topologies . items ( ) :
    result [ cluster ] = dict ( )
    for environ , environ_value in cluster_value . items ( ) :
        result [ cluster ] [ environ ] = dict ( )
        for topology , topology_value in environ_value . items ( ) :
            if "jobname" not in topology_value or topology_value [ "jobname" ] is None :
                continue
            if "submission_time" in topology_value :
                topology_value [ "submission_time" ] = topology_value [ "submission_time" ]
            else :
                topology_value [ "submission_time" ] = '-'
            result [ cluster ] [ environ ] [ topology ] = topology_value
self . write ( result )


signature:
def get ( self , cluster , environ , topology ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : return :'''

implementation:
start_time = time . time ( )
lplan = yield access . get_logical_plan ( cluster , environ , topology )
# construct the result
result = dict ( status = "success" , message = "" , version = common . VERSION , executiontime = time . time ( ) - start_time , result = lplan )
self . write ( result )


implementation:
start_time = time . time ( )
pplan = yield access . get_physical_plan ( cluster , environ , topology )
result_map = dict ( status = "success" , message = "" , version = common . VERSION , executiontime = time . time ( ) - start_time , result = pplan )
self . write ( result_map )


signature:
def get ( self , cluster , environ , topology , component ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param component :
 : return :'''

implementation:
start_time = time . time ( )
futures = yield access . get_component_exceptions ( cluster , environ , topology , component )
result_map = dict ( status = 'success' , executiontime = time . time ( ) - start_time , result = futures )
self . write ( json . dumps ( result_map ) )


signature:
def get ( self , cluster , environ , topology , instance ) :
    ''': param cluster :
 : param environ :
 : param topology :
 : param instance :
 : return :'''

implementation:
pplan = yield access . get_physical_plan ( cluster , environ , topology )
host = pplan [ 'stmgrs' ] [ pplan [ 'instances' ] [ instance ] [ 'stmgrId' ] ] [ 'host' ]
result = json . loads ( ( yield access . get_instance_pid ( cluster , environ , topology , instance ) ) )
self . write ( '

$%s>: %s

%s
' % ( host , tornado . escape . xhtml_escape ( result [ 'command' ] ) , tornado . escape . xhtml_escape ( result [ 'stdout' ] ) ) )


signature:
def add_context ( self , err_context , succ_context = None ) :
    """Prepend msg to add some context information
 : param pmsg : context info
 : return : None"""

implementation:
self . err_context = err_context
self . succ_context = succ_context


signature:
def renderProcessStdErr ( self , stderr_line ) :
    """render stderr of shelled - out process
 stderr could be error message of failure of invoking process or
 normal stderr output from successfully shelled - out process .
 In the first case , ` ` Popen ' ' should fail fast and we should be able to
 get return code immediately . We then render the failure message .
 In the second case , we simply print stderr line in stderr .
 The way to handle the first case is shaky but should be the best we can
 do since we have conflicts of design goals here .
 : param stderr _ line : one line from shelled - out process
 : return :"""

implementation:
retcode = self . process . poll ( )
if retcode is not None and status_type ( retcode ) == Status . InvocationError :
    self . _do_log ( Log . error , stderr_line )
else :
    self . _do_print ( sys . stderr , stderr_line )


signature:
def renderProcessStdOut ( self , stdout ) :
    """render stdout of shelled - out process
 stdout always contains information Java process wants to
 propagate back to cli , so we do special rendering here
 : param stdout : all lines from shelled - out process
 : return :"""

implementation:
# since we render stdout line based on Java process return code ,
# ` ` status ' ' has to be already set
assert self . status is not None
# remove pending newline
if self . status == Status . Ok :
    self . _do_log ( Log . info , stdout )
elif self . status == Status . HeronError : # remove last newline since logging will append newline
    self . _do_log ( Log . error , stdout )
# No need to prefix [ INFO ] here . We want to display dry - run response in a clean way
elif self . status == Status . DryRun :
    self . _do_print ( sys . stdout , stdout )
elif self . status == Status . InvocationError :
    self . _do_print ( sys . stdout , stdout )
else :
    raise RuntimeError ( "Unknown status type of value %d. Expected value: %s" % ( self . status . value , list ( Status ) ) )


signature:
def is_host_port_reachable ( self ) :
    """Returns true if the host is reachable . In some cases , it may not be reachable a tunnel
 must be used ."""

implementation:
for hostport in self . hostportlist :
    try :
        socket . create_connection ( hostport , StateManager . TIMEOUT_SECONDS )
        return True
    except :
        LOG . info ( "StateManager %s Unable to connect to host: %s port %i" % ( self . name , hostport [ 0 ] , hostport [ 1 ] ) )
        continue
return False


signature:
def pick_unused_port ( self ) :
    """Pick an unused port . There is a slight chance that this wont work ."""

implementation:
s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM )
s . bind ( ( '127.0.0.1' , 0 ) )
_ , port = s . getsockname ( )
s . close ( )
return port


signature:
def establish_ssh_tunnel ( self ) :
    """Establish an ssh tunnel for each local host and port
 that can be used to communicate with the state host ."""

implementation:
localportlist = [ ]
for ( host , port ) in self . hostportlist :
    localport = self . pick_unused_port ( )
    self . tunnel . append ( subprocess . Popen ( ( 'ssh' , self . tunnelhost , '-NL127.0.0.1:%d:%s:%d' % ( localport , host , port ) ) ) )
    localportlist . append ( ( '127.0.0.1' , localport ) )
return localportlist


signature:
def delete_topology_from_zk ( self , topologyName ) :
    """Removes the topology entry from :
 1 . topologies list ,
 2 . pplan ,
 3 . execution _ state , and"""

implementation:
self . delete_pplan ( topologyName )
self . delete_execution_state ( topologyName )
self . delete_topology ( topologyName )


signature:
def get_topologies ( self , callback = None ) :
    """get topologies"""

implementation:
if callback :
    self . topologies_watchers . append ( callback )
else :
    topologies_path = self . get_topologies_path ( )
    return filter ( lambda f : os . path . isfile ( os . path . join ( topologies_path , f ) ) , os . listdir ( topologies_path ) )


signature:
def get_topology ( self , topologyName , callback = None ) :
    """get topology"""

implementation:
if callback :
    self . topology_watchers [ topologyName ] . append ( callback )
else :
    topology_path = self . get_topology_path ( topologyName )
    with open ( topology_path ) as f :
        data = f . read ( )
        topology = Topology ( )
        topology . ParseFromString ( data )
        return topology


signature:
def get_packing_plan ( self , topologyName , callback = None ) :
    """get packing plan"""

implementation:
if callback :
    self . packing_plan_watchers [ topologyName ] . append ( callback )
else :
    packing_plan_path = self . get_packing_plan_path ( topologyName )
    with open ( packing_plan_path ) as f :
        data = f . read ( )
        packing_plan = PackingPlan ( )
        packing_plan . ParseFromString ( data )


signature:
def get_pplan ( self , topologyName , callback = None ) :
    """Get physical plan of a topology"""

implementation:
if callback :
    self . pplan_watchers [ topologyName ] . append ( callback )
else :
    pplan_path = self . get_pplan_path ( topologyName )
    with open ( pplan_path ) as f :
        data = f . read ( )
        pplan = PhysicalPlan ( )
        pplan . ParseFromString ( data )
        return pplan


signature:
def get_execution_state ( self , topologyName , callback = None ) :
    """Get execution state"""

implementation:
if callback :
    self . execution_state_watchers [ topologyName ] . append ( callback )
else :
    execution_state_path = self . get_execution_state_path ( topologyName )
    with open ( execution_state_path ) as f :
        data = f . read ( )
        executionState = ExecutionState ( )
        executionState . ParseFromString ( data )
        return executionState


signature:
def get_tmaster ( self , topologyName , callback = None ) :
    """Get tmaster"""

implementation:
if callback :
    self . tmaster_watchers [ topologyName ] . append ( callback )
else :
    tmaster_path = self . get_tmaster_path ( topologyName )
    with open ( tmaster_path ) as f :
        data = f . read ( )
        tmaster = TMasterLocation ( )
        tmaster . ParseFromString ( data )
        return tmaster


signature:
def get_scheduler_location ( self , topologyName , callback = None ) :
    """Get scheduler location"""

implementation:
if callback :
    self . scheduler_location_watchers [ topologyName ] . append ( callback )
else :
    scheduler_location_path = self . get_scheduler_location_path ( topologyName )
    with open ( scheduler_location_path ) as f :
        data = f . read ( )
        scheduler_location = SchedulerLocation ( )
        scheduler_location . ParseFromString ( data )
        return scheduler_location


signature:
def get ( self , pid ) :
    '''get method'''

implementation:
body = utils . str_cmd ( [ 'jmap' , '-histo' , pid ] , None , None )
self . content_type = 'application/json'
self . write ( json . dumps ( body ) )
self . finish ( )


signature:
def create_socket_options ( ) :
    """Creates SocketOptions object from a given sys _ config dict"""

implementation:
sys_config = system_config . get_sys_config ( )
opt_list = [ const . INSTANCE_NETWORK_WRITE_BATCH_SIZE_BYTES , const . INSTANCE_NETWORK_WRITE_BATCH_TIME_MS , const . INSTANCE_NETWORK_READ_BATCH_SIZE_BYTES , const . INSTANCE_NETWORK_READ_BATCH_TIME_MS , const . INSTANCE_NETWORK_OPTIONS_SOCKET_RECEIVED_BUFFER_SIZE_BYTES , const . INSTANCE_NETWORK_OPTIONS_SOCKET_SEND_BUFFER_SIZE_BYTES ]
Log . debug ( "In create_socket_options()" )
try :
    value_lst = [ int ( sys_config [ opt ] ) for opt in opt_list ]
    sock_opt = SocketOptions ( * value_lst )
    return sock_opt
except ValueError as e : # couldn ' t convert to int
    raise ValueError ( "Invalid value in sys_config: %s" % str ( e ) )
except KeyError as e : # option key was not found
    raise KeyError ( "Incomplete sys_config: %s" % str ( e ) )


signature:
def class_dict_to_specs ( mcs , class_dict ) :
    """Takes a class ` _ _ dict _ _ ` and returns ` HeronComponentSpec ` entries"""

implementation:
specs = { }
for name , spec in class_dict . items ( ) :
    if isinstance ( spec , HeronComponentSpec ) : # Use the variable name as the specification name .
        if spec . name is None :
            spec . name = name
        if spec . name in specs :
            raise ValueError ( "Duplicate component name: %s" % spec . name )
        else :
            specs [ spec . name ] = spec
return specs


signature:
def class_dict_to_topo_config ( mcs , class_dict ) :
    """Takes a class ` _ _ dict _ _ ` and returns a map containing topology - wide
 configuration .
 The returned dictionary is a sanitized ` dict ` of type ` < str - >
 ( str | object ) > ` .
 This classmethod firsts insert default topology configuration and then
 overrides it with a given topology - wide configuration . Note that this
 configuration will be overriden by a component - specific configuration at
 runtime ."""

implementation:
topo_config = { }
# add defaults
topo_config . update ( mcs . DEFAULT_TOPOLOGY_CONFIG )
for name , custom_config in class_dict . items ( ) :
    if name == 'config' and isinstance ( custom_config , dict ) :
        sanitized_dict = mcs . _sanitize_config ( custom_config )
        topo_config . update ( sanitized_dict )
return topo_config


signature:
def init_topology ( mcs , classname , class_dict ) :
    """Initializes a topology protobuf"""

implementation:
if classname == 'Topology' : # Base class can ' t initialize protobuf
    return
heron_options = TopologyType . get_heron_options_from_env ( )
initial_state = heron_options . get ( "cmdline.topology.initial.state" , "RUNNING" )
tmp_directory = heron_options . get ( "cmdline.topologydefn.tmpdirectory" )
if tmp_directory is None :
    raise RuntimeError ( "Topology definition temp directory not specified" )
topology_name = heron_options . get ( "cmdline.topology.name" , classname )
topology_id = topology_name + str ( uuid . uuid4 ( ) )
# create protobuf
topology = topology_pb2 . Topology ( )
topology . id = topology_id
topology . name = topology_name
topology . state = topology_pb2 . TopologyState . Value ( initial_state )
topology . topology_config . CopyFrom ( TopologyType . get_topology_config_protobuf ( class_dict ) )
TopologyType . add_bolts_and_spouts ( topology , class_dict )
class_dict [ 'topology_name' ] = topology_name
class_dict [ 'topology_id' ] = topology_id
class_dict [ 'protobuf_topology' ] = topology
class_dict [ 'topologydefn_tmpdir' ] = tmp_directory
class_dict [ 'heron_runtime_options' ] = heron_options


signature:
def get_heron_options_from_env ( ) :
    """Retrieves heron options from the ` HERON _ OPTIONS ` environment variable .
 Heron options have the following format :
 cmdline . topologydefn . tmpdirectory = / var / folders / tmpdir
 cmdline . topology . initial . state = PAUSED
 In this case , the returned map will contain :
 # ! json
 " cmdline . topologydefn . tmpdirectory " : " / var / folders / tmpdir " ,
 " cmdline . topology . initial . state " : " PAUSED "
 Currently supports the following options natively :
 - ` cmdline . topologydefn . tmpdirectory ` : ( required ) the directory to which this
 topology ' s defn file is written
 - ` cmdline . topology . initial . state ` : ( default : " RUNNING " ) the initial state of the topology
 - ` cmdline . topology . name ` : ( default : class name ) topology name on deployment
 Returns : map mapping from key to value"""

implementation:
heron_options_raw = os . environ . get ( "HERON_OPTIONS" )
if heron_options_raw is None :
    raise RuntimeError ( "HERON_OPTIONS environment variable not found" )
options = { }
for option_line in heron_options_raw . replace ( "%%%%" , " " ) . split ( ',' ) :
    key , sep , value = option_line . partition ( "=" )
    if sep :
        options [ key ] = value
    else :
        raise ValueError ( "Invalid HERON_OPTIONS part %r" % option_line )
return options


signature:
def add_spec ( self , * specs ) :
    """Add specs to the topology
 : type specs : HeronComponentSpec
 : param specs : specs to add to the topology"""

implementation:
for spec in specs :
    if not isinstance ( spec , HeronComponentSpec ) :
        raise TypeError ( "Argument to add_spec needs to be HeronComponentSpec, given: %s" % str ( spec ) )
    if spec . name is None :
        raise ValueError ( "TopologyBuilder cannot take a spec without name" )
    if spec . name == "config" :
        raise ValueError ( "config is a reserved name" )
    if spec . name in self . _specs :
        raise ValueError ( "Attempting to add duplicate spec name: %r %r" % ( spec . name , spec ) )
    self . _specs [ spec . name ] = spec


signature:
def add_spout ( self , name , spout_cls , par , config = None , optional_outputs = None ) :
    """Add a spout to the topology"""

implementation:
spout_spec = spout_cls . spec ( name = name , par = par , config = config , optional_outputs = optional_outputs )
self . add_spec ( spout_spec )
return spout_spec


signature:
def add_bolt ( self , name , bolt_cls , par , inputs , config = None , optional_outputs = None ) :
    """Add a bolt to the topology"""

implementation:
bolt_spec = bolt_cls . spec ( name = name , par = par , inputs = inputs , config = config , optional_outputs = optional_outputs )
self . add_spec ( bolt_spec )
return bolt_spec


signature:
def set_config ( self , config ) :
    """Set topology - wide configuration to the topology
 : type config : dict
 : param config : topology - wide config"""

implementation:
if not isinstance ( config , dict ) :
    raise TypeError ( "Argument to set_config needs to be dict, given: %s" % str ( config ) )
self . _topology_config = config


signature:
def build_and_submit ( self ) :
    """Builds the topology and submits to the destination"""

implementation:
class_dict = self . _construct_topo_class_dict ( )
topo_cls = TopologyType ( self . topology_name , ( Topology , ) , class_dict )
topo_cls . write ( )


signature:
def fetch_url_as_json ( fetch_url , default_value = None ) :
    '''Fetch the given url and convert the response to json .
 : param fetch _ url : URL to fetch
 : param default _ value : value to return in case of failure
 : return :'''

implementation:
# assign empty dict for optional param
if default_value is None :
    default_value = dict ( )
Log . debug ( "fetching url %s" , fetch_url )
ret = default_value
# time the duration of the fetch
start = time . time ( )
# fetch the URL asynchronously
http_response = yield tornado . httpclient . AsyncHTTPClient ( ) . fetch ( fetch_url )
# handle http errors , and return if any
if http_response . error :
    Log . error ( "Unable to get response from %s. Error %s" , fetch_url , http_response . error )
    raise tornado . gen . Return ( ret )
# load response and handle return errors , if any
response = json . loads ( http_response . body )
if not 'result' in response :
    Log . error ( "Empty response from %s" , fetch_url )
    raise tornado . gen . Return ( ret )
# get the response and execution time on server side
ret = response [ 'result' ]
execution = 1000 * response [ 'executiontime' ]
# calculate the time
end = time . time ( )
duration = 1000 * ( end - start )
Log . debug ( "TIME: url fetch took %.2f ms server time %s" , execution , fetch_url )
Log . debug ( "TIME: url fetch took %.2f ms round trip  %s" , duration , fetch_url )
# convert future to value
raise tornado . gen . Return ( ret )


signature:
def create_parser ( subparsers ) :
    """create parser"""

implementation:
parser = subparsers . add_parser ( 'version' , help = 'Display version' , usage = "%(prog)s" , add_help = False )
args . add_titles ( parser )
parser . set_defaults ( subcommand = 'version' )
return parser


signature:
def queries_map ( ) :
    """map from query parameter to query name"""

implementation:
qs = _all_metric_queries ( )
return dict ( zip ( qs [ 0 ] , qs [ 1 ] ) + zip ( qs [ 2 ] , qs [ 3 ] ) )


signature:
def get_clusters ( ) :
    """Synced API call to get all cluster names"""

implementation:
instance = tornado . ioloop . IOLoop . instance ( )
# pylint : disable = unnecessary - lambda
try :
    return instance . run_sync ( lambda : API . get_clusters ( ) )
except Exception :
    Log . debug ( traceback . format_exc ( ) )
    raise


signature:
def get_logical_plan ( cluster , env , topology , role ) :
    """Synced API call to get logical plans"""

implementation:
instance = tornado . ioloop . IOLoop . instance ( )
try :
    return instance . run_sync ( lambda : API . get_logical_plan ( cluster , env , topology , role ) )
except Exception :
    Log . debug ( traceback . format_exc ( ) )
    raise


signature:
def get_topology_info ( * args ) :
    """Synced API call to get topology information"""

implementation:
instance = tornado . ioloop . IOLoop . instance ( )
try :
    return instance . run_sync ( lambda : API . get_topology_info ( * args ) )
except Exception :
    Log . debug ( traceback . format_exc ( ) )
    raise


signature:
def get_component_metrics ( component , cluster , env , topology , role ) :
    """Synced API call to get component metrics"""

implementation:
all_queries = metric_queries ( )
try :
    result = get_topology_metrics ( cluster , env , topology , component , [ ] , all_queries , [ 0 , - 1 ] , role )
    return result [ "metrics" ]
except Exception :
    Log . debug ( traceback . format_exc ( ) )
    raise


signature:
def configure ( level = logging . INFO , logfile = None ) :
    """Configure logger which dumps log on terminal
 : param level : logging level : info , warning , verbose . . .
 : type level : logging level
 : param logfile : log file name , default to None
 : type logfile : string
 : return : None
 : rtype : None"""

implementation:
# Remove all the existing StreamHandlers to avoid duplicate
for handler in Log . handlers :
    if isinstance ( handler , logging . StreamHandler ) :
        Log . handlers . remove ( handler )
Log . setLevel ( level )
# if logfile is specified , FileHandler is used
if logfile is not None :
    log_format = "[%(asctime)s] [%(levelname)s]: %(message)s"
    formatter = logging . Formatter ( fmt = log_format , datefmt = date_format )
    file_handler = logging . FileHandler ( logfile )
    file_handler . setFormatter ( formatter )
    Log . addHandler ( file_handler )
# otherwise , use StreamHandler to output to stream ( stdout , stderr . . . )
else :
    log_format = "[%(asctime)s] %(log_color)s[%(levelname)s]%(reset)s: %(message)s"
    # pylint : disable = redefined - variable - type
    formatter = colorlog . ColoredFormatter ( fmt = log_format , datefmt = date_format )
    stream_handler = logging . StreamHandler ( )
    stream_handler . setFormatter ( formatter )
    Log . addHandler ( stream_handler )


signature:
def init_rotating_logger ( level , logfile , max_files , max_bytes ) :
    """Initializes a rotating logger
 It also makes sure that any StreamHandler is removed , so as to avoid stdout / stderr
 constipation issues"""

implementation:
logging . basicConfig ( )
root_logger = logging . getLogger ( )
log_format = "[%(asctime)s] [%(levelname)s] %(filename)s: %(message)s"
root_logger . setLevel ( level )
handler = RotatingFileHandler ( logfile , maxBytes = max_bytes , backupCount = max_files )
handler . setFormatter ( logging . Formatter ( fmt = log_format , datefmt = date_format ) )
root_logger . addHandler ( handler )
for handler in root_logger . handlers :
    root_logger . debug ( "Associated handlers - " + str ( handler ) )
    if isinstance ( handler , logging . StreamHandler ) :
        root_logger . debug ( "Removing StreamHandler: " + str ( handler ) )
        root_logger . handlers . remove ( handler )


signature:
def set_logging_level ( cl_args ) :
    """simply set verbose level based on command - line args
 : param cl _ args : CLI arguments
 : type cl _ args : dict
 : return : None
 : rtype : None"""

implementation:
if 'verbose' in cl_args and cl_args [ 'verbose' ] :
    configure ( logging . DEBUG )
else :
    configure ( logging . INFO )


signature:
def _get_spout ( self ) :
    """Returns Spout protobuf message"""

implementation:
spout = topology_pb2 . Spout ( )
spout . comp . CopyFrom ( self . _get_base_component ( ) )
# Add output streams
self . _add_out_streams ( spout )
return spout


signature:
def _get_bolt ( self ) :
    """Returns Bolt protobuf message"""

implementation:
bolt = topology_pb2 . Bolt ( )
bolt . comp . CopyFrom ( self . _get_base_component ( ) )
# Add streams
self . _add_in_streams ( bolt )
self . _add_out_streams ( bolt )
return bolt


signature:
def _get_base_component ( self ) :
    """Returns Component protobuf message"""

implementation:
comp = topology_pb2 . Component ( )
comp . name = self . name
comp . spec = topology_pb2 . ComponentObjectSpec . Value ( "PYTHON_CLASS_NAME" )
comp . class_name = self . python_class_path
comp . config . CopyFrom ( self . _get_comp_config ( ) )
return comp


signature:
def _get_comp_config ( self ) :
    """Returns component - specific Config protobuf message
 It first adds ` ` topology . component . parallelism ` ` , and is overriden by
 a user - defined component - specific configuration , specified by spec ( ) ."""

implementation:
proto_config = topology_pb2 . Config ( )
# first add parallelism
key = proto_config . kvs . add ( )
key . key = TOPOLOGY_COMPONENT_PARALLELISM
key . value = str ( self . parallelism )
key . type = topology_pb2 . ConfigValueType . Value ( "STRING_VALUE" )
# iterate through self . custom _ config
if self . custom_config is not None :
    sanitized = self . _sanitize_config ( self . custom_config )
    for key , value in sanitized . items ( ) :
        if isinstance ( value , str ) :
            kvs = proto_config . kvs . add ( )
            kvs . key = key
            kvs . value = value
            kvs . type = topology_pb2 . ConfigValueType . Value ( "STRING_VALUE" )
        else : # need to serialize
            kvs = proto_config . kvs . add ( )
            kvs . key = key
            kvs . serialized_value = default_serializer . serialize ( value )
            kvs . type = topology_pb2 . ConfigValueType . Value ( "PYTHON_SERIALIZED_VALUE" )
return proto_config


signature:
def _sanitize_config ( custom_config ) :
    """Checks whether ` ` custom _ config ` ` is sane and returns a sanitized dict < str - > ( str | object ) >
 It checks if keys are all strings and sanitizes values of a given dictionary as follows :
 - If string , number or boolean is given as a value , it is converted to string .
 For string and number ( int , float ) , it is converted to string by a built - in ` ` str ( ) ` ` method .
 For a boolean value , ` ` True ` ` is converted to " true " instead of " True " , and ` ` False ` ` is
 converted to " false " instead of " False " , in order to keep the consistency with
 Java configuration .
 - If neither of the above is given as a value , it is inserted into the sanitized dict as it is .
 These values will need to be serialized before adding to a protobuf message ."""

implementation:
if not isinstance ( custom_config , dict ) :
    raise TypeError ( "Component-specific configuration must be given as a dict type, given: %s" % str ( type ( custom_config ) ) )
sanitized = { }
for key , value in custom_config . items ( ) :
    if not isinstance ( key , str ) :
        raise TypeError ( "Key for component-specific configuration must be string, given: %s:%s" % ( str ( type ( key ) ) , str ( key ) ) )
    if isinstance ( value , bool ) :
        sanitized [ key ] = "true" if value else "false"
    elif isinstance ( value , ( str , int , float ) ) :
        sanitized [ key ] = str ( value )
    else :
        sanitized [ key ] = value
return sanitized


signature:
def _add_in_streams ( self , bolt ) :
    """Adds inputs to a given protobuf Bolt message"""

implementation:
if self . inputs is None :
    return
# sanitize inputs and get a map < GlobalStreamId - > Grouping >
input_dict = self . _sanitize_inputs ( )
for global_streamid , gtype in input_dict . items ( ) :
    in_stream = bolt . inputs . add ( )
    in_stream . stream . CopyFrom ( self . _get_stream_id ( global_streamid . component_id , global_streamid . stream_id ) )
    if isinstance ( gtype , Grouping . FIELDS ) : # it ' s a field grouping
        in_stream . gtype = gtype . gtype
        in_stream . grouping_fields . CopyFrom ( self . _get_stream_schema ( gtype . fields ) )
    elif isinstance ( gtype , Grouping . CUSTOM ) : # it ' s a custom grouping
        in_stream . gtype = gtype . gtype
        in_stream . custom_grouping_object = gtype . python_serialized
        in_stream . type = topology_pb2 . CustomGroupingObjectType . Value ( "PYTHON_OBJECT" )
    else :
        in_stream . gtype = gtype


signature:
def _sanitize_inputs ( self ) :
    """Sanitizes input fields and returns a map < GlobalStreamId - > Grouping >"""

implementation:
ret = { }
if self . inputs is None :
    return
if isinstance ( self . inputs , dict ) : # inputs are dictionary , must be either < HeronComponentSpec - > Grouping > or
# < GlobalStreamId - > Grouping >
    for key , grouping in self . inputs . items ( ) :
        if not Grouping . is_grouping_sane ( grouping ) :
            raise ValueError ( 'A given grouping is not supported' )
        if isinstance ( key , HeronComponentSpec ) : # use default streamid
            if key . name is None : # should not happen as TopologyType metaclass sets name attribute
            # before calling this method
                raise RuntimeError ( "In _sanitize_inputs(): HeronComponentSpec doesn't have a name" )
            global_streamid = GlobalStreamId ( key . name , Stream . DEFAULT_STREAM_ID )
            ret [ global_streamid ] = grouping
        elif isinstance ( key , GlobalStreamId ) :
            ret [ key ] = grouping
        else :
            raise ValueError ( "%s is not supported as a key to inputs" % str ( key ) )
elif isinstance ( self . inputs , ( list , tuple ) ) : # inputs are lists , must be either a list of HeronComponentSpec or GlobalStreamId
# will use SHUFFLE grouping
    for input_obj in self . inputs :
        if isinstance ( input_obj , HeronComponentSpec ) :
            if input_obj . name is None : # should not happen as TopologyType metaclass sets name attribute
            # before calling this method
                raise RuntimeError ( "In _sanitize_inputs(): HeronComponentSpec doesn't have a name" )
            global_streamid = GlobalStreamId ( input_obj . name , Stream . DEFAULT_STREAM_ID )
            ret [ global_streamid ] = Grouping . SHUFFLE
        elif isinstance ( input_obj , GlobalStreamId ) :
            ret [ input_obj ] = Grouping . SHUFFLE
        else :
            raise ValueError ( "%s is not supported as an input" % str ( input_obj ) )
else :
    raise TypeError ( "Inputs must be a list, dict, or None, given: %s" % str ( self . inputs ) )
return ret


signature:
def _add_out_streams ( self , spbl ) :
    """Adds outputs to a given protobuf Bolt or Spout message"""

implementation:
if self . outputs is None :
    return
# sanitize outputs and get a map < stream _ id - > out fields >
output_map = self . _sanitize_outputs ( )
for stream_id , out_fields in output_map . items ( ) :
    out_stream = spbl . outputs . add ( )
    out_stream . stream . CopyFrom ( self . _get_stream_id ( self . name , stream_id ) )
    out_stream . schema . CopyFrom ( self . _get_stream_schema ( out_fields ) )


signature:
def _sanitize_outputs ( self ) :
    """Sanitizes output fields and returns a map < stream _ id - > list of output fields >"""

implementation:
ret = { }
if self . outputs is None :
    return
if not isinstance ( self . outputs , ( list , tuple ) ) :
    raise TypeError ( "Argument to outputs must be either list or tuple, given: %s" % str ( type ( self . outputs ) ) )
for output in self . outputs :
    if not isinstance ( output , ( str , Stream ) ) :
        raise TypeError ( "Outputs must be a list of strings or Streams, given: %s" % str ( output ) )
    if isinstance ( output , str ) : # it ' s a default stream
        if Stream . DEFAULT_STREAM_ID not in ret :
            ret [ Stream . DEFAULT_STREAM_ID ] = list ( )
        ret [ Stream . DEFAULT_STREAM_ID ] . append ( output )
    else : # output is a Stream object
        if output . stream_id == Stream . DEFAULT_STREAM_ID and Stream . DEFAULT_STREAM_ID in ret : # some default stream fields are already in there
            ret [ Stream . DEFAULT_STREAM_ID ] . extend ( output . fields )
        else :
            ret [ output . stream_id ] = output . fields
return ret


signature:
def get_out_streamids ( self ) :
    """Returns a set of output stream ids registered for this component"""

implementation:
if self . outputs is None :
    return set ( )
if not isinstance ( self . outputs , ( list , tuple ) ) :
    raise TypeError ( "Argument to outputs must be either list or tuple, given: %s" % str ( type ( self . outputs ) ) )
ret_lst = [ ]
for output in self . outputs :
    if not isinstance ( output , ( str , Stream ) ) :
        raise TypeError ( "Outputs must be a list of strings or Streams, given: %s" % str ( output ) )
    ret_lst . append ( Stream . DEFAULT_STREAM_ID if isinstance ( output , str ) else output . stream_id )
return set ( ret_lst )


signature:
def _get_stream_id ( comp_name , stream_id ) :
    """Returns a StreamId protobuf message"""

implementation:
proto_stream_id = topology_pb2 . StreamId ( )
proto_stream_id . id = stream_id
proto_stream_id . component_name = comp_name
return proto_stream_id


signature:
def _get_stream_schema ( fields ) :
    """Returns a StreamSchema protobuf message"""

implementation:
stream_schema = topology_pb2 . StreamSchema ( )
for field in fields :
    key = stream_schema . keys . add ( )
    key . key = field
    key . type = topology_pb2 . Type . Value ( "OBJECT" )
return stream_schema


signature:
def component_id ( self ) :
    """Returns component _ id of this GlobalStreamId
 Note that if HeronComponentSpec is specified as componentId and its name is not yet
 available ( i . e . when ` ` name ` ` argument was not given in ` ` spec ( ) ` ` method in Bolt or Spout ) ,
 this property returns a message with uuid . However , this is provided only for safety
 with _ _ eq _ _ ( ) , _ _ str _ _ ( ) , and _ _ hash _ _ ( ) methods , and not meant to be called explicitly
 before TopologyType class finally sets the name attribute of HeronComponentSpec ."""

implementation:
if isinstance ( self . _component_id , HeronComponentSpec ) :
    if self . _component_id . name is None : # HeronComponentSpec instance ' s name attribute might not be available until
    # TopologyType metaclass finally sets it . This statement is to support _ _ eq _ _ ( ) ,
    # _ _ hash _ _ ( ) and _ _ str _ _ ( ) methods with safety , as raising Exception is not
    # appropriate this case .
        return "<No name available for HeronComponentSpec yet, uuid: %s>" % self . _component_id . uuid
    return self . _component_id . name
elif isinstance ( self . _component_id , str ) :
    return self . _component_id
else :
    raise ValueError ( "Component Id for this GlobalStreamId is not properly set: <%s:%s>" % ( str ( type ( self . _component_id ) ) , str ( self . _component_id ) ) )


signature:
def write_error ( self , status_code , ** kwargs ) :
    ''': param status _ code :
 : param kwargs :
 : return :'''

implementation:
if "exc_info" in kwargs :
    exc_info = kwargs [ "exc_info" ]
    error = exc_info [ 1 ]
    errormessage = "%s: %s" % ( status_code , error )
    self . render ( "error.html" , errormessage = errormessage )
else :
    errormessage = "%s" % ( status_code )
    self . render ( "error.html" , errormessage = errormessage )


signature:
def register_metric ( self , name , metric , time_bucket_in_sec ) :
    """Registers a new metric to this context"""

implementation:
collector = self . get_metrics_collector ( )
collector . register_metric ( name , metric , time_bucket_in_sec )


signature:
def get_sources ( self , component_id ) :
    """Returns the declared inputs to specified component
 : return : map < streamId namedtuple ( same structure as protobuf msg ) - > gtype > , or
 None if not found"""

implementation:
# this is necessary because protobuf message is not hashable
StreamId = namedtuple ( 'StreamId' , 'id, component_name' )
if component_id in self . inputs :
    ret = { }
    for istream in self . inputs . get ( component_id ) :
        key = StreamId ( id = istream . stream . id , component_name = istream . stream . component_name )
        ret [ key ] = istream . gtype
    return ret
else :
    return None


signature:
def get_component_tasks ( self , component_id ) :
    """Returns the task ids allocated for the given component id"""

implementation:
ret = [ ]
for task_id , comp_id in self . task_to_component_map . items ( ) :
    if comp_id == component_id :
        ret . append ( task_id )
return ret


signature:
def add_task_hook ( self , task_hook ) :
    """Registers a specified task hook to this context
 : type task _ hook : heron . instance . src . python . utils . topology . ITaskHook
 : param task _ hook : Implementation of ITaskHook"""

implementation:
if not isinstance ( task_hook , ITaskHook ) :
    raise TypeError ( "In add_task_hook(): attempt to add non ITaskHook instance, given: %s" % str ( type ( task_hook ) ) )
self . task_hooks . append ( task_hook )


signature:
def get_metrics_collector ( self ) :
    """Returns this context ' s metrics collector"""

implementation:
if self . metrics_collector is None or not isinstance ( self . metrics_collector , MetricsCollector ) :
    raise RuntimeError ( "Metrics collector is not registered in this context" )
return self . metrics_collector


signature:
def invoke_hook_prepare ( self ) :
    """invoke task hooks for after the spout / bolt ' s initialize ( ) method"""

implementation:
for task_hook in self . task_hooks :
    task_hook . prepare ( self . get_cluster_config ( ) , self )


signature:
def invoke_hook_emit ( self , values , stream_id , out_tasks ) :
    """invoke task hooks for every time a tuple is emitted in spout / bolt
 : type values : list
 : param values : values emitted
 : type stream _ id : str
 : param stream _ id : stream id into which tuple is emitted
 : type out _ tasks : list
 : param out _ tasks : list of custom grouping target task id"""

implementation:
if len ( self . task_hooks ) > 0 :
    emit_info = EmitInfo ( values = values , stream_id = stream_id , task_id = self . get_task_id ( ) , out_tasks = out_tasks )
    for task_hook in self . task_hooks :
        task_hook . emit ( emit_info )


signature:
def invoke_hook_spout_ack ( self , message_id , complete_latency_ns ) :
    """invoke task hooks for every time spout acks a tuple
 : type message _ id : str
 : param message _ id : message id to which an acked tuple was anchored
 : type complete _ latency _ ns : float
 : param complete _ latency _ ns : complete latency in nano seconds"""

implementation:
if len ( self . task_hooks ) > 0 :
    spout_ack_info = SpoutAckInfo ( message_id = message_id , spout_task_id = self . get_task_id ( ) , complete_latency_ms = complete_latency_ns * system_constants . NS_TO_MS )
    for task_hook in self . task_hooks :
        task_hook . spout_ack ( spout_ack_info )


signature:
def invoke_hook_spout_fail ( self , message_id , fail_latency_ns ) :
    """invoke task hooks for every time spout fails a tuple
 : type message _ id : str
 : param message _ id : message id to which a failed tuple was anchored
 : type fail _ latency _ ns : float
 : param fail _ latency _ ns : fail latency in nano seconds"""

implementation:
if len ( self . task_hooks ) > 0 :
    spout_fail_info = SpoutFailInfo ( message_id = message_id , spout_task_id = self . get_task_id ( ) , fail_latency_ms = fail_latency_ns * system_constants . NS_TO_MS )
    for task_hook in self . task_hooks :
        task_hook . spout_fail ( spout_fail_info )


signature:
def invoke_hook_bolt_execute ( self , heron_tuple , execute_latency_ns ) :
    """invoke task hooks for every time bolt processes a tuple
 : type heron _ tuple : HeronTuple
 : param heron _ tuple : tuple that is executed
 : type execute _ latency _ ns : float
 : param execute _ latency _ ns : execute latency in nano seconds"""

implementation:
if len ( self . task_hooks ) > 0 :
    bolt_execute_info = BoltExecuteInfo ( heron_tuple = heron_tuple , executing_task_id = self . get_task_id ( ) , execute_latency_ms = execute_latency_ns * system_constants . NS_TO_MS )
    for task_hook in self . task_hooks :
        task_hook . bolt_execute ( bolt_execute_info )


signature:
def invoke_hook_bolt_ack ( self , heron_tuple , process_latency_ns ) :
    """invoke task hooks for every time bolt acks a tuple
 : type heron _ tuple : HeronTuple
 : param heron _ tuple : tuple that is acked
 : type process _ latency _ ns : float
 : param process _ latency _ ns : process latency in nano seconds"""

implementation:
if len ( self . task_hooks ) > 0 :
    bolt_ack_info = BoltAckInfo ( heron_tuple = heron_tuple , acking_task_id = self . get_task_id ( ) , process_latency_ms = process_latency_ns * system_constants . NS_TO_MS )
    for task_hook in self . task_hooks :
        task_hook . bolt_ack ( bolt_ack_info )


signature:
def invoke_hook_bolt_fail ( self , heron_tuple , fail_latency_ns ) :
    """invoke task hooks for every time bolt fails a tuple
 : type heron _ tuple : HeronTuple
 : param heron _ tuple : tuple that is failed
 : type fail _ latency _ ns : float
 : param fail _ latency _ ns : fail latency in nano seconds"""

implementation:
if len ( self . task_hooks ) > 0 :
    bolt_fail_info = BoltFailInfo ( heron_tuple = heron_tuple , failing_task_id = self . get_task_id ( ) , fail_latency_ms = fail_latency_ns * system_constants . NS_TO_MS )
    for task_hook in self . task_hooks :
        task_hook . bolt_fail ( bolt_fail_info )


signature:
def create_parser ( subparsers ) :
    '''Create a subparser for the submit command
 : param subparsers :
 : return :'''

implementation:
parser = subparsers . add_parser ( 'submit' , help = 'Submit a topology' , usage = "%(prog)s [options] cluster/[role]/[env] " + "topology-file-name topology-class-name [topology-args]" , add_help = True )
cli_args . add_titles ( parser )
cli_args . add_cluster_role_env ( parser )
cli_args . add_topology_file ( parser )
cli_args . add_topology_class ( parser )
cli_args . add_config ( parser )
cli_args . add_deactive_deploy ( parser )
cli_args . add_dry_run ( parser )
cli_args . add_extra_launch_classpath ( parser )
cli_args . add_release_yaml_file ( parser )
cli_args . add_service_url ( parser )
cli_args . add_system_property ( parser )
cli_args . add_verbose ( parser )
parser . set_defaults ( subcommand = 'submit' )
return parser


signature:
def launch_a_topology ( cl_args , tmp_dir , topology_file , topology_defn_file , topology_name ) :
    '''Launch a topology given topology jar , its definition file and configurations
 : param cl _ args :
 : param tmp _ dir :
 : param topology _ file :
 : param topology _ defn _ file :
 : param topology _ name :
 : return :'''

implementation:
# get the normalized path for topology . tar . gz
topology_pkg_path = config . normalized_class_path ( os . path . join ( tmp_dir , 'topology.tar.gz' ) )
# get the release yaml file
release_yaml_file = cl_args [ 'release_yaml_file' ]
# create a tar package with the cluster configuration and generated config files
config_path = cl_args [ 'config_path' ]
tar_pkg_files = [ topology_file , topology_defn_file ]
generated_config_files = [ release_yaml_file , cl_args [ 'override_config_file' ] ]
config . create_tar ( topology_pkg_path , tar_pkg_files , config_path , generated_config_files )
# pass the args to submitter main
args = [ "--cluster" , cl_args [ 'cluster' ] , "--role" , cl_args [ 'role' ] , "--environment" , cl_args [ 'environ' ] , "--submit_user" , cl_args [ 'submit_user' ] , "--heron_home" , config . get_heron_dir ( ) , "--config_path" , config_path , "--override_config_file" , cl_args [ 'override_config_file' ] , "--release_file" , release_yaml_file , "--topology_package" , topology_pkg_path , "--topology_defn" , topology_defn_file , "--topology_bin" , os . path . basename ( topology_file ) # pex / cpp file if pex / cpp specified
]
if Log . getEffectiveLevel ( ) == logging . DEBUG :
    args . append ( "--verbose" )
if cl_args [ "dry_run" ] :
    args . append ( "--dry_run" )
    if "dry_run_format" in cl_args :
        args += [ "--dry_run_format" , cl_args [ "dry_run_format" ] ]
lib_jars = config . get_heron_libs ( jars . scheduler_jars ( ) + jars . uploader_jars ( ) + jars . statemgr_jars ( ) + jars . packing_jars ( ) )
extra_jars = cl_args [ 'extra_launch_classpath' ] . split ( ':' )
# invoke the submitter to submit and launch the topology
main_class = 'org.apache.heron.scheduler.SubmitterMain'
res = execute . heron_class ( class_name = main_class , lib_jars = lib_jars , extra_jars = extra_jars , args = args , java_defines = [ ] )
err_ctxt = "Failed to launch topology '%s' %s" % ( topology_name , launch_mode_msg ( cl_args ) )
succ_ctxt = "Successfully launched topology '%s' %s" % ( topology_name , launch_mode_msg ( cl_args ) )
res . add_context ( err_ctxt , succ_ctxt )
return res


signature:
def launch_topology_server ( cl_args , topology_file , topology_defn_file , topology_name ) :
    '''Launch a topology given topology jar , its definition file and configurations
 : param cl _ args :
 : param topology _ file :
 : param topology _ defn _ file :
 : param topology _ name :
 : return :'''

implementation:
service_apiurl = cl_args [ 'service_url' ] + rest . ROUTE_SIGNATURES [ 'submit' ] [ 1 ]
service_method = rest . ROUTE_SIGNATURES [ 'submit' ] [ 0 ]
data = dict ( name = topology_name , cluster = cl_args [ 'cluster' ] , role = cl_args [ 'role' ] , environment = cl_args [ 'environ' ] , user = cl_args [ 'submit_user' ] , )
Log . info ( "" + str ( cl_args ) )
overrides = dict ( )
if 'config_property' in cl_args :
    overrides = config . parse_override_config ( cl_args [ 'config_property' ] )
if overrides :
    data . update ( overrides )
if cl_args [ 'dry_run' ] :
    data [ "dry_run" ] = True
files = dict ( definition = open ( topology_defn_file , 'rb' ) , topology = open ( topology_file , 'rb' ) , )
err_ctxt = "Failed to launch topology '%s' %s" % ( topology_name , launch_mode_msg ( cl_args ) )
succ_ctxt = "Successfully launched topology '%s' %s" % ( topology_name , launch_mode_msg ( cl_args ) )
try :
    r = service_method ( service_apiurl , data = data , files = files )
    ok = r . status_code is requests . codes . ok
    created = r . status_code is requests . codes . created
    s = Status . Ok if created or ok else Status . HeronError
    if s is Status . HeronError :
        Log . error ( r . json ( ) . get ( 'message' , "Unknown error from API server %d" % r . status_code ) )
    elif ok : # this case happens when we request a dry _ run
        print ( r . json ( ) . get ( "response" ) )
except ( requests . exceptions . ConnectionError , requests . exceptions . HTTPError ) as err :
    Log . error ( err )
    return SimpleResult ( Status . HeronError , err_ctxt , succ_ctxt )
return SimpleResult ( s , err_ctxt , succ_ctxt )


signature:
def launch_topologies ( cl_args , topology_file , tmp_dir ) :
    '''Launch topologies
 : param cl _ args :
 : param topology _ file :
 : param tmp _ dir :
 : return : list ( Responses )'''

implementation:
# the submitter would have written the . defn file to the tmp _ dir
defn_files = glob . glob ( tmp_dir + '/*.defn' )
if len ( defn_files ) == 0 :
    return SimpleResult ( Status . HeronError , "No topologies found under %s" % tmp_dir )
results = [ ]
for defn_file in defn_files : # load the topology definition from the file
    topology_defn = topology_pb2 . Topology ( )
    try :
        handle = open ( defn_file , "rb" )
        topology_defn . ParseFromString ( handle . read ( ) )
        handle . close ( )
    except Exception as e :
        err_context = "Cannot load topology definition '%s': %s" % ( defn_file , e )
        return SimpleResult ( Status . HeronError , err_context )
    # launch the topology
    Log . info ( "Launching topology: \'%s\'%s" , topology_defn . name , launch_mode_msg ( cl_args ) )
    # check if we have to do server or direct based deployment
    if cl_args [ 'deploy_mode' ] == config . SERVER_MODE :
        res = launch_topology_server ( cl_args , topology_file , defn_file , topology_defn . name )
    else :
        res = launch_a_topology ( cl_args , tmp_dir , topology_file , defn_file , topology_defn . name )
    results . append ( res )
return results


signature:
def submit_fatjar ( cl_args , unknown_args , tmp_dir ) :
    '''We use the packer to make a package for the jar and dump it
 to a well - known location . We then run the main method of class
 with the specified arguments . We pass arguments as an environment variable HERON _ OPTIONS .
 This will run the jar file with the topology _ class _ name . The submitter
 inside will write out the topology defn file to a location that
 we specify . Then we write the topology defn file to a well known
 location . We then write to appropriate places in zookeeper
 and launch the scheduler jobs
 : param cl _ args :
 : param unknown _ args :
 : param tmp _ dir :
 : return :'''

implementation:
# execute main of the topology to create the topology definition
topology_file = cl_args [ 'topology-file-name' ]
main_class = cl_args [ 'topology-class-name' ]
res = execute . heron_class ( class_name = main_class , lib_jars = config . get_heron_libs ( jars . topology_jars ( ) ) , extra_jars = [ topology_file ] , args = tuple ( unknown_args ) , java_defines = cl_args [ 'topology_main_jvm_property' ] )
result . render ( res )
if not result . is_successful ( res ) :
    err_context = ( "Failed to create topology definition " "file when executing class '%s' of file '%s'" ) % ( main_class , topology_file )
    res . add_context ( err_context )
    return res
results = launch_topologies ( cl_args , topology_file , tmp_dir )
return results


signature:
def submit_tar ( cl_args , unknown_args , tmp_dir ) :
    '''Extract and execute the java files inside the tar and then add topology
 definition file created by running submitTopology
 We use the packer to make a package for the tar and dump it
 to a well - known location . We then run the main method of class
 with the specified arguments . We pass arguments as an environment variable HERON _ OPTIONS .
 This will run the jar file with the topology class name .
 The submitter inside will write out the topology defn file to a location
 that we specify . Then we write the topology defn file to a well known
 packer location . We then write to appropriate places in zookeeper
 and launch the aurora jobs
 : param cl _ args :
 : param unknown _ args :
 : param tmp _ dir :
 : return :'''

implementation:
# execute main of the topology to create the topology definition
topology_file = cl_args [ 'topology-file-name' ]
java_defines = cl_args [ 'topology_main_jvm_property' ]
main_class = cl_args [ 'topology-class-name' ]
res = execute . heron_tar ( main_class , topology_file , tuple ( unknown_args ) , tmp_dir , java_defines )
result . render ( res )
if not result . is_successful ( res ) :
    err_context = ( "Failed to create topology definition " "file when executing class '%s' of file '%s'" ) % ( main_class , topology_file )
    res . add_context ( err_context )
    return res
return launch_topologies ( cl_args , topology_file , tmp_dir )


signature:
def run ( command , parser , cl_args , unknown_args ) :
    '''Submits the topology to the scheduler
 * Depending on the topology file name extension , we treat the file as a
 fatjar ( if the ext is . jar ) or a tar file ( if the ext is . tar / . tar . gz ) .
 * We upload the topology file to the packer , update zookeeper and launch
 scheduler jobs representing that topology
 * You can see your topology in Heron UI
 : param command :
 : param parser :
 : param cl _ args :
 : param unknown _ args :
 : return :'''

implementation:
Log . debug ( "Submit Args %s" , cl_args )
# get the topology file name
topology_file = cl_args [ 'topology-file-name' ]
if urlparse . urlparse ( topology_file ) . scheme :
    cl_args [ 'topology-file-name' ] = download ( topology_file , cl_args [ 'cluster' ] )
    topology_file = cl_args [ 'topology-file-name' ]
    Log . debug ( "download uri to local file: %s" , topology_file )
# check to see if the topology file exists
if not os . path . isfile ( topology_file ) :
    err_context = "Topology file '%s' does not exist" % topology_file
    return SimpleResult ( Status . InvocationError , err_context )
# check if it is a valid file type
jar_type = topology_file . endswith ( ".jar" )
tar_type = topology_file . endswith ( ".tar" ) or topology_file . endswith ( ".tar.gz" )
pex_type = topology_file . endswith ( ".pex" )
cpp_type = topology_file . endswith ( ".dylib" ) or topology_file . endswith ( ".so" )
if not ( jar_type or tar_type or pex_type or cpp_type ) :
    _ , ext_name = os . path . splitext ( topology_file )
    err_context = "Unknown file type '%s'. Please use .tar " "or .tar.gz or .jar or .pex or .dylib or .so file" % ext_name
    return SimpleResult ( Status . InvocationError , err_context )
# check if extra launch classpath is provided and if it is validate
if cl_args [ 'extra_launch_classpath' ] :
    valid_classpath = classpath . valid_java_classpath ( cl_args [ 'extra_launch_classpath' ] )
    if not valid_classpath :
        err_context = "One of jar or directory in extra launch classpath does not exist: %s" % cl_args [ 'extra_launch_classpath' ]
        return SimpleResult ( Status . InvocationError , err_context )
# create a temporary directory for topology definition file
tmp_dir = tempfile . mkdtemp ( )
opts . cleaned_up_files . append ( tmp_dir )
# if topology needs to be launched in deactivated state , do it so
if cl_args [ 'deploy_deactivated' ] :
    initial_state = topology_pb2 . TopologyState . Name ( topology_pb2 . PAUSED )
else :
    initial_state = topology_pb2 . TopologyState . Name ( topology_pb2 . RUNNING )
# set the tmp dir and deactivated state in global options
opts . set_config ( 'cmdline.topologydefn.tmpdirectory' , tmp_dir )
opts . set_config ( 'cmdline.topology.initial.state' , initial_state )
opts . set_config ( 'cmdline.topology.role' , cl_args [ 'role' ] )
opts . set_config ( 'cmdline.topology.environment' , cl_args [ 'environ' ] )
# Use CLI release yaml file if the release _ yaml _ file config is empty
if not cl_args [ 'release_yaml_file' ] :
    cl_args [ 'release_yaml_file' ] = config . get_heron_release_file ( )
# check the extension of the file name to see if it is tar / jar file .
if jar_type :
    return submit_fatjar ( cl_args , unknown_args , tmp_dir )
elif tar_type :
    return submit_tar ( cl_args , unknown_args , tmp_dir )
elif cpp_type :
    return submit_cpp ( cl_args , unknown_args , tmp_dir )
else :
    return submit_pex ( cl_args , unknown_args , tmp_dir )


signature:
def get ( self ) :
    """get method"""

implementation:
try :
    cluster = self . get_argument_cluster ( )
    role = self . get_argument_role ( )
    environ = self . get_argument_environ ( )
    topology_name = self . get_argument_topology ( )
    container = self . get_argument ( constants . PARAM_CONTAINER )
    path = self . get_argument ( constants . PARAM_PATH )
    offset = self . get_argument_offset ( )
    length = self . get_argument_length ( )
    topology_info = self . tracker . getTopologyInfo ( topology_name , cluster , role , environ )
    stmgr_id = "stmgr-" + container
    stmgr = topology_info [ "physical_plan" ] [ "stmgrs" ] [ stmgr_id ]
    host = stmgr [ "host" ]
    shell_port = stmgr [ "shell_port" ]
    file_data_url = "http://%s:%d/filedata/%s?offset=%s&length=%s" % ( host , shell_port , path , offset , length )
    http_client = tornado . httpclient . AsyncHTTPClient ( )
    response = yield http_client . fetch ( file_data_url )
    self . write_success_response ( json . loads ( response . body ) )
    self . finish ( )
except Exception as e :
    Log . debug ( traceback . format_exc ( ) )
    self . write_error_response ( e )


signature:
def setup ( self , context ) :
    """Implements TextFile Generator ' s setup method"""

implementation:
myindex = context . get_partition_index ( )
self . _files_to_consume = self . _files [ myindex : : context . get_num_partitions ( ) ]
self . logger . info ( "TextFileSpout files to consume %s" % self . _files_to_consume )
self . _lines_to_consume = self . _get_next_lines ( )
self . _emit_count = 0


signature:
def add_config ( parser ) :
    """add config"""

implementation:
# the default config path
default_config_path = config . get_heron_conf_dir ( )
parser . add_argument ( '--config-path' , metavar = '(a string; path to cluster config; default: "' + default_config_path + '")' , default = os . path . join ( config . get_heron_dir ( ) , default_config_path ) )
return parser


signature:
def add_verbose ( parser ) :
    """add optional verbose argument"""

implementation:
parser . add_argument ( '--verbose' , metavar = '(a boolean; default: "false")' , type = bool , default = False )
return parser


signature:
def add_tracker_url ( parser ) :
    """add optional tracker _ url argument"""

implementation:
parser . add_argument ( '--tracker_url' , metavar = '(tracker url; default: "' + DEFAULT_TRACKER_URL + '")' , type = str , default = DEFAULT_TRACKER_URL )
return parser


signature:
def hex_escape ( bin_str ) :
    """Hex encode a binary string"""

implementation:
printable = string . ascii_letters + string . digits + string . punctuation + ' '
return '' . join ( ch if ch in printable else r'0x{0:02x}' . format ( ord ( ch ) ) for ch in bin_str )


signature:
def make_shell_endpoint ( topologyInfo , instance_id ) :
    """Makes the http endpoint for the heron shell
 if shell port is present , otherwise returns None ."""

implementation:
# Format : container _ < id > _ < instance _ id >
pplan = topologyInfo [ "physical_plan" ]
stmgrId = pplan [ "instances" ] [ instance_id ] [ "stmgrId" ]
host = pplan [ "stmgrs" ] [ stmgrId ] [ "host" ]
shell_port = pplan [ "stmgrs" ] [ stmgrId ] [ "shell_port" ]
return "http://%s:%d" % ( host , shell_port )


signature:
def make_shell_logfiles_url ( host , shell_port , _ , instance_id = None ) :
    """Make the url for log - files in heron - shell
 from the info stored in stmgr .
 If no instance _ id is provided , the link will
 be to the dir for the whole container .
 If shell port is not present , it returns None ."""

implementation:
if not shell_port :
    return None
if not instance_id :
    return "http://%s:%d/browse/log-files" % ( host , shell_port )
else :
    return "http://%s:%d/file/log-files/%s.log.0" % ( host , shell_port , instance_id )


signature:
def make_shell_logfile_data_url ( host , shell_port , instance_id , offset , length ) :
    """Make the url for log - file data in heron - shell
 from the info stored in stmgr ."""

implementation: return "http://%s:%d/filedata/log-files/%s.log.0?offset=%s&length=%s" % ( host , shell_port , instance_id , offset , length )


signature:
def cygpath ( x ) :
    """This will return the path of input arg for windows
 : return : the path in windows"""

implementation:
command = [ 'cygpath' , '-wp' , x ]
p = subprocess . Popen ( command , stdout = subprocess . PIPE )
output , _ = p . communicate ( )
lines = output . split ( "\n" )
return lines [ 0 ]


signature:
def get_heron_tracker_dir ( ) :
    """This will extract heron tracker directory from . pex file .
 : return : root location for heron - tools ."""

implementation:
path = "/" . join ( os . path . realpath ( __file__ ) . split ( '/' ) [ : - 8 ] )
return normalized_class_path ( path )


signature:
def parse_config_file ( config_file ) :
    """This will parse the config file for the tracker
 : return : the config or None if the file is not found"""

implementation:
expanded_config_file_path = os . path . expanduser ( config_file )
if not os . path . lexists ( expanded_config_file_path ) :
    return None
configs = { }
# Read the configuration file
with open ( expanded_config_file_path , 'r' ) as f :
    configs = yaml . load ( f )
return configs


signature:
def _handle_register_response ( self , response ) :
    """Called when a register response ( RegisterInstanceResponse ) arrives"""

implementation:
if response . status . status != common_pb2 . StatusCode . Value ( "OK" ) :
    raise RuntimeError ( "Stream Manager returned a not OK response for register" )
Log . info ( "We registered ourselves to the Stream Manager" )
self . is_registered = True
if response . HasField ( "pplan" ) :
    Log . info ( "Handling assignment message from response" )
    self . _handle_assignment_message ( response . pplan )
else :
    Log . debug ( "Received a register response with no pplan" )


signature:
def _handle_assignment_message ( self , pplan ) :
    """Called when new NewInstanceAssignmentMessage arrives"""

implementation:
Log . debug ( "In handle_assignment_message() of STStmgrClient, Physical Plan: \n%s" , str ( pplan ) )
self . heron_instance_cls . handle_assignment_msg ( pplan )


signature:
def decode_packet ( packet ) :
    """Decodes an IncomingPacket object and returns ( typename , reqid , serialized message )"""

implementation:
if not packet . is_complete :
    raise RuntimeError ( "In decode_packet(): Packet corrupted" )
data = packet . data
len_typename = HeronProtocol . unpack_int ( data [ : 4 ] )
data = data [ 4 : ]
typename = data [ : len_typename ]
data = data [ len_typename : ]
reqid = REQID . unpack ( data [ : REQID . REQID_SIZE ] )
data = data [ REQID . REQID_SIZE : ]
len_msg = HeronProtocol . unpack_int ( data [ : 4 ] )
data = data [ 4 : ]
serialized_msg = data [ : len_msg ]
return typename , reqid , serialized_msg


signature:
def create_packet ( reqid , message ) :
    """Creates Outgoing Packet from a given reqid and message
 : param reqid : REQID object
 : param message : protocol buffer object"""

implementation:
assert message . IsInitialized ( )
packet = ''
# calculate the totla size of the packet incl . header
typename = message . DESCRIPTOR . full_name
datasize = HeronProtocol . get_size_to_pack_string ( typename ) + REQID . REQID_SIZE + HeronProtocol . get_size_to_pack_message ( message )
# first write out how much data is there as the header
packet += HeronProtocol . pack_int ( datasize )
# next write the type string
packet += HeronProtocol . pack_int ( len ( typename ) )
packet += typename
# reqid
packet += reqid . pack ( )
# add the proto
packet += HeronProtocol . pack_int ( message . ByteSize ( ) )
packet += message . SerializeToString ( )
return OutgoingPacket ( packet )


signature:
def send ( self , dispatcher ) :
    """Sends this outgoing packet to dispatcher ' s socket"""

implementation:
if self . sent_complete :
    return
sent = dispatcher . send ( self . to_send )
self . to_send = self . to_send [ sent : ]