get all workers heartbeats of the supervisor

@param conf conf
@throws IOException
430962
public Map<String, WorkerHeartbeat> readWorkerHeartbeats(Map conf) throws Exception {
        Map<String, WorkerHeartbeat> workerHeartbeats = new HashMap<>();

        // get the path: STORM-LOCAL-DIR/workers
        String path = StormConfig.worker_root(conf);

        List<String> workerIds = PathUtils.read_dir_contents(path);

        if (workerIds == null) {
            LOG.info("No worker dir under " + path);
            return workerHeartbeats;
        }

        for (String workerId : workerIds) {
            WorkerHeartbeat whb = readWorkerHeartbeat(conf, workerId);
            // ATTENTION: whb can be null
            workerHeartbeats.put(workerId, whb);
        }
        return workerHeartbeats;
    }
1172140
public void heartbeat(final List<Metric> metrics) throws IOException {
    connect();
    try {
      MetricsHeartbeatPRequest.Builder request = MetricsHeartbeatPRequest.newBuilder();
      request.setClientId(IdUtils.createOrGetAppIdFromConfig(mContext.getConf()));
      request.setHostname(NetworkAddressUtils.getClientHostName(mContext.getConf()));
      request.setOptions(MetricsHeartbeatPOptions.newBuilder().addAllMetrics(metrics).build());
      mClient.metricsHeartbeat(request.build());
    } catch (io.grpc.StatusRuntimeException e) {
      throw new UnavailableException(e);
    }
  }
183733
private SupervisorWorkers getSupervisorWorkersByHostOrId(String host, String id) throws TException {
        long start = System.nanoTime();

        if (StringUtils.isBlank(id) && StringUtils.isBlank(host)) {
            throw new TException("Must specify host or supervisor id!");
        }

        try {
            StormClusterState stormClusterState = data.getStormClusterState();

            // all supervisors
            Map<String, SupervisorInfo> supervisorInfos = Cluster.get_all_SupervisorInfo(stormClusterState, null);
            SupervisorInfo supervisorInfo = null;
            String ip;
            if (!StringUtils.isBlank(id)) {
                supervisorInfo = supervisorInfos.get(id);
                host = supervisorInfo.getHostName();
                ip = NetWorkUtils.host2Ip(host);
            } else {
                ip = NetWorkUtils.host2Ip(host);
                for (Entry<String, SupervisorInfo> entry : supervisorInfos.entrySet()) {
                    SupervisorInfo info = entry.getValue();
                    if (info.getHostName().equals(host) || info.getHostName().equals(ip)) {
                        id = entry.getKey();
                        supervisorInfo = info;
                        break;
                    }
                }
            }
            if (supervisorInfo == null) {
                throw new TException("unknown supervisor id:" + id);
            }

            Map<String, Assignment> assignments = Cluster.get_all_assignment(stormClusterState, null);
            Map<Integer, WorkerSummary> portWorkerSummaries = new TreeMap<>();
            int usedSlotNumber = 0;
            Map<String, Map<Integer, String>> topologyTaskToComponent = new HashMap<>();

            Map<String, MetricInfo> metricInfoMap = new HashMap<>();
            for (Entry<String, Assignment> entry : assignments.entrySet()) {
                String topologyId = entry.getKey();
                Assignment assignment = entry.getValue();

                Set<ResourceWorkerSlot> workers = assignment.getWorkers();
                for (ResourceWorkerSlot worker : workers) {
                    if (!id.equals(worker.getNodeId())) {
                        continue;
                    }
                    usedSlotNumber++;

                    Integer port = worker.getPort();
                    WorkerSummary workerSummary = portWorkerSummaries.get(port);
                    if (workerSummary == null) {
                        workerSummary = new WorkerSummary();
                        workerSummary.set_port(port);
                        workerSummary.set_topology(topologyId);
                        workerSummary.set_tasks(new ArrayList<TaskComponent>());

                        portWorkerSummaries.put(port, workerSummary);
                    }

                    Map<Integer, String> taskToComponent = topologyTaskToComponent.get(topologyId);
                    if (taskToComponent == null) {
                        taskToComponent = Cluster.get_all_task_component(stormClusterState, topologyId, null);
                        topologyTaskToComponent.put(topologyId, taskToComponent);
                    }

                    int earliest = TimeUtils.current_time_secs();
                    for (Integer taskId : worker.getTasks()) {
                        TaskComponent taskComponent = new TaskComponent();
                        taskComponent.set_component(taskToComponent.get(taskId));
                        taskComponent.set_taskId(taskId);
                        Integer startTime = assignment.getTaskStartTimeSecs().get(taskId);
                        if (startTime != null && startTime < earliest) {
                            earliest = startTime;
                        }

                        workerSummary.add_to_tasks(taskComponent);
                    }

                    workerSummary.set_uptime(TimeUtils.time_delta(earliest));

                    String workerSlotName = getWorkerSlotName(ip, port);
                    List<MetricInfo> workerMetricInfoList = this.data.getMetricCache().getMetricData(topologyId, MetaType.WORKER);
                    if (workerMetricInfoList.size() > 0) {
                        MetricInfo workerMetricInfo = workerMetricInfoList.get(0);
                        // remove metrics that don't belong to current worker
                        for (Iterator<String> itr = workerMetricInfo.get_metrics().keySet().iterator();
                             itr.hasNext(); ) {
                            String metricName = itr.next();
                            if (!metricName.contains(ip)) {
                                itr.remove();
                            }
                        }
                        metricInfoMap.put(workerSlotName, workerMetricInfo);
                    }
                }
            }

            List<WorkerSummary> workerList = new ArrayList<>();
            workerList.addAll(portWorkerSummaries.values());

            Map<String, Integer> supervisorToUsedSlotNum = new HashMap<>();
            supervisorToUsedSlotNum.put(id, usedSlotNumber);
            SupervisorSummary supervisorSummary = NimbusUtils.mkSupervisorSummary(supervisorInfo, id, supervisorToUsedSlotNum);

            return new SupervisorWorkers(supervisorSummary, workerList, metricInfoMap);
        } catch (TException e) {
            LOG.info("Failed to get ClusterSummary ", e);
            throw e;
        } catch (Exception e) {
            LOG.info("Failed to get ClusterSummary ", e);
            throw new TException(e);
        } finally {
            long end = System.nanoTime();
            SimpleJStormMetric.updateNimbusHistogram("getSupervisorWorkers", (end - start) / TimeUtils.NS_PER_US);
        }

    }
607206
def do_workers(self, args):
        ''''''
        workers = self.task_master.workers(alive=not args.all)
        for k in sorted(workers.iterkeys()):
            self.stdout.write('{0} ({1})\n'.format(k, workers[k]))
            if args.details:
                heartbeat = self.task_master.get_heartbeat(k)
                for hk, hv in heartbeat.iteritems():
                    self.stdout.write('  {0}: {1}\n'.format(hk, hv))
441353
@Override
    public void heartbeat() throws IOException {
        isOpen();
        int ret = WDT.ping(fd);
        if(ret<0) {
            throw new IOException("Heartbeat error. File " + filename + " got " + ret + " back.");
        }
    }
507740
@SuppressWarnings("rawtypes")
    public SupervisorManger mkSupervisor(Map conf, IContext sharedContext) throws Exception {

        LOG.info("Starting Supervisor with conf " + conf);

        /**
         * Step 1: cleanup all files in /storm-local-dir/supervisor/tmp
         */
        String path = StormConfig.supervisorTmpDir(conf);
        FileUtils.cleanDirectory(new File(path));

        /**
         * Step 2: create ZK operation instance StormClusterState
         */

        StormClusterState stormClusterState = Cluster.mk_storm_cluster_state(conf);

        String hostName = JStormServerUtils.getHostName(conf);
        WorkerReportError workerReportError = new WorkerReportError(stormClusterState, hostName);

        /**
         * Step 3, create LocalStat (a simple KV store)
         * 3.1 create LocalState instance;
         * 3.2 get supervisorId, if there's no supervisorId, create one
         */

        LocalState localState = StormConfig.supervisorState(conf);

        String supervisorId = (String) localState.get(Common.LS_ID);
        if (supervisorId == null) {
            supervisorId = UUID.randomUUID().toString();
            localState.put(Common.LS_ID, supervisorId);
        }
        //clean LocalStat's zk-assignment & versions
        localState.remove(Common.LS_LOCAl_ZK_ASSIGNMENTS);
        localState.remove(Common.LS_LOCAL_ZK_ASSIGNMENT_VERSION);

        Vector<AsyncLoopThread> threads = new Vector<>();

        // Step 4 create HeartBeat
        // every supervisor.heartbeat.frequency.secs, write SupervisorInfo to ZK
        // sync heartbeat to nimbus
        Heartbeat hb = new Heartbeat(conf, stormClusterState, supervisorId, localState);
        hb.update();

        AsyncLoopThread heartbeat = new AsyncLoopThread(hb, false, null, Thread.MIN_PRIORITY, true);
        threads.add(heartbeat);

        // Sync heartbeat to Apsara Container
        AsyncLoopThread syncContainerHbThread = SyncContainerHb.mkSupervisorInstance(conf);
        if (syncContainerHbThread != null) {
            threads.add(syncContainerHbThread);
        }

        // Step 5 create and start sync Supervisor thread
        // every supervisor.monitor.frequency.secs second run SyncSupervisor
        ConcurrentHashMap<String, String> workerThreadPids = new ConcurrentHashMap<>();
        SyncProcessEvent syncProcessEvent = new SyncProcessEvent(
                supervisorId, conf, localState, workerThreadPids, sharedContext, workerReportError, stormClusterState);

        EventManagerImp syncSupEventManager = new EventManagerImp();
        AsyncLoopThread syncSupEventThread = new AsyncLoopThread(syncSupEventManager);
        threads.add(syncSupEventThread);

        SyncSupervisorEvent syncSupervisorEvent = new SyncSupervisorEvent(
                supervisorId, conf, syncSupEventManager, stormClusterState, localState, syncProcessEvent, hb);

        int syncFrequency = JStormUtils.parseInt(conf.get(Config.SUPERVISOR_MONITOR_FREQUENCY_SECS));
        EventManagerPusher syncSupervisorPusher = new EventManagerPusher(
                syncSupEventManager, syncSupervisorEvent, syncFrequency);
        AsyncLoopThread syncSupervisorThread = new AsyncLoopThread(syncSupervisorPusher);
        threads.add(syncSupervisorThread);

        // Step 6 start httpserver
        Httpserver httpserver = null;
        if (!StormConfig.local_mode(conf)) {
            int port = ConfigExtension.getSupervisorDeamonHttpserverPort(conf);
            httpserver = new Httpserver(port, conf);
            httpserver.start();
        }

        //Step 7 check supervisor
        if (!StormConfig.local_mode(conf)) {
            if (ConfigExtension.isEnableCheckSupervisor(conf)) {
                SupervisorHealth supervisorHealth = new SupervisorHealth(conf, hb, supervisorId);
                AsyncLoopThread healthThread = new AsyncLoopThread(supervisorHealth, false, null, Thread.MIN_PRIORITY, true);
                threads.add(healthThread);
            }

            // init refresh config thread
            AsyncLoopThread refreshConfigThread = new AsyncLoopThread(new SupervisorRefreshConfig(conf));
            threads.add(refreshConfigThread);
        }

        // create SupervisorManger which can shutdown all supervisor and workers
        return new SupervisorManger(
                conf, supervisorId, threads, syncSupEventManager, httpserver, stormClusterState, workerThreadPids);
    }
509315
@Override
    public GenThread[] prepare(JobConf conf, Text key, Text value)
        throws IOException {
      this.rtc = new RunTimeConstants();
      super.prepare(conf, key, value, rtc);
      Path basePath = new Path(key.toString());
      LOG.info("base path is " + basePath);
      Path checksumPath = null;
      FileSystem fs = FileSystem.newInstance(conf);
      if (value.toString().length() != 0) {
        checksumPath = new Path(value.toString());
      }
      HashMap<String, Long> checksumMap = null;
      boolean verifyChecksum = false;
      if (fs.exists(checksumPath)) {
        LOG.info("checksum path is " + checksumPath);
        verifyChecksum = true;
        checksumMap = new HashMap<String, Long>();
        SequenceFile.Reader reader = null;
        try {
          reader = new SequenceFile.Reader(fs, checksumPath, conf);
          Writable dir = (Writable) ReflectionUtils.newInstance(
            reader.getKeyClass(), conf);
          Writable checksum = (Writable) ReflectionUtils.newInstance(
            reader.getValueClass(), conf);
          while(reader.next(dir, checksum)) {
            LOG.info("dir: " + dir.toString() + " checksum: " + checksum);
            checksumMap.put(
                fs.makeQualified(new Path(dir.toString())).toUri().getPath(),
                Long.parseLong(checksum.toString()));
          }
        } catch(Exception e) {
          LOG.error(e);
          throw new IOException(e);
        } finally {
          IOUtils.closeStream(reader);
        }
      }
      
      FileStatus[] baseDirs = fs.listStatus(basePath);
      if (rtc.nthreads != baseDirs.length) {
        throw new IOException("Number of directory under " + basePath + 
            "(" + baseDirs.length + ") doesn't match number of threads " + 
            "(" + rtc.nthreads + ").");
      }
      
      GenReaderThread[] threads = new GenReaderThread[(int)rtc.nthreads];

      for (int i=0; i < rtc.nthreads; i++) {
        long checksum = 0;
        if (verifyChecksum) {
          String basePathStr = baseDirs[i].getPath().toUri().getPath();
          checksum = checksumMap.get(basePathStr);
        }
        threads[i] = new GenReaderThread(conf, baseDirs[i].getPath(), 
            checksum, verifyChecksum, rtc);
      }
      return threads;
    }
315908
public static void main(String[] args) {
    if (args.length != 0) {
      LOG.info("java -cp {} {}", RuntimeConstants.ALLUXIO_JAR,
          AlluxioWorker.class.getCanonicalName());
      System.exit(-1);
    }

    if (!ConfigurationUtils.masterHostConfigured(ServerConfiguration.global())) {
      ProcessUtils.fatalError(LOG,
          ConfigurationUtils.getMasterHostNotConfiguredMessage("Alluxio worker"));
    }

    CommonUtils.PROCESS_TYPE.set(CommonUtils.ProcessType.WORKER);
    MasterInquireClient masterInquireClient =
        MasterInquireClient.Factory.create(ServerConfiguration.global());
    try {
      RetryUtils.retry("load cluster default configuration with master", () -> {
        InetSocketAddress masterAddress = masterInquireClient.getPrimaryRpcAddress();
        ServerConfiguration.loadClusterDefaults(masterAddress);
      }, RetryUtils.defaultWorkerMasterClientRetry(
          ServerConfiguration.getDuration(PropertyKey.WORKER_MASTER_CONNECT_RETRY_TIMEOUT)));
    } catch (IOException e) {
      ProcessUtils.fatalError(LOG,
          "Failed to load cluster default configuration for worker: %s", e.getMessage());
    }
    WorkerProcess process = WorkerProcess.Factory.create();
    ProcessUtils.run(process);
  }
193066
public int computeDatanodeWork() throws IOException {
    int workFound = 0;
    int blocksToProcess = 0;
    int nodesToProcess = 0;
    // blocks should not be replicated or removed if safe mode is on
    if (isInSafeMode()) {
      updateReplicationCounts(workFound);
      return workFound;
    }
    synchronized (heartbeats) {
      blocksToProcess = (int) (heartbeats.size()
        * ReplicationConfigKeys.replicationWorkMultiplier);
      nodesToProcess = (int) Math.ceil((double) heartbeats.size()
        * ReplicationConfigKeys.INVALIDATE_WORK_PCT_PER_ITERATION / 100);
    }

    workFound = computeReplicationWork(blocksToProcess);

    // Update FSNamesystemMetrics counters
    updateReplicationCounts(workFound);

    workFound += computeInvalidateWork(nodesToProcess);
    return workFound;
  }
561642
public void sessionHeartbeat(long sessionId) {
    synchronized (mSessions) {
      if (mSessions.containsKey(sessionId)) {
        mSessions.get(sessionId).heartbeat();
      } else {
        int sessionTimeoutMs = (int) ServerConfiguration
            .getMs(PropertyKey.WORKER_SESSION_TIMEOUT_MS);
        mSessions.put(sessionId, new SessionInfo(sessionId, sessionTimeoutMs));
      }
    }
  }
1317567
def __get_supervisor(self):
        
        options = supervisorctl.ClientOptions()
        options.realize(args=['-c', self.supervisord_conf_path])
        return supervisorctl.Controller(options).get_supervisor()
748137
public static void main(String[] args) {
    if (args.length != 0) {
      LOG.info("java -cp {} {}", RuntimeConstants.ALLUXIO_JAR,
          AlluxioWorkerMonitor.class.getCanonicalName());
      LOG.warn("ignoring arguments");
    }
    AlluxioConfiguration conf = new InstancedConfiguration(ConfigurationUtils.defaults());

    HealthCheckClient client = new WorkerHealthCheckClient(
        NetworkAddressUtils.getConnectAddress(NetworkAddressUtils.ServiceType.WORKER_RPC, conf),
        ONE_MIN_EXP_BACKOFF, conf);
    if (!client.isServing()) {
      System.exit(1);
    }
    System.exit(0);
  }
1009646
def read_worker_status(options)
      on_activity = options[:on_activity]
      until @master_socket.nil?
        IO.select([@master_socket], nil, nil, 0) or return
        case line = @master_socket.readline
        when /^\+(\d+),(.*)$/
          worker_pids[$1] = $2
          on_activity.worker_started($1) if on_activity
        when /^-(\d+)$/
          worker_pids.delete($1)
          on_activity.worker_finished($1) if on_activity
        when /^RUNNING/
          on_activity.listener_running(self) if on_activity
        when ''
          break
        else
          log "Malformed data from listener: #{line.inspect}"
        end
      end
    rescue EOFError, Errno::ECONNRESET
      @master_socket.close
      @master_socket = nil
    end
639382
def _handle_getconfig(self,cfg_file,*args,**options):
        """"""
        if args:
            raise CommandError("supervisor getconfig takes no arguments")
        print cfg_file.read()
        return 0
621468
def heartbeat(self):
        """"""
        while self._manager.is_active and not self._stop_event.is_set():
            self._manager.heartbeat()
            _LOGGER.debug("Sent heartbeat.")
            self._stop_event.wait(timeout=self._period)

        _LOGGER.info("%s exiting.", _HEARTBEAT_WORKER_NAME)
1090575
private void recordHeartbeat(RaftMemberContext member, long timestamp) {
    raft.checkThread();

    // Update the member's heartbeat time. This will be used when calculating the quorum heartbeat time.
    member.setHeartbeatTime(timestamp);
    member.setResponseTime(System.currentTimeMillis());

    // Compute the quorum heartbeat time.
    long heartbeatTime = computeHeartbeatTime();
    long currentTimestamp = System.currentTimeMillis();

    // Iterate through pending timestamped heartbeat futures and complete all futures where the timestamp
    // is greater than the last timestamp a quorum of the cluster was contacted.
    Iterator<TimestampedFuture<Long>> iterator = heartbeatFutures.iterator();
    while (iterator.hasNext()) {
      TimestampedFuture<Long> future = iterator.next();

      // If the future is timestamped prior to the last heartbeat to a majority of the cluster, complete the future.
      if (future.timestamp < heartbeatTime) {
        future.complete(null);
        iterator.remove();
      }
      // If the future is more than an election timeout old, fail it with a protocol exception.
      else if (currentTimestamp - future.timestamp > electionTimeout) {
        future.completeExceptionally(new RaftException.ProtocolException("Failed to reach consensus"));
        iterator.remove();
      }
      // Otherwise, we've reached recent heartbeat futures. Break out of the loop.
      else {
        break;
      }
    }

    // If heartbeat futures are still pending, attempt to send heartbeats.
    if (!heartbeatFutures.isEmpty()) {
      sendHeartbeats();
    }
  }
70936
private void checkSystemInput()
        throws IOException
    {
        while ( System.in.available() > 0 )
        {
            int input = System.in.read();
            if ( input >= 0 )
            {
                char c = (char) input;
                if ( c == '\n' )
                {
                    synchronized ( lock )
                    {
                        finished = true;
                        lock.notifyAll();
                    }
                }
            }
            else
            {
                synchronized ( lock )
                {
                    finished = true;
                    lock.notifyAll();
                }
            }
        }
    }