heron/executor/src/python/heron_executor.py
def _get_tmaster_processes(self):
    ''' get the command to start the tmaster processes '''
    retval = {}
    tmaster_cmd_lst = [
        self.tmaster_binary,
        '--topology_name=%s' % self.topology_name,
        '--topology_id=%s' % self.topology_id,
        '--zkhostportlist=%s' % self.state_manager_connection,
        '--zkroot=%s' % self.state_manager_root,
        '--myhost=%s' % self.master_host,
        '--master_port=%s' % str(self.master_port),
        '--controller_port=%s' % str(self.tmaster_controller_port),
        '--stats_port=%s' % str(self.tmaster_stats_port),
        '--config_file=%s' % self.heron_internals_config_file,
        '--override_config_file=%s' % self.override_config_file,
        '--metrics_sinks_yaml=%s' % self.metrics_sinks_config_file,
        '--metricsmgr_port=%s' % str(self.metrics_manager_port),
        '--ckptmgr_port=%s' % str(self.checkpoint_manager_port)]

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

    retval["heron-tmaster"] = tmaster_cmd

    if self.metricscache_manager_mode.lower() != "disabled":
      retval["heron-metricscache"] = self._get_metrics_cache_cmd()

    if self.health_manager_mode.lower() != "disabled":
      retval["heron-healthmgr"] = self._get_healthmgr_cmd()

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

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

    return retval
def _get_tmaster_processes(self):
    ''' get the command to start the tmaster processes '''
    retval = {}
    tmaster_cmd_lst = [
        self.tmaster_binary,
        '--topology_name=%s' % self.topology_name,
        '--topology_id=%s' % self.topology_id,
        '--zkhostportlist=%s' % self.state_manager_connection,
        '--zkroot=%s' % self.state_manager_root,
        '--myhost=%s' % self.master_host,
        '--master_port=%s' % str(self.master_port),
        '--controller_port=%s' % str(self.tmaster_controller_port),
        '--stats_port=%s' % str(self.tmaster_stats_port),
        '--config_file=%s' % self.heron_internals_config_file,
        '--override_config_file=%s' % self.override_config_file,
        '--metrics_sinks_yaml=%s' % self.metrics_sinks_config_file,
        '--metricsmgr_port=%s' % str(self.metrics_manager_port),
        '--ckptmgr_port=%s' % str(self.checkpoint_manager_port)]

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

    retval["heron-tmaster"] = tmaster_cmd

    if self.metricscache_manager_mode.lower() != "disabled":
      retval["heron-metricscache"] = self._get_metrics_cache_cmd()

    if self.health_manager_mode.lower() != "disabled":
      retval["heron-healthmgr"] = self._get_healthmgr_cmd()

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

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

    return retval
def _get_tmaster_processes ( self ) : retval = { } tmaster_cmd_lst = [ self . tmaster_binary , '--topology_name=%s' % self . topology_name , '--topology_id=%s' % self . topology_id , '--zkhostportlist=%s' % self . state_manager_connection , '--zkroot=%s' % self . state_manager_root , '--myhost=%s' % self . master_host , '--master_port=%s' % str ( self . master_port ) , '--controller_port=%s' % str ( self . tmaster_controller_port ) , '--stats_port=%s' % str ( self . tmaster_stats_port ) , '--config_file=%s' % self . heron_internals_config_file , '--override_config_file=%s' % self . override_config_file , '--metrics_sinks_yaml=%s' % self . metrics_sinks_config_file , '--metricsmgr_port=%s' % str ( self . metrics_manager_port ) , '--ckptmgr_port=%s' % str ( self . checkpoint_manager_port ) ] tmaster_env = self . shell_env . copy ( ) if self . shell_env is not None else { } tmaster_cmd = Command ( tmaster_cmd_lst , tmaster_env ) if os . environ . get ( 'ENABLE_HEAPCHECK' ) is not None : tmaster_cmd . env . update ( { 'LD_PRELOAD' : "/usr/lib/libtcmalloc.so" , 'HEAPCHECK' : "normal" } ) retval [ "heron-tmaster" ] = tmaster_cmd if self . metricscache_manager_mode . lower ( ) != "disabled" : retval [ "heron-metricscache" ] = self . _get_metrics_cache_cmd ( ) if self . health_manager_mode . lower ( ) != "disabled" : retval [ "heron-healthmgr" ] = self . _get_healthmgr_cmd ( ) retval [ self . metricsmgr_ids [ 0 ] ] = self . _get_metricsmgr_cmd ( self . metricsmgr_ids [ 0 ] , self . metrics_sinks_config_file , self . metrics_manager_port ) if self . is_stateful_topology : retval . update ( self . _get_ckptmgr_process ( ) ) return retval
get the command to start the tmaster processes
get the command to start the tmaster processes