heron/executor/src/python/heron_executor.py
def _get_streaming_processes(self):
    '''
    Returns the processes to handle streams, including the stream-mgr and the user code containing
    the stream logic of the topology
    '''
    retval = {}
    instance_plans = self._get_instance_plans(self.packing_plan, self.shard)
    instance_info = []
    for instance_plan in instance_plans:
      global_task_id = instance_plan.task_id
      component_index = instance_plan.component_index
      component_name = instance_plan.component_name
      instance_id = "container_%s_%s_%d" % (str(self.shard), component_name, global_task_id)
      instance_info.append((instance_id, component_name, global_task_id, component_index))

    stmgr_cmd_lst = [
        self.stmgr_binary,
        '--topology_name=%s' % self.topology_name,
        '--topology_id=%s' % self.topology_id,
        '--topologydefn_file=%s' % self.topology_defn_file,
        '--zkhostportlist=%s' % self.state_manager_connection,
        '--zkroot=%s' % self.state_manager_root,
        '--stmgr_id=%s' % self.stmgr_ids[self.shard],
        '--instance_ids=%s' % ','.join(map(lambda x: x[0], instance_info)),
        '--myhost=%s' % self.master_host,
        '--data_port=%s' % str(self.master_port),
        '--local_data_port=%s' % str(self.tmaster_controller_port),
        '--metricsmgr_port=%s' % str(self.metrics_manager_port),
        '--shell_port=%s' % str(self.shell_port),
        '--config_file=%s' % self.heron_internals_config_file,
        '--override_config_file=%s' % self.override_config_file,
        '--ckptmgr_port=%s' % str(self.checkpoint_manager_port),
        '--ckptmgr_id=%s' % self.ckptmgr_ids[self.shard],
        '--metricscachemgr_mode=%s' % self.metricscache_manager_mode.lower()]

    stmgr_env = self.shell_env.copy() if self.shell_env is not None else {}
    stmgr_cmd = Command(stmgr_cmd_lst, stmgr_env)
    if os.environ.get('ENABLE_HEAPCHECK') is not None:
      stmgr_cmd.env.update({
          'LD_PRELOAD': "/usr/lib/libtcmalloc.so",
          'HEAPCHECK': "normal"
      })

    retval[self.stmgr_ids[self.shard]] = stmgr_cmd

    # metricsmgr_metrics_sink_config_file = 'metrics_sinks.yaml'

    retval[self.metricsmgr_ids[self.shard]] = self._get_metricsmgr_cmd(
        self.metricsmgr_ids[self.shard],
        self.metrics_sinks_config_file,
        self.metrics_manager_port
    )

    if self.is_stateful_topology:
      retval.update(self._get_ckptmgr_process())

    if self.pkg_type == 'jar' or self.pkg_type == 'tar':
      retval.update(self._get_java_instance_cmd(instance_info))
    elif self.pkg_type == 'pex':
      retval.update(self._get_python_instance_cmd(instance_info))
    elif self.pkg_type == 'so':
      retval.update(self._get_cpp_instance_cmd(instance_info))
    elif self.pkg_type == 'dylib':
      retval.update(self._get_cpp_instance_cmd(instance_info))
    else:
      raise ValueError("Unrecognized package type: %s" % self.pkg_type)

    return retval
def _get_streaming_processes(self):
    '''
    Returns the processes to handle streams, including the stream-mgr and the user code containing
    the stream logic of the topology
    '''
    retval = {}
    instance_plans = self._get_instance_plans(self.packing_plan, self.shard)
    instance_info = []
    for instance_plan in instance_plans:
      global_task_id = instance_plan.task_id
      component_index = instance_plan.component_index
      component_name = instance_plan.component_name
      instance_id = "container_%s_%s_%d" % (str(self.shard), component_name, global_task_id)
      instance_info.append((instance_id, component_name, global_task_id, component_index))

    stmgr_cmd_lst = [
        self.stmgr_binary,
        '--topology_name=%s' % self.topology_name,
        '--topology_id=%s' % self.topology_id,
        '--topologydefn_file=%s' % self.topology_defn_file,
        '--zkhostportlist=%s' % self.state_manager_connection,
        '--zkroot=%s' % self.state_manager_root,
        '--stmgr_id=%s' % self.stmgr_ids[self.shard],
        '--instance_ids=%s' % ','.join(map(lambda x: x[0], instance_info)),
        '--myhost=%s' % self.master_host,
        '--data_port=%s' % str(self.master_port),
        '--local_data_port=%s' % str(self.tmaster_controller_port),
        '--metricsmgr_port=%s' % str(self.metrics_manager_port),
        '--shell_port=%s' % str(self.shell_port),
        '--config_file=%s' % self.heron_internals_config_file,
        '--override_config_file=%s' % self.override_config_file,
        '--ckptmgr_port=%s' % str(self.checkpoint_manager_port),
        '--ckptmgr_id=%s' % self.ckptmgr_ids[self.shard],
        '--metricscachemgr_mode=%s' % self.metricscache_manager_mode.lower()]

    stmgr_env = self.shell_env.copy() if self.shell_env is not None else {}
    stmgr_cmd = Command(stmgr_cmd_lst, stmgr_env)
    if os.environ.get('ENABLE_HEAPCHECK') is not None:
      stmgr_cmd.env.update({
          'LD_PRELOAD': "/usr/lib/libtcmalloc.so",
          'HEAPCHECK': "normal"
      })

    retval[self.stmgr_ids[self.shard]] = stmgr_cmd

    # metricsmgr_metrics_sink_config_file = 'metrics_sinks.yaml'

    retval[self.metricsmgr_ids[self.shard]] = self._get_metricsmgr_cmd(
        self.metricsmgr_ids[self.shard],
        self.metrics_sinks_config_file,
        self.metrics_manager_port
    )

    if self.is_stateful_topology:
      retval.update(self._get_ckptmgr_process())

    if self.pkg_type == 'jar' or self.pkg_type == 'tar':
      retval.update(self._get_java_instance_cmd(instance_info))
    elif self.pkg_type == 'pex':
      retval.update(self._get_python_instance_cmd(instance_info))
    elif self.pkg_type == 'so':
      retval.update(self._get_cpp_instance_cmd(instance_info))
    elif self.pkg_type == 'dylib':
      retval.update(self._get_cpp_instance_cmd(instance_info))
    else:
      raise ValueError("Unrecognized package type: %s" % self.pkg_type)

    return retval
def _get_streaming_processes ( self ) : retval = { } instance_plans = self . _get_instance_plans ( self . packing_plan , self . shard ) instance_info = [ ] for instance_plan in instance_plans : global_task_id = instance_plan . task_id component_index = instance_plan . component_index component_name = instance_plan . component_name instance_id = "container_%s_%s_%d" % ( str ( self . shard ) , component_name , global_task_id ) instance_info . append ( ( instance_id , component_name , global_task_id , component_index ) ) stmgr_cmd_lst = [ self . stmgr_binary , '--topology_name=%s' % self . topology_name , '--topology_id=%s' % self . topology_id , '--topologydefn_file=%s' % self . topology_defn_file , '--zkhostportlist=%s' % self . state_manager_connection , '--zkroot=%s' % self . state_manager_root , '--stmgr_id=%s' % self . stmgr_ids [ self . shard ] , '--instance_ids=%s' % ',' . join ( map ( lambda x : x [ 0 ] , instance_info ) ) , '--myhost=%s' % self . master_host , '--data_port=%s' % str ( self . master_port ) , '--local_data_port=%s' % str ( self . tmaster_controller_port ) , '--metricsmgr_port=%s' % str ( self . metrics_manager_port ) , '--shell_port=%s' % str ( self . shell_port ) , '--config_file=%s' % self . heron_internals_config_file , '--override_config_file=%s' % self . override_config_file , '--ckptmgr_port=%s' % str ( self . checkpoint_manager_port ) , '--ckptmgr_id=%s' % self . ckptmgr_ids [ self . shard ] , '--metricscachemgr_mode=%s' % self . metricscache_manager_mode . lower ( ) ] stmgr_env = self . shell_env . copy ( ) if self . shell_env is not None else { } stmgr_cmd = Command ( stmgr_cmd_lst , stmgr_env ) if os . environ . get ( 'ENABLE_HEAPCHECK' ) is not None : stmgr_cmd . env . update ( { 'LD_PRELOAD' : "/usr/lib/libtcmalloc.so" , 'HEAPCHECK' : "normal" } ) retval [ self . stmgr_ids [ self . shard ] ] = stmgr_cmd # metricsmgr_metrics_sink_config_file = 'metrics_sinks.yaml' retval [ self . metricsmgr_ids [ self . shard ] ] = self . _get_metricsmgr_cmd ( self . metricsmgr_ids [ self . shard ] , self . metrics_sinks_config_file , self . metrics_manager_port ) if self . is_stateful_topology : retval . update ( self . _get_ckptmgr_process ( ) ) if self . pkg_type == 'jar' or self . pkg_type == 'tar' : retval . update ( self . _get_java_instance_cmd ( instance_info ) ) elif self . pkg_type == 'pex' : retval . update ( self . _get_python_instance_cmd ( instance_info ) ) elif self . pkg_type == 'so' : retval . update ( self . _get_cpp_instance_cmd ( instance_info ) ) elif self . pkg_type == 'dylib' : retval . update ( self . _get_cpp_instance_cmd ( instance_info ) ) else : raise ValueError ( "Unrecognized package type: %s" % self . pkg_type ) return retval
Returns the processes to handle streams, including the stream-mgr and the user code containing
    the stream logic of the topology
Returns the processes to handle streams including the stream - mgr and the user code containing the stream logic of the topology