| /* |
| * 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 org.apache.uima.ducc.agent.launcher; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.PrintWriter; |
| import java.io.StringWriter; |
| import java.lang.reflect.Method; |
| import java.net.URL; |
| import java.net.URLClassLoader; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| import org.apache.uima.ducc.agent.NodeAgent; |
| import org.apache.uima.ducc.agent.launcher.ManagedProcess.StopPriority; |
| import org.apache.uima.ducc.common.IDuccUser; |
| import org.apache.uima.ducc.common.container.FlagsHelper; |
| import org.apache.uima.ducc.common.exception.DuccRuntimeException; |
| import org.apache.uima.ducc.common.utils.DuccLogger; |
| import org.apache.uima.ducc.common.utils.TimeStamp; |
| import org.apache.uima.ducc.common.utils.Utils; |
| import org.apache.uima.ducc.common.utils.id.DuccId; |
| import org.apache.uima.ducc.transport.agent.ProcessStateUpdate; |
| import org.apache.uima.ducc.transport.cmdline.ACommandLine; |
| import org.apache.uima.ducc.transport.cmdline.ICommandLine; |
| import org.apache.uima.ducc.transport.cmdline.JavaCommandLine; |
| import org.apache.uima.ducc.transport.cmdline.NonJavaCommandLine; |
| import org.apache.uima.ducc.transport.event.common.IDuccProcess; |
| import org.apache.uima.ducc.transport.event.common.IDuccProcess.ReasonForStoppingProcess; |
| import org.apache.uima.ducc.transport.event.common.IDuccProcessType.ProcessType; |
| import org.apache.uima.ducc.transport.event.common.IProcessState.ProcessState; |
| import org.apache.uima.ducc.transport.event.common.ITimeWindow; |
| import org.apache.uima.ducc.transport.event.common.TimeWindow; |
| import org.apache.uima.ducc.user.common.PrivateClassLoader; |
| |
| public class DuccCommandExecutor extends CommandExecutor { |
| DuccLogger logger = DuccLogger.getLogger(this.getClass(), NodeAgent.COMPONENT_NAME); |
| |
| @SuppressWarnings("unused") |
| private static AtomicInteger nextPort = new AtomicInteger(30000); |
| |
| public DuccCommandExecutor(NodeAgent agent, ICommandLine cmdLine, String host, String ip, |
| Process managedProcess) throws Exception { |
| super(agent, cmdLine, host, ip, managedProcess); |
| } |
| |
| public DuccCommandExecutor(ICommandLine cmdLine, String host, String ip, Process managedProcess) |
| throws Exception { |
| super(null, cmdLine, host, ip, managedProcess); |
| } |
| |
| private boolean useDuccSpawn() { |
| if (super.managedProcess.isAgentProcess() || Utils.isWindows()) { |
| return false; |
| } |
| // On non-windows check if we should spawn the process via ducc_ling |
| String useSpawn = System.getProperty("ducc.agent.launcher.use.ducc_spawn"); |
| if (useSpawn != null && useSpawn.toLowerCase().equals("true")) { |
| return true; |
| } |
| // default |
| return false; |
| } |
| |
| private boolean createCGroupContainer(IDuccProcess duccProcess, String containerId, String owner) |
| throws Exception { |
| // create cgroups container and assign limits |
| if (agent.cgroupsManager.createContainer(containerId, System.getProperty("user.name"), |
| agent.cgroupsManager.getUserGroupName(System.getProperty("user.name")), |
| useDuccSpawn())) { |
| logger.info("createCGroupContainer", null, |
| "Calculating CPU shares \nProcess Max Memory=" |
| + duccProcess.getCGroup().getMaxMemoryLimit() + "\nNode Memory Total=" |
| + agent.getNodeInfo().getNodeMetrics().getNodeMemory().getMemTotal()); |
| long cpuShares = duccProcess.getCGroup().getMaxMemoryLimit() |
| / agent.getNodeInfo().getNodeMetrics().getNodeMemory().getMemTotal(); |
| logger.info("createCGroupContainer", null, "\nCalculated Shares=" + cpuShares); |
| agent.cgroupsManager.setContainerCpuShares(containerId, owner, useDuccSpawn(), cpuShares); |
| long swappiness = 10; // default |
| if (agent.configurationFactory.nodeSwappiness != null) { |
| swappiness = Long.valueOf(agent.configurationFactory.nodeSwappiness); |
| } |
| agent.cgroupsManager.setContainerSwappiness(containerId, owner, useDuccSpawn(), swappiness); |
| |
| return agent.cgroupsManager.setContainerMaxMemoryLimit(containerId, owner, useDuccSpawn(), |
| duccProcess.getCGroup().getMaxMemoryLimit()); |
| } |
| return false; |
| } |
| |
| private String getContainerId() { |
| String containerId; |
| if (((ManagedProcess) super.managedProcess).getDuccProcess().getProcessType() |
| .equals(ProcessType.Service)) { |
| containerId = String |
| .valueOf(((ManagedProcess) managedProcess).getDuccProcess().getCGroup().getId()); |
| } else { |
| containerId = ((ManagedProcess) managedProcess).getWorkDuccId().getFriendly() + "." |
| + ((ManagedProcess) managedProcess).getDuccProcess().getCGroup().getId(); |
| } |
| return containerId; |
| } |
| |
| public Process exec(ICommandLine cmdLine, Map<String, String> processEnv) throws Exception { |
| String methodName = "exec"; |
| try { |
| String[] cmd = getDeployableCommandLine(cmdLine, processEnv); |
| if (isKillCommand(cmdLine)) { |
| logger.debug(methodName, null, "Killing process"); |
| stopProcess(cmdLine, cmd); |
| } else { |
| IDuccProcess duccProcess = ((ManagedProcess) managedProcess).getDuccProcess(); |
| // If running a real agent on a node, collect swap info and |
| // assign max swap usage threshold |
| // for each process. In virtual mode, where there are multiple |
| // agents per node, we dont |
| // set nor enforce swap limits. |
| if (!agent.isVirtual()) { |
| // Calculate how much swap space the process is allowed to |
| // use. The calculation is based on |
| // the percentage of real memory the process is assigned. |
| // The process is entitled the |
| // same percentage of the swap. |
| // Normalize node's total memory as it is expressed in KB. |
| // The calculation below is based on bytes. |
| double percentOfTotal = ((double) duccProcess.getCGroup().getMaxMemoryLimit()) |
| / (agent.getNodeInfo().getNodeMetrics().getNodeMemory().getMemTotal() * 1024); // need |
| // bytes |
| |
| // substract 1Gig from total swap on this node to |
| // accommodate OS needs for swapping. The |
| // getSwapTotal() returns swap space in KBs so normalize |
| // 1Gig |
| long adjustedTotalSwapAvailable = agent.getNodeInfo().getNodeMetrics().getNodeMemory() |
| .getSwapTotal() - 1048576; |
| // calculate the portion (in bytes) of swap this process is |
| // entitled to |
| long maxProcessSwapUsage = (long) (adjustedTotalSwapAvailable * percentOfTotal) * 1024; |
| // assigned how much swap this process is entitled to. If it |
| // exceeds this number the Agent |
| // will kill the process. |
| ((ManagedProcess) managedProcess).setMaxSwapThreshold(maxProcessSwapUsage); |
| logger.info(methodName, null, "---Process DuccId:" + duccProcess.getDuccId() |
| + " CGroup.getMaxMemoryLimit():" |
| + ((duccProcess.getCGroup().getMaxMemoryLimit() / 1024) / 1024) + " MBs" |
| + " Node Memory Total:" |
| + (agent.getNodeInfo().getNodeMetrics().getNodeMemory().getMemTotal() / 1024) |
| + " MBs" + " Percentage Of Real Memory:" + percentOfTotal |
| + " Adjusted Total Swap Available On Node:" + adjustedTotalSwapAvailable / 1024 |
| + " MBs" + " Process Entitled To Max:" + (maxProcessSwapUsage / 1024) / 1024 |
| + " MBs of Swap"); |
| |
| // logger.info(methodName, null, |
| // "The Process With ID:"+duccProcess.getDuccId()+" is Entitled to the Max "+( |
| // (maxProcessSwapUsage/1024)/1024)+" Megs of Swap Space"); |
| // if configured to use cgroups and the process is the |
| // cgroup owner, create a cgroup |
| // using Process DuccId as a name. Additional processes may |
| // be injected into the |
| // cgroup by declaring cgroup owner id. |
| if (agent.useCgroups) { |
| // JDs are of type Pop (Plain Old Process). JDs run in a |
| // reservation. The cgroup container |
| // is created for the reservation and we co-locate as |
| // many JDs as we can fit in it. |
| // String containerId = ((ManagedProcess) |
| // managedProcess).getWorkDuccId()+"."+duccProcess.getCGroup().getId().getFriendly(); |
| String containerId = getContainerId(); |
| logger.info(methodName, null, "Checking for CGroup Existance with ID:" + containerId); |
| if (!agent.cgroupsManager.cgroupExists( |
| agent.cgroupsManager.getDuccCGroupBaseDir() + "/" + containerId)) { |
| logger.info(methodName, null, "No CGroup with ID:" + containerId + " Found"); |
| boolean failed = false; |
| // create cgroup container for JDs |
| try { |
| if (createCGroupContainer(duccProcess, containerId, |
| ((ManagedProcess) super.managedProcess).getOwner())) { |
| logger.info( |
| methodName, null, "Created CGroup with ID:" + containerId |
| + " With Memory Limit=" + ((ManagedProcess) super.managedProcess) |
| .getDuccProcess().getCGroup().getMaxMemoryLimit() |
| + " Bytes"); |
| } else { |
| logger.info(methodName, null, "Failed To Create CGroup with ID:" + containerId); |
| duccProcess.setProcessState(ProcessState.Failed); |
| duccProcess.setReasonForStoppingProcess("CGroupCreationFailed"); |
| failed = true; |
| // agent.stop(); |
| } |
| } catch (Exception e) { |
| logger.error(methodName, null, e); |
| failed = true; |
| duccProcess.setProcessState(ProcessState.Failed); |
| duccProcess.setReasonForStoppingProcess("CGroupCreationFailed"); |
| // agent.stop(); |
| } |
| if (failed) { |
| logger.error(methodName, null, |
| new RuntimeException( |
| "The Agent is Unable To Create A CGroup with Container ID: " |
| + containerId + ". Rejecting Deployment of Process with ID:" |
| + duccProcess.getDuccId())); |
| return managedProcess; |
| } |
| } else { |
| logger.info(methodName, null, "CGroup Exists with ID:" + containerId); |
| |
| } |
| |
| String[] cgroupCmd = new String[cmd.length + 3]; |
| cgroupCmd[0] = agent.cgroupsManager.getCGroupsUtilsDir() + "/cgexec"; |
| cgroupCmd[1] = "-g"; |
| cgroupCmd[2] = agent.cgroupsManager.getSubsystems() + containerId; // UIMA-5405 |
| // subsystems |
| // includes the |
| // "ducc" id |
| int inx = 3; |
| for (String cmdPart : cmd) { |
| cgroupCmd[inx++] = cmdPart; |
| } |
| startProcess(cmdLine, cgroupCmd, processEnv); |
| } else { |
| // Not configured to use CGroups |
| startProcess(cmdLine, cmd, processEnv); |
| } |
| } else { |
| // dont use CGroups on virtual agents |
| startProcess(cmdLine, cmd, processEnv); |
| } |
| |
| } |
| return managedProcess; |
| } catch (Throwable e) { |
| if (((ManagedProcess) super.managedProcess).getDuccProcess() != null) { |
| DuccId duccId = ((ManagedProcess) super.managedProcess).getDuccId(); |
| logger.error(methodName, duccId, |
| ((ManagedProcess) super.managedProcess).getDuccProcess().getDuccId(), e, |
| new Object[] {}); |
| } |
| logger.error(methodName, null, e); |
| throw e; |
| } |
| } |
| |
| private boolean processInRunningOrInitializingState() { |
| return (((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.Running) |
| || ((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.Initializing) |
| || ((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.Starting) |
| || ((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.Stopping) |
| || ((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.Started)); |
| } |
| |
| private void stopProcess(ICommandLine cmdLine, String[] cmd) throws Exception { |
| String methodName = "stopProcess"; |
| |
| if (processInRunningOrInitializingState()) { |
| Future<?> future = ((ManagedProcess) managedProcess).getFuture(); |
| if (future == null) { |
| throw new Exception("Future Object not Found. Unable to Stop Process with PID:" |
| + ((ManagedProcess) managedProcess).getPid()); |
| } |
| // for stop to work, PID must be provided |
| if (((ManagedProcess) managedProcess).getDuccProcess().getPID() == null |
| || ((ManagedProcess) managedProcess).getDuccProcess().getPID().trim().length() == 0) { |
| throw new Exception("Process Stop Command Failed. PID not provided."); |
| } |
| try { |
| // NEW Code |
| logger.debug(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), |
| ">>>>>>>>>>>>>>> Stopping Process:" + ((ManagedProcess) managedProcess).getPid()); |
| ICommandLine cmdL; |
| if (Utils.isWindows()) { |
| cmdL = new NonJavaCommandLine("taskkill"); |
| cmdL.addArgument("/PID"); |
| } else { |
| cmdL = new NonJavaCommandLine("/bin/kill"); |
| cmdL.addArgument("-15"); |
| } |
| cmdL.addArgument(((ManagedProcess) managedProcess).getDuccProcess().getPID()); |
| |
| String[] sigTermCmdLine = getDeployableCommandLine(cmdL, new HashMap<String, String>()); |
| doExec(new ProcessBuilder(sigTermCmdLine), sigTermCmdLine, true); |
| // if agent receives admin STOP request, all managed processes should |
| // be stopped without each waiting for 60 secs. The agent |
| // blasts SIGTERM in parallel to all running child processes |
| if (!StopPriority.DONT_WAIT.equals(((ManagedProcess) managedProcess).getStopPriority())) { |
| long maxTimeToWaitForProcessToStop = 60000; // default 1 minute |
| if (super.agent.configurationFactory.processStopTimeout != null) { |
| maxTimeToWaitForProcessToStop = Long |
| .valueOf(super.agent.configurationFactory.processStopTimeout); |
| } |
| |
| try { |
| logger.info(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), |
| "------------ Agent Starting Killer Timer Task For Process with PID:" |
| + ((ManagedProcess) managedProcess).getDuccProcess().getPID() |
| + " Process State: " |
| + ((ManagedProcess) managedProcess).getDuccProcess().getProcessState()); |
| future.get(maxTimeToWaitForProcessToStop, TimeUnit.MILLISECONDS); |
| |
| } catch (TimeoutException te) { |
| if (!((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.Stopped)) { |
| logger.info(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), |
| "------------ Agent Timed-out Waiting for Process with PID:" |
| + ((ManagedProcess) managedProcess).getDuccProcess().getPID() |
| + " to Stop. Process State:" |
| + ((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| + " .Process did not stop in allotted time of " |
| + maxTimeToWaitForProcessToStop + " millis"); |
| logger.info(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), |
| ">>>>>>>>>>>>>>> Killing Process:" |
| + ((ManagedProcess) managedProcess).getDuccProcess().getPID() |
| + " .Process State:" + ((ManagedProcess) managedProcess) |
| .getDuccProcess().getProcessState()); |
| doExec(new ProcessBuilder(cmd), cmd, true); |
| |
| } |
| |
| } |
| |
| } |
| } catch (Exception e) { // InterruptedException, ExecutionException |
| logger.error(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), e, |
| new Object[] {}); |
| } |
| |
| } |
| } |
| |
| private void startProcess(ICommandLine cmdLine, String[] cmd, Map<String, String> processEnv) |
| throws Exception { |
| String methodName = "startProcess"; |
| |
| ProcessBuilder pb = new ProcessBuilder(cmd); |
| |
| Map<String, String> env = pb.environment(); |
| // Dont enherit agent's environment |
| env.clear(); |
| // enrich Process environment |
| env.putAll(processEnv); |
| if (cmdLine instanceof ACommandLine) { |
| // enrich Process environment with one from a given command line |
| env.putAll(((ACommandLine) cmdLine).getEnvironment()); |
| } |
| if (logger.isTrace()) { |
| for (Entry<String, String> entry : env.entrySet()) { |
| String message = "key:" + entry.getKey() + " " + "value:" + entry.getValue(); |
| logger.trace(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), message); |
| |
| } |
| } |
| try { |
| doExec(pb, cmd, isKillCommand(cmdLine)); |
| } catch (Exception e) { |
| throw e; |
| } finally { |
| // millis = TimeStamp.getCurrentMillis(); |
| // twr.setEnd(millis); |
| } |
| } |
| |
| /** |
| * Checks if a given process is AP. The code checks if process type is POP and it is *not* JD |
| * |
| * @param process |
| * - process instance |
| * @return - true if AP, false otherwise |
| */ |
| private boolean isAP(ManagedProcess process) { |
| if (!process.isJd() && process.getDuccProcess().getProcessType().equals(ProcessType.Pop)) { |
| return true; |
| } else { |
| return false; |
| } |
| } |
| |
| public void doExec(ProcessBuilder pb, String[] cmd, boolean isKillCmd) throws Exception { |
| String methodName = "doExec"; |
| int exitCode = 0; |
| boolean failed = false; |
| try { |
| StringBuilder sb = new StringBuilder( |
| (isKillCommand(cmdLine) ? "--->Killing Process " : "---> Launching Process:") |
| + " Using command line:"); |
| int inx = 0; |
| for (String cmdPart : cmd) { |
| sb.append("\n\t[").append(inx++).append("]") |
| .append(Utils.resolvePlaceholderIfExists(cmdPart, System.getProperties())); |
| // Not sure why place-holders are replaced just for this msg? |
| } |
| logger.info(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), sb.toString()); |
| |
| java.lang.Process process = pb.start(); |
| // Drain process streams |
| postExecStep(process, logger, isKillCmd); |
| // block waiting for the process to terminate. |
| exitCode = process.waitFor(); |
| if (!isKillCommand(cmdLine)) { |
| logger.info(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), |
| ">>>>>>>>>>>>> Process with PID:" |
| + ((ManagedProcess) super.managedProcess).getDuccProcess().getPID() |
| + " Terminated. Exit Code:" + exitCode); |
| // Process is dead, determine if the cgroup container should be |
| // destroyed as well. |
| if (agent.useCgroups) { |
| String containerId = getContainerId(); |
| String userId = ((ManagedProcess) super.managedProcess).getOwner(); |
| // before destroying the container the code checks if there |
| // are processes still running in it. This could be true if |
| // user code launched child processes. If there are child |
| // processes still running, the code kills each one at a |
| // time and at the end the container is removed. |
| agent.cgroupsManager.destroyContainer(containerId, userId, NodeAgent.SIGKILL); |
| logger.info(methodName, null, "Removed CGroup Container with ID:" + containerId); |
| } |
| } |
| |
| } catch (NullPointerException ex) { |
| ((ManagedProcess) super.managedProcess).getDuccProcess().setProcessState(ProcessState.Failed); |
| StringBuffer sb = new StringBuffer(); |
| sb.setLength(0); |
| sb.append( |
| "\n\tJava ProcessBuilder Failed to Launch Process due to NullPointerException. An Entry in the Command Array Must be Null. Look at Command Array Below:\n"); |
| for (String cmdPart : cmd) { |
| if (cmdPart != null) { |
| sb.append("\n\t").append(cmdPart); |
| } |
| } |
| logger.info(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), sb.toString()); |
| ((ManagedProcess) super.managedProcess).getDuccProcess().setProcessState(ProcessState.Failed); |
| throw ex; |
| } catch (Exception ex) { |
| ((ManagedProcess) super.managedProcess).getDuccProcess() |
| .setProcessState(ProcessState.LaunchFailed); |
| |
| ((ManagedProcess) super.managedProcess).getDuccProcess().setProcessExitCode(-1); // overwrite |
| // process |
| // exit code |
| // if stderr |
| // has a msg |
| |
| StringWriter stackTraceBuffer = new StringWriter(); |
| ex.printStackTrace(new PrintWriter(stackTraceBuffer)); |
| |
| ((ManagedProcess) managedProcess).getDuccProcess() |
| .setReasonForStoppingProcess(stackTraceBuffer.toString()); |
| failed = true; |
| logger.info(methodName, ((ManagedProcess) super.managedProcess).getDuccId(), |
| "Failed to launch Process - Reason:" + stackTraceBuffer.toString()); |
| |
| throw ex; |
| } finally { |
| if (isKillCmd) { |
| // the kill command process has been launched. Nothing else to do. |
| // We now wait for the process to die. |
| return; |
| } |
| if (!failed) { |
| // associate exit code |
| ((ManagedProcess) managedProcess).getDuccProcess().setProcessExitCode(exitCode); |
| |
| } |
| |
| // Per team discussion on Aug 31 2011, the process is stopped by an |
| // agent when initialization |
| // times out or initialization failed. Both Initialization_Timeout |
| // and FailedIntialization imply |
| // that the process is stopped. If the process is AP and it exited |
| // it should be marked |
| // as Stopped. If the exit was due to Ducc kill mark reason as |
| // KilledByDucc otherwise we have |
| // no way of knowing why the process exited and in such case reason |
| // is Other. |
| if ((isAP((ManagedProcess) super.managedProcess))) { |
| |
| // if failed to execute the command line, the process state is already set to Failed |
| if (!((ManagedProcess) super.managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.LaunchFailed)) { |
| |
| ((ManagedProcess) managedProcess).getDuccProcess().setProcessState(ProcessState.Stopped); |
| } |
| |
| if (((ManagedProcess) super.managedProcess).doKill()) { // killed |
| // by |
| // agent/ducc |
| ((ManagedProcess) managedProcess).getDuccProcess() |
| .setReasonForStoppingProcess(ReasonForStoppingProcess.KilledByDucc.toString()); |
| } else if (!failed) { |
| |
| ((ManagedProcess) managedProcess).getDuccProcess() |
| .setReasonForStoppingProcess(ReasonForStoppingProcess.Other.toString()); |
| } |
| |
| } else if (!isKillCommand(cmdLine) |
| && !((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.InitializationTimeout) |
| && !((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.FailedInitialization) |
| && !((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.Failed) |
| && !((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.LaunchFailed) |
| && !((ManagedProcess) managedProcess).getDuccProcess().getProcessState() |
| .equals(ProcessState.Killed)) { |
| ((ManagedProcess) managedProcess).getDuccProcess().setProcessState(ProcessState.Stopped); |
| } |
| } |
| |
| } |
| |
| private String[] getDeployableCommandLine(ICommandLine cmdLine, Map<String, String> processEnv) |
| throws Exception { |
| // String methodName = "getDeployableCommandLine"; |
| String[] cmd = new String[0]; |
| boolean uimaBasedAP = false; |
| try { |
| // lock using Agent single permit semaphore. The |
| // Utils.concatAllArrays() |
| // uses native call (for efficiency) which appears not thread safe. |
| NodeAgent.lock(); |
| // Use ducc_ling (c code) as a launcher for the actual process. The |
| // ducc_ling |
| // allows the process to run as a specified user in order to write |
| // out logs in |
| // user's space as oppose to ducc space. |
| String c_launcher_path = Utils.resolvePlaceholderIfExists( |
| System.getProperty("ducc.agent.launcher.ducc_spawn_path"), System.getProperties()); |
| |
| // if the command line is kill, don't provide any logging info to |
| // the ducc_ling. Otherwise, |
| // ducc_ling creates and empty log for each time we are killing a |
| // process |
| if (isKillCommand(cmdLine)) { |
| // Duccling, with no logging, always run by ducc, no need for |
| // workingdir |
| String[] duccling_nolog = new String[] { c_launcher_path, "-u", |
| ((ManagedProcess) super.managedProcess).getOwner(), "--" }; |
| if (useDuccSpawn()) { |
| cmd = Utils.concatAllArrays(duccling_nolog, new String[] { cmdLine.getExecutable() }, |
| cmdLine.getCommandLine()); |
| } else { |
| cmd = Utils.concatAllArrays(new String[] { cmdLine.getExecutable() }, |
| cmdLine.getCommandLine()); |
| } |
| |
| } else { |
| String processType = "-UIMA-"; |
| // If Java then may run many JPs from the same cmdLine so make a local copy that we can |
| // modify |
| if (cmdLine instanceof JavaCommandLine) { |
| cmdLine = ((JavaCommandLine) cmdLine).copy(); |
| } |
| |
| logger.info("getDeployableCommandLine", ((ManagedProcess) super.managedProcess).getDuccId(), |
| "Deploying Process Type:" + ((ManagedProcess) super.managedProcess).getDuccProcess() |
| .getProcessType().name()); |
| |
| switch (((ManagedProcess) super.managedProcess).getDuccProcess().getProcessType()) { |
| case Pop: |
| // Both JD and POP arbitrary process are POPs. Assume this |
| // is an arbitrary process |
| processType = "-POP-"; |
| if (cmdLine instanceof JavaCommandLine) { |
| for (String option : cmdLine.getOptions()) { |
| |
| logger.info("getDeployableCommandLine", |
| ((ManagedProcess) super.managedProcess).getDuccId(), |
| "POP CmdLine Option:" + option); |
| |
| // Both services and JD have processType=POP. |
| // However, only the JD |
| // will have -Dducc.deploy.components option set. |
| if (option.startsWith("-Dducc.deploy.components=")) { |
| processType = "-JD-"; |
| ((ManagedProcess) super.managedProcess).setIsJD(); // mark this process as JD |
| // break; |
| } else if (option.startsWith("-Dducc.deploy.JpType=uima")) { |
| // this is an AP with JP-based "nature". Meaning its using |
| // a JP process configuration and its pull based model. |
| uimaBasedAP = true; |
| } |
| |
| } |
| } |
| break; |
| case Service: |
| // processType = "-AP-"; |
| break; |
| case Job_Uima_AS_Process: |
| processType = "-UIMA-"; |
| boolean isDucc20JpProcess = false; |
| boolean isDucc20ServiceProcess = false; |
| // determine if we are launching Ducc2.0 or Ducc1.+ JP |
| List<String> options = cmdLine.getOptions(); |
| for (String option : options) { |
| if (option.indexOf(FlagsHelper.Name.JpType.pname()) > -1) { |
| isDucc20JpProcess = true; |
| } |
| if (option.indexOf("ducc.deploy.components=service") > -1) { |
| isDucc20ServiceProcess = true; |
| } |
| // break; |
| if (option.trim().equals("-Dducc.deploy.JpType=uima")) { |
| // this is an AP with JP-based "nature". Meaning its using |
| // a JP process configuration and its pull based model. |
| uimaBasedAP = true; |
| } else if (option.trim().equals("-Dducc.deploy.JpType=uima-as")) { |
| ((ManagedProcess) super.managedProcess).setUimaAs(); |
| } |
| } |
| |
| // Add main class and component type to the command line |
| if (isDucc20JpProcess) { |
| if (!isDucc20ServiceProcess) { |
| cmdLine.addOption("-Dducc.deploy.components=job-process"); |
| } |
| // add port where an agent listens for process state update events |
| processEnv.put(IDuccUser.EnvironmentVariable.DUCC_UPDATE_PORT.value(), |
| System.getProperty("AGENT_AP_STATE_UPDATE_PORT")); |
| ((JavaCommandLine) cmdLine) |
| .setClassName("org.apache.uima.ducc.user.common.main.DuccJobService"); |
| // ((JavaCommandLine)cmdLine).setClassName("org.apache.uima.ducc.ps.service.main.ServiceWrapper"); |
| } else { |
| ((ManagedProcess) super.managedProcess).setUimaAs(); |
| cmdLine.addOption("-Dducc.deploy.components=uima-as"); |
| processEnv.put(IDuccUser.EnvironmentVariable.DUCC_UPDATE_PORT.value(), |
| System.getProperty(ProcessStateUpdate.ProcessStateUpdatePort)); |
| ((JavaCommandLine) cmdLine) |
| .setClassName("org.apache.uima.ducc.common.main.DuccService"); |
| } |
| break; |
| } |
| String processLogDir = ((ManagedProcess) super.managedProcess).getProcessInfo() |
| .getLogDirectory() |
| + (((ManagedProcess) super.managedProcess).getProcessInfo().getLogDirectory() |
| .endsWith(File.separator) ? "" : File.separator) |
| + ((ManagedProcess) super.managedProcess).getWorkDuccId() + File.separator; |
| String processLogFile = ((ManagedProcess) super.managedProcess).getWorkDuccId() |
| + processType + host; |
| String workingDir = ((ManagedProcess) super.managedProcess).getProcessInfo() |
| .getWorkingDirectory(); |
| if (workingDir == null) { |
| workingDir = "NONE"; |
| } |
| |
| // Duccling, with logging |
| String[] duccling = new String[] { c_launcher_path, "-f", processLogDir + processLogFile, |
| "-w", workingDir, "-u", ((ManagedProcess) super.managedProcess).getOwner(), "--" }; |
| |
| String executable = cmdLine.getExecutable(); |
| // Check if user specified which java to use to launch the |
| // process. If not provided, |
| // use the same java that the agent is running with |
| if (executable == null || executable.trim().length() == 0) { |
| executable = System.getProperty("java.home") + File.separator + "bin" + File.separator |
| + "java"; |
| } |
| boolean jd = ((ManagedProcess) super.managedProcess).isJd(); |
| |
| if (cmdLine instanceof JavaCommandLine) { |
| String classpath = ""; |
| if (jd) { |
| for (String option : ((JavaCommandLine) cmdLine).getOptions()) { |
| if (option.startsWith("-Dducc.deploy.UserClasspath")) { |
| classpath = option.split("=")[1]; |
| break; |
| } |
| } |
| } else { |
| classpath = ((JavaCommandLine) cmdLine).getClasspath(); |
| } |
| |
| String duccHomePath = Utils.findDuccHome(); |
| /* |
| * String[] jars = classpath.split(":"); URLClassLoader clsLoader = newClassLoader(jars); |
| * Class<?> cls = clsLoader.loadClass("org.apache.uima.impl.UimaVersion"); Method |
| * majorVersionMethod = cls.getMethod("getMajorVersion"); short majorVersion = |
| * (short)majorVersionMethod.invoke(null); if ( !duccHomePath.trim().endsWith("/") ) { |
| * duccHomePath = duccHomePath.concat("/"); } String workItemJarDir = |
| * duccHomePath+"lib/uima-ducc/workitem/uima-ducc-workitem-"; if ( majorVersion < 3 ) { |
| * classpath = workItemJarDir+"v2.jar:"+classpath; } else if ( majorVersion >= 3 ) { |
| * classpath = workItemJarDir+"v3.jar:"+classpath; } else { throw new |
| * DuccRuntimeException("Unknown version of UIMA - majorVersion="+majorVersion); } |
| * |
| * if ( jd ) { // JD uses classloader separation to run user specified jars. |
| * ((JavaCommandLine) cmdLine).replaceOption("-Dducc.deploy.UserClasspath", classpath); } |
| * else { ((JavaCommandLine) cmdLine).setClasspath(classpath); } |
| */ |
| cmdLine.addOption("-DDUCC_HOME=" + duccHomePath); |
| cmdLine.addOption( |
| "-Dducc.deploy.configuration=" + System.getProperty("ducc.deploy.configuration")); |
| // UIMA-4935 Following moved from CommandExecutor to avoid duplications in the shared |
| // cmdLine |
| cmdLine.addOption("-Dducc.deploy.JpUniqueId=" |
| + ((ManagedProcess) managedProcess).getDuccId().getUnique()); |
| |
| // NOTE - These are redundant since the information is also |
| // in the environment for both Java and non-Java processes |
| cmdLine.addOption("-Dducc.process.log.dir=" + processLogDir); |
| cmdLine.addOption("-Dducc.process.log.basename=" + processLogFile); |
| cmdLine.addOption( |
| "-Dducc.job.id=" + ((ManagedProcess) super.managedProcess).getWorkDuccId()); |
| |
| } |
| |
| if (useDuccSpawn()) { |
| cmd = Utils.concatAllArrays(duccling, new String[] { executable }, |
| cmdLine.getCommandLine()); |
| } else { |
| cmd = Utils.concatAllArrays(new String[] { executable }, cmdLine.getCommandLine()); |
| } |
| // add JobId and the log prefix to the env so additional |
| // similarly-named log files can be created |
| // Also put the state update port in the environment for all processes ... instead of just |
| // some as a system property |
| processEnv.put(IDuccUser.EnvironmentVariable.DUCC_ID_JOB.value(), String |
| .valueOf(((ManagedProcess) super.managedProcess).getWorkDuccId().getFriendly())); |
| processEnv.put(IDuccUser.EnvironmentVariable.DUCC_LOG_PREFIX.value(), |
| processLogDir + processLogFile); |
| |
| for (String part : cmd) { |
| if (part.startsWith("-Dducc.deploy.JpType=uima")) { |
| // this is an AP with JP-based "nature". Meaning its using |
| // a JP process configuration and its pull based model. |
| uimaBasedAP = true; |
| break; |
| } |
| } |
| |
| // Currently agent has two ports where it listens for process state updates. One is |
| // for JPs and the other is for APs and JDs. The latter use a simplified state update |
| // protocol |
| // which is String based. The JPs actually serialize a more complex state Object. |
| // There is a way to deploy an AP with a JP "nature". Meaning an AP whose internal |
| // components looks just like a JP. Such AP must report its state to the same |
| // agent port as a JP. Only non-uima based APs will report to the other port. |
| if (jd || (isAP((ManagedProcess) super.managedProcess) && !uimaBasedAP)) { |
| logger.info("getDeployableCommandLine", |
| ((ManagedProcess) super.managedProcess).getDuccId(), |
| "Deploying Process Type:" |
| + ((ManagedProcess) super.managedProcess).getDuccProcess() |
| .getProcessType().name() |
| + " This process will report state update to AP specific socket listener running on port:" |
| + System.getProperty("AGENT_AP_STATE_UPDATE_PORT")); |
| |
| ITimeWindow twi = new TimeWindow(); |
| ((ManagedProcess) managedProcess).getDuccProcess().setTimeWindowInit(twi); |
| String millis = TimeStamp.getCurrentMillis(); |
| twi.setStart(millis); |
| twi.setEnd(millis); |
| |
| ITimeWindow twr = new TimeWindow(); |
| ((ManagedProcess) managedProcess).getDuccProcess().setTimeWindowRun(twr); |
| twr.setStart(millis); |
| processEnv.put(IDuccUser.EnvironmentVariable.DUCC_UPDATE_PORT.value(), |
| System.getProperty("AGENT_AP_STATE_UPDATE_PORT")); |
| } else { |
| logger.info("getDeployableCommandLine", |
| ((ManagedProcess) super.managedProcess).getDuccId(), |
| "Deploying Process Type:" |
| + ((ManagedProcess) super.managedProcess).getDuccProcess() |
| .getProcessType().name() |
| + " This process will report state update to Camel-Mina listener running on port:" |
| + System.getProperty(ProcessStateUpdate.ProcessStateUpdatePort)); |
| // processEnv.put(IDuccUser.EnvironmentVariable.DUCC_UPDATE_PORT.value(), |
| // System.getProperty(ProcessStateUpdate.ProcessStateUpdatePort)); |
| } |
| } |
| |
| // Replace the reserved DUCC variable with the architecture of this node (ppc64 or amd64 or |
| // ...) |
| // (could have been done in getCommandLine if that did return the full cmd line!) |
| String osArch = System.getProperty("os.arch"); |
| StringBuilder sb = new StringBuilder(); |
| for (int i = 0; i < cmd.length; ++i) { |
| if (cmd[i].contains("${DUCC_OS_ARCH}")) { |
| cmd[i] = cmd[i].replace("${DUCC_OS_ARCH}", osArch); |
| } |
| sb.append(cmd[i]).append(" "); |
| } |
| logger.info("getDeployableCommandLine --------------- ", null, sb.toString()); |
| |
| return cmd; |
| |
| } catch (Exception ex) { |
| ((ManagedProcess) super.managedProcess).getDuccProcess().setProcessState(ProcessState.Failed); |
| throw ex; |
| } finally { |
| NodeAgent.unlock(); |
| } |
| |
| } |
| |
| public URLClassLoader newClassLoader(String[] classPathElements) throws IOException { |
| ArrayList<URL> urlList = new ArrayList<URL>(classPathElements.length); |
| for (String element : classPathElements) { |
| if (element.endsWith("*")) { |
| File dir = new File(element.substring(0, element.length() - 1)); |
| File[] files = dir.listFiles(); // Will be null if missing or not a dir |
| if (files != null) { |
| for (File f : files) { |
| if (f.getName().endsWith(".jar")) { |
| urlList.add(f.getCanonicalFile().toURI().toURL()); |
| } |
| } |
| } |
| } else { |
| File f = new File(element); |
| if (f.exists()) { |
| urlList.add(f.getCanonicalFile().toURI().toURL()); |
| } |
| } |
| } |
| URL[] urls = new URL[urlList.size()]; |
| return new URLClassLoader(urlList.toArray(urls), |
| ClassLoader.getSystemClassLoader().getParent()); |
| } |
| |
| public void stop() { |
| } |
| } |
