| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package com.alibaba.jstorm.daemon.supervisor; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.util.Collection; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| import java.util.regex.Pattern; |
| |
| import com.alibaba.jstorm.daemon.worker.*; |
| import org.apache.commons.lang.StringUtils; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import backtype.storm.Config; |
| import backtype.storm.GenericOptionsParser; |
| import backtype.storm.messaging.IContext; |
| import backtype.storm.utils.LocalState; |
| |
| import com.alibaba.jstorm.client.ConfigExtension; |
| import com.alibaba.jstorm.cluster.Common; |
| import com.alibaba.jstorm.cluster.StormConfig; |
| import com.alibaba.jstorm.utils.JStormUtils; |
| import com.alibaba.jstorm.utils.Pair; |
| import com.alibaba.jstorm.utils.PathUtils; |
| import com.alibaba.jstorm.utils.TimeFormat; |
| import com.alibaba.jstorm.utils.TimeUtils; |
| |
| /** |
| * SyncProcesses (1) kill bad worker (2) start new worker |
| * @author Johnfang (xiaojian.fxj@alibaba-inc.com) |
| */ |
| class SyncProcessEvent extends ShutdownWork { |
| private static Logger LOG = LoggerFactory.getLogger(SyncProcessEvent.class); |
| |
| private LocalState localState; |
| |
| private Map conf; |
| |
| private ConcurrentHashMap<String, String> workerThreadPids; |
| |
| private String supervisorId; |
| |
| private IContext sharedContext; |
| |
| private CgroupManager cgroupManager; |
| |
| private SandBoxMaker sandBoxMaker; |
| |
| /** |
| * Due to the worker startTime is put in Supervisor memory, When supervisor restart, the starting worker is likely to be killed |
| */ |
| private Map<String, Pair<Integer, Integer>> workerIdToStartTimeAndPort; |
| /** |
| * workerIdToStartTimeAndPort ensure workId is unique, but don't ensure workId for workerPort |
| */ |
| private Map<Integer, String> portToWorkerId = new HashMap<Integer, String>(); |
| |
| private AtomicReference<Set> needDownloadTopologys; |
| |
| // workers under killing, Map<workerId, time of kill start>; |
| private Map<String, Integer> killingWorkers; |
| |
| // private Supervisor supervisor; |
| private int lastTime; |
| |
| private WorkerReportError workerReportError; |
| |
| /** |
| * @param conf |
| * @param localState |
| * @param workerThreadPids |
| * @param supervisorId |
| * @param sharedContext |
| * @param workerThreadPidsReadLock |
| * @param workerThreadPidsWriteLock |
| */ |
| public SyncProcessEvent(String supervisorId, Map conf, LocalState localState, ConcurrentHashMap<String, String> workerThreadPids, |
| IContext sharedContext, WorkerReportError workerReportError) { |
| |
| this.supervisorId = supervisorId; |
| |
| this.conf = conf; |
| |
| this.localState = localState; |
| |
| this.workerThreadPids = workerThreadPids; |
| |
| // right now, sharedContext is null |
| this.sharedContext = sharedContext; |
| |
| this.sandBoxMaker = new SandBoxMaker(conf); |
| |
| this.workerIdToStartTimeAndPort = new HashMap<String, Pair<Integer, Integer>>(); |
| |
| this.needDownloadTopologys = new AtomicReference<Set>(); |
| |
| if (ConfigExtension.isEnableCgroup(conf)) { |
| cgroupManager = new CgroupManager(conf); |
| } |
| |
| killingWorkers = new HashMap<String, Integer>(); |
| this.workerReportError = workerReportError; |
| } |
| |
| /** |
| * @@@ Change the old logic In the old logic, it will store LS_LOCAL_ASSIGNMENTS Map<String, Integer> into LocalState |
| * |
| * But I don't think LS_LOCAL_ASSIGNMENTS is useful, so remove this logic |
| */ |
| @SuppressWarnings("unchecked") |
| @Override |
| public void run() { |
| |
| } |
| |
| public void run(Map<Integer, LocalAssignment> localAssignments, Set<String> downloadFailedTopologyIds ) { |
| LOG.debug("Syncing processes, interval seconds:" + TimeUtils.time_delta(lastTime)); |
| lastTime = TimeUtils.current_time_secs(); |
| try { |
| |
| /** |
| * Step 1: get assigned tasks from localstat Map<port(type Integer), LocalAssignment> |
| */ |
| if (localAssignments == null) { |
| localAssignments = new HashMap<Integer, LocalAssignment>(); |
| } |
| LOG.debug("Assigned tasks: " + localAssignments); |
| |
| /** |
| * Step 2: get local WorkerStats from local_dir/worker/ids/heartbeat Map<workerid [WorkerHeartbeat, state]> |
| */ |
| Map<String, StateHeartbeat> localWorkerStats = null; |
| try { |
| localWorkerStats = getLocalWorkerStats(conf, localState, localAssignments); |
| } catch (Exception e) { |
| LOG.error("Failed to get Local worker stats"); |
| throw e; |
| } |
| LOG.debug("Allocated: " + localWorkerStats); |
| |
| /** |
| * Step 3: kill Invalid Workers and remove killed worker from localWorkerStats |
| */ |
| Map<String, Integer> taskCleaupTimeoutMap = null; |
| Set<Integer> keepPorts = null; |
| try { |
| taskCleaupTimeoutMap = (Map<String, Integer>) localState.get(Common.LS_TASK_CLEANUP_TIMEOUT); |
| keepPorts = killUselessWorkers(localWorkerStats, localAssignments, taskCleaupTimeoutMap); |
| localState.put(Common.LS_TASK_CLEANUP_TIMEOUT, taskCleaupTimeoutMap); |
| } catch (IOException e) { |
| LOG.error("Failed to kill workers", e); |
| } |
| |
| // check new workers |
| checkNewWorkers(conf); |
| |
| // check which topology need update |
| checkNeedUpdateTopologys(localWorkerStats, localAssignments); |
| |
| // start new workers |
| startNewWorkers(keepPorts, localAssignments, downloadFailedTopologyIds); |
| |
| } catch (Exception e) { |
| LOG.error("Failed Sync Process", e); |
| // throw e |
| } |
| |
| } |
| |
| /** |
| * check all workers is failed or not |
| */ |
| @SuppressWarnings("unchecked") |
| public void checkNeedUpdateTopologys(Map<String, StateHeartbeat> localWorkerStats, Map<Integer, LocalAssignment> localAssignments) throws Exception { |
| Set<String> topologys = new HashSet<String>(); |
| Map<String, Long> topologyAssignTimeStamps = new HashMap<String, Long>(); |
| |
| for (Entry<Integer, LocalAssignment> entry : localAssignments.entrySet()) { |
| topologys.add(entry.getValue().getTopologyId()); |
| topologyAssignTimeStamps.put(entry.getValue().getTopologyId(), entry.getValue().getTimeStamp()); |
| } |
| |
| for (StateHeartbeat stateHb : localWorkerStats.values()) { |
| State state = stateHb.getState(); |
| if (!state.equals(State.notStarted)) { |
| String topologyId = stateHb.getHeartbeat().getTopologyId(); |
| topologys.remove(topologyId); |
| } |
| } |
| long currTime = System.currentTimeMillis(); |
| Set<String> needRemoveTopologys = new HashSet<String>(); |
| for (String topologyId : topologys) { |
| try { |
| long newAssignTime = topologyAssignTimeStamps.get(topologyId); |
| if ((currTime - newAssignTime) / 1000 < (JStormUtils.MIN_1 * 2)) { |
| LOG.debug("less 2 minite ,so removed " + topologyId); |
| needRemoveTopologys.add(topologyId); |
| } |
| } catch (Exception e) { |
| LOG.error("Failed to get the time of file last modification for topology" + topologyId, e); |
| needRemoveTopologys.add(topologyId); |
| } |
| } |
| topologys.removeAll(needRemoveTopologys); |
| |
| if (topologys.size() > 0) { |
| LOG.debug("Following topologys is going to re-download the jars, " + topologys); |
| } |
| needDownloadTopologys.set(topologys); |
| } |
| |
| /** |
| * mark all new Workers |
| * |
| * @param workerIds |
| * @pdOid 52b11418-7474-446d-bff5-0ecd68f4954f |
| */ |
| public void markAllNewWorkers(Map<Integer, String> workerIds) { |
| |
| int startTime = TimeUtils.current_time_secs(); |
| |
| for (Entry<Integer, String> entry : workerIds.entrySet()) { |
| String oldWorkerIds = portToWorkerId.get(entry.getKey()); |
| if (oldWorkerIds != null) { |
| workerIdToStartTimeAndPort.remove(oldWorkerIds); |
| // update portToWorkerId |
| LOG.info("exit port is still occupied by old wokerId, so remove unuseful " + oldWorkerIds + " form workerIdToStartTimeAndPort"); |
| } |
| portToWorkerId.put(entry.getKey(), entry.getValue()); |
| workerIdToStartTimeAndPort.put(entry.getValue(), new Pair<Integer, Integer>(startTime, entry.getKey())); |
| } |
| } |
| |
| /** |
| * check new workers if the time is not > * SUPERVISOR_WORKER_START_TIMEOUT_SECS, otherwise info failed |
| * |
| * @param conf |
| * @pdOid f0a6ab43-8cd3-44e1-8fd3-015a2ec51c6a |
| */ |
| public void checkNewWorkers(Map conf) throws IOException, InterruptedException { |
| |
| Set<String> workers = new HashSet<String>(); |
| for (Entry<String, Pair<Integer, Integer>> entry : workerIdToStartTimeAndPort.entrySet()) { |
| String workerId = entry.getKey(); |
| int startTime = entry.getValue().getFirst(); |
| LocalState ls = StormConfig.worker_state(conf, workerId); |
| WorkerHeartbeat whb = (WorkerHeartbeat) ls.get(Common.LS_WORKER_HEARTBEAT); |
| if (whb == null) { |
| if ((TimeUtils.current_time_secs() - startTime) < JStormUtils.parseInt(conf.get(Config.SUPERVISOR_WORKER_START_TIMEOUT_SECS))) { |
| LOG.info(workerId + " still hasn't started"); |
| } else { |
| LOG.error("Failed to start Worker " + workerId); |
| workers.add(workerId); |
| } |
| } else { |
| LOG.info("Successfully start worker " + workerId); |
| workers.add(workerId); |
| } |
| } |
| for (String workerId : workers) { |
| Integer port = this.workerIdToStartTimeAndPort.get(workerId).getSecond(); |
| this.workerIdToStartTimeAndPort.remove(workerId); |
| this.portToWorkerId.remove(port); |
| } |
| } |
| |
| public Map<Integer, String> getPortToWorkerId() { |
| return portToWorkerId; |
| } |
| |
| /** |
| * get localstat approved workerId's map |
| * |
| * @return Map<workerid [workerheart, state]> [workerheart, state] is also a map, key is "workheartbeat" and "state" |
| * @param conf |
| * @param localState |
| * @param assignedTasks |
| * @throws IOException |
| * @pdOid 11c9bebb-d082-4c51-b323-dd3d5522a649 |
| */ |
| @SuppressWarnings("unchecked") |
| public Map<String, StateHeartbeat> getLocalWorkerStats(Map conf, LocalState localState, Map<Integer, LocalAssignment> assignedTasks) throws Exception { |
| |
| Map<String, StateHeartbeat> workeridHbstate = new HashMap<String, StateHeartbeat>(); |
| |
| int now = TimeUtils.current_time_secs(); |
| |
| /** |
| * Get Map<workerId, WorkerHeartbeat> from local_dir/worker/ids/heartbeat |
| */ |
| Map<String, WorkerHeartbeat> idToHeartbeat = readWorkerHeartbeats(conf); |
| for (Entry<String, WorkerHeartbeat> entry : idToHeartbeat.entrySet()) { |
| |
| String workerid = entry.getKey().toString(); |
| |
| WorkerHeartbeat whb = entry.getValue(); |
| |
| State state = null; |
| |
| if (whb == null) { |
| state = State.notStarted; |
| Pair<Integer, Integer> timeToPort = this.workerIdToStartTimeAndPort.get(workerid); |
| if (timeToPort != null) { |
| LocalAssignment localAssignment = assignedTasks.get(timeToPort.getSecond()); |
| if (localAssignment == null) { |
| LOG.info("Following worker don't exit assignment, so remove this port=" + timeToPort.getSecond()); |
| state = State.disallowed; |
| // workerId is disallowed ,so remove it from workerIdToStartTimeAndPort |
| Integer port = this.workerIdToStartTimeAndPort.get(workerid).getSecond(); |
| this.workerIdToStartTimeAndPort.remove(workerid); |
| this.portToWorkerId.remove(port); |
| } |
| } |
| } else if (matchesAssignment(whb, assignedTasks) == false) { |
| |
| // workerId isn't approved or |
| // isn't assigned task |
| state = State.disallowed; |
| |
| } else if ((now - whb.getTimeSecs()) > JStormUtils.parseInt(conf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS))) { |
| if (killingWorkers.containsKey(workerid) == false) { |
| String outTimeInfo = " it is likely to be out of memory, the worker is time out "; |
| workerReportError.report(whb.getTopologyId(), whb.getPort(), |
| whb.getTaskIds(), outTimeInfo); |
| } |
| |
| state = State.timedOut; |
| } else { |
| if (isWorkerDead(workerid)) { |
| if (killingWorkers.containsKey(workerid) == false){ |
| String workeDeadInfo = "Worker is dead "; |
| workerReportError.report(whb.getTopologyId(), whb.getPort(), |
| whb.getTaskIds(), workeDeadInfo); |
| } |
| state = State.timedOut; |
| } else { |
| state = State.valid; |
| } |
| } |
| |
| if (state != State.valid) { |
| if (killingWorkers.containsKey(workerid) == false) |
| LOG.info("Worker:" + workerid + " state:" + state + " WorkerHeartbeat:" + whb + " assignedTasks:" + assignedTasks |
| + " at supervisor time-secs " + now); |
| } else { |
| LOG.debug("Worker:" + workerid + " state:" + state + " WorkerHeartbeat: " + whb + " at supervisor time-secs " + now); |
| } |
| |
| workeridHbstate.put(workerid, new StateHeartbeat(state, whb)); |
| } |
| |
| return workeridHbstate; |
| } |
| |
| /** |
| * check whether the workerheartbeat is allowed in the assignedTasks |
| * |
| * @param whb : WorkerHeartbeat |
| * @param assignedTasks |
| * @return boolean if true, the assignments(LS-LOCAL-ASSIGNMENTS) is match with workerheart if fasle, is not matched |
| */ |
| public boolean matchesAssignment(WorkerHeartbeat whb, Map<Integer, LocalAssignment> assignedTasks) { |
| |
| boolean isMatch = true; |
| LocalAssignment localAssignment = assignedTasks.get(whb.getPort()); |
| |
| if (localAssignment == null) { |
| LOG.debug("Following worker has been removed, port=" + whb.getPort() + ", assignedTasks=" + assignedTasks); |
| isMatch = false; |
| } else if (!whb.getTopologyId().equals(localAssignment.getTopologyId())) { |
| // topology id not equal |
| LOG.info("topology id not equal whb=" + whb.getTopologyId() + ",localAssignment=" + localAssignment.getTopologyId()); |
| isMatch = false; |
| }/* |
| * else if (!(whb.getTaskIds().equals(localAssignment.getTaskIds()))) { // task-id isn't equal LOG.info("task-id isn't equal whb=" + whb.getTaskIds() + |
| * ",localAssignment=" + localAssignment.getTaskIds()); isMatch = false; } |
| */ |
| |
| return isMatch; |
| } |
| |
| /** |
| * get all workers heartbeats of the supervisor |
| * |
| * @param conf |
| * @return Map<workerId, WorkerHeartbeat> |
| * @throws IOException |
| * @throws IOException |
| */ |
| public Map<String, WorkerHeartbeat> readWorkerHeartbeats(Map conf) throws Exception { |
| |
| Map<String, WorkerHeartbeat> workerHeartbeats = new HashMap<String, WorkerHeartbeat>(); |
| |
| // 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; |
| } |
| |
| /** |
| * get worker heartbeat by workerid |
| * |
| * @param conf |
| * @param workerId |
| * @returns WorkerHeartbeat |
| * @throws IOException |
| */ |
| public WorkerHeartbeat readWorkerHeartbeat(Map conf, String workerId) throws Exception { |
| |
| try { |
| LocalState ls = StormConfig.worker_state(conf, workerId); |
| |
| return (WorkerHeartbeat) ls.get(Common.LS_WORKER_HEARTBEAT); |
| } catch (Exception e) { |
| LOG.error("Failed to get worker Heartbeat", e); |
| return null; |
| } |
| |
| } |
| |
| /** |
| * launch a worker in local mode |
| * |
| * @param conf |
| * @param sharedcontext |
| * @param topologyId |
| * @param supervisorId |
| * @param port |
| * @param workerId |
| * @param workerThreadPidsAtom |
| * @throws Exception |
| */ |
| public void launchWorker(Map conf, IContext sharedcontext, String topologyId, String supervisorId, Integer port, String workerId, |
| ConcurrentHashMap<String, String> workerThreadPidsAtom) throws Exception { |
| |
| String pid = UUID.randomUUID().toString(); |
| |
| WorkerShutdown worker = Worker.mk_worker(conf, sharedcontext, topologyId, supervisorId, port, workerId, null); |
| |
| ProcessSimulator.registerProcess(pid, worker); |
| |
| workerThreadPidsAtom.put(workerId, pid); |
| |
| } |
| |
| // filter conflict jar |
| private Set<String> setFilterJars(Map totalConf) { |
| Set<String> filterJars = new HashSet<String>(); |
| |
| boolean enableClassloader = ConfigExtension.isEnableTopologyClassLoader(totalConf); |
| if (enableClassloader == false) { |
| // avoid logback vs log4j conflict |
| boolean enableLog4j = false; |
| String userDefLog4jConf = ConfigExtension.getUserDefinedLog4jConf(totalConf); |
| if (StringUtils.isBlank(userDefLog4jConf) == false) { |
| enableLog4j = true; |
| } |
| |
| if (enableLog4j == true) { |
| filterJars.add("log4j-over-slf4j"); |
| filterJars.add("logback-core"); |
| filterJars.add("logback-classic"); |
| |
| } else { |
| filterJars.add("slf4j-log4j"); |
| filterJars.add("log4j"); |
| } |
| } |
| |
| LOG.info("Remove jars " + filterJars); |
| return filterJars; |
| } |
| |
| public static boolean isKeyContain(Collection<String> collection, String jar) { |
| if (collection == null) { |
| return false; |
| } |
| File file = new File(jar); |
| String fileName = file.getName(); |
| for (String item : collection) { |
| |
| String regex = item + "[-._0-9]*.jar"; |
| Pattern p = Pattern.compile(regex); |
| |
| if (p.matcher(fileName).matches() == true) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| public AtomicReference<Set> getTopologyIdNeedDownload() { |
| return needDownloadTopologys; |
| } |
| |
| private String getClassPath(String stormjar, String stormHome, Map totalConf) { |
| |
| // String classpath = JStormUtils.current_classpath() + ":" + stormjar; |
| // return classpath; |
| |
| String classpath = JStormUtils.current_classpath(); |
| |
| String[] classpathes = classpath.split(":"); |
| |
| Set<String> classSet = new HashSet<String>(); |
| |
| for (String classJar : classpathes) { |
| if (StringUtils.isBlank(classJar) == true) { |
| continue; |
| } |
| classSet.add(classJar); |
| } |
| |
| if (stormHome != null) { |
| List<String> stormHomeFiles = PathUtils.read_dir_contents(stormHome); |
| |
| for (String file : stormHomeFiles) { |
| if (file.endsWith(".jar")) { |
| classSet.add(stormHome + File.separator + file); |
| } |
| } |
| |
| List<String> stormLibFiles = PathUtils.read_dir_contents(stormHome + File.separator + "lib"); |
| for (String file : stormLibFiles) { |
| if (file.endsWith(".jar")) { |
| classSet.add(stormHome + File.separator + "lib" + File.separator + file); |
| } |
| } |
| |
| } |
| |
| Set<String> filterJars = setFilterJars(totalConf); |
| |
| StringBuilder sb = new StringBuilder(); |
| for (String jar : classSet) { |
| if (isKeyContain(filterJars, jar)) { |
| LOG.info("Remove " + jar); |
| continue; |
| } |
| sb.append(jar + ":"); |
| } |
| |
| if (ConfigExtension.isEnableTopologyClassLoader(totalConf)) { |
| return sb.toString().substring(0, sb.length() - 1); |
| } else { |
| sb.append(stormjar); |
| return sb.toString(); |
| } |
| |
| } |
| |
| public String getChildOpts(Map stormConf) { |
| String childopts = " "; |
| |
| if (stormConf.get(Config.TOPOLOGY_WORKER_CHILDOPTS) != null) { |
| childopts += (String) stormConf.get(Config.TOPOLOGY_WORKER_CHILDOPTS); |
| } else if (ConfigExtension.getWorkerGc(stormConf) != null) { |
| childopts += ConfigExtension.getWorkerGc(stormConf); |
| } |
| |
| return childopts; |
| } |
| |
| public String getLogParameter(Map conf, String stormHome, String topologyName, int port) { |
| final String LOGBACK_CONF_TAG = "logback.configurationFile"; |
| final String LOGBACK_CONF_TAG_CMD = " -D" + LOGBACK_CONF_TAG + "="; |
| final String DEFAULT_LOG_CONF = "jstorm.logback.xml"; |
| |
| String logFileName = JStormUtils.genLogName(topologyName, port); |
| // String logFileName = topologyId + "-worker-" + port + ".log"; |
| |
| |
| StringBuilder commandSB = new StringBuilder(); |
| commandSB.append(" -Dlogfile.name="); |
| commandSB.append(logFileName); |
| commandSB.append(" -Dtopology.name=").append(topologyName); |
| |
| // commandSB.append(" -Dlog4j.ignoreTCL=true"); |
| |
| String userDefLogbackConf = ConfigExtension.getUserDefinedLogbackConf(conf); |
| String logConf = System.getProperty(LOGBACK_CONF_TAG); |
| |
| if (StringUtils.isBlank(userDefLogbackConf) == false) { |
| LOG.info("Use user fined back conf " + userDefLogbackConf); |
| commandSB.append(LOGBACK_CONF_TAG_CMD).append(userDefLogbackConf); |
| } else if (StringUtils.isBlank(logConf) == false) { |
| commandSB.append(LOGBACK_CONF_TAG_CMD).append(logConf); |
| } else if (StringUtils.isBlank(stormHome) == false) { |
| commandSB.append(LOGBACK_CONF_TAG_CMD).append(stormHome).append(File.separator).append("conf").append(File.separator).append(DEFAULT_LOG_CONF); |
| } else { |
| commandSB.append(LOGBACK_CONF_TAG_CMD + DEFAULT_LOG_CONF); |
| } |
| |
| final String LOG4J_CONF_TAG = "log4j.configuration"; |
| String userDefLog4jConf = ConfigExtension.getUserDefinedLog4jConf(conf); |
| if (StringUtils.isBlank(userDefLog4jConf) == false) { |
| LOG.info("Use user fined log4j conf " + userDefLog4jConf); |
| commandSB.append(" -D" + LOG4J_CONF_TAG + "=").append(userDefLog4jConf); |
| } |
| |
| return commandSB.toString(); |
| } |
| |
| private String getGcDumpParam(String topologyName, Map totalConf) { |
| // String gcPath = ConfigExtension.getWorkerGcPath(totalConf); |
| String gcPath = JStormUtils.getLogDir(); |
| |
| Date now = new Date(); |
| String nowStr = TimeFormat.getSecond(now); |
| |
| StringBuilder gc = new StringBuilder(256); |
| gc.append(" -Xloggc:"); |
| gc.append(gcPath).append(File.separator); |
| gc.append(topologyName).append(File.separator); |
| gc.append("%TOPOLOGYID%-worker-%ID%"); |
| gc.append("-gc.log -verbose:gc -XX:HeapDumpPath="); |
| gc.append(gcPath).append(File.separator).append(topologyName).append(File.separator).append("java-%TOPOLOGYID%-").append(nowStr).append(".hprof"); |
| gc.append(" "); |
| |
| |
| return gc.toString(); |
| } |
| |
| /** |
| * launch a worker in distributed mode |
| * |
| * @param conf |
| * @param sharedcontext |
| * @param topologyId |
| * @param supervisorId |
| * @param port |
| * @param workerId |
| * @throws IOException |
| * @pdOid 6ea369dd-5ce2-4212-864b-1f8b2ed94abb |
| */ |
| public void launchWorker(Map conf, IContext sharedcontext, String topologyId, String supervisorId, Integer port, String workerId, LocalAssignment assignment) |
| throws IOException { |
| |
| // STORM-LOCAL-DIR/supervisor/stormdist/topologyId |
| String stormroot = StormConfig.supervisor_stormdist_root(conf, topologyId); |
| |
| // STORM-LOCAL-DIR/supervisor/stormdist/topologyId/stormjar.jar |
| String stormjar = StormConfig.stormjar_path(stormroot); |
| |
| // get supervisor conf |
| Map stormConf = StormConfig.read_supervisor_topology_conf(conf, topologyId); |
| |
| Map totalConf = new HashMap(); |
| totalConf.putAll(conf); |
| totalConf.putAll(stormConf); |
| |
| // get classpath |
| // String[] param = new String[1]; |
| // param[0] = stormjar; |
| // String classpath = JStormUtils.add_to_classpath( |
| // JStormUtils.current_classpath(), param); |
| |
| // get child process parameter |
| |
| String stormhome = System.getProperty("jstorm.home"); |
| |
| long memSize = assignment.getMem(); |
| long memMinSize = ConfigExtension.getMemMinSizePerWorker(totalConf); |
| int cpuNum = assignment.getCpu(); |
| long memGsize = memSize / JStormUtils.SIZE_1_G; |
| int gcThreadsNum = memGsize > 4 ? (int) (memGsize * 1.5) : 4; |
| String childopts = getChildOpts(totalConf); |
| |
| childopts += getGcDumpParam(Common.getTopologyNameById(topologyId), totalConf); |
| |
| Map<String, String> environment = new HashMap<String, String>(); |
| |
| if (ConfigExtension.getWorkerRedirectOutput(totalConf)) { |
| environment.put("REDIRECT", "true"); |
| } else { |
| environment.put("REDIRECT", "false"); |
| } |
| |
| environment.put("LD_LIBRARY_PATH", (String) totalConf.get(Config.JAVA_LIBRARY_PATH)); |
| |
| StringBuilder commandSB = new StringBuilder(); |
| |
| try { |
| if (this.cgroupManager != null) { |
| commandSB.append(cgroupManager.startNewWorker(totalConf, cpuNum, workerId)); |
| } |
| } catch (Exception e) { |
| LOG.error("fail to prepare cgroup to workerId: " + workerId, e); |
| return; |
| } |
| |
| // commandSB.append("java -server -Xdebug -Xrunjdwp:transport=dt_socket,address=8000,server=y,suspend=n "); |
| commandSB.append("java -server "); |
| commandSB.append(" -Xms" + memMinSize); |
| commandSB.append(" -Xmx" + memSize + " "); |
| if (memMinSize < (memSize / 2)) |
| commandSB.append(" -Xmn" + memMinSize + " "); |
| else |
| commandSB.append(" -Xmn" + memSize / 2 + " "); |
| if (memGsize >= 2) { |
| commandSB.append(" -XX:PermSize=" + memSize / 32); |
| } else { |
| commandSB.append(" -XX:PermSize=" + memSize / 16); |
| } |
| commandSB.append(" -XX:MaxPermSize=" + memSize / 16); |
| commandSB.append(" -XX:ParallelGCThreads=" + gcThreadsNum); |
| commandSB.append(" " + childopts); |
| commandSB.append(" " + (assignment.getJvm() == null ? "" : assignment.getJvm())); |
| |
| commandSB.append(" -Djava.library.path="); |
| commandSB.append((String) totalConf.get(Config.JAVA_LIBRARY_PATH)); |
| |
| if (stormhome != null) { |
| commandSB.append(" -Djstorm.home="); |
| commandSB.append(stormhome); |
| } |
| |
| String logDir = System.getProperty("jstorm.log.dir"); |
| if (logDir != null) |
| commandSB.append(" -Djstorm.log.dir=").append(logDir); |
| commandSB.append(getLogParameter(totalConf, stormhome, assignment.getTopologyName(), port)); |
| |
| String classpath = getClassPath(stormjar, stormhome, totalConf); |
| String workerClassPath = (String) totalConf.get(Config.TOPOLOGY_CLASSPATH); |
| List<String> otherLibs = (List<String>) stormConf.get(GenericOptionsParser.TOPOLOGY_LIB_NAME); |
| StringBuilder sb = new StringBuilder(); |
| if (otherLibs != null) { |
| for (String libName : otherLibs) { |
| sb.append(StormConfig.stormlib_path(stormroot, libName)).append(":"); |
| } |
| } |
| workerClassPath = workerClassPath + ":" + sb.toString(); |
| |
| Map<String, String> policyReplaceMap = new HashMap<String, String>(); |
| String realClassPath = classpath + ":" + workerClassPath; |
| policyReplaceMap.put(SandBoxMaker.CLASS_PATH_KEY, realClassPath); |
| commandSB.append(sandBoxMaker.sandboxPolicy(workerId, policyReplaceMap)); |
| |
| commandSB.append(" -cp "); |
| // commandSB.append(workerClassPath + ":"); |
| commandSB.append(classpath); |
| if (!ConfigExtension.isEnableTopologyClassLoader(totalConf)) |
| commandSB.append(":").append(workerClassPath); |
| |
| commandSB.append(" com.alibaba.jstorm.daemon.worker.Worker "); |
| commandSB.append(topologyId); |
| |
| commandSB.append(" "); |
| commandSB.append(supervisorId); |
| |
| commandSB.append(" "); |
| commandSB.append(port); |
| |
| commandSB.append(" "); |
| commandSB.append(workerId); |
| |
| commandSB.append(" "); |
| commandSB.append(workerClassPath + ":" + stormjar); |
| |
| String cmd = commandSB.toString(); |
| cmd = cmd.replace("%ID%", port.toString()); |
| cmd = cmd.replace("%TOPOLOGYID%", topologyId); |
| if (stormhome != null) { |
| cmd = cmd.replace("%JSTORM_HOME%", stormhome); |
| } else { |
| cmd = cmd.replace("%JSTORM_HOME%", "./"); |
| } |
| |
| LOG.info("Launching worker with command: " + cmd); |
| LOG.info("Environment:" + environment.toString()); |
| |
| JStormUtils.launch_process(cmd, environment, true); |
| } |
| |
| private Set<Integer> killUselessWorkers(Map<String, StateHeartbeat> localWorkerStats, Map<Integer, LocalAssignment> localAssignments, |
| Map<String, Integer> taskCleanupTimeoutMap) { |
| Map<String, String> removed = new HashMap<String, String>(); |
| Set<Integer> keepPorts = new HashSet<Integer>(); |
| |
| for (Entry<String, StateHeartbeat> entry : localWorkerStats.entrySet()) { |
| |
| String workerid = entry.getKey(); |
| StateHeartbeat hbstate = entry.getValue(); |
| if (workerIdToStartTimeAndPort.containsKey(workerid) && hbstate.getState().equals(State.notStarted)) |
| continue; |
| |
| if (hbstate.getState().equals(State.valid)) { |
| // hbstate.getHeartbeat() won't be null |
| keepPorts.add(hbstate.getHeartbeat().getPort()); |
| } else { |
| if (hbstate.getHeartbeat() != null) { |
| removed.put(workerid, hbstate.getHeartbeat().getTopologyId()); |
| } else { |
| removed.put(workerid, null); |
| } |
| |
| if (killingWorkers.containsKey(workerid) == false) { |
| |
| StringBuilder sb = new StringBuilder(); |
| sb.append("Shutting down and clearing state for id "); |
| sb.append(workerid); |
| sb.append(";State:"); |
| sb.append(hbstate); |
| |
| LOG.info(sb.toString()); |
| } |
| } |
| } |
| |
| shutWorker(conf, supervisorId, removed, workerThreadPids, cgroupManager, false, killingWorkers, taskCleanupTimeoutMap); |
| Set<String> activeTopologys = new HashSet<String>(); |
| if (killingWorkers.size() == 0) { |
| // When all workers under killing are killed successfully, |
| // clean the task cleanup timeout map correspondingly. |
| for (Entry<Integer, LocalAssignment> entry : localAssignments.entrySet()) { |
| activeTopologys.add(entry.getValue().getTopologyId()); |
| } |
| |
| Set<String> obsoleteTopologys = new HashSet<String>(); |
| for (String topologyId : taskCleanupTimeoutMap.keySet()) { |
| if (activeTopologys.contains(topologyId) == false) { |
| obsoleteTopologys.add(topologyId); |
| } |
| } |
| for (String topologyId : obsoleteTopologys) { |
| taskCleanupTimeoutMap.remove(topologyId); |
| } |
| } |
| |
| for (String removedWorkerId : removed.keySet()) { |
| localWorkerStats.remove(removedWorkerId); |
| } |
| // Keep the workers which are still under starting |
| for (Entry<String, Pair<Integer, Integer>> entry : workerIdToStartTimeAndPort.entrySet()) { |
| String workerId = entry.getKey(); |
| StateHeartbeat hbstate = localWorkerStats.get(workerId); |
| if (hbstate != null) |
| if (hbstate.getState().equals(State.notStarted)) { |
| keepPorts.add(entry.getValue().getSecond()); |
| } |
| } |
| return keepPorts; |
| } |
| |
| private void startNewWorkers(Set<Integer> keepPorts, Map<Integer, LocalAssignment> localAssignments, Set<String> downloadFailedTopologyIds) |
| throws Exception { |
| /** |
| * Step 4: get reassigned tasks, which is in assignedTasks, but not in keeperPorts Map<port(type Integer), LocalAssignment> |
| */ |
| Map<Integer, LocalAssignment> newWorkers = JStormUtils.select_keys_pred(keepPorts, localAssignments); |
| |
| /** |
| * Step 5: generate new work ids |
| */ |
| Map<Integer, String> newWorkerIds = new HashMap<Integer, String>(); |
| |
| for (Entry<Integer, LocalAssignment> entry : newWorkers.entrySet()) { |
| Integer port = entry.getKey(); |
| LocalAssignment assignment = entry.getValue(); |
| if (assignment != null && assignment.getTopologyId() != null && downloadFailedTopologyIds.contains(assignment.getTopologyId())) { |
| LOG.info("Can't start this worker: " + port + " about the topology: " + assignment.getTopologyId() + ", due to the damaged binary !!"); |
| continue; |
| } |
| String workerId = UUID.randomUUID().toString(); |
| |
| newWorkerIds.put(port, workerId); |
| |
| // create new worker Id directory |
| // LOCALDIR/workers/newworkid/pids |
| try { |
| StormConfig.worker_pids_root(conf, workerId); |
| } catch (IOException e1) { |
| LOG.error("Failed to create " + workerId + " localdir", e1); |
| throw e1; |
| } |
| |
| StringBuilder sb = new StringBuilder(); |
| sb.append("Launching worker with assiangment "); |
| sb.append(assignment.toString()); |
| sb.append(" for the supervisor "); |
| sb.append(supervisorId); |
| sb.append(" on port "); |
| sb.append(port); |
| sb.append(" with id "); |
| sb.append(workerId); |
| LOG.info(sb.toString()); |
| |
| try { |
| String clusterMode = StormConfig.cluster_mode(conf); |
| |
| if (clusterMode.equals("distributed")) { |
| launchWorker(conf, sharedContext, assignment.getTopologyId(), supervisorId, port, workerId, assignment); |
| } else if (clusterMode.equals("local")) { |
| launchWorker(conf, sharedContext, assignment.getTopologyId(), supervisorId, port, workerId, workerThreadPids); |
| } |
| } catch (Exception e) { |
| workerReportError.report(assignment.getTopologyId(), port, |
| assignment.getTaskIds(), new String(JStormUtils.getErrorInfo(e))); |
| String errorMsg = "Failed to launchWorker workerId:" + workerId + ":" + port; |
| LOG.error(errorMsg, e); |
| throw e; |
| } |
| |
| } |
| |
| /** |
| * FIXME, workerIds should be Set, not Collection, but here simplify the logic |
| */ |
| markAllNewWorkers(newWorkerIds); |
| // try { |
| // waitForWorkersLaunch(conf, workerIds); |
| // } catch (IOException e) { |
| // LOG.error(e + " waitForWorkersLaunch failed"); |
| // } catch (InterruptedException e) { |
| // LOG.error(e + " waitForWorkersLaunch failed"); |
| // } |
| } |
| |
| boolean isWorkerDead(String workerId) { |
| if (ConfigExtension.isCheckWorkerAliveBySystemInfo(conf) == false) { |
| return false; |
| } |
| |
| try { |
| List<String> pids = getPid(conf, workerId); |
| if (pids == null || pids.size() == 0) { |
| // local mode doesn't exist pid |
| return false; |
| } |
| // if all pid in pids are dead, then the worker is dead |
| for (String pid : pids) { |
| boolean isDead = JStormUtils.isProcDead(pid); |
| if (isDead == true) { |
| LOG.info("Found " + workerId + " is dead "); |
| } else { |
| return false; |
| } |
| } |
| |
| return true; |
| } catch (IOException e) { |
| LOG.info("Failed to check whether worker is dead through /proc/pid", e); |
| return false; |
| } |
| |
| } |
| |
| } |
