signature:
def server_mode_cluster_role_env ( cluster_role_env , config_map ) :
    """Check cluster / [ role ] / [ environ ] , if they are required"""

implementation:
cmap = config_map [ cluster_role_env [ 0 ] ]
# if role is required but not provided , raise exception
role_present = True if len ( cluster_role_env [ 1 ] ) > 0 else False
if ROLE_KEY in cmap and cmap [ ROLE_KEY ] and not role_present :
    raise Exception ( "role required but not provided (cluster/role/env = %s)." % ( cluster_role_env ) )
# if environ is required but not provided , raise exception
environ_present = True if len ( cluster_role_env [ 2 ] ) > 0 else False
if ENVIRON_KEY in cmap and cmap [ ENVIRON_KEY ] and not environ_present :
    raise Exception ( "environ required but not provided (cluster/role/env = %s)." % ( cluster_role_env ) )
return True


signature:
def defaults_cluster_role_env ( cluster_role_env ) :
    """if role is not provided , supply userid
 if environ is not provided , supply ' default '"""

implementation:
if len ( cluster_role_env [ 1 ] ) == 0 and len ( cluster_role_env [ 2 ] ) == 0 :
    return ( cluster_role_env [ 0 ] , getpass . getuser ( ) , ENVIRON )
return ( cluster_role_env [ 0 ] , cluster_role_env [ 1 ] , cluster_role_env [ 2 ] )


signature:
def parse_override_config_and_write_file ( namespace ) :
    """Parse the command line for overriding the defaults and
 create an override file ."""

implementation:
overrides = parse_override_config ( namespace )
try :
    tmp_dir = tempfile . mkdtemp ( )
    override_config_file = os . path . join ( tmp_dir , OVERRIDE_YAML )
    with open ( override_config_file , 'w' ) as f :
        f . write ( yaml . dump ( overrides ) )
    return override_config_file
except Exception as e :
    raise Exception ( "Failed to parse override config: %s" % str ( e ) )


signature:
def parse_override_config ( namespace ) :
    """Parse the command line for overriding the defaults"""

implementation:
overrides = dict ( )
for config in namespace :
    kv = config . split ( "=" )
    if len ( kv ) != 2 :
        raise Exception ( "Invalid config property format (%s) expected key=value" % config )
    if kv [ 1 ] in [ 'true' , 'True' , 'TRUE' ] :
        overrides [ kv [ 0 ] ] = True
    elif kv [ 1 ] in [ 'false' , 'False' , 'FALSE' ] :
        overrides [ kv [ 0 ] ] = False
    else :
        overrides [ kv [ 0 ] ] = kv [ 1 ]
return overrides


signature:
def get_java_path ( ) :
    """Get the path of java executable"""

implementation:
java_home = os . environ . get ( "JAVA_HOME" )
return os . path . join ( java_home , BIN_DIR , "java" )


signature:
def check_java_home_set ( ) :
    """Check if the java home set"""

implementation:
# check if environ variable is set
if "JAVA_HOME" not in os . environ :
    Log . error ( "JAVA_HOME not set" )
    return False
# check if the value set is correct
java_path = get_java_path ( )
if os . path . isfile ( java_path ) and os . access ( java_path , os . X_OK ) :
    return True
Log . error ( "JAVA_HOME/bin/java either does not exist or not an executable" )
return False


signature:
def check_release_file_exists ( ) :
    """Check if the release . yaml file exists"""

implementation:
release_file = get_heron_release_file ( )
# if the file does not exist and is not a file
if not os . path . isfile ( release_file ) :
    Log . error ( "Required file not found: %s" % release_file )
    return False
return True


signature:
def print_build_info ( zipped_pex = False ) :
    """Print build _ info from release . yaml
 : param zipped _ pex : True if the PEX file is built with flag ` zip _ safe = False ' ."""

implementation:
if zipped_pex :
    release_file = get_zipped_heron_release_file ( )
else :
    release_file = get_heron_release_file ( )
with open ( release_file ) as release_info :
    release_map = yaml . load ( release_info )
    release_items = sorted ( release_map . items ( ) , key = lambda tup : tup [ 0 ] )
    for key , value in release_items :
        print ( "%s : %s" % ( key , value ) )


signature:
def get_version_number ( zipped_pex = False ) :
    """Print version from release . yaml
 : param zipped _ pex : True if the PEX file is built with flag ` zip _ safe = False ' ."""

implementation:
if zipped_pex :
    release_file = get_zipped_heron_release_file ( )
else :
    release_file = get_heron_release_file ( )
with open ( release_file ) as release_info :
    for line in release_info :
        trunks = line [ : - 1 ] . split ( ' ' )
        if trunks [ 0 ] == 'heron.build.version' :
            return trunks [ - 1 ] . replace ( "'" , "" )
    return 'unknown'


signature:
def insert_bool ( param , command_args ) :
    ''': param param :
 : param command _ args :
 : return :'''

implementation:
index = 0
found = False
for lelem in command_args :
    if lelem == '--' and not found :
        break
    if lelem == param :
        found = True
        break
    index = index + 1
if found :
    command_args . insert ( index + 1 , 'True' )
return command_args


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

implementation:
# get the command for detailed help
command_help = args [ 'help-command' ]
# if no command is provided , just print main help
if command_help == 'help' :
    parser . print_help ( )
    return True
# get the subparser for the specific command
subparser = config . get_subparser ( parser , command_help )
if subparser :
    print ( subparser . format_help ( ) )
    return True
else :
    Log . error ( "Unknown subcommand \'%s\'" % command_help )
    return False


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

implementation:
try :
    cluster = self . get_argument_cluster ( )
    environ = self . get_argument_environ ( )
    role = self . get_argument_role ( )
    topology_name = self . get_argument_topology ( )
    component = self . get_argument_component ( )
    topology = self . tracker . getTopologyByClusterRoleEnvironAndName ( cluster , role , environ , topology_name )
    instances = self . get_arguments ( constants . PARAM_INSTANCE )
    exceptions_summary = yield tornado . gen . Task ( self . getComponentExceptionSummary , topology . tmaster , component , instances )
    self . write_success_response ( exceptions_summary )
except Exception as e :
    Log . debug ( traceback . format_exc ( ) )
    self . write_error_response ( e )


signature:
def getComponentExceptionSummary ( self , tmaster , component_name , instances = [ ] , callback = None ) :
    """Get the summary of exceptions for component _ name and list of instances .
 Empty instance list will fetch all exceptions ."""

implementation:
if not tmaster or not tmaster . host or not tmaster . stats_port :
    return
exception_request = tmaster_pb2 . ExceptionLogRequest ( )
exception_request . component_name = component_name
if len ( instances ) > 0 :
    exception_request . instances . extend ( instances )
request_str = exception_request . SerializeToString ( )
port = str ( tmaster . stats_port )
host = tmaster . host
url = "http://{0}:{1}/exceptionsummary" . format ( host , port )
Log . debug ( "Creating request object." )
request = tornado . httpclient . HTTPRequest ( url , body = request_str , method = 'POST' , request_timeout = 5 )
Log . debug ( 'Making HTTP call to fetch exceptionsummary url: %s' , url )
try :
    client = tornado . httpclient . AsyncHTTPClient ( )
    result = yield client . fetch ( request )
    Log . debug ( "HTTP call complete." )
except tornado . httpclient . HTTPError as e :
    raise Exception ( str ( e ) )
# Check the response code - error if it is in 400s or 500s
responseCode = result . code
if responseCode >= 400 :
    message = "Error in getting exceptions from Tmaster, code: " + responseCode
    Log . error ( message )
    raise tornado . gen . Return ( { "message" : message } )
# Parse the response from tmaster .
exception_response = tmaster_pb2 . ExceptionLogResponse ( )
exception_response . ParseFromString ( result . body )
if exception_response . status . status == common_pb2 . NOTOK :
    if exception_response . status . HasField ( "message" ) :
        raise tornado . gen . Return ( { "message" : exception_response . status . message } )
# Send response
ret = [ ]
for exception_log in exception_response . exceptions :
    ret . append ( { 'class_name' : exception_log . stacktrace , 'lasttime' : exception_log . lasttime , 'firsttime' : exception_log . firsttime , 'count' : str ( exception_log . count ) } )
raise tornado . gen . Return ( ret )


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

implementation:
# pylint : disable = no - member
options = dict ( cluster = cluster , environ = environ , topology = topology , active = "topologies" , function = common . className , baseUrl = self . baseUrl )
self . render ( "config.html" , ** options )


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

implementation:
clusters = yield access . get_clusters ( )
# pylint : disable = no - member
options = dict ( topologies = [ ] , # no topologies
clusters = [ str ( cluster ) for cluster in clusters ] , active = "topologies" , # active icon the nav bar
function = common . className , baseUrl = self . baseUrl )
# send the all topologies page
self . render ( "topologies.html" , ** options )


implementation:
# fetch the execution of the topology asynchronously
execution_state = yield access . get_execution_state ( cluster , environ , topology )
# fetch scheduler location of the topology
scheduler_location = yield access . get_scheduler_location ( cluster , environ , topology )
job_page_link = scheduler_location [ "job_page_link" ]
# convert the topology launch time to display format
launched_at = datetime . utcfromtimestamp ( execution_state [ 'submission_time' ] )
launched_time = launched_at . strftime ( '%Y-%m-%d %H:%M:%S UTC' )
# pylint : disable = no - member
options = dict ( cluster = cluster , environ = environ , topology = topology , execution_state = execution_state , launched = launched_time , status = "running" if random . randint ( 0 , 1 ) else "errors" , active = "topologies" , job_page_link = job_page_link , function = common . className , baseUrl = self . baseUrl )
# send the single topology page
self . render ( "topology.html" , ** options )


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

implementation:
path = self . get_argument ( "path" )
options = dict ( cluster = cluster , environ = environ , topology = topology , container = container , path = path , baseUrl = self . baseUrl )
self . render ( "file.html" , ** options )


implementation:
offset = self . get_argument ( "offset" )
length = self . get_argument ( "length" )
path = self . get_argument ( "path" )
data = yield access . get_container_file_data ( cluster , environ , topology , container , path , offset , length )
self . write ( data )
self . finish ( )


implementation:
path = self . get_argument ( "path" , default = "." )
data = yield access . get_filestats ( cluster , environ , topology , container , path )
options = dict ( cluster = cluster , environ = environ , topology = topology , container = container , path = path , filestats = data , baseUrl = self . baseUrl )
self . render ( "browse.html" , ** options )


implementation:
# If the file is large , we want to abandon downloading
# if user cancels the requests .
# pylint : disable = attribute - defined - outside - init
self . connection_closed = False
path = self . get_argument ( "path" )
filename = path . split ( "/" ) [ - 1 ]
self . set_header ( "Content-Disposition" , "attachment; filename=%s" % filename )
# Download the files in chunks . We are downloading from Tracker ,
# which in turns downloads from heron - shell . This much indirection
# means that if we use static file downloading , the whole files would
# be cached in memory before it can be sent downstream . Hence , we reuse
# the file data API to read in chunks until the EOF , or until the download
# is cancelled by user .
# 4 MB gives good enough chunk size giving good speed for small files .
# If files are large , a single threaded download may not be enough .
file_download_url = access . get_container_file_download_url ( cluster , environ , topology , container , path )
Log . debug ( "file download url: %s" , str ( file_download_url ) )
def streaming_callback ( chunk ) :
    self . write ( chunk )
    self . flush ( )
http_client = tornado . httpclient . AsyncHTTPClient ( )
yield http_client . fetch ( file_download_url , streaming_callback = streaming_callback )
self . finish ( )


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

implementation:
path = tornado . escape . url_unescape ( path )
if not path :
    path = "."
# User should not be able to access anything outside
# of the dir that heron - shell is running in . This ensures
# sandboxing . So we don ' t allow absolute paths and parent
# accessing .
if not utils . check_path ( path ) :
    self . write ( "Only relative paths are allowed" )
    self . set_status ( 403 )
    self . finish ( )
    return
listing = utils . get_listing ( path )
file_stats = { }
for fn in listing :
    try :
        is_dir = False
        formatted_stat = utils . format_prefix ( fn , utils . get_stat ( path , fn ) )
        if stat . S_ISDIR ( utils . get_stat ( path , fn ) . st_mode ) :
            is_dir = True
        file_stats [ fn ] = { "formatted_stat" : formatted_stat , "is_dir" : is_dir , "path" : tornado . escape . url_escape ( os . path . join ( path , fn ) ) , }
        if fn == ".." :
            path_fragments = path . split ( "/" )
            if not path_fragments :
                file_stats [ fn ] [ "path" ] = "."
            else :
                file_stats [ fn ] [ "path" ] = tornado . escape . url_escape ( "/" . join ( path_fragments [ : - 1 ] ) )
    except :
        continue
self . write ( json . dumps ( file_stats ) )
self . finish ( )


signature:
def register_watch ( self , callback ) :
    """Returns the UUID with which the watch is
 registered . This UUID can be used to unregister
 the watch .
 Returns None if watch could not be registered .
 The argument ' callback ' must be a function that takes
 exactly one argument , the topology on which
 the watch was triggered .
 Note that the watch will be unregistered in case
 it raises any Exception the first time .
 This callback is also called at the time
 of registration ."""

implementation:
RETRY_COUNT = 5
# Retry in case UID is previously
# generated , just in case . . .
for _ in range ( RETRY_COUNT ) : # Generate a random UUID .
    uid = uuid . uuid4 ( )
    if uid not in self . watches :
        Log . info ( "Registering a watch with uid: " + str ( uid ) )
        try :
            callback ( self )
        except Exception as e :
            Log . error ( "Caught exception while triggering callback: " + str ( e ) )
            Log . debug ( traceback . format_exc ( ) )
            return None
        self . watches [ uid ] = callback
        return uid
return None


signature:
def unregister_watch ( self , uid ) :
    """Unregister the watch with the given UUID ."""

implementation:
# Do not raise an error if UUID is
# not present in the watches .
Log . info ( "Unregister a watch with uid: " + str ( uid ) )
self . watches . pop ( uid , None )


signature:
def trigger_watches ( self ) :
    """Call all the callbacks .
 If any callback raises an Exception ,
 unregister the corresponding watch ."""

implementation:
to_remove = [ ]
for uid , callback in self . watches . items ( ) :
    try :
        callback ( self )
    except Exception as e :
        Log . error ( "Caught exception while triggering callback: " + str ( e ) )
        Log . debug ( traceback . format_exc ( ) )
        to_remove . append ( uid )
for uid in to_remove :
    self . unregister_watch ( uid )


signature:
def set_physical_plan ( self , physical_plan ) :
    """set physical plan"""

implementation:
if not physical_plan :
    self . physical_plan = None
    self . id = None
else :
    self . physical_plan = physical_plan
    self . id = physical_plan . topology . id
self . trigger_watches ( )


signature:
def set_packing_plan ( self , packing_plan ) :
    """set packing plan"""

implementation:
if not packing_plan :
    self . packing_plan = None
    self . id = None
else :
    self . packing_plan = packing_plan
    self . id = packing_plan . id
self . trigger_watches ( )


signature:
def set_execution_state ( self , execution_state ) :
    """set exectuion state"""

implementation:
if not execution_state :
    self . execution_state = None
    self . cluster = None
    self . environ = None
else :
    self . execution_state = execution_state
    cluster , environ = self . get_execution_state_dc_environ ( execution_state )
    self . cluster = cluster
    self . environ = environ
    self . zone = cluster
self . trigger_watches ( )


signature:
def num_instances ( self ) :
    """Number of spouts + bolts"""

implementation:
num = 0
# Get all the components
components = self . spouts ( ) + self . bolts ( )
# Get instances for each worker
for component in components :
    config = component . comp . config
    for kvs in config . kvs :
        if kvs . key == api_constants . TOPOLOGY_COMPONENT_PARALLELISM :
            num += int ( kvs . value )
            break
return num


signature:
def get_machines ( self ) :
    """Get all the machines that this topology is running on .
 These are the hosts of all the stmgrs ."""

implementation:
if self . physical_plan :
    stmgrs = list ( self . physical_plan . stmgrs )
    return map ( lambda s : s . host_name , stmgrs )
return [ ]


signature:
def get_status ( self ) :
    """Get the current state of this topology .
 The state values are from the topology . proto
 RUNNING = 1 , PAUSED = 2 , KILLED = 3
 if the state is None " Unknown " is returned ."""

implementation:
status = None
if self . physical_plan and self . physical_plan . topology :
    status = self . physical_plan . topology . state
if status == 1 :
    return "Running"
elif status == 2 :
    return "Paused"
elif status == 3 :
    return "Killed"
else :
    return "Unknown"


signature:
def convert_pb_kvs ( kvs , include_non_primitives = True ) :
    """converts pb kvs to dict"""

implementation:
config = { }
for kv in kvs :
    if kv . value :
        config [ kv . key ] = kv . value
    elif kv . serialized_value : # add serialized _ value support for python values ( fixme )
    # is this a serialized java object
        if topology_pb2 . JAVA_SERIALIZED_VALUE == kv . type :
            jv = _convert_java_value ( kv , include_non_primitives = include_non_primitives )
            if jv is not None :
                config [ kv . key ] = jv
        else :
            config [ kv . key ] = _raw_value ( kv )
return config


signature:
def getTopologyByClusterRoleEnvironAndName ( self , cluster , role , environ , topologyName ) :
    """Find and return the topology given its cluster , environ , topology name , and
 an optional role .
 Raises exception if topology is not found , or more than one are found ."""

implementation:
topologies = list ( filter ( lambda t : t . name == topologyName and t . cluster == cluster and ( not role or t . execution_state . role == role ) and t . environ == environ , self . topologies ) )
if not topologies or len ( topologies ) > 1 :
    if role is not None :
        raise Exception ( "Topology not found for {0}, {1}, {2}, {3}" . format ( cluster , role , environ , topologyName ) )
    else :
        raise Exception ( "Topology not found for {0}, {1}, {2}" . format ( cluster , environ , topologyName ) )
# There is only one topology which is returned .
return topologies [ 0 ]


signature:
def getTopologiesForStateLocation ( self , name ) :
    """Returns all the topologies for a given state manager ."""

implementation: return filter ( lambda t : t . state_manager_name == name , self . topologies )


signature:
def removeTopology ( self , topology_name , state_manager_name ) :
    """Removes the topology from the local cache ."""

implementation:
topologies = [ ]
for top in self . topologies :
    if ( top . name == topology_name and top . state_manager_name == state_manager_name ) : # Remove topologyInfo
        if ( topology_name , state_manager_name ) in self . topologyInfos :
            self . topologyInfos . pop ( ( topology_name , state_manager_name ) )
    else :
        topologies . append ( top )
self . topologies = topologies


signature:
def extract_execution_state ( self , topology ) :
    """Returns the repesentation of execution state that will
 be returned from Tracker ."""

implementation:
execution_state = topology . execution_state
executionState = { "cluster" : execution_state . cluster , "environ" : execution_state . environ , "role" : execution_state . role , "jobname" : topology . name , "submission_time" : execution_state . submission_time , "submission_user" : execution_state . submission_user , "release_username" : execution_state . release_state . release_username , "release_tag" : execution_state . release_state . release_tag , "release_version" : execution_state . release_state . release_version , "has_physical_plan" : None , "has_tmaster_location" : None , "has_scheduler_location" : None , "extra_links" : [ ] , }
for extra_link in self . config . extra_links :
    link = extra_link . copy ( )
    link [ "url" ] = self . config . get_formatted_url ( executionState , link [ EXTRA_LINK_FORMATTER_KEY ] )
    executionState [ "extra_links" ] . append ( link )
return executionState


signature:
def extract_scheduler_location ( self , topology ) :
    """Returns the representation of scheduler location that will
 be returned from Tracker ."""

implementation:
schedulerLocation = { "name" : None , "http_endpoint" : None , "job_page_link" : None , }
if topology . scheduler_location :
    schedulerLocation [ "name" ] = topology . scheduler_location . topology_name
    schedulerLocation [ "http_endpoint" ] = topology . scheduler_location . http_endpoint
    schedulerLocation [ "job_page_link" ] = topology . scheduler_location . job_page_link [ 0 ] if len ( topology . scheduler_location . job_page_link ) > 0 else ""
return schedulerLocation


signature:
def extract_tmaster ( self , topology ) :
    """Returns the representation of tmaster that will
 be returned from Tracker ."""

implementation:
tmasterLocation = { "name" : None , "id" : None , "host" : None , "controller_port" : None , "master_port" : None , "stats_port" : None , }
if topology . tmaster :
    tmasterLocation [ "name" ] = topology . tmaster . topology_name
    tmasterLocation [ "id" ] = topology . tmaster . topology_id
    tmasterLocation [ "host" ] = topology . tmaster . host
    tmasterLocation [ "controller_port" ] = topology . tmaster . controller_port
    tmasterLocation [ "master_port" ] = topology . tmaster . master_port
    tmasterLocation [ "stats_port" ] = topology . tmaster . stats_port
return tmasterLocation


signature:
def extract_logical_plan ( self , topology ) :
    """Returns the representation of logical plan that will
 be returned from Tracker ."""

implementation:
logicalPlan = { "spouts" : { } , "bolts" : { } , }
# Add spouts .
for spout in topology . spouts ( ) :
    spoutName = spout . comp . name
    spoutType = "default"
    spoutSource = "NA"
    spoutVersion = "NA"
    spoutConfigs = spout . comp . config . kvs
    for kvs in spoutConfigs :
        if kvs . key == "spout.type" :
            spoutType = javaobj . loads ( kvs . serialized_value )
        elif kvs . key == "spout.source" :
            spoutSource = javaobj . loads ( kvs . serialized_value )
        elif kvs . key == "spout.version" :
            spoutVersion = javaobj . loads ( kvs . serialized_value )
    spoutPlan = { "config" : convert_pb_kvs ( spoutConfigs , include_non_primitives = False ) , "type" : spoutType , "source" : spoutSource , "version" : spoutVersion , "outputs" : [ ] }
    for outputStream in list ( spout . outputs ) :
        spoutPlan [ "outputs" ] . append ( { "stream_name" : outputStream . stream . id } )
    logicalPlan [ "spouts" ] [ spoutName ] = spoutPlan
# Add bolts .
for bolt in topology . bolts ( ) :
    boltName = bolt . comp . name
    boltPlan = { "config" : convert_pb_kvs ( bolt . comp . config . kvs , include_non_primitives = False ) , "outputs" : [ ] , "inputs" : [ ] }
    for outputStream in list ( bolt . outputs ) :
        boltPlan [ "outputs" ] . append ( { "stream_name" : outputStream . stream . id } )
    for inputStream in list ( bolt . inputs ) :
        boltPlan [ "inputs" ] . append ( { "stream_name" : inputStream . stream . id , "component_name" : inputStream . stream . component_name , "grouping" : topology_pb2 . Grouping . Name ( inputStream . gtype ) } )
    logicalPlan [ "bolts" ] [ boltName ] = boltPlan
return logicalPlan


signature:
def extract_physical_plan ( self , topology ) :
    """Returns the representation of physical plan that will
 be returned from Tracker ."""

implementation:
physicalPlan = { "instances" : { } , "instance_groups" : { } , "stmgrs" : { } , "spouts" : { } , "bolts" : { } , "config" : { } , "components" : { } }
if not topology . physical_plan :
    return physicalPlan
spouts = topology . spouts ( )
bolts = topology . bolts ( )
stmgrs = None
instances = None
# Physical Plan
stmgrs = list ( topology . physical_plan . stmgrs )
instances = list ( topology . physical_plan . instances )
# Configs
if topology . physical_plan . topology . topology_config :
    physicalPlan [ "config" ] = convert_pb_kvs ( topology . physical_plan . topology . topology_config . kvs )
for spout in spouts :
    spout_name = spout . comp . name
    physicalPlan [ "spouts" ] [ spout_name ] = [ ]
    if spout_name not in physicalPlan [ "components" ] :
        physicalPlan [ "components" ] [ spout_name ] = { "config" : convert_pb_kvs ( spout . comp . config . kvs ) }
for bolt in bolts :
    bolt_name = bolt . comp . name
    physicalPlan [ "bolts" ] [ bolt_name ] = [ ]
    if bolt_name not in physicalPlan [ "components" ] :
        physicalPlan [ "components" ] [ bolt_name ] = { "config" : convert_pb_kvs ( bolt . comp . config . kvs ) }
for stmgr in stmgrs :
    host = stmgr . host_name
    cwd = stmgr . cwd
    shell_port = stmgr . shell_port if stmgr . HasField ( "shell_port" ) else None
    physicalPlan [ "stmgrs" ] [ stmgr . id ] = { "id" : stmgr . id , "host" : host , "port" : stmgr . data_port , "shell_port" : shell_port , "cwd" : cwd , "pid" : stmgr . pid , "joburl" : utils . make_shell_job_url ( host , shell_port , cwd ) , "logfiles" : utils . make_shell_logfiles_url ( host , shell_port , cwd ) , "instance_ids" : [ ] }
instance_groups = collections . OrderedDict ( )
for instance in instances :
    instance_id = instance . instance_id
    stmgrId = instance . stmgr_id
    name = instance . info . component_name
    stmgrInfo = physicalPlan [ "stmgrs" ] [ stmgrId ]
    host = stmgrInfo [ "host" ]
    cwd = stmgrInfo [ "cwd" ]
    shell_port = stmgrInfo [ "shell_port" ]
    # instance _ id format container _ < index > _ component _ 1
    # group name is container _ < index >
    group_name = instance_id . rsplit ( "_" , 2 ) [ 0 ]
    igroup = instance_groups . get ( group_name , list ( ) )
    igroup . append ( instance_id )
    instance_groups [ group_name ] = igroup
    physicalPlan [ "instances" ] [ instance_id ] = { "id" : instance_id , "name" : name , "stmgrId" : stmgrId , "logfile" : utils . make_shell_logfiles_url ( host , shell_port , cwd , instance . instance_id ) , }
    physicalPlan [ "stmgrs" ] [ stmgrId ] [ "instance_ids" ] . append ( instance_id )
    if name in physicalPlan [ "spouts" ] :
        physicalPlan [ "spouts" ] [ name ] . append ( instance_id )
    else :
        physicalPlan [ "bolts" ] [ name ] . append ( instance_id )
physicalPlan [ "instance_groups" ] = instance_groups
return physicalPlan


signature:
def extract_packing_plan ( self , topology ) :
    """Returns the representation of packing plan that will
 be returned from Tracker ."""

implementation:
packingPlan = { "id" : "" , "container_plans" : [ ] }
if not topology . packing_plan :
    return packingPlan
container_plans = topology . packing_plan . container_plans
containers = [ ]
for container_plan in container_plans :
    instances = [ ]
    for instance_plan in container_plan . instance_plans :
        instance_resources = { "cpu" : instance_plan . resource . cpu , "ram" : instance_plan . resource . ram , "disk" : instance_plan . resource . disk }
        instance = { "component_name" : instance_plan . component_name , "task_id" : instance_plan . task_id , "component_index" : instance_plan . component_index , "instance_resources" : instance_resources }
        instances . append ( instance )
    required_resource = { "cpu" : container_plan . requiredResource . cpu , "ram" : container_plan . requiredResource . ram , "disk" : container_plan . requiredResource . disk }
    scheduled_resource = { }
    if container_plan . scheduledResource :
        scheduled_resource = { "cpu" : container_plan . scheduledResource . cpu , "ram" : container_plan . scheduledResource . ram , "disk" : container_plan . scheduledResource . disk }
    container = { "id" : container_plan . id , "instances" : instances , "required_resources" : required_resource , "scheduled_resources" : scheduled_resource }
    containers . append ( container )
packingPlan [ "id" ] = topology . packing_plan . id
packingPlan [ "container_plans" ] = containers
return json . dumps ( packingPlan )


signature:
def setTopologyInfo ( self , topology ) :
    """Extracts info from the stored proto states and
 convert it into representation that is exposed using
 the API .
 This method is called on any change for the topology .
 For example , when a container moves and its host or some
 port changes . All the information is parsed all over
 again and cache is updated ."""

implementation:
# Execution state is the most basic info .
# If there is no execution state , just return
# as the rest of the things don ' t matter .
if not topology . execution_state :
    Log . info ( "No execution state found for: " + topology . name )
    return
Log . info ( "Setting topology info for topology: " + topology . name )
has_physical_plan = True
if not topology . physical_plan :
    has_physical_plan = False
Log . info ( "Setting topology info for topology: " + topology . name )
has_packing_plan = True
if not topology . packing_plan :
    has_packing_plan = False
has_tmaster_location = True
if not topology . tmaster :
    has_tmaster_location = False
has_scheduler_location = True
if not topology . scheduler_location :
    has_scheduler_location = False
topologyInfo = { "name" : topology . name , "id" : topology . id , "logical_plan" : None , "physical_plan" : None , "packing_plan" : None , "execution_state" : None , "tmaster_location" : None , "scheduler_location" : None , }
executionState = self . extract_execution_state ( topology )
executionState [ "has_physical_plan" ] = has_physical_plan
executionState [ "has_packing_plan" ] = has_packing_plan
executionState [ "has_tmaster_location" ] = has_tmaster_location
executionState [ "has_scheduler_location" ] = has_scheduler_location
executionState [ "status" ] = topology . get_status ( )
topologyInfo [ "metadata" ] = self . extract_metadata ( topology )
topologyInfo [ "runtime_state" ] = self . extract_runtime_state ( topology )
topologyInfo [ "execution_state" ] = executionState
topologyInfo [ "logical_plan" ] = self . extract_logical_plan ( topology )
topologyInfo [ "physical_plan" ] = self . extract_physical_plan ( topology )
topologyInfo [ "packing_plan" ] = self . extract_packing_plan ( topology )
topologyInfo [ "tmaster_location" ] = self . extract_tmaster ( topology )
topologyInfo [ "scheduler_location" ] = self . extract_scheduler_location ( topology )
self . topologyInfos [ ( topology . name , topology . state_manager_name ) ] = topologyInfo


signature:
def getTopologyInfo ( self , topologyName , cluster , role , environ ) :
    """Returns the JSON representation of a topology
 by its name , cluster , environ , and an optional role parameter .
 Raises exception if no such topology is found ."""

implementation:
# Iterate over the values to filter the desired topology .
for ( topology_name , _ ) , topologyInfo in self . topologyInfos . items ( ) :
    executionState = topologyInfo [ "execution_state" ]
    if ( topologyName == topology_name and cluster == executionState [ "cluster" ] and environ == executionState [ "environ" ] ) : # If role is specified , first try to match " role " field . If " role " field
    # does not exist , try to match " submission _ user " field .
        if not role or executionState . get ( "role" ) == role :
            return topologyInfo
if role is not None :
    Log . info ( "Could not find topology info for topology: %s," "cluster: %s, role: %s, and environ: %s" , topologyName , cluster , role , environ )
else :
    Log . info ( "Could not find topology info for topology: %s," "cluster: %s and environ: %s" , topologyName , cluster , environ )
raise Exception ( "No topology found" )


signature:
def load_configs ( self ) :
    """load config files"""

implementation:
self . statemgr_config . set_state_locations ( self . configs [ STATEMGRS_KEY ] )
if EXTRA_LINKS_KEY in self . configs :
    for extra_link in self . configs [ EXTRA_LINKS_KEY ] :
        self . extra_links . append ( self . validate_extra_link ( extra_link ) )


signature:
def validate_extra_link ( self , extra_link ) :
    """validate extra link"""

implementation:
if EXTRA_LINK_NAME_KEY not in extra_link or EXTRA_LINK_FORMATTER_KEY not in extra_link :
    raise Exception ( "Invalid extra.links format. " + "Extra link must include a 'name' and 'formatter' field" )
self . validated_formatter ( extra_link [ EXTRA_LINK_FORMATTER_KEY ] )
return extra_link


signature:
def validated_formatter ( self , url_format ) :
    """validate visualization url format"""

implementation:
# We try to create a string by substituting all known
# parameters . If an unknown parameter is present , an error
# will be thrown
valid_parameters = { "${CLUSTER}" : "cluster" , "${ENVIRON}" : "environ" , "${TOPOLOGY}" : "topology" , "${ROLE}" : "role" , "${USER}" : "user" , }
dummy_formatted_url = url_format
for key , value in valid_parameters . items ( ) :
    dummy_formatted_url = dummy_formatted_url . replace ( key , value )
# All $ signs must have been replaced
if '$' in dummy_formatted_url :
    raise Exception ( "Invalid viz.url.format: %s" % ( url_format ) )
# No error is thrown , so the format is valid .
return url_format


signature:
def emit ( self , tup , tup_id = None , stream = Stream . DEFAULT_STREAM_ID , direct_task = None , need_task_ids = False ) :
    """Emits a new tuple from this Spout
 It is compatible with StreamParse API .
 : type tup : list or tuple
 : param tup : the new output Tuple to send from this spout ,
 should contain only serializable data .
 : type tup _ id : str or object
 : param tup _ id : the ID for the Tuple . Leave this blank for an unreliable emit .
 ( Same as messageId in Java )
 : type stream : str
 : param stream : the ID of the stream this Tuple should be emitted to .
 Leave empty to emit to the default stream .
 : type direct _ task : int
 : param direct _ task : the task to send the Tuple to if performing a direct emit .
 : type need _ task _ ids : bool
 : param need _ task _ ids : indicate whether or not you would like the task IDs the Tuple was emitted ."""

implementation:
# first check whether this tuple is sane
self . pplan_helper . check_output_schema ( stream , tup )
# get custom grouping target task ids ; get empty list if not custom grouping
custom_target_task_ids = self . pplan_helper . choose_tasks_for_custom_grouping ( stream , tup )
self . pplan_helper . context . invoke_hook_emit ( tup , stream , None )
data_tuple = tuple_pb2 . HeronDataTuple ( )
data_tuple . key = 0
if direct_task is not None :
    if not isinstance ( direct_task , int ) :
        raise TypeError ( "direct_task argument needs to be an integer, given: %s" % str ( type ( direct_task ) ) )
    # performing emit - direct
    data_tuple . dest_task_ids . append ( direct_task )
elif custom_target_task_ids is not None : # for custom grouping
    for task_id in custom_target_task_ids :
        data_tuple . dest_task_ids . append ( task_id )
if tup_id is not None :
    tuple_info = TupleHelper . make_root_tuple_info ( stream , tup_id )
    if self . acking_enabled : # this message is rooted
        root = data_tuple . roots . add ( )
        root . taskid = self . pplan_helper . my_task_id
        root . key = tuple_info . key
        self . in_flight_tuples [ tuple_info . key ] = tuple_info
    else :
        self . immediate_acks . append ( tuple_info )
tuple_size_in_bytes = 0
start_time = time . time ( )
# Serialize
for obj in tup :
    serialized = self . serializer . serialize ( obj )
    data_tuple . values . append ( serialized )
    tuple_size_in_bytes += len ( serialized )
serialize_latency_ns = ( time . time ( ) - start_time ) * system_constants . SEC_TO_NS
self . spout_metrics . serialize_data_tuple ( stream , serialize_latency_ns )
super ( SpoutInstance , self ) . admit_data_tuple ( stream_id = stream , data_tuple = data_tuple , tuple_size_in_bytes = tuple_size_in_bytes )
self . total_tuples_emitted += 1
self . spout_metrics . update_emit_count ( stream )
if need_task_ids :
    sent_task_ids = custom_target_task_ids or [ ]
    if direct_task is not None :
        sent_task_ids . append ( direct_task )
    return sent_task_ids


signature:
def _is_continue_to_work ( self ) :
    """Checks whether we still need to do more work
 When the topology state is RUNNING :
 1 . if the out _ queue is not full and ack is not enabled , we could wake up next time to
 produce more tuples and push to the out _ queue
 2 . if the out _ queue is not full but the acking is enabled , we need to make sure that
 the number of pending tuples is smaller than max _ spout _ pending
 3 . if there are more to read , we will wake up itself next time ."""

implementation:
if not self . _is_topology_running ( ) :
    return False
max_spout_pending = self . pplan_helper . context . get_cluster_config ( ) . get ( api_constants . TOPOLOGY_MAX_SPOUT_PENDING )
if not self . acking_enabled and self . output_helper . is_out_queue_available ( ) :
    return True
elif self . acking_enabled and self . output_helper . is_out_queue_available ( ) and len ( self . in_flight_tuples ) < max_spout_pending :
    return True
elif self . acking_enabled and not self . in_stream . is_empty ( ) :
    return True
else :
    return False


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

implementation:
components_parser = subparsers . add_parser ( 'components' , help = 'Display information of a topology\'s components' , usage = "%(prog)s cluster/[role]/[env] topology-name [options]" , add_help = False )
args . add_cluster_role_env ( components_parser )
args . add_topology_name ( components_parser )
args . add_spouts ( components_parser )
args . add_bolts ( components_parser )
args . add_verbose ( components_parser )
args . add_tracker_url ( components_parser )
args . add_config ( components_parser )
components_parser . set_defaults ( subcommand = 'components' )
return subparsers


signature:
def to_table ( components , topo_info ) :
    """normalize raw logical plan info to table"""

implementation:
inputs , outputs = defaultdict ( list ) , defaultdict ( list )
for ctype , component in components . items ( ) :
    if ctype == 'bolts' :
        for component_name , component_info in component . items ( ) :
            for input_stream in component_info [ 'inputs' ] :
                input_name = input_stream [ 'component_name' ]
                inputs [ component_name ] . append ( input_name )
                outputs [ input_name ] . append ( component_name )
info = [ ]
spouts_instance = topo_info [ 'physical_plan' ] [ 'spouts' ]
bolts_instance = topo_info [ 'physical_plan' ] [ 'bolts' ]
for ctype , component in components . items ( ) : # stages is an int so keep going
    if ctype == "stages" :
        continue
    for component_name , component_info in component . items ( ) :
        row = [ ctype [ : - 1 ] , component_name ]
        if ctype == 'spouts' :
            row . append ( len ( spouts_instance [ component_name ] ) )
        else :
            row . append ( len ( bolts_instance [ component_name ] ) )
        row . append ( ',' . join ( inputs . get ( component_name , [ '-' ] ) ) )
        row . append ( ',' . join ( outputs . get ( component_name , [ '-' ] ) ) )
        info . append ( row )
header = [ 'type' , 'name' , 'parallelism' , 'input' , 'output' ]
return info , header


signature:
def filter_bolts ( table , header ) :
    """filter to keep bolts"""

implementation:
bolts_info = [ ]
for row in table :
    if row [ 0 ] == 'bolt' :
        bolts_info . append ( row )
return bolts_info , header


signature:
def filter_spouts ( table , header ) :
    """filter to keep spouts"""

implementation:
spouts_info = [ ]
for row in table :
    if row [ 0 ] == 'spout' :
        spouts_info . append ( row )
return spouts_info , header


signature:
def run ( cl_args , compo_type ) :
    """run command"""

implementation:
cluster , role , env = cl_args [ 'cluster' ] , cl_args [ 'role' ] , cl_args [ 'environ' ]
topology = cl_args [ 'topology-name' ]
spouts_only , bolts_only = cl_args [ 'spout' ] , cl_args [ 'bolt' ]
try :
    components = tracker_access . get_logical_plan ( cluster , env , topology , role )
    topo_info = tracker_access . get_topology_info ( cluster , env , topology , role )
    table , header = to_table ( components , topo_info )
    if spouts_only == bolts_only :
        print ( tabulate ( table , headers = header ) )
    elif spouts_only :
        table , header = filter_spouts ( table , header )
        print ( tabulate ( table , headers = header ) )
    else :
        table , header = filter_bolts ( table , header )
        print ( tabulate ( table , headers = header ) )
    return True
except :
    Log . error ( "Fail to connect to tracker: \'%s\'" , cl_args [ "tracker_url" ] )
    return False


signature:
def create_topology ( self , topologyName , topology ) :
    """crate topology"""

implementation:
if not topology or not topology . IsInitialized ( ) :
    raise_ ( StateException ( "Topology protobuf not init properly" , StateException . EX_TYPE_PROTOBUF_ERROR ) , sys . exc_info ( ) [ 2 ] )
path = self . get_topology_path ( topologyName )
LOG . info ( "Adding topology: {0} to path: {1}" . format ( topologyName , path ) )
topologyString = topology . SerializeToString ( )
try :
    self . client . create ( path , value = topologyString , makepath = True )
    return True
except NoNodeError :
    raise_ ( StateException ( "NoNodeError while creating topology" , StateException . EX_TYPE_NO_NODE_ERROR ) , sys . exc_info ( ) [ 2 ] )
except NodeExistsError :
    raise_ ( StateException ( "NodeExistsError while creating topology" , StateException . EX_TYPE_NODE_EXISTS_ERROR ) , sys . exc_info ( ) [ 2 ] )
except ZookeeperError :
    raise_ ( StateException ( "Zookeeper while creating topology" , StateException . EX_TYPE_ZOOKEEPER_ERROR ) , sys . exc_info ( ) [ 2 ] )
except Exception : # Just re raise the exception .
    raise


signature:
def delete_topology ( self , topologyName ) :
    """delete topology"""

implementation:
path = self . get_topology_path ( topologyName )
LOG . info ( "Removing topology: {0} from path: {1}" . format ( topologyName , path ) )
try :
    self . client . delete ( path )
    return True
except NoNodeError :
    raise_ ( StateException ( "NoNodeError while deteling topology" , StateException . EX_TYPE_NO_NODE_ERROR ) , sys . exc_info ( ) [ 2 ] )
except NotEmptyError :
    raise_ ( StateException ( "NotEmptyError while deleting topology" , StateException . EX_TYPE_NOT_EMPTY_ERROR ) , sys . exc_info ( ) [ 2 ] )
except ZookeeperError :
    raise_ ( StateException ( "Zookeeper while deleting topology" , StateException . EX_TYPE_ZOOKEEPER_ERROR ) , sys . exc_info ( ) [ 2 ] )
except Exception : # Just re raise the exception .
    raise


signature:
def create_pplan ( self , topologyName , pplan ) :
    """create physical plan"""

implementation:
if not pplan or not pplan . IsInitialized ( ) :
    raise_ ( StateException ( "Physical Plan protobuf not init properly" , StateException . EX_TYPE_PROTOBUF_ERROR ) , sys . exc_info ( ) [ 2 ] )
path = self . get_pplan_path ( topologyName )
LOG . info ( "Adding topology: {0} to path: {1}" . format ( topologyName , path ) )
pplanString = pplan . SerializeToString ( )
try :
    self . client . create ( path , value = pplanString , makepath = True )
    return True
except NoNodeError :
    raise_ ( StateException ( "NoNodeError while creating pplan" , StateException . EX_TYPE_NO_NODE_ERROR ) , sys . exc_info ( ) [ 2 ] )
except NodeExistsError :
    raise_ ( StateException ( "NodeExistsError while creating pplan" , StateException . EX_TYPE_NODE_EXISTS_ERROR ) , sys . exc_info ( ) [ 2 ] )
except ZookeeperError :
    raise_ ( StateException ( "Zookeeper while creating pplan" , StateException . EX_TYPE_ZOOKEEPER_ERROR ) , sys . exc_info ( ) [ 2 ] )
except Exception : # Just re raise the exception .
    raise


signature:
def create_execution_state ( self , topologyName , executionState ) :
    """create execution state"""

implementation:
if not executionState or not executionState . IsInitialized ( ) :
    raise_ ( StateException ( "Execution State protobuf not init properly" , StateException . EX_TYPE_PROTOBUF_ERROR ) , sys . exc_info ( ) [ 2 ] )
path = self . get_execution_state_path ( topologyName )
LOG . info ( "Adding topology: {0} to path: {1}" . format ( topologyName , path ) )
executionStateString = executionState . SerializeToString ( )
try :
    self . client . create ( path , value = executionStateString , makepath = True )
    return True
except NoNodeError :
    raise_ ( StateException ( "NoNodeError while creating execution state" , StateException . EX_TYPE_NO_NODE_ERROR ) , sys . exc_info ( ) [ 2 ] )
except NodeExistsError :
    raise_ ( StateException ( "NodeExistsError while creating execution state" , StateException . EX_TYPE_NODE_EXISTS_ERROR ) , sys . exc_info ( ) [ 2 ] )
except ZookeeperError :
    raise_ ( StateException ( "Zookeeper while creating execution state" , StateException . EX_TYPE_ZOOKEEPER_ERROR ) , sys . exc_info ( ) [ 2 ] )
except Exception : # Just re raise the exception .
    raise


signature:
def load ( file_object ) :
    """Deserializes Java primitive data and objects serialized by ObjectOutputStream
 from a file - like object ."""

implementation:
marshaller = JavaObjectUnmarshaller ( file_object )
marshaller . add_transformer ( DefaultObjectTransformer ( ) )
return marshaller . readObject ( )


signature:
def loads ( string ) :
    """Deserializes Java objects and primitive data serialized by ObjectOutputStream
 from a string ."""

implementation:
f = StringIO . StringIO ( string )
marshaller = JavaObjectUnmarshaller ( f )
marshaller . add_transformer ( DefaultObjectTransformer ( ) )
return marshaller . readObject ( )


signature:
def copy ( self , new_object ) :
    """copy an object"""

implementation:
new_object . classdesc = self . classdesc
for name in self . classdesc . fields_names :
    new_object . __setattr__ ( name , getattr ( self , name ) )


signature:
def readObject ( self ) :
    """read object"""

implementation:
try :
    _ , res = self . _read_and_exec_opcode ( ident = 0 )
    position_bak = self . object_stream . tell ( )
    the_rest = self . object_stream . read ( )
    if len ( the_rest ) :
        log_error ( "Warning!!!!: Stream still has %s bytes left.\
Enable debug mode of logging to see the hexdump." % len ( the_rest ) )
        log_debug ( self . _create_hexdump ( the_rest ) )
    else :
        log_debug ( "Java Object unmarshalled succesfully!" )
    self . object_stream . seek ( position_bak )
    return res
except Exception :
    self . _oops_dump_state ( )
    raise


signature:
def do_classdesc ( self , parent = None , ident = 0 ) :
    """do _ classdesc"""

implementation:
# TC _ CLASSDESC className serialVersionUID newHandle classDescInfo
# classDescInfo :
# classDescFlags fields classAnnotation superClassDesc
# classDescFlags :
# ( byte ) / / Defined in Terminal Symbols and Constants
# fields :
# ( short ) < count > fieldDesc [ count ]
# fieldDesc :
# primitiveDesc
# objectDesc
# primitiveDesc :
# prim _ typecode fieldName
# objectDesc :
# obj _ typecode fieldName className1
clazz = JavaClass ( )
log_debug ( "[classdesc]" , ident )
ba = self . _readString ( )
clazz . name = ba
log_debug ( "Class name: %s" % ba , ident )
( serialVersionUID , newHandle , classDescFlags ) = self . _readStruct ( ">LLB" )
clazz . serialVersionUID = serialVersionUID
clazz . flags = classDescFlags
self . _add_reference ( clazz )
log_debug ( "Serial: 0x%X newHandle: 0x%X.\
classDescFlags: 0x%X" % ( serialVersionUID , newHandle , classDescFlags ) , ident )
( length , ) = self . _readStruct ( ">H" )
log_debug ( "Fields num: 0x%X" % length , ident )
clazz . fields_names = [ ]
clazz . fields_types = [ ]
for _ in range ( length ) :
    ( typecode , ) = self . _readStruct ( ">B" )
    field_name = self . _readString ( )
    field_type = None
    field_type = self . _convert_char_to_type ( typecode )
    if field_type == self . TYPE_ARRAY :
        _ , field_type = self . _read_and_exec_opcode ( ident = ident + 1 , expect = [ self . TC_STRING , self . TC_REFERENCE ] )
        assert isinstance ( field_type , str )
    # if field _ type is not None :
    # field _ type = " array of " + field _ type
    # else :
    # field _ type = " array of None "
    elif field_type == self . TYPE_OBJECT :
        _ , field_type = self . _read_and_exec_opcode ( ident = ident + 1 , expect = [ self . TC_STRING , self . TC_REFERENCE ] )
        assert isinstance ( field_type , str )
    log_debug ( "FieldName: 0x%X" % typecode + " " + str ( field_name ) + " " + str ( field_type ) , ident )
    assert field_name is not None
    assert field_type is not None
    clazz . fields_names . append ( field_name )
    clazz . fields_types . append ( field_type )
# pylint : disable = protected - access
if parent :
    parent . __fields = clazz . fields_names
    parent . __types = clazz . fields_types
# classAnnotation
( opid , ) = self . _readStruct ( ">B" )
log_debug ( "OpCode: 0x%X" % opid , ident )
if opid != self . TC_ENDBLOCKDATA :
    raise NotImplementedError ( "classAnnotation isn't implemented yet" )
# superClassDesc
_ , superclassdesc = self . _read_and_exec_opcode ( ident = ident + 1 , expect = [ self . TC_CLASSDESC , self . TC_NULL , self . TC_REFERENCE ] )
log_debug ( str ( superclassdesc ) , ident )
clazz . superclass = superclassdesc
return clazz


signature:
def getMetricsTimeline ( tmaster , component_name , metric_names , instances , start_time , end_time , callback = None ) :
    """Get the specified metrics for the given component name of this topology .
 Returns the following dict on success :
 " timeline " : {
 < metricname > : {
 < instance > : {
 < start _ time > : < numeric value > ,
 < start _ time > : < numeric value > ,
 " starttime " : < numeric value > ,
 " endtime " : < numeric value > ,
 " component " : " . . . "
 Returns the following dict on failure :
 " message " : " . . . " """

implementation:
# Tmaster is the proto object and must have host and port for stats .
if not tmaster or not tmaster . host or not tmaster . stats_port :
    raise Exception ( "No Tmaster found" )
host = tmaster . host
port = tmaster . stats_port
# Create the proto request object to get metrics .
metricRequest = tmaster_pb2 . MetricRequest ( )
metricRequest . component_name = component_name
# If no instances are give , metrics for all instances
# are fetched by default .
if len ( instances ) > 0 :
    for instance in instances :
        metricRequest . instance_id . append ( instance )
for metricName in metric_names :
    metricRequest . metric . append ( metricName )
metricRequest . explicit_interval . start = start_time
metricRequest . explicit_interval . end = end_time
metricRequest . minutely = True
# Serialize the metricRequest to send as a payload
# with the HTTP request .
metricRequestString = metricRequest . SerializeToString ( )
# Form and send the http request .
url = "http://{0}:{1}/stats" . format ( host , port )
request = tornado . httpclient . HTTPRequest ( url , body = metricRequestString , method = 'POST' , request_timeout = 5 )
Log . debug ( "Making HTTP call to fetch metrics" )
Log . debug ( "url: " + url )
try :
    client = tornado . httpclient . AsyncHTTPClient ( )
    result = yield client . fetch ( request )
    Log . debug ( "HTTP call complete." )
except tornado . httpclient . HTTPError as e :
    raise Exception ( str ( e ) )
# Check the response code - error if it is in 400s or 500s
responseCode = result . code
if responseCode >= 400 :
    message = "Error in getting metrics from Tmaster, code: " + responseCode
    Log . error ( message )
    raise Exception ( message )
# Parse the response from tmaster .
metricResponse = tmaster_pb2 . MetricResponse ( )
metricResponse . ParseFromString ( result . body )
if metricResponse . status . status == common_pb2 . NOTOK :
    if metricResponse . status . HasField ( "message" ) :
        Log . warn ( "Received response from Tmaster: %s" , metricResponse . status . message )
# Form the response .
ret = { }
ret [ "starttime" ] = start_time
ret [ "endtime" ] = end_time
ret [ "component" ] = component_name
ret [ "timeline" ] = { }
# Loop through all the metrics
# One instance corresponds to one metric , which can have
# multiple IndividualMetrics for each metricname requested .
for metric in metricResponse . metric :
    instance = metric . instance_id
    # Loop through all individual metrics .
    for im in metric . metric :
        metricname = im . name
        if metricname not in ret [ "timeline" ] :
            ret [ "timeline" ] [ metricname ] = { }
        if instance not in ret [ "timeline" ] [ metricname ] :
            ret [ "timeline" ] [ metricname ] [ instance ] = { }
        # We get minutely metrics .
        # Interval - values correspond to the minutely mark for which
        # this metric value corresponds to .
        for interval_value in im . interval_values :
            ret [ "timeline" ] [ metricname ] [ instance ] [ interval_value . interval . start ] = interval_value . value
raise tornado . gen . Return ( ret )


signature:
def create_parser ( subparsers ) :
    ''': param subparsers :
 : return :'''

implementation:
parser = subparsers . add_parser ( 'version' , help = 'Print version of heron-cli' , usage = "%(prog)s [options] [cluster]" , add_help = True )
add_version_titles ( parser )
parser . add_argument ( 'cluster' , nargs = '?' , type = str , default = "" , help = 'Name of the cluster' )
cli_args . add_service_url ( parser )
parser . set_defaults ( subcommand = 'version' )
return parser


signature:
def run ( command , parser , cl_args , unknown_args ) :
    ''': param command :
 : param parser :
 : param args :
 : param unknown _ args :
 : return :'''

implementation:
cluster = cl_args [ 'cluster' ]
# server mode
if cluster :
    config_file = config . heron_rc_file ( )
    client_confs = dict ( )
    # Read the cluster definition , if not found
    client_confs = cdefs . read_server_mode_cluster_definition ( cluster , cl_args , config_file )
    if not client_confs [ cluster ] :
        Log . error ( 'Neither service url nor %s cluster definition in %s file' , cluster , config_file )
        return SimpleResult ( Status . HeronError )
    # if cluster definition exists , but service _ url is not set , it is an error
    if not 'service_url' in client_confs [ cluster ] :
        Log . error ( 'No service url for %s cluster in %s' , cluster , config_file )
        sys . exit ( 1 )
    service_endpoint = cl_args [ 'service_url' ]
    service_apiurl = service_endpoint + rest . ROUTE_SIGNATURES [ command ] [ 1 ]
    service_method = rest . ROUTE_SIGNATURES [ command ] [ 0 ]
    try :
        r = service_method ( service_apiurl )
        if r . status_code != requests . codes . ok :
            Log . error ( r . json ( ) . get ( 'message' , "Unknown error from API server %d" % r . status_code ) )
        sorted_items = sorted ( r . json ( ) . items ( ) , key = lambda tup : tup [ 0 ] )
        for key , value in sorted_items :
            print ( "%s : %s" % ( key , value ) )
    except ( requests . exceptions . ConnectionError , requests . exceptions . HTTPError ) as err :
        Log . error ( err )
        return SimpleResult ( Status . HeronError )
else :
    config . print_build_info ( )
return SimpleResult ( Status . Ok )


signature:
def validate_state_locations ( self ) :
    """Names of all state locations must be unique ."""

implementation:
names = map ( lambda loc : loc [ "name" ] , self . locations )
assert len ( names ) == len ( set ( names ) ) , "Names of state locations must be unique"


signature:
def add_arguments ( parser ) :
    ''': param parser :
 : return :'''

implementation:
parser . add_argument ( '--tracker_url' , metavar = '(a url; path to tracker; default: "' + consts . DEFAULT_TRACKER_URL + '")' , default = consts . DEFAULT_TRACKER_URL )
parser . add_argument ( '--address' , metavar = '(an string; address to listen; default: "' + consts . DEFAULT_ADDRESS + '")' , default = consts . DEFAULT_ADDRESS )
parser . add_argument ( '--port' , metavar = '(an integer; port to listen; default: ' + str ( consts . DEFAULT_PORT ) + ')' , type = int , default = consts . DEFAULT_PORT )
parser . add_argument ( '--base_url' , metavar = '(a string; the base url path if operating behind proxy; default: ' + str ( consts . DEFAULT_BASE_URL ) + ')' , default = consts . DEFAULT_BASE_URL )
return parser


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

implementation:
cluster = self . get_argument ( "cluster" )
environ = self . get_argument ( "environ" )
topology = self . get_argument ( "topology" )
component = self . get_argument ( "component" , default = None )
metricnames = self . get_arguments ( "metricname" )
instances = self . get_arguments ( "instance" )
interval = self . get_argument ( "interval" , default = - 1 )
time_range = ( 0 , interval )
compnames = [ component ] if component else ( yield access . get_comps ( cluster , environ , topology ) )
# fetch the metrics
futures = { }
for comp in compnames :
    future = access . get_comp_metrics ( cluster , environ , topology , comp , instances , metricnames , time_range )
    futures [ comp ] = future
results = yield futures
self . write ( results [ component ] if component else results )


implementation:
cluster = self . get_argument ( "cluster" )
environ = self . get_argument ( "environ" )
topology = self . get_argument ( "topology" )
component = self . get_argument ( "component" , default = None )
metric = self . get_argument ( "metric" )
instances = self . get_argument ( "instance" )
start = self . get_argument ( "starttime" )
end = self . get_argument ( "endtime" )
maxquery = self . get_argument ( "max" , default = False )
timerange = ( start , end )
compnames = [ component ]
# fetch the metrics
futures = { }
if metric == "backpressure" :
    for comp in compnames :
        future = query_handler . fetch_backpressure ( cluster , metric , topology , component , instances , timerange , maxquery , environ )
        futures [ comp ] = future
else :
    fetch = query_handler . fetch_max if maxquery else query_handler . fetch
    for comp in compnames :
        future = fetch ( cluster , metric , topology , component , instances , timerange , environ )
        futures [ comp ] = future
results = yield futures
self . write ( results [ component ] if component else results )


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

implementation:
self . content_type = 'application/json'
self . write ( json . dumps ( utils . chain ( [ [ 'ps' , 'auxwwww' ] , [ 'grep' , instance_id ] , [ 'grep' , 'java' ] , [ 'awk' , '\'{print $2}\'' ] ] ) ) . strip ( ) )
self . finish ( )


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

implementation:
if path is None :
    return { }
if not utils . check_path ( path ) :
    self . write ( "Only relative paths are allowed" )
    self . set_status ( 403 )
    self . finish ( )
    return
offset = self . get_argument ( "offset" , default = - 1 )
length = self . get_argument ( "length" , default = - 1 )
if not os . path . isfile ( path ) :
    return { }
data = utils . read_chunk ( path , offset = offset , length = length , escape_data = True )
self . write ( json . dumps ( data ) )
self . finish ( )


signature:
def initialize ( self , config , context ) :
    """Implements Pulsar Spout ' s initialize method"""

implementation:
self . logger . info ( "Initializing PulsarSpout with the following" )
self . logger . info ( "Component-specific config: \n%s" % str ( config ) )
self . logger . info ( "Context: \n%s" % str ( context ) )
self . emit_count = 0
self . ack_count = 0
self . fail_count = 0
if not PulsarSpout . serviceUrl in config or not PulsarSpout . topicName in config :
    self . logger . fatal ( "Need to specify both serviceUrl and topicName" )
self . pulsar_cluster = str ( config [ PulsarSpout . serviceUrl ] )
self . topic = str ( config [ PulsarSpout . topicName ] )
mode = config [ api_constants . TOPOLOGY_RELIABILITY_MODE ]
if mode == api_constants . TopologyReliabilityMode . ATLEAST_ONCE :
    self . acking_timeout = 1000 * int ( config [ api_constants . TOPOLOGY_MESSAGE_TIMEOUT_SECS ] )
else :
    self . acking_timeout = 30000
if PulsarSpout . receiveTimeoutMs in config :
    self . receive_timeout_ms = config [ PulsarSpout . receiveTimeoutMs ]
else :
    self . receive_timeout_ms = 10
if PulsarSpout . deserializer in config :
    self . deserializer = config [ PulsarSpout . deserializer ]
    if not callable ( self . deserializer ) :
        self . logger . fatal ( "Pulsar Message Deserializer needs to be callable" )
else :
    self . deserializer = self . default_deserializer
# First generate the config
self . logConfFileName = GenerateLogConfig ( context )
self . logger . info ( "Generated LogConf at %s" % self . logConfFileName )
# We currently use the high level consumer API
# For supporting effectively once , we will need to switch
# to using lower level Reader API , when it becomes
# available in python
self . client = pulsar . Client ( self . pulsar_cluster , log_conf_file_path = self . logConfFileName )
self . logger . info ( "Setup Client with cluster %s" % self . pulsar_cluster )
try :
    self . consumer = self . client . subscribe ( self . topic , context . get_topology_name ( ) , consumer_type = pulsar . ConsumerType . Failover , unacked_messages_timeout_ms = self . acking_timeout )
except Exception as e :
    self . logger . fatal ( "Pulsar client subscription failed: %s" % str ( e ) )
self . logger . info ( "Subscribed to topic %s" % self . topic )


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 )
role = self . get_argument_role ( )
ret = { }
topologies = self . tracker . topologies
for topology in topologies :
    cluster = topology . cluster
    environ = topology . environ
    if not cluster or not environ :
        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
    if cluster not in ret :
        ret [ cluster ] = { }
    if environ not in ret [ cluster ] :
        ret [ cluster ] [ environ ] = { }
    try :
        topology_info = self . tracker . getTopologyInfo ( topology . name , cluster , role , environ )
        if topology_info and "execution_state" in topology_info :
            ret [ cluster ] [ environ ] [ topology . name ] = topology_info [ "execution_state" ]
    except Exception : # Do nothing
        pass
self . write_success_response ( ret )


signature:
def getInstanceJstack ( self , topology_info , instance_id ) :
    """Fetches Instance jstack from heron - shell ."""

implementation:
pid_response = yield getInstancePid ( topology_info , instance_id )
try :
    http_client = tornado . httpclient . AsyncHTTPClient ( )
    pid_json = json . loads ( pid_response )
    pid = pid_json [ 'stdout' ] . strip ( )
    if pid == '' :
        raise Exception ( 'Failed to get pid' )
    endpoint = utils . make_shell_endpoint ( topology_info , instance_id )
    url = "%s/jstack/%s" % ( endpoint , pid )
    response = yield http_client . fetch ( url )
    Log . debug ( "HTTP call for url: %s" , url )
    raise tornado . gen . Return ( response . body )
except tornado . httpclient . HTTPError as e :
    raise Exception ( str ( e ) )


signature:
def create_parser ( subparsers ) :
    """Create the parse for the update command"""

implementation:
parser = subparsers . add_parser ( 'update' , help = 'Update a topology' , usage = "%(prog)s [options] cluster/[role]/[env] <topology-name> " + "[--component-parallelism <name:value>] " + "[--container-number value] " + "[--runtime-config [component:]<name:value>]" , add_help = True )
args . add_titles ( parser )
args . add_cluster_role_env ( parser )
args . add_topology ( parser )
args . add_config ( parser )
args . add_dry_run ( parser )
args . add_service_url ( parser )
args . add_verbose ( parser )
# Special parameters for update command
def parallelism_type ( value ) :
    pattern = re . compile ( r"^[\w\.-]+:[\d]+$" )
    if not pattern . match ( value ) :
        raise argparse . ArgumentTypeError ( "Invalid syntax for component parallelism (<component_name:value>): %s" % value )
    return value
parser . add_argument ( '--component-parallelism' , action = 'append' , type = parallelism_type , required = False , help = 'Component name and the new parallelism value ' + 'colon-delimited: <component_name>:<parallelism>' )
def runtime_config_type ( value ) :
    pattern = re . compile ( r"^([\w\.-]+:){1,2}[\w\.-]+$" )
    if not pattern . match ( value ) :
        raise argparse . ArgumentTypeError ( "Invalid syntax for runtime config ([component:]<name:value>): %s" % value )
    return value
parser . add_argument ( '--runtime-config' , action = 'append' , type = runtime_config_type , required = False , help = 'Runtime configurations for topology and components ' + 'colon-delimited: [component:]<name>:<value>' )
def container_number_type ( value ) :
    pattern = re . compile ( r"^\d+$" )
    if not pattern . match ( value ) :
        raise argparse . ArgumentTypeError ( "Invalid syntax for container number (value): %s" % value )
    return value
parser . add_argument ( '--container-number' , action = 'append' , type = container_number_type , required = False , help = 'Number of containers <value>' )
parser . set_defaults ( subcommand = 'update' )
return parser


signature:
def build_extra_args_dict ( cl_args ) :
    """Build extra args map"""

implementation:
# Check parameters
component_parallelism = cl_args [ 'component_parallelism' ]
runtime_configs = cl_args [ 'runtime_config' ]
container_number = cl_args [ 'container_number' ]
# Users need to provide either ( component - parallelism | | container _ number ) or runtime - config
if ( component_parallelism and runtime_configs ) or ( container_number and runtime_configs ) :
    raise Exception ( "(component-parallelism or container_num) and runtime-config " + "can't be updated at the same time" )
dict_extra_args = { }
nothing_set = True
if component_parallelism :
    dict_extra_args . update ( { 'component_parallelism' : component_parallelism } )
    nothing_set = False
if container_number :
    dict_extra_args . update ( { 'container_number' : container_number } )
    nothing_set = False
if runtime_configs :
    dict_extra_args . update ( { 'runtime_config' : runtime_configs } )
    nothing_set = False
if nothing_set :
    raise Exception ( "Missing arguments --component-parallelism or --runtime-config or --container-number" )
if cl_args [ 'dry_run' ] :
    dict_extra_args . update ( { 'dry_run' : True } )
    if 'dry_run_format' in cl_args :
        dict_extra_args . update ( { 'dry_run_format' : cl_args [ "dry_run_format" ] } )
return dict_extra_args


signature:
def convert_args_dict_to_list ( dict_extra_args ) :
    """flatten extra args"""

implementation:
list_extra_args = [ ]
if 'component_parallelism' in dict_extra_args :
    list_extra_args += [ "--component_parallelism" , ',' . join ( dict_extra_args [ 'component_parallelism' ] ) ]
if 'runtime_config' in dict_extra_args :
    list_extra_args += [ "--runtime_config" , ',' . join ( dict_extra_args [ 'runtime_config' ] ) ]
if 'container_number' in dict_extra_args :
    list_extra_args += [ "--container_number" , ',' . join ( dict_extra_args [ 'container_number' ] ) ]
if 'dry_run' in dict_extra_args and dict_extra_args [ 'dry_run' ] :
    list_extra_args += [ '--dry_run' ]
if 'dry_run_format' in dict_extra_args :
    list_extra_args += [ '--dry_run_format' , dict_extra_args [ 'dry_run_format' ] ]
return list_extra_args


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

implementation:
Log . debug ( "Update Args: %s" , cl_args )
# Build jar list
extra_lib_jars = jars . packing_jars ( )
action = "update topology%s" % ( ' in dry-run mode' if cl_args [ "dry_run" ] else '' )
# Build extra args
dict_extra_args = { }
try :
    dict_extra_args = build_extra_args_dict ( cl_args )
except Exception as err :
    return SimpleResult ( Status . InvocationError , err . message )
# Execute
if cl_args [ 'deploy_mode' ] == config . SERVER_MODE :
    return cli_helper . run_server ( command , cl_args , action , dict_extra_args )
else : # Convert extra argument to commandline format and then execute
    list_extra_args = convert_args_dict_to_list ( dict_extra_args )
    return cli_helper . run_direct ( command , cl_args , action , list_extra_args , extra_lib_jars )


signature:
def getInstancePid ( topology_info , instance_id ) :
    """This method is used by other modules , and so it
 is not a part of the class .
 Fetches Instance pid from heron - shell ."""

implementation:
try :
    http_client = tornado . httpclient . AsyncHTTPClient ( )
    endpoint = utils . make_shell_endpoint ( topology_info , instance_id )
    url = "%s/pid/%s" % ( endpoint , instance_id )
    Log . debug ( "HTTP call for url: %s" , url )
    response = yield http_client . fetch ( url )
    raise tornado . gen . Return ( response . body )
except tornado . httpclient . HTTPError as e :
    raise Exception ( str ( e ) )


implementation:
try :
    cluster = self . get_argument_cluster ( )
    role = self . get_argument_role ( )
    environ = self . get_argument_environ ( )
    topology_name = self . get_argument_topology ( )
    instance = self . get_argument_instance ( )
    topology_info = self . tracker . getTopologyInfo ( topology_name , cluster , role , environ )
    result = yield getInstancePid ( topology_info , instance )
    self . write_success_response ( result )
except Exception as e :
    Log . debug ( traceback . format_exc ( ) )
    self . write_error_response ( e )


signature:
def is_grouping_sane ( cls , gtype ) :
    """Checks if a given gtype is sane"""

implementation:
if gtype == cls . SHUFFLE or gtype == cls . ALL or gtype == cls . LOWEST or gtype == cls . NONE :
    return True
elif isinstance ( gtype , cls . FIELDS ) :
    return gtype . gtype == topology_pb2 . Grouping . Value ( "FIELDS" ) and gtype . fields is not None
elif isinstance ( gtype , cls . CUSTOM ) :
    return gtype . gtype == topology_pb2 . Grouping . Value ( "CUSTOM" ) and gtype . python_serialized is not None
else : # pylint : disable = fixme
# TODO : DIRECT are not supported yet
    return False


signature:
def fields ( cls , * fields ) :
    """Field grouping"""

implementation:
if len ( fields ) == 1 and isinstance ( fields [ 0 ] , list ) :
    fields = fields [ 0 ]
else :
    fields = list ( fields )
for i in fields :
    if not isinstance ( i , str ) :
        raise TypeError ( "Non-string cannot be specified in fields" )
if not fields :
    raise ValueError ( "List cannot be empty for fields grouping" )
return cls . FIELDS ( gtype = topology_pb2 . Grouping . Value ( "FIELDS" ) , fields = fields )


signature:
def custom ( cls , customgrouper ) :
    """Custom grouping from a given implementation of ICustomGrouping
 : param customgrouper : The ICustomGrouping implemention to use"""

implementation:
if customgrouper is None :
    raise TypeError ( "Argument to custom() must be ICustomGrouping instance or classpath" )
if not isinstance ( customgrouper , ICustomGrouping ) and not isinstance ( customgrouper , str ) :
    raise TypeError ( "Argument to custom() must be ICustomGrouping instance or classpath" )
serialized = default_serializer . serialize ( customgrouper )
return cls . custom_serialized ( serialized , is_java = False )


signature:
def custom_serialized ( cls , serialized , is_java = True ) :
    """Custom grouping from a given serialized string
 This class is created for compatibility with ` ` custom _ serialized ( cls , java _ serialized ) ` ` method
 of StreamParse API , although its functionality is not yet implemented ( Java - serialized ) .
 Currently only custom grouping implemented in Python is supported , and ` ` custom ( ) ` ` method
 should be used to indicate its classpath , rather than directly to use this method .
 In the future , users can directly specify Java - serialized object with ` ` is _ java = True ` ` in order
 to use a custom grouping implemented in Java for python topology .
 : param serialized : serialized classpath to custom grouping class to use ( if python )
 : param is _ java : indicate whether this is Java serialized , or python serialized"""

implementation:
if not isinstance ( serialized , bytes ) :
    raise TypeError ( "Argument to custom_serialized() must be " "a serialized Python class as bytes, given: %s" % str ( serialized ) )
if not is_java :
    return cls . CUSTOM ( gtype = topology_pb2 . Grouping . Value ( "CUSTOM" ) , python_serialized = serialized )
else :
    raise NotImplementedError ( "Custom grouping implemented in Java for Python topology" "is not yet supported." )


signature:
def register_metrics ( self , metrics_collector , interval ) :
    """Registers its metrics to a given metrics collector with a given interval"""

implementation:
for field , metrics in self . metrics . items ( ) :
    metrics_collector . register_metric ( field , metrics , interval )


signature:
def update_count ( self , name , incr_by = 1 , key = None ) :
    """Update the value of CountMetric or MultiCountMetric
 : type name : str
 : param name : name of the registered metric to be updated .
 : type incr _ by : int
 : param incr _ by : specifies how much to increment . Default is 1.
 : type key : str or None
 : param key : specifies a key for MultiCountMetric . Needs to be ` None ` for updating CountMetric ."""

implementation:
if name not in self . metrics :
    Log . error ( "In update_count(): %s is not registered in the metric" , name )
if key is None and isinstance ( self . metrics [ name ] , CountMetric ) :
    self . metrics [ name ] . incr ( incr_by )
elif key is not None and isinstance ( self . metrics [ name ] , MultiCountMetric ) :
    self . metrics [ name ] . incr ( key , incr_by )
else :
    Log . error ( "In update_count(): %s is registered but not supported with this method" , name )


signature:
def update_reduced_metric ( self , name , value , key = None ) :
    """Update the value of ReducedMetric or MultiReducedMetric
 : type name : str
 : param name : name of the registered metric to be updated .
 : param value : specifies a value to be reduced .
 : type key : str or None
 : param key : specifies a key for MultiReducedMetric . Needs to be ` None ` for updating
 ReducedMetric ."""

implementation:
if name not in self . metrics :
    Log . error ( "In update_reduced_metric(): %s is not registered in the metric" , name )
if key is None and isinstance ( self . metrics [ name ] , ReducedMetric ) :
    self . metrics [ name ] . update ( value )
elif key is not None and isinstance ( self . metrics [ name ] , MultiReducedMetric ) :
    self . metrics [ name ] . update ( key , value )
else :
    Log . error ( "In update_count(): %s is registered but not supported with this method" , name )


signature:
def update_received_packet ( self , received_pkt_size_bytes ) :
    """Update received packet metrics"""

implementation:
self . update_count ( self . RECEIVED_PKT_COUNT )
self . update_count ( self . RECEIVED_PKT_SIZE , incr_by = received_pkt_size_bytes )


signature:
def update_sent_packet ( self , sent_pkt_size_bytes ) :
    """Update sent packet metrics"""

implementation:
self . update_count ( self . SENT_PKT_COUNT )
self . update_count ( self . SENT_PKT_SIZE , incr_by = sent_pkt_size_bytes )


signature:
def register_metrics ( self , context ) :
    """Registers metrics to context
 : param context : Topology Context"""

implementation:
sys_config = system_config . get_sys_config ( )
interval = float ( sys_config [ constants . HERON_METRICS_EXPORT_INTERVAL_SEC ] )
collector = context . get_metrics_collector ( )
super ( ComponentMetrics , self ) . register_metrics ( collector , interval )


signature:
def serialize_data_tuple ( self , stream_id , latency_in_ns ) :
    """Apply update to serialization metrics"""

implementation: self . update_count ( self . TUPLE_SERIALIZATION_TIME_NS , incr_by = latency_in_ns , key = stream_id )


signature:
def _init_multi_count_metrics ( self , pplan_helper ) :
    """Initializes the default values for a necessary set of MultiCountMetrics"""

implementation:
to_init = [ self . metrics [ i ] for i in self . to_multi_init if i in self . metrics and isinstance ( self . metrics [ i ] , MultiCountMetric ) ]
for out_stream in pplan_helper . get_my_spout ( ) . outputs :
    stream_id = out_stream . stream . id
    for metric in to_init :
        metric . add_key ( stream_id )


signature:
def next_tuple ( self , latency_in_ns ) :
    """Apply updates to the next tuple metrics"""

implementation:
self . update_reduced_metric ( self . NEXT_TUPLE_LATENCY , latency_in_ns )
self . update_count ( self . NEXT_TUPLE_COUNT )


signature:
def acked_tuple ( self , stream_id , complete_latency_ns ) :
    """Apply updates to the ack metrics"""

implementation:
self . update_count ( self . ACK_COUNT , key = stream_id )
self . update_reduced_metric ( self . COMPLETE_LATENCY , complete_latency_ns , key = stream_id )


signature:
def failed_tuple ( self , stream_id , fail_latency_ns ) :
    """Apply updates to the fail metrics"""

implementation:
self . update_count ( self . FAIL_COUNT , key = stream_id )
self . update_reduced_metric ( self . FAIL_LATENCY , fail_latency_ns , key = stream_id )


implementation:
# inputs
to_in_init = [ self . metrics [ i ] for i in self . inputs_init if i in self . metrics and isinstance ( self . metrics [ i ] , MultiCountMetric ) ]
for in_stream in pplan_helper . get_my_bolt ( ) . inputs :
    stream_id = in_stream . stream . id
    global_stream_id = in_stream . stream . component_name + "/" + stream_id
    for metric in to_in_init :
        metric . add_key ( stream_id )
        metric . add_key ( global_stream_id )
# outputs
to_out_init = [ self . metrics [ i ] for i in self . outputs_init if i in self . metrics and isinstance ( self . metrics [ i ] , MultiCountMetric ) ]
for out_stream in pplan_helper . get_my_bolt ( ) . outputs :
    stream_id = out_stream . stream . id
    for metric in to_out_init :
        metric . add_key ( stream_id )


signature:
def execute_tuple ( self , stream_id , source_component , latency_in_ns ) :
    """Apply updates to the execute metrics"""

implementation:
self . update_count ( self . EXEC_COUNT , key = stream_id )
self . update_reduced_metric ( self . EXEC_LATENCY , latency_in_ns , stream_id )
self . update_count ( self . EXEC_TIME_NS , incr_by = latency_in_ns , key = stream_id )
global_stream_id = source_component + "/" + stream_id
self . update_count ( self . EXEC_COUNT , key = global_stream_id )
self . update_reduced_metric ( self . EXEC_LATENCY , latency_in_ns , global_stream_id )
self . update_count ( self . EXEC_TIME_NS , incr_by = latency_in_ns , key = global_stream_id )


signature:
def deserialize_data_tuple ( self , stream_id , source_component , latency_in_ns ) :
    """Apply updates to the deserialization metrics"""

implementation:
self . update_count ( self . TUPLE_DESERIALIZATION_TIME_NS , incr_by = latency_in_ns , key = stream_id )
global_stream_id = source_component + "/" + stream_id
self . update_count ( self . TUPLE_DESERIALIZATION_TIME_NS , incr_by = latency_in_ns , key = global_stream_id )


signature:
def acked_tuple ( self , stream_id , source_component , latency_in_ns ) :
    """Apply updates to the ack metrics"""

implementation:
self . update_count ( self . ACK_COUNT , key = stream_id )
self . update_reduced_metric ( self . PROCESS_LATENCY , latency_in_ns , stream_id )
global_stream_id = source_component + '/' + stream_id
self . update_count ( self . ACK_COUNT , key = global_stream_id )
self . update_reduced_metric ( self . PROCESS_LATENCY , latency_in_ns , global_stream_id )


signature:
def failed_tuple ( self , stream_id , source_component , latency_in_ns ) :
    """Apply updates to the fail metrics"""

implementation:
self . update_count ( self . FAIL_COUNT , key = stream_id )
self . update_reduced_metric ( self . FAIL_LATENCY , latency_in_ns , stream_id )
global_stream_id = source_component + '/' + stream_id
self . update_count ( self . FAIL_COUNT , key = global_stream_id )
self . update_reduced_metric ( self . FAIL_LATENCY , latency_in_ns , global_stream_id )


signature:
def register_metric ( self , name , metric , time_bucket_in_sec ) :
    """Registers a given metric
 : param name : name of the metric
 : param metric : IMetric object to be registered
 : param time _ bucket _ in _ sec : time interval for update to the metrics manager"""

implementation:
if name in self . metrics_map :
    raise RuntimeError ( "Another metric has already been registered with name: %s" % name )
Log . debug ( "Register metric: %s, with interval: %s" , name , str ( time_bucket_in_sec ) )
self . metrics_map [ name ] = metric
if time_bucket_in_sec in self . time_bucket_in_sec_to_metrics_name :
    self . time_bucket_in_sec_to_metrics_name [ time_bucket_in_sec ] . append ( name )
else :
    self . time_bucket_in_sec_to_metrics_name [ time_bucket_in_sec ] = [ name ]
    self . _register_timer_task ( time_bucket_in_sec )


signature:
def poll ( self ) :
    """Poll from the buffer
 It is a non - blocking operation , and when the buffer is empty , it raises Queue . Empty exception"""

implementation:
try : # non - blocking
    ret = self . _buffer . get ( block = False )
    if self . _producer_callback is not None :
        self . _producer_callback ( )
    return ret
except Queue . Empty :
    Log . debug ( "%s: Empty in poll()" % str ( self ) )
    raise Queue . Empty


signature:
def offer ( self , item ) :
    """Offer to the buffer
 It is a non - blocking operation , and when the buffer is full , it raises Queue . Full exception"""

implementation:
try : # non - blocking
    self . _buffer . put ( item , block = False )
    if self . _consumer_callback is not None :
        self . _consumer_callback ( )
    return True
except Queue . Full :
    Log . debug ( "%s: Full in offer()" % str ( self ) )
    raise Queue . Full


signature:
def parse ( version ) :
    """Parse version to major , minor , patch , pre - release , build parts ."""

implementation:
match = _REGEX . match ( version )
if match is None :
    raise ValueError ( '%s is not valid SemVer string' % version )
verinfo = match . groupdict ( )
verinfo [ 'major' ] = int ( verinfo [ 'major' ] )
verinfo [ 'minor' ] = int ( verinfo [ 'minor' ] )
verinfo [ 'patch' ] = int ( verinfo [ 'patch' ] )
return verinfo


signature:
def create_parser ( subparsers , action , help_arg ) :
    ''': param subparsers :
 : param action :
 : param help _ arg :
 : return :'''

implementation:
parser = subparsers . add_parser ( action , help = help_arg , usage = "%(prog)s [options] cluster/[role]/[env] <topology-name>" , add_help = True )
args . add_titles ( parser )
args . add_cluster_role_env ( parser )
args . add_topology ( parser )
args . add_config ( parser )
args . add_service_url ( parser )
args . add_verbose ( parser )
parser . set_defaults ( subcommand = action )
return parser


signature:
def run_server ( command , cl_args , action , extra_args = dict ( ) ) :
    '''helper function to take action on topologies using REST API
 : param command :
 : param cl _ args :
 : param action : description of action taken
 : return :'''

implementation:
topology_name = cl_args [ 'topology-name' ]
service_endpoint = cl_args [ 'service_url' ]
apiroute = rest . ROUTE_SIGNATURES [ command ] [ 1 ] % ( cl_args [ 'cluster' ] , cl_args [ 'role' ] , cl_args [ 'environ' ] , topology_name )
service_apiurl = service_endpoint + apiroute
service_method = rest . ROUTE_SIGNATURES [ command ] [ 0 ]
# convert the dictionary to a list of tuples
data = flatten_args ( extra_args )
err_msg = "Failed to %s: %s" % ( action , topology_name )
succ_msg = "Successfully %s: %s" % ( action , topology_name )
try :
    r = service_method ( service_apiurl , data = data )
    s = Status . Ok if r . status_code == requests . codes . ok else Status . HeronError
    if r . status_code != requests . codes . ok :
        Log . error ( r . json ( ) . get ( 'message' , "Unknown error from API server %d" % r . status_code ) )
except ( requests . exceptions . ConnectionError , requests . exceptions . HTTPError ) as err :
    Log . error ( err )
    return SimpleResult ( Status . HeronError , err_msg , succ_msg )
return SimpleResult ( s , err_msg , succ_msg )


signature:
def run_direct ( command , cl_args , action , extra_args = [ ] , extra_lib_jars = [ ] ) :
    '''helper function to take action on topologies
 : param command :
 : param cl _ args :
 : param action : description of action taken
 : return :'''

implementation:
topology_name = cl_args [ 'topology-name' ]
new_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" , cl_args [ 'config_path' ] , "--override_config_file" , cl_args [ 'override_config_file' ] , "--release_file" , config . get_heron_release_file ( ) , "--topology_name" , topology_name , "--command" , command , ]
new_args += extra_args
lib_jars = config . get_heron_libs ( jars . scheduler_jars ( ) + jars . statemgr_jars ( ) )
lib_jars += extra_lib_jars
if Log . getEffectiveLevel ( ) == logging . DEBUG :
    new_args . append ( "--verbose" )
# invoke the runtime manager to kill the topology
result = execute . heron_class ( 'org.apache.heron.scheduler.RuntimeManagerMain' , lib_jars , extra_jars = [ ] , args = new_args )
err_msg = "Failed to %s: %s" % ( action , topology_name )
succ_msg = "Successfully %s: %s" % ( action , topology_name )
result . add_context ( err_msg , succ_msg )
return result


signature:
def get_all_zk_state_managers ( conf ) :
    """Creates all the zookeeper state _ managers and returns
 them in a list"""

implementation:
state_managers = [ ]
state_locations = conf . get_state_locations_of_type ( "zookeeper" )
for location in state_locations :
    name = location [ 'name' ]
    hostport = location [ 'hostport' ]
    hostportlist = [ ]
    for hostportpair in hostport . split ( ',' ) :
        host = None
        port = None
        if ':' in hostport :
            hostandport = hostportpair . split ( ':' )
            if len ( hostandport ) == 2 :
                host = hostandport [ 0 ]
                port = int ( hostandport [ 1 ] )
        if not host or not port :
            raise Exception ( "Hostport for %s must be of the format 'host:port'." % ( name ) )
        hostportlist . append ( ( host , port ) )
    tunnelhost = location [ 'tunnelhost' ]
    rootpath = location [ 'rootpath' ]
    LOG . info ( "Connecting to zk hostports: " + str ( hostportlist ) + " rootpath: " + rootpath )
    state_manager = ZkStateManager ( name , hostportlist , rootpath , tunnelhost )
    state_managers . append ( state_manager )
return state_managers


signature:
def get_all_file_state_managers ( conf ) :
    """Returns all the file state _ managers ."""

implementation:
state_managers = [ ]
state_locations = conf . get_state_locations_of_type ( "file" )
for location in state_locations :
    name = location [ 'name' ]
    rootpath = os . path . expanduser ( location [ 'rootpath' ] )
    LOG . info ( "Connecting to file state with rootpath: " + rootpath )
    state_manager = FileStateManager ( name , rootpath )
    state_managers . append ( state_manager )
return state_managers


signature:
def incr ( self , key , to_add = 1 ) :
    """Increments the value of a given key by ` ` to _ add ` `"""

implementation:
if key not in self . value :
    self . value [ key ] = CountMetric ( )
self . value [ key ] . incr ( to_add )


signature:
def update ( self , key , value ) :
    """Updates a value of a given key and apply reduction"""

implementation:
if key not in self . value :
    self . value [ key ] = ReducedMetric ( self . reducer )
self . value [ key ] . update ( value )


signature:
def add_key ( self , key ) :
    """Adds a new key to this metric"""

implementation:
if key not in self . value :
    self . value [ key ] = ReducedMetric ( self . reducer )


signature:
def add_data_tuple ( self , stream_id , new_data_tuple , tuple_size_in_bytes ) :
    """Add a new data tuple to the currently buffered set of tuples"""

implementation:
if ( self . current_data_tuple_set is None ) or ( self . current_data_tuple_set . stream . id != stream_id ) or ( len ( self . current_data_tuple_set . tuples ) >= self . data_tuple_set_capacity ) or ( self . current_data_tuple_size_in_bytes >= self . max_data_tuple_size_in_bytes ) :
    self . _init_new_data_tuple ( stream_id )
added_tuple = self . current_data_tuple_set . tuples . add ( )
added_tuple . CopyFrom ( new_data_tuple )
self . current_data_tuple_size_in_bytes += tuple_size_in_bytes
self . total_data_emitted_in_bytes += tuple_size_in_bytes


signature:
def add_control_tuple ( self , new_control_tuple , tuple_size_in_bytes , is_ack ) :
    """Add a new control ( Ack / Fail ) tuple to the currently buffered set of tuples
 : param is _ ack : ` ` True ` ` if Ack , ` ` False ` ` if Fail"""

implementation:
if self . current_control_tuple_set is None :
    self . _init_new_control_tuple ( )
elif is_ack and ( len ( self . current_control_tuple_set . fails ) > 0 or len ( self . current_control_tuple_set . acks ) >= self . control_tuple_set_capacity ) :
    self . _init_new_control_tuple ( )
elif not is_ack and ( len ( self . current_control_tuple_set . acks ) > 0 or len ( self . current_control_tuple_set . fails ) >= self . control_tuple_set_capacity ) :
    self . _init_new_control_tuple ( )
if is_ack :
    added_tuple = self . current_control_tuple_set . acks . add ( )
else :
    added_tuple = self . current_control_tuple_set . fails . add ( )
added_tuple . CopyFrom ( new_control_tuple )
self . total_data_emitted_in_bytes += tuple_size_in_bytes


signature:
def add_ckpt_state ( self , ckpt_id , ckpt_state ) :
    """Add the checkpoint state message to be sent back the stmgr
 : param ckpt _ id : The id of the checkpoint
 : ckpt _ state : The checkpoint state"""

implementation:
# first flush any buffered tuples
self . _flush_remaining ( )
msg = ckptmgr_pb2 . StoreInstanceStateCheckpoint ( )
istate = ckptmgr_pb2 . InstanceStateCheckpoint ( )
istate . checkpoint_id = ckpt_id
istate . state = ckpt_state
msg . state . CopyFrom ( istate )
self . _push_tuple_to_stream ( msg )


signature:
def create_parser ( subparsers ) :
    ''': param subparsers :
 : return :'''

implementation:
parser = subparsers . add_parser ( 'config' , help = 'Config properties for a cluster' , usage = "%(prog)s [cluster]" , add_help = True )
parser . add_argument ( 'cluster' , help = 'Cluster to configure' )
ex_subparsers = parser . add_subparsers ( title = "Commands" , description = None )
# add config list parser
list_parser = ex_subparsers . add_parser ( 'list' , help = 'List config properties for a cluster' , usage = "%(prog)s" , add_help = True )
list_parser . set_defaults ( configcommand = 'list' )
# add config set parser
set_parser = ex_subparsers . add_parser ( 'set' , help = 'Set a cluster config property' , usage = "%(prog)s [property] [value]" , add_help = True )
set_parser . add_argument ( 'property' , help = 'Config property to set' )
set_parser . add_argument ( 'value' , help = 'Value of config property' )
set_parser . set_defaults ( configcommand = 'set' )
# add config unset parser
unset_parser = ex_subparsers . add_parser ( 'unset' , help = 'Unset a cluster config property' , usage = "%(prog)s [property]" , add_help = True )
unset_parser . add_argument ( 'property' , help = 'Config property to unset' )
unset_parser . set_defaults ( configcommand = 'unset' )
parser . set_defaults ( subcommand = 'config' )
return parser


signature:
def run ( command , parser , cl_args , unknown_args ) :
    ''': param command :
 : param parser :
 : param args :
 : param unknown _ args :
 : return :'''

implementation:
configcommand = cl_args . get ( 'configcommand' , None )
if configcommand == 'set' :
    return _set ( cl_args )
elif configcommand == 'unset' :
    return _unset ( cl_args )
else :
    return _list ( cl_args )


signature:
def valid_path ( path ) :
    '''Check if an entry in the class path exists as either a directory or a file'''

implementation:
# check if the suffic of classpath suffix exists as directory
if path . endswith ( '*' ) :
    Log . debug ( 'Checking classpath entry suffix as directory: %s' , path [ : - 1 ] )
    if os . path . isdir ( path [ : - 1 ] ) :
        return True
    return False
# check if the classpath entry is a directory
Log . debug ( 'Checking classpath entry as directory: %s' , path )
if os . path . isdir ( path ) :
    return True
else : # check if the classpath entry is a file
    Log . debug ( 'Checking classpath entry as file: %s' , path )
    if os . path . isfile ( path ) :
        return True
return False


signature:
def valid_java_classpath ( classpath ) :
    '''Given a java classpath , check whether the path entries are valid or not'''

implementation:
paths = classpath . split ( ':' )
for path_entry in paths :
    if not valid_path ( path_entry . strip ( ) ) :
        return False
return True


signature:
def add_edge ( self , U , V ) :
    ''': param U :
 : param V :
 : return :'''

implementation:
if not U in self . edges :
    self . edges [ U ] = set ( )
if not V in self . edges :
    self . edges [ V ] = set ( )
if not V in self . edges [ U ] :
    self . edges [ U ] . add ( V )


signature:
def bfs_depth ( self , U ) :
    '''Returns the maximum distance between any vertex and U in the connected
 component containing U
 : param U :
 : return :'''

implementation:
bfs_queue = [ [ U , 0 ] ]
# Stores the vertices whose BFS hadn ' t been completed .
visited = set ( )
max_depth = 0
while bfs_queue :
    [ V , depth ] = bfs_queue . pop ( )
    if max_depth < depth :
        max_depth = depth
    visited . add ( V )
    adj_set = self . edges [ V ]
    for W in adj_set :
        if W not in visited :
            bfs_queue . append ( [ W , depth + 1 ] )
return max_depth


signature:
def diameter ( self ) :
    '''Returns the maximum distance between any vertex and U in the connected
 component containing U
 : return :'''

implementation:
diameter = 0
for U in self . edges :
    depth = self . bfs_depth ( U )
    if depth > diameter :
        diameter = depth
return diameter


signature:
def _get_deps_list ( abs_path_to_pex ) :
    """Get a list of paths to included dependencies in the specified pex file
 Note that dependencies are located under ` . deps ` directory"""

implementation:
pex = zipfile . ZipFile ( abs_path_to_pex , mode = 'r' )
deps = list ( set ( [ re . match ( egg_regex , i ) . group ( 1 ) for i in pex . namelist ( ) if re . match ( egg_regex , i ) is not None ] ) )
return deps


signature:
def load_pex ( path_to_pex , include_deps = True ) :
    """Loads pex file and its dependencies to the current python path"""

implementation:
abs_path_to_pex = os . path . abspath ( path_to_pex )
Log . debug ( "Add a pex to the path: %s" % abs_path_to_pex )
if abs_path_to_pex not in sys . path :
    sys . path . insert ( 0 , os . path . dirname ( abs_path_to_pex ) )
# add dependencies to path
if include_deps :
    for dep in _get_deps_list ( abs_path_to_pex ) :
        to_join = os . path . join ( os . path . dirname ( abs_path_to_pex ) , dep )
        if to_join not in sys . path :
            Log . debug ( "Add a new dependency to the path: %s" % dep )
            sys . path . insert ( 0 , to_join )
Log . debug ( "Python path: %s" % str ( sys . path ) )