uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/NodeAgent.java - uima-ducc - Git at Google

 /*
  * 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;

 import java.io.BufferedReader;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileReader;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;

 import org.apache.camel.CamelContext;
 import org.apache.camel.Processor;
 import org.apache.camel.Route;
 import org.apache.camel.model.RouteDefinition;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.commons.lang.SerializationUtils;
 import org.apache.hadoop.conf.Configured;
 import org.apache.uima.ducc.agent.config.AgentConfiguration;
 import org.apache.uima.ducc.agent.event.AgentEventListener;
 import org.apache.uima.ducc.agent.event.ProcessLifecycleObserver;
 import org.apache.uima.ducc.agent.launcher.CGroupsManager;
 import org.apache.uima.ducc.agent.launcher.DefunctProcessDetector;
 import org.apache.uima.ducc.agent.launcher.ICommand;
 import org.apache.uima.ducc.agent.launcher.Launcher;
 import org.apache.uima.ducc.agent.launcher.ManagedProcess;
 import org.apache.uima.ducc.agent.launcher.ManagedProcess.StopPriority;
 import org.apache.uima.ducc.agent.launcher.SigKillCommand;
 import org.apache.uima.ducc.agent.launcher.SigTermCommand;
 import org.apache.uima.ducc.agent.metrics.collectors.NodeUsersCollector;
 import org.apache.uima.ducc.agent.processors.DefaultNodeInventoryProcessor;
 import org.apache.uima.ducc.common.Node;
 import org.apache.uima.ducc.common.NodeIdentity;
 import org.apache.uima.ducc.common.admin.event.DuccAdminEvent;
 import org.apache.uima.ducc.common.admin.event.DuccAdminEventQuiesceAndStop;
 import org.apache.uima.ducc.common.admin.event.DuccAdminEventStop;
 import org.apache.uima.ducc.common.admin.event.DuccAdminEventStopMetrics;
 import org.apache.uima.ducc.common.boot.DuccDaemonRuntimeProperties;
 import org.apache.uima.ducc.common.component.AbstractDuccComponent;
 import org.apache.uima.ducc.common.main.DuccService;
 import org.apache.uima.ducc.common.node.metrics.NodeUsersInfo;
 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.IDuccLoggerComponents.Daemon;
 import org.apache.uima.ducc.common.utils.id.DuccId;
 import org.apache.uima.ducc.common.utils.id.IDuccId;
 import org.apache.uima.ducc.transport.agent.IUimaPipelineAEComponent;
 import org.apache.uima.ducc.transport.agent.ProcessStateUpdate;
 import org.apache.uima.ducc.transport.cmdline.ICommandLine;
 import org.apache.uima.ducc.transport.cmdline.NonJavaCommandLine;
 import org.apache.uima.ducc.transport.dispatcher.DuccEventDispatcher;
 import org.apache.uima.ducc.transport.event.AgentProcessLifecycleReportDuccEvent;
 import org.apache.uima.ducc.transport.event.AgentProcessLifecycleReportDuccEvent.LifecycleEvent;
 import org.apache.uima.ducc.transport.event.DuccEvent.EventType;
 import org.apache.uima.ducc.transport.event.DaemonDuccEvent;
 import org.apache.uima.ducc.transport.event.DuccEvent;
 import org.apache.uima.ducc.transport.event.NodeInventoryUpdateDuccEvent;
 import org.apache.uima.ducc.transport.event.ProcessStateUpdateDuccEvent;
 import org.apache.uima.ducc.transport.event.common.DuccProcess;
 import org.apache.uima.ducc.transport.event.common.DuccReservation;
 import org.apache.uima.ducc.transport.event.common.DuccReservationMap;
 import org.apache.uima.ducc.transport.event.common.DuccUserReservation;
 import org.apache.uima.ducc.transport.event.common.IDuccJobDeployment;
 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.IDuccReservation;
 import org.apache.uima.ducc.transport.event.common.IDuccReservationMap;
 import org.apache.uima.ducc.transport.event.common.IDuccStandardInfo;
 import org.apache.uima.ducc.transport.event.common.IProcessState.ProcessState;
 import org.apache.uima.ducc.transport.event.common.IResourceState.ResourceState;
 import org.apache.uima.ducc.transport.event.common.ITimeWindow;
 import org.apache.uima.ducc.transport.event.common.ProcessMemoryAssignment;
 import org.apache.uima.ducc.transport.event.common.TimeWindow;

 public class NodeAgent extends AbstractDuccComponent implements Agent, ProcessLifecycleObserver {
   public static DuccLogger logger = DuccLogger.getLogger(NodeAgent.class, COMPONENT_NAME);

   private static DuccId jobid = null;
   // Replaced by duplicate in org.apache.uima.ducc.transport.agent.ProcessStateUpdate
   // public static final String ProcessStateUpdatePort = "ducc.agent.process.state.update.port";

   public static int SIGKILL = 9;

   public static int SIGTERM = 15;

   // for LinuxNodeMetrics logging
   public static AtomicLong logCounter = new AtomicLong();

   public static String cgroupFailureReason;

   // Map of known processes this agent is managing. This map is published
   // at regular intervals as part of agent's inventory update.
   private Map<DuccId, IDuccProcess> inventory = new ConcurrentHashMap<>();

   // Semaphore controlling access to inventory Map
   private Semaphore inventorySemaphore = new Semaphore(1);

   List<ManagedProcess> deployedProcesses = new ArrayList<ManagedProcess>();

   // This agent's identity ( host name and IP address)
   private NodeIdentity nodeIdentity;

   // Component to launch processes
   private Launcher launcher;

   // Reference to the Agent's configuration factory component. This is where
   // Agent
   // dependencies are instantiated and injected via Agent's c'tor.
   public AgentConfiguration configurationFactory;

   private static Semaphore agentLock = new Semaphore(1);

   private DuccEventDispatcher commonProcessDispatcher;

   private DuccEventDispatcher ORDispatcher;

   private Object monitor = new Object();

   boolean duccLingExists = false;

   boolean runWithDuccLing = false;

   private List<DuccUserReservation> reservations = new ArrayList<DuccUserReservation>();

   private Semaphore reservationsSemaphore = new Semaphore(1);

   // private AgentMonitor nodeMonitor;

   private volatile boolean stopping = false;

   private Object stopLock = new Object();

   private RogueProcessReaper rogueProcessReaper = new RogueProcessReaper(logger, 5, 10);

   public volatile boolean useCgroups = false;

   public CGroupsManager cgroupsManager = null;

   public Node node = null;

   public volatile boolean excludeAPs = false;

   public int shareQuantum;

   private boolean agentVirtual = System.getProperty("ducc.agent.virtual") == null ? false : true;

   // This flag, when true, forces the agent to be "real"
   // regardless of the value of ducc.agent.virtual.
   // In the future, this flag and support for ducc.agent.virtual
   // should be removed altogether.
   // This is to support CGroups for all agents,
   // which was previously disabled for virtual ones.
   private boolean agentRealOnly = true;

   public boolean pageSizeFetched = false;

   public int pageSize = 4096; // default

   public int cpuClockRate = 100;

   public int numProcessors = 0;

   ExecutorService defunctDetectorExecutor = Executors.newCachedThreadPool();

   private AgentEventListener eventListener;

   // indicates whether or not this agent received at least one publication
   // from the PM. This flag is used to determine if the agent should use
   // rogue process detector. The detector will be used if this flag is true.
   public volatile boolean receivedDuccState = false;

   private String stateChangeEndpoint;

   long maxTimeToWaitForProcessToStop = 60000; // default 1 minute

   public void setStateChangeEndpoint(String stateChangeEndpoint) {
     this.stateChangeEndpoint = stateChangeEndpoint;
   }

   /**
    * Ctor used exclusively for black-box testing of this class.
    */
   public NodeAgent() {
     super(COMPONENT_NAME, null);
   }

   public NodeAgent(NodeIdentity ni) {
     this();
     this.nodeIdentity = ni;
     Utils.findDuccHome(); // add DUCC_HOME to System.properties

     if (configurationFactory.processStopTimeout != null) {
       maxTimeToWaitForProcessToStop = Long.valueOf(configurationFactory.processStopTimeout);
     }
   }

   public long getLastORSequence() {
     long lastORSequence = 0;
     if (eventListener != null) {
       lastORSequence = eventListener.getLastSequence();
     }
     return lastORSequence;
   }

   public AgentEventListener getEventListener() {
     return eventListener;
   }

   public void setAgentEventListener(AgentEventListener listener) {
     eventListener = listener;
   }

   public boolean isVirtual() {
     boolean retVal = agentVirtual;
     if (agentRealOnly) {
       retVal = false;
     }
     return retVal;
   }

   /**
    * Tell Orchestrator about state change for recording into system-events.log
    */
   private void stateChange(EventType eventType) {
     String methodName = "stateChange";
     try {
       Daemon daemon = Daemon.Agent;
       NodeIdentity nodeIdentity = new NodeIdentity();
       DaemonDuccEvent ev = new DaemonDuccEvent(daemon, eventType, nodeIdentity);
       ORDispatcher.dispatch(stateChangeEndpoint, ev, "");
       logger.info(methodName, null, stateChangeEndpoint, eventType.name(),
               nodeIdentity.getCanonicalName());
     } catch (Exception e) {
       logger.error(methodName, null, e);
     }
   }

   /*
    * Report process lifecycle events on same channel that inventory is reported
    */
   public DuccEventDispatcher getProcessLifecycleReportDispatcher() {
     return ORDispatcher;
   }

   /*
    * Send process lifecycle event to interested listeners
    */
   private void sendProcessLifecycleEventReport(IDuccProcess process,
           LifecycleEvent lifecycleEvent) {
     String location = "sendProcessLifecycleEventReport";
     try {
       NodeIdentity nodeIdentity = getIdentity();
       AgentProcessLifecycleReportDuccEvent duccEvent = new AgentProcessLifecycleReportDuccEvent(
               process, nodeIdentity, lifecycleEvent);
       DuccEventDispatcher dispatcher = getProcessLifecycleReportDispatcher();
       dispatcher.dispatch(duccEvent);
       StringBuffer sb = new StringBuffer();
       sb.append("id:" + process.getDuccId().toString() + " ");
       sb.append("lifecycleEvent:" + lifecycleEvent.name() + " ");
       String args = sb.toString().trim();
       logger.info(location, jobid, args);
     } catch (Exception e) {
       logger.error(location, jobid, e);
     }
   }

   /*
    * Add ManagedProcess to map and send lifecycle event
    */
   private void processDeploy(ManagedProcess mp) {
     String location = "processDeploy";
     if (mp != null) {
       if (deployedProcesses.contains(mp)) {
         String args = "mp:" + mp.getProcessId();
         logger.error(location, jobid, args);
       } else {
         String args = "mp:" + mp.getProcessId();
         logger.debug(location, jobid, args);
         deployedProcesses.add(mp);
         IDuccProcess process = mp.getDuccProcess();
         sendProcessLifecycleEventReport(process, LifecycleEvent.Launch);
       }
     } else {
       String args = "mp:" + mp;
       logger.error(location, jobid, args);
     }
   }

   /*
    * Remove ManagedProcess from map and send lifecycle event
    */
   private void processUndeploy(ManagedProcess mp, Iterator<ManagedProcess> it) {
     String location = "processUndeploy";
     if (mp != null) {
       if (!deployedProcesses.contains(mp)) {
         String args = "mp:" + mp.getProcessId();
         logger.error(location, jobid, args);
       } else {
         String args = "mp:" + mp.getProcessId();
         logger.debug(location, jobid, args);
         // deployedProcesses.remove(mp);
         // it.remove();
         IDuccProcess process = mp.getDuccProcess();
         sendProcessLifecycleEventReport(process, LifecycleEvent.Terminate);
       }
     } else {
       String args = "mp:" + mp;
       logger.error(location, jobid, args);
     }
   }

   /**
    * C'tor for dependecy injection
    *
    * @param nodeIdentity
    *          - this Agent's identity
    * @param launcher
    *          - component to launch processes
    * @param context
    *          - camel context
    */
   public NodeAgent(NodeIdentity nodeIdentity, Launcher launcher, CamelContext context,
           AgentConfiguration factory) throws Exception {
     super(COMPONENT_NAME, context);

     Utils.findDuccHome(); // add DUCC_HOME to System.properties

     // Running a real agent
     agentVirtual = System.getProperty("ducc.agent.virtual") == null ? false : true;

     this.nodeIdentity = nodeIdentity;
     this.launcher = launcher;
     this.configurationFactory = factory;
     this.commonProcessDispatcher = factory.getCommonProcessDispatcher(context);
     this.ORDispatcher = factory.getORDispatcher(context);

     // fetch Page Size from the OS and cache it
     pageSize = getOSPageSize();

     numProcessors = getNodeProcessors();

     logger.info("NodeAgent", null, "OS Page Size:" + pageSize);

     cpuClockRate = getOSClockRate();
     logger.info("NodeAgent", null, "OS Clock Rate:" + cpuClockRate);

     if (System.getProperty("ducc.rm.share.quantum") != null
             && System.getProperty("ducc.rm.share.quantum").trim().length() > 0) {
       shareQuantum = Integer.parseInt(System.getProperty("ducc.rm.share.quantum").trim());
     }
     /* Enable CGROUPS */
     String cgroups;
     String cgUtilsPath = null;
     boolean excludeNodeFromCGroups = false;
     if (!isVirtual()
             && (cgroups = System.getProperty("ducc.agent.launcher.cgroups.enable")) != null) {
       if (cgroups.equalsIgnoreCase("true")) {
         logger.info("nodeAgent", null, "ducc.properties [ducc.agent.launcher.cgroups.enable=true]");
         // Load exclusion file. Some nodes may be excluded from cgroups
         String exclusionFile;

         // get the name of the exclusion file from ducc.properties
         if ((exclusionFile = System.getProperty("ducc.agent.exclusion.file")) != null) {
           logger.info("nodeAgent", null,
                   "Ducc configured with cgroup node exclusion file - ducc.properties [ducc.agent.exclusion.file="
                           + exclusionFile + "]");
           // Parse node exclusion file and determine if cgroups and AP
           // deployment
           // is allowed on this node
           NodeExclusionParser exclusionParser = new NodeExclusionParser();
           exclusionParser.parse(exclusionFile);
           excludeNodeFromCGroups = exclusionParser.cgroupsExcluded();
           excludeAPs = exclusionParser.apExcluded();
           if (excludeNodeFromCGroups) {
             logger.info("nodeAgent", null,
                     "------- Node Explicitly Excluded From Using CGroups. Check File:"
                             + exclusionFile);
             cgroupFailureReason = "------- Node Explicitly Excluded From Using CGroups. Check File:"
                     + exclusionFile;
           }
           System.out.println(
                   "excludeNodeFromCGroups=" + excludeNodeFromCGroups + " excludeAPs=" + excludeAPs);
         } else {
           logger.info("nodeAgent", null, "Agent node *not* excluded from using cgroups");
         }
         // node not in the exclusion list for cgroups
         if (!excludeNodeFromCGroups) {
           // fetch a list of paths the agent will search to find cgroups utils
           // like cgexec. The default location is /usr/bin
           logger.info("nodeAgent", null,
                   "Testing cgroups to check if runtime utilities (cgexec) exist in expected locations in the filesystem");
           String cgroupsUtilsDirs = System.getProperty("ducc.agent.launcher.cgroups.utils.dir");
           if (cgroupsUtilsDirs == null) {
             cgUtilsPath = "/usr/bin"; // default
           } else {
             String[] paths = cgroupsUtilsDirs.split(",");
             for (String path : paths) {
               File file = new File(path.trim() + "/cgexec");
               if (file.exists()) {
                 cgUtilsPath = path;
                 break;
               }
             }
           }
           // scan /proc/mounts for base cgroup dir
           String cgroupsBaseDir = fetchCgroupsBaseDir("/proc/mounts");

           if (cgUtilsPath == null) {
             useCgroups = false;
             logger.info("nodeAgent", null,
                     "------- CGroups Disabled - Unable to Find Cgroups Utils Directory. Add/Modify ducc.agent.launcher.cgroups.utils.dir property in ducc.properties");
           } else if (cgroupsBaseDir == null || cgroupsBaseDir.trim().length() == 0) {
             useCgroups = false;
             logger.info("nodeAgent", null,
                     "------- CGroups Disabled - Unable to Find Cgroups Root Directory in /proc/mounts");

           } else {
             logger.info("nodeAgent", null, "Agent found cgroups runtime in " + cgUtilsPath
                     + " cgroups base dir=" + cgroupsBaseDir);
             // if cpuacct is configured in cgroups, the subsystems list will be updated
             String cgroupsSubsystems = "memory,cpu";

             long maxTimeToWaitForProcessToStop = 60000; // default 1 minute
             if (configurationFactory.processStopTimeout != null) {
               maxTimeToWaitForProcessToStop = Long.valueOf(configurationFactory.processStopTimeout);
             }

             cgroupsManager = new CGroupsManager(cgUtilsPath, cgroupsBaseDir, cgroupsSubsystems,
                     logger, maxTimeToWaitForProcessToStop);
             cgroupsManager.configure(this);
             // check if cgroups base directory exists in the filesystem
             // which means that cgroups
             // and cgroups convenience package are installed and the
             // daemon is up and running.
             if (cgroupsManager.cgroupExists(cgroupsBaseDir)) {
               logger.info("nodeAgent", null,
                       "Agent found cgroup base directory in " + cgroupsBaseDir);
               try {
                 String containerId = "test";
                 // validate cgroups by creating a dummy cgroup. The code checks if cgroup actually
                 // got created by
                 // verifying existence of test cgroup file. The second step in verification is to
                 // check if
                 // CPU control is working. Configured in cgconfig.conf, the CPU control allows for
                 // setting
                 // cpu.shares. The code will attempt to set the shares and subsequently tries to
                 // read the
                 // value from cpu.shares file to make sure the values match. Any exception in the
                 // above steps
                 // will cause cgroups to be disabled.
                 //
                 cgroupsManager.validator(cgroupsBaseDir, containerId,
                         System.getProperty("user.name"), false).cgcreate().cgset(100); // write
                                                                                        // cpu.shares=100
                                                                                        // and
                                                                                        // validate

                 // cleanup dummy cgroup
                 cgroupsManager.destroyContainer(containerId, System.getProperty("user.name"),
                         SIGKILL);
                 useCgroups = true;
               } catch (CGroupsManager.CGroupsException ee) {
                 logger.info("nodeAgent", null, ee);
                 cgroupFailureReason = ee.getMessage();
                 useCgroups = false;
               }
               if (useCgroups) {
                 try {
                   // remove stale CGroups
                   cgroupsManager.cleanup();
                 } catch (Exception e) {
                   logger.error("nodeAgent", null, e);
                   useCgroups = false;
                   logger.info("nodeAgent", null,
                           "Agent cgroup cleanup failed on this machine base directory in "
                                   + cgroupsBaseDir
                                   + ". Check if cgroups is installed on this node, Agent has correct permissions (consistent with cgconfig.conf), and the cgroup daemon is running");
                   cgroupFailureReason = "------- CGroups Not Working on this Machine";
                 }
               } else {
                 logger.info("nodeAgent", null,
                         "Agent cgroup test failed on this machine base directory in "
                                 + cgroupsBaseDir
                                 + ". Check if cgroups is installed on this node, Agent has correct permissions (consistent with cgconfig.conf), and the cgroup daemon is running");
                 cgroupFailureReason = "------- CGroups Not Working on this Machine";
               }

             } else {
               logger.info("nodeAgent", null, "Agent failed to find cgroup base directory in "
                       + cgroupsBaseDir
                       + ". Check if cgroups is installed on this node and the cgroup daemon is running");
               // logger.info("nodeAgent", null, "------- CGroups Not Installed on this Machine");
               cgroupFailureReason = "------- CGroups Not Installed on this Machine";
             }
           }
         }
       }
     } else {
       logger.info("nodeAgent", null, "------- CGroups Not Enabled on this Machine");
       cgroupFailureReason = "------- CGroups Not Enabled on this Machine - check ducc.properties: ducc.agent.launcher.cgroups.enable ";
     }

     // begin publishing node metrics
     factory.startNodeMetrics(this);

     logger.info("nodeAgent", null,
             "CGroup Support=" + useCgroups + " excludeNodeFromCGroups=" + excludeNodeFromCGroups
                     + " excludeAPs=" + excludeAPs + " CGroups utils Dir:" + cgUtilsPath);

     String useSpawn = System.getProperty("ducc.agent.launcher.use.ducc_spawn");
     if (useSpawn != null && useSpawn.toLowerCase().equals("true")) {
       runWithDuccLing = true;
       String c_launcher_path = Utils.resolvePlaceholderIfExists(
               System.getProperty("ducc.agent.launcher.ducc_spawn_path"), System.getProperties());
       try {
         File duccLing = new File(c_launcher_path);
         if (duccLing.exists()) {
           duccLingExists = true;
         }
       } catch (Exception e) {
         logger.info("nodeAgent", null,
                 "------- Agent failed while checking for existence of ducc_ling", e);
       }

     }
   }

   private String fetchCgroupsBaseDir(String mounts) {
     String cbaseDir = null;
     BufferedReader br = null;
     try {
       FileInputStream fis = new FileInputStream(mounts);
       // Construct BufferedReader from InputStreamReader
       br = new BufferedReader(new InputStreamReader(fis));

       String line = null;
       while ((line = br.readLine()) != null) {
         System.out.println(line);
         if (line.trim().startsWith("cgroup")) {
           String[] cgroupsInfo = line.split(" ");
           if (cgroupsInfo[1].trim().equals("/cgroup")) {
             cbaseDir = cgroupsInfo[1].trim();
             break;
           } else if (cgroupsInfo[1].trim().endsWith("/memory")) {
             // return the mount point minus the memory part
             cbaseDir = cgroupsInfo[1].substring(0, cgroupsInfo[1].indexOf("/memory"));
             break;
           }
         }
       } // while

     } catch (Exception e) {
       logger.info("nodeAgent", null,
               "------- Agent failed while checking for existence of CGroups", e.getMessage());
     } finally {
       if (br != null) {
         try {
           br.close();
         } catch (Exception ex) {
         }
       }
     }
     return cbaseDir;
   }

   public int getNodeProcessors() {
     return runOSCommand(new String[] { "/usr/bin/getconf", "_NPROCESSORS_ONLN" });
   }

   public int getOSPageSize() {
     return runOSCommand(new String[] { "/usr/bin/getconf", "PAGESIZE" });
   }

   public int getOSClockRate() {
     return runOSCommand(new String[] { "/usr/bin/getconf", "CLK_TCK" });
   }

   private int runOSCommand(String[] cmd) {
     InputStreamReader in = null;
     BufferedReader reader = null;
     int retVal = 0;
     try {
       ProcessBuilder pb = new ProcessBuilder();
       pb.command(cmd);
       pb.redirectErrorStream(true);
       Process p = pb.start();
       in = new InputStreamReader(p.getInputStream());
       reader = new BufferedReader(in);
       String line = null;

       while ((line = reader.readLine()) != null) {
         retVal = Integer.parseInt(line.trim());
       }
     } catch (Exception e) {
       logger.error("runOSCommand", null, e);
     } finally {
       if (reader != null) {
         try {
           reader.close();
         } catch (Exception ex) {
         }
       }
     }
     return retVal;
   }

   public void setNodeInfo(Node node) {
     this.node = node;
   }

   public Node getNodeInfo() {
     return node;
   }

   public int getNodeTotalNumberOfShares() {
     int shareQuantum = 0;
     int shares = 1;
     if (System.getProperty("ducc.rm.share.quantum") != null
             && System.getProperty("ducc.rm.share.quantum").trim().length() > 0) {
       shareQuantum = Integer.parseInt(System.getProperty("ducc.rm.share.quantum").trim());
       shares = (int) getNodeInfo().getNodeMetrics().getNodeMemory().getMemTotal() / shareQuantum; // get
                                                                                                   // number
                                                                                                   // of
                                                                                                   // shares
       if ((getNodeInfo().getNodeMetrics().getNodeMemory().getMemTotal() % shareQuantum) > 0)
         shares++; // ciel
     }

     return shares;
   }

   public void start(DuccService service) throws Exception {
     super.start(service, null);
     String methodName = "start";
     String name = nodeIdentity.getShortName();
     String ip = nodeIdentity.getIp();
     String jmxUrl = getProcessJmxUrl();
     DuccDaemonRuntimeProperties.getInstance().bootAgent(name, ip, jmxUrl);
     String key = "ducc.broker.url";
     String value = System.getProperty(key);
     logger.info(methodName, null, key + "=" + value);
     stateChange(EventType.BOOT);
   }

   public DuccEventDispatcher getEventDispatcherForRemoteProcess() {
     return commonProcessDispatcher;
   }

   public boolean duccLingExists() {
     return duccLingExists;
   }

   public void duccLingExists(boolean duccLingExists) {
     this.duccLingExists = duccLingExists;
   }

   public boolean runWithDuccLing() {
     return runWithDuccLing;
   }

   public void runWithDuccLing(boolean runWithDuccLing) {
     this.runWithDuccLing = runWithDuccLing;
   }

   /**
    * Returns deep copy (by way of java serialization) of the Agents inventory.
    */
   @SuppressWarnings("unchecked")
   public Map<DuccId, IDuccProcess> getInventoryCopy() {
     Object deepCopy = null;
     try {
       inventorySemaphore.acquire();
       deepCopy = SerializationUtils.clone((ConcurrentHashMap<DuccId, IDuccProcess>) inventory);
     } catch (InterruptedException e) {
     } finally {
       inventorySemaphore.release();
     }
     return (Map<DuccId, IDuccProcess>) deepCopy;
   }

   /**
    * Returns shallow copy of the Agent's inventory
    */
   public Map<DuccId, IDuccProcess> getInventoryRef() {
     return inventory;
   }

   /*
    * Check if both the command and its args are missing, since the command defaults to the DUCC JVM.
    */
   private boolean invalidCommand(ICommandLine commandLine) {
     if (commandLine != null) {
       if (commandLine.getExecutable() != null && commandLine.getExecutable().length() > 0)
         return false;
       if (commandLine.getCommandLine() != null && commandLine.getCommandLine().length > 0)
         return false;
     }
     return true;
   }

   private boolean isProcessDeallocated(IDuccProcess process) {
     return (process.getProcessState().equals(ProcessState.Undefined) && process.isDeallocated());
   }

   /**
    * Stops any process that is in agent's inventory but not in provided job list sent by the PM.
    *
    * @param lifecycleController
    *          - instance implementing stopProcess() method
    * @param jobDeploymentList
    *          - all DUCC jobs sent by PM
    */
   public void takeDownProcessWithNoJob(ProcessLifecycleController lifecycleController,
           List<IDuccJobDeployment> jobDeploymentList) {
     String methodName = "takeDownProcessWithNoJob";
     try {
       inventorySemaphore.acquire();
       List<IDuccProcess> purgeList = new ArrayList<IDuccProcess>();
       boolean hasAjob = false;
       // Check if every process in agent's inventory is associated with a
       // job in a given
       // jobDeploymentList
       for (Entry<DuccId, IDuccProcess> processEntry : getInventoryRef().entrySet()) {
         // if a job list is empty, take down all agent processes that
         // are in the inventory
         if (jobDeploymentList.isEmpty()) {
           logger.info(methodName, null, "...Agent Process:" + processEntry.getValue().getDuccId()
                   + " Not in JobDeploymentList. Ducc Currently Has No Jobs Running");
           hasAjob = false;
         } else {
           // iterate over all jobs
           for (IDuccJobDeployment job : jobDeploymentList) {
             // check if current process is a JD
             if (job.getJdProcess() != null) {
               if (processEntry.getValue().getDuccId().equals(job.getJdProcess().getDuccId())) {
                 hasAjob = true;
                 break;
               }
             }
             // check if current process is a JP
             for (IDuccProcess jProcess : job.getJpProcessList()) {
               if (processEntry.getValue().getDuccId().equals(jProcess.getDuccId())) {
                 hasAjob = true;
                 break;
               }
             }
             if (hasAjob) {
               break;
             }
           }
         }
         if (!hasAjob) {
           // if a process in agent inventory has no job and is still
           // alive, stop it
           if (isAlive(processEntry.getValue())) {
             logger.error(methodName, null,
                     "<<<<<<<<< Stopping Process with no Job Assignement (Ghost Process) - DuccId:"
                             + processEntry.getValue().getDuccId() + " PID:"
                             + processEntry.getValue().getPID());
             processEntry.getValue().setReasonForStoppingProcess(
                     ReasonForStoppingProcess.JPHasNoActiveJob.toString());
             lifecycleController.stopProcess(processEntry.getValue());
           } else {
             // add process to purge list
             purgeList.add(processEntry.getValue());
           }
         } else {
           hasAjob = false;
         }
       }
       for (IDuccProcess processToPurge : purgeList) {
         logger.error(methodName, null,
                 "XXXXXXXXXX Purging Process:" + processToPurge.getDuccId() + " Process State:"
                         + processToPurge.getProcessState() + " Process Resource State:"
                         + processToPurge.getResourceState());
         getInventoryRef().remove(processToPurge.getDuccId());
       }
     } catch (Exception e) {

     } finally {
       inventorySemaphore.release();
     }
   }

   private void stopProcessIfAlive(IDuccProcess process,
           ProcessLifecycleController lifecycleController) {
     String methodName = "stopProcessIfAlive";
     if (isAlive(process)) {
       logger.error(methodName, null,
               "<<<<<<<<< Stopping Process with no Job Assignement (Ghost Process) - DuccId:"
                       + process.getDuccId() + " PID:" + process.getPID());
       process.setReasonForStoppingProcess(ReasonForStoppingProcess.JPHasNoActiveJob.toString());
       lifecycleController.stopProcess(process);
     } else {
       logger.error(methodName, null,
               "XXXXXXXXXX Purging Process:" + process.getDuccId() + " Process State:"
                       + process.getProcessState() + " Process Resource State:"
                       + process.getResourceState());

       getInventoryRef().remove(process.getDuccId());

       Iterator<ManagedProcess> it = deployedProcesses.iterator();
       while (it.hasNext()) {
         ManagedProcess deployedProcess = it.next();
         if (deployedProcess.getDuccProcess().getDuccId().equals(process.getDuccId())) {
           it.remove();
           break;
         }
       }
       logger.info(methodName, null, "After Purge Inventory size:" + inventory.size()
               + " deployedProcesses size:" + deployedProcesses.size());
     }
   }

   /*
    * Valid process exists in agent inventory and in an incoming OR state. If process exists in agent
    * inventory but not in OR state than such process is invalid
    */
   private boolean validProcess(IDuccProcess process, List<IDuccJobDeployment> jobDeploymentList) {
     // iterate over all jobs
     for (IDuccJobDeployment job : jobDeploymentList) {
       // check if current process is a JD
       if ((job.getJdProcess() != null
               && process.getDuccId().equals(job.getJdProcess().getDuccId()))) {
         return true;
       } else {
         // check if current process is a JP
         for (IDuccProcess jProcess : job.getJpProcessList()) {
           if (process.getDuccId().equals(jProcess.getDuccId())) {
             return true;
           }
         }
       }
     }
     return false;
   }

   public void takeDownProcessWithNoJobV2(ProcessLifecycleController lifecycleController,
           List<IDuccJobDeployment> jobDeploymentList) {
     String methodName = "takeDownProcessWithNoJobV2";
     try {
       inventorySemaphore.acquire();

       // iterate over all processes in agent inventory
       for (Entry<DuccId, IDuccProcess> processEntry : getInventoryRef().entrySet()) {
         // if OR deployment list is empty, take down all agent processes that
         // are in the inventory
         if (jobDeploymentList.isEmpty()
                 || !validProcess(processEntry.getValue(), jobDeploymentList)) {
           logger.info(methodName, null, "...Agent Process:" + processEntry.getValue().getDuccId()
                   + " Not in OR JobDeploymentList");
           stopProcessIfAlive(processEntry.getValue(), lifecycleController);
         }
       }
     } catch (Exception e) {
       logger.error(methodName, null, e);
     } finally {
       inventorySemaphore.release();
     }
   }

   /**
    * Reconciles agent inventory with job processes sent by PM
    *
    * @param lifecycleController
    *          - instance implementing stopProcess and startProcess
    * @param process
    *          - job process from a Job List
    * @param commandLine
    *          - in case this process is not in agents inventory we need cmd line to start it
    * @param info
    *          - DUCC common info including user log dir, user name, etc
    * @param workDuccId
    *          - job id
    */
   public void reconcileProcessStateAndTakeAction(ProcessLifecycleController lifecycleController,
           IDuccProcess process, ICommandLine commandLine, IDuccStandardInfo info,
           ProcessMemoryAssignment processMemoryAssignment, DuccId workDuccId,
           boolean isPreemptable) {
     String methodName = "reconcileProcessStateAndTakeAction";
     try {
       inventorySemaphore.acquire();
       // Check if process exists in agent's inventory
       if (getInventoryRef().containsKey(process.getDuccId())) {
         IDuccProcess agentManagedProcess = getInventoryRef().get(process.getDuccId());
         // check if process is Running, Initializing, Started, or Starting
         if (isAlive(agentManagedProcess)) {
           // Stop the process if it has been deallocated
           if (process.isDeallocated()) {
             // if agent is in stopping state, it will try to stop
             // its processes.
             if (stopping && isPreemptable) {
               logger.info(methodName, workDuccId, ">>>>>>>> Agent is stopping - process with PID:"
                       + process.getPID() + " is stopping");
               return; // agent is stopping. All processes are stopping
             } else {
               agentManagedProcess.setResourceState(ResourceState.Deallocated);
               logger.info(methodName, workDuccId,
                       "<<<<<<<< Agent Stopping Process:" + process.getDuccId() + " PID:"
                               + process.getPID() + " Reason: Ducc Deallocated the Process.");
               lifecycleController.stopProcess(agentManagedProcess);
             }
           }
           // else nothing to do. Process has been deallocated
         }
       } else { // Process not in agent's inventory
         // Add this process to the inventory so that it gets published.
         getInventoryRef().put(process.getDuccId(), process);
         if (process.isFailed()) {
           // When a process scheduling class is invalid, the AgentEventListener will
           // tag it as FAILED
           process.setReasonForStoppingProcess(
                   IDuccProcess.ReasonForStoppingProcess.InvalidSchedulingClass.name());
           ITimeWindow twr = new TimeWindow();

           process.setTimeWindowRun(twr);
           twr.setStartLong(0);
           twr.setEndLong(0);

           ITimeWindow twi = new TimeWindow();
           process.setTimeWindowInit(twi);
           twi.setStartLong(0);
           twi.setEndLong(0);

         } else if (process.isDeallocated()) {
           // process not in agent's inventory and it is marked as
           // deallocated. This can happen when an agent is restarted
           // while the rest of DUCC is running.
           // markAsStopped(process);
           process.setProcessState(ProcessState.Stopped);
         } else if (process.getResourceState().equals(ResourceState.Allocated)) {
           // check if OR thinks that this process is still running. If
           // so, this agent was restarted while the OR was running and
           // we need to mark the process as Failed.
           if (process.getProcessState().equals(ProcessState.Initializing)
                   || process.getProcessState().equals(ProcessState.Running)) {
             process.setProcessState(ProcessState.Failed);
           } else {
             // enforce presence of command line
             if (invalidCommand(commandLine)) {
               process.setProcessState(ProcessState.Failed);
               logger.info(methodName, workDuccId,
                       "Rejecting Process Start Request. Command line not provided for Process ID:"
                               + process.getDuccId());
               process.setReasonForStoppingProcess(
                       IDuccProcess.ReasonForStoppingProcess.CommandLineMissing.name());
             } else {
               if (stopping) {
                 process.setProcessState(ProcessState.Rejected);
                 logger.info(methodName, workDuccId, ">>>>>>> Agent Rejected Process:"
                         + process.getDuccId() + " Start Request - Agent is stopping");
                 return;
               }
               process.setProcessState(ProcessState.Starting);
               logger.info(methodName, workDuccId,
                       ">>>>>>> Agent Starting Process:" + process.getDuccId() + " Process State:"
                               + process.getProcessState() + " Process Resource State:"
                               + process.getResourceState());
               lifecycleController.startProcess(process, commandLine, info, workDuccId,
                       processMemoryAssignment, isPreemptable);
             }
           }
         }
       }
     } catch (Exception e) {
       logger.error(methodName, workDuccId, e);
     } finally {
       inventorySemaphore.release();
     }
   }

   public void doStartProcess(IDuccProcess process, ICommandLine commandLine, IDuccStandardInfo info,
           DuccId workDuccId, boolean isPreemptable) {
     String methodName = "doStartProcess";
     try {
       inventorySemaphore.acquire();
       if (getInventoryRef().containsKey(process.getDuccId())) {
         logger.error(methodName, null, "Rejecting Process Start Request. Process with a Ducc ID:"
                 + process.getDuccId() + " is already in agent's inventory.");
         return;
       }
       startProcess(process, commandLine, info, workDuccId, new ProcessMemoryAssignment(),
               isPreemptable);
     } catch (InterruptedException e) {
       logger.error(methodName, null, e);
     } finally {
       inventorySemaphore.release();
     }
   }

   private boolean isProcessRunning(IDuccProcess process) {
     if (process.getProcessState().equals(ProcessState.Running)
             || process.getProcessState().equals(ProcessState.Initializing)) {
       return true;
     }
     return false;
   }

   private boolean isOverSwapLimit(IDuccProcess process) {
     Iterator<ManagedProcess> it = deployedProcesses.iterator();
     while (it.hasNext()) {
       ManagedProcess deployedProcess = it.next();
       // for (ManagedProcess deployedProcess : deployedProcesses) {
       // Check if this process exceeds its alloted max swap usage
       if (deployedProcess.getDuccProcess().getDuccId().equals(process.getDuccId())
               && process.getSwapUsage() > deployedProcess.getMaxSwapThreshold()) {
         return true;
       }
     }
     return false;
   }

   private long getSwapOverLimit(IDuccProcess process) {
     long overLimit = 0;
     Iterator<ManagedProcess> it = deployedProcesses.iterator();
     while (it.hasNext()) {
       // for (ManagedProcess deployedProcess : deployedProcesses) {
       ManagedProcess deployedProcess = it.next();
       if (deployedProcess.getDuccProcess().getDuccId().equals(process.getDuccId())) {
         overLimit = deployedProcess.getMaxSwapThreshold() - process.getSwapUsage();
       }
     }
     if (overLimit < 0) {
       overLimit = 0;
     }
     return overLimit;
   }

   /**
    * Called when swap space on a node reached minimum as defined by ducc.node.min.swap.threshold in
    * ducc.properties. The agent will find the biggest (in terms of memory) process in its inventory
    * and stop it.
    */
   public void killProcessDueToLowSwapSpace(long minSwapThreshold) {
     String methodName = "killProcessDueToLowSwapSpace";
     IDuccProcess biggestProcess = null;
     try {
       inventorySemaphore.acquire();
       // find the fattest process in terms of absolute use of swap over the process limit
       for (Entry<DuccId, IDuccProcess> processEntry : getInventoryRef().entrySet()) {
         if (isProcessRunning(processEntry.getValue()) && isOverSwapLimit(processEntry.getValue())
                 && (biggestProcess == null || getSwapOverLimit(biggestProcess) < getSwapOverLimit(
                         processEntry.getValue()))) {
           biggestProcess = processEntry.getValue();
         }
       }
     } catch (InterruptedException e) {
       logger.error(methodName, null, e);
     } finally {
       inventorySemaphore.release();
     }
     if (biggestProcess != null) {
       biggestProcess.setReasonForStoppingProcess(ReasonForStoppingProcess.LowSwapSpace.toString());
       logger.info(methodName, null, "Stopping Process:" + biggestProcess.getDuccId() + " PID:"
               + biggestProcess.getPID()
               + " Due to a low swap space. Process' RSS exceeds configured swap threshold of "
               + minSwapThreshold
               + " Defined in ducc.properties. Check ducc.node.min.swap.threshold property");
       stopProcess(biggestProcess);
     }
   }

   public void interruptThreadInWaitFor(String pid) throws Exception {
     String methodName = "interruptZombieProcess";
     synchronized (monitor) {
       Iterator<ManagedProcess> it = deployedProcesses.iterator();
       while (it.hasNext()) {
         // for (ManagedProcess dProcess : deployedProcesses) {
         ManagedProcess dProcess = it.next();
         if (dProcess.getPid() != null && dProcess.getPid().equals(pid)) {
           Future<?> future = dProcess.getFuture();
           if (future != null && !future.isDone() && !future.isCancelled()) {
             future.cancel(true); // interrupt the thread blocked on waitFor()
             logger.info(methodName, dProcess.getDuccProcess().getDuccId(),
                     "Interrupted Thread - Zombie Process with PID:" + dProcess.getPid());
           }
           break;
         }
       }
     }
   }

   /**
    * Called when Agent receives request to start a process.
    *
    * @param process
    *          - IDuccProcess instance with identity (DuccId)
    * @param commandLine
    *          - fully defined command line that will be used to exec the process.
    *
    */
   public void startProcess(IDuccProcess process, ICommandLine commandLine, IDuccStandardInfo info,
           DuccId workDuccId, ProcessMemoryAssignment processMemoryAssignment,
           boolean isPreemptable) {
     String methodName = "startProcess";

     try {
       // Add process to the Agent's inventory before it is started
       getInventoryRef().put(process.getDuccId(), process);
       // enforce presence of command line
       if (invalidCommand(commandLine)) {
         process.setProcessState(ProcessState.Failed);
         logger.info(methodName, null,
                 "Rejecting Process Start Request. Command line not provided for Process ID:"
                         + process.getDuccId());
       } else if (isProcessDeallocated(process)) {
         process.setProcessState(ProcessState.Stopped);
         logger.info(methodName, null, "Rejecting Process Start Request. Process ID:"
                 + process.getDuccId() + " hava already been deallocated due to Shrink");
       } else {
         deployProcess(process, commandLine, info, workDuccId, processMemoryAssignment,
                 isPreemptable);
       }

     } catch (Exception e) {
       logger.error(methodName, null, e);
     }

   }

   public boolean isAlive(IDuccProcess invProcess) {
     return invProcess.getProcessState().equals(ProcessState.Initializing)
             || invProcess.getProcessState().equals(ProcessState.Running)
             || invProcess.getProcessState().equals(ProcessState.Stopping)
             || invProcess.getProcessState().equals(ProcessState.Starting)
             || invProcess.getProcessState().equals(ProcessState.Started);
   }

   public void doStopProcess(IDuccProcess process) {
     String methodName = "stopProcess";
     try {
       inventorySemaphore.acquire();
       stopProcess(process);
     } catch (InterruptedException e) {
       logger.error(methodName, null, e);
     } finally {
       inventorySemaphore.release();
     }

   }

   /**
    * Called when Agent receives request to stop a process.
    *
    * @param process
    *          - IDuccProcess instance with identity (DuccId)
    *
    */
   public void stopProcess(IDuccProcess process) {
     String methodName = "stopProcess";
     try {
       IDuccProcess invProcess = null;
       if ((invProcess = getInventoryRef().get(process.getDuccId())) != null
               && isAlive(invProcess)) {
         logger.info(methodName, null, "Undeploing Process with PID:" + process.getPID());
         undeployProcess(process);
       } else if (invProcess == null) { // process not in inventory
         logger.info(methodName, null,
                 "Agent received Stop request for a process which is not in the Agent's inventory. "
                         + "It looks like this Agent was killed along with its child processes. Adding stale process to the inventory. PID:"
                         + process.getPID() + " DuccId:" + process.getDuccId() + "");
         // Received a request to stop a process that this is not in the
         // current
         // inventory. Most likely this agent was killed while its
         // processes were
         // still running. Add the process to the agent's inventory so
         // that its
         // included in the next published inventory. This is done so
         // that the
         // orchestrator can cleanup its state.
         if (process.getProcessState() != ProcessState.Stopped
                 && process.getProcessState() != ProcessState.Failed
                 && process.getProcessState() != ProcessState.InitializationTimeout
                 && process.getProcessState() != ProcessState.FailedInitialization) {
           // Force the Stopped state if not already stopped or failed
           process.setProcessState(ProcessState.Stopped);
         }
         // Add stale process to the inventory. This will eventually be
         // cleaned up
         // when the PM sends purge request.
         getInventoryRef().put(process.getDuccId(), process);
       }
     } catch (Exception e) {
       logger.error(methodName, null, e);
     }
   }

   /**
    * Checks if process with a given PID has already registered memory collector Camel route. This
    * route periodically fetches resident memory of a process with a given PID. Each process
    * collector route is identified by process PID.
    *
    * @param pid
    *          - process PID also id of its route
    * @return - true if memory collector route has already been created. False, otherwise
    */
   private boolean addProcessMemoryCollector(String pid) {
     // search all camel routes for one with a given id
     for (Route route : super.getContext().getRoutes()) {
       if (route.getId().equals(pid)) {
         return false;
       }
     }
     return true;
   }

   /**
    * Remove given process from Agent's inventory
    *
    * @param process
    *          - process to purge from inventory
    * @throws Exception
    */
   public void purgeProcess(IDuccProcess process) throws Exception {
     String methodName = "purgeProcess";
     DuccId key = null;
     String pid = "";
     try {
       inventorySemaphore.acquire();
       for (Entry<DuccId, IDuccProcess> processEntry : getInventoryRef().entrySet()) {
         // Check if process with a given unique DuccId exist in the
         // local map
         if (processEntry.getKey().equals(process.getDuccId())) {
           key = processEntry.getKey();
           pid = processEntry.getValue().getPID();
           break;
         }
       }
       if (key != null) {
         getInventoryRef().remove(key);
         logger.info(methodName, null, ">>>> Agent Purged Process with PID:" + pid);
       }
     } catch (InterruptedException e) {
     } finally {
       inventorySemaphore.release();
     }
     Iterator<ManagedProcess> it = deployedProcesses.iterator();
     while (it.hasNext()) {
       // for (ManagedProcess deployedProcess : deployedProcesses) {
       ManagedProcess deployedProcess = it.next();
       // Find ManagedProcess instance the DuccProcess instance is
       // associated with
       if (deployedProcess.getDuccProcess().getDuccId().equals(process.getDuccId())) {
         processUndeploy(deployedProcess, it);
         break;
       }
     }
   }

   private boolean changeState(ProcessState state) {
     switch (state) {
       case FailedInitialization:
       case InitializationTimeout:
       case Stopped:
       case Stopping:
         return false;
       case Starting:
       case Started:
       case Initializing:
       case Running:
         return true;
       default:
         break;
     }
     return false;
   }

   /**
    * Called when a service wrapper sends status update
    *
    * @param duccEvent
    *          - Ducc event object
    *
    * @throws Exception
    */

   public void updateProcessStatus(ProcessStateUpdateDuccEvent duccEvent) throws Exception {
     String methodName = "updateProcessStatus";

     try {
       inventorySemaphore.acquire();
       for (Entry<DuccId, IDuccProcess> processEntry : getInventoryRef().entrySet()) {
         // Check if process with a given unique DuccId exist in the
         // local map
         if (processEntry.getKey().getUnique().equals(duccEvent.getDuccProcessId())) {
           // found it. Update pid and state of the process
           if (duccEvent.getPid() != null && processEntry.getValue().getPID() == null) {
             processEntry.getValue().setPID(duccEvent.getPid());
           }

           if (duccEvent.getProcessJmxUrl() != null
                   && processEntry.getValue().getProcessJmxUrl() == null) {
             processEntry.getValue().setProcessJmxUrl(duccEvent.getProcessJmxUrl());
           }
           ITimeWindow tw = processEntry.getValue().getTimeWindowInit();
           if (tw != null) {
             if (!duccEvent.getState().equals(ProcessState.Initializing)) {
               // Mark the time the process ended initialization. It also
               // covers a case when the process terminates while initializing
               tw.setEnd(TimeStamp.getCurrentMillis());
               if (duccEvent.getState().equals(ProcessState.Running)) {
                 ITimeWindow twr = new TimeWindow();
                 String millis;
                 millis = TimeStamp.getCurrentMillis();
                 // Mark the time the process started running
                 processEntry.getValue().setTimeWindowRun(twr);
                 twr.setStart(millis);
               }
             }
           } else {
             logger.info(methodName, null,
                     "++++++++++++ Agent Init TimeWindow not available - tw==null");
           }
           ManagedProcess deployedProcess = null;
           synchronized (monitor) {
             Iterator<ManagedProcess> it = deployedProcesses.iterator();
             while (it.hasNext()) {
               // Find ManagedProcess instance the DuccProcess
               // instance is associated with
               ManagedProcess dProcess = it.next();
               if (dProcess.getDuccProcess().getDuccId().getUnique()
                       .equals(duccEvent.getDuccProcessId())) {
                 deployedProcess = dProcess;
                 break;
               }
             }
           }
           if (processEntry.getValue().getProcessState() != ProcessState.Running
                   && duccEvent.getState().equals(ProcessState.Running) && deployedProcess != null) {
             // cancel process initialization timer.
             deployedProcess.stopInitializationTimer();
           }

           logger.info(methodName, null,
                   ">>>> Agent Handling Process Update - Ducc Id: "
                           + processEntry.getValue().getDuccId() + " PID:"
                           + processEntry.getValue().getPID() + " Status:" + duccEvent.getState()
                           + " Deallocated:" + processEntry.getValue().isDeallocated());
           if (deployedProcess != null && deployedProcess.getSocketEndpoint() == null
                   && duccEvent.getServiceEdnpoint() != null) {
             deployedProcess.setSocketEndpoint(duccEvent.getServiceEdnpoint());
           }

           // This is a delayed stop. Previously a request to stop the
           // process was received
           // but the PID was not available yet. Instead a flag was set
           // to initiate a
           // stop after the process reports the PID.
           if (deployedProcess != null && deployedProcess.killAfterLaunch()) {
             logger.info(methodName, null,
                     ">>>> Process Ducc Id:" + processEntry.getValue().getDuccId()
                             + " Was Previously Tagged for Kill While It Was Starting");
             undeployProcess(processEntry.getValue());
           } else if (deployedProcess != null && deployedProcess.doKill() && deployedProcess
                   .getDuccProcess().getProcessState().equals(ProcessState.Stopped)) {
             deployedProcess.getDuccProcess()
                     .setReasonForStoppingProcess(ReasonForStoppingProcess.KilledByDucc.toString());
           } else if (deployedProcess != null && (deployedProcess.doKill()
                   || deployedProcess.getDuccProcess().getProcessState().equals(ProcessState.Failed)
                   || deployedProcess.getDuccProcess().getProcessState()
                           .equals(ProcessState.Killed))) {
             // The process has already stopped, but managed to send
             // the last update before dying. Ignore the update
             return;
           } else if (changeState(processEntry.getValue().getProcessState())) {
             logger.info(methodName, null, "=============== PID:" + processEntry.getValue().getPID()
                     + " Changing State - current state:" + processEntry.getValue().getProcessState()
                     + " New State:" + duccEvent.getState());
             processEntry.getValue().setProcessState(duccEvent.getState());
             DuccEventDispatcher dispatcher = configurationFactory
                     .getORDispatcher(super.getContext());
             try {
               DefaultNodeInventoryProcessor processor = configurationFactory
                       .nodeInventoryProcessor(this);
               Map<DuccId, IDuccProcess> inventoryCopy = (Map<DuccId, IDuccProcess>) SerializationUtils
                       .clone((ConcurrentHashMap<DuccId, IDuccProcess>) inventory);

               processor.dispatchInventoryUpdate(dispatcher,
                       configurationFactory.getInventoryUpdateEndpoint(), inventoryCopy);
               logger.info(methodName, null, "Sent Node Inventory Update to the OR - process PID:"
                       + processEntry.getValue().getPID());

             } catch (Exception e) {
               logger.warn("", null, e);
             }

             // if the process is Stopping, it must have hit an error threshold
           }
           // Check if MemoryCollector should be created for this
           // process. It collects
           // resident memory of the process at regular intervals.
           // Should only be added
           // once for each process. This route will have its id set to
           // process PID.
           if (addProcessMemoryCollector(duccEvent.getPid())
                   && (duccEvent.getState().equals(ProcessState.Initializing)
                           || duccEvent.getState().equals(ProcessState.Running))) {
             if (duccEvent.getState().equals(ProcessState.Running)) {
               if (processEntry.getValue().getUimaPipelineComponents() != null
                       && processEntry.getValue().getUimaPipelineComponents().size() > 0) {
                 processEntry.getValue().getUimaPipelineComponents().clear();
                 if (duccEvent.getUimaPipeline() != null) {
                   duccEvent.getUimaPipeline().clear();
                 }
               }
             }

           } else if (duccEvent.getState().equals(ProcessState.Stopped)
                   || duccEvent.getState().equals(ProcessState.Failed)
                   || duccEvent.getState().equals(ProcessState.Killed)) {
             if (deployedProcess.getMetricsProcessor() != null) {
               deployedProcess.getMetricsProcessor().close(); // close open fds (stat and statm
                                                              // files)
             }
             logger.info(methodName, null,
                     "----------- Agent Stopped ProcessMemoryUsagePollingRouter for Process:"
                             + duccEvent.getPid());
           } else if (duccEvent.getState().equals(ProcessState.FailedInitialization)) {
             logger.info(methodName, null,
                     ">>>> Agent Handling Process FailedInitialization. PID:" + duccEvent.getPid());
             deployedProcess.getDuccProcess().setReasonForStoppingProcess(
                     ReasonForStoppingProcess.FailedInitialization.toString());
             deployedProcess.getDuccProcess().setProcessState(ProcessState.Stopping);
             deployedProcess.setStopping();

             deployedProcess.kill();
             logger.info(methodName, null, ">>>> Agent Handling Process FailedInitialization. PID:"
                     + duccEvent.getPid() + " Killing Process");

             undeployProcess(processEntry.getValue());

           } else if (duccEvent.getState().equals(ProcessState.InitializationTimeout)) {
             deployedProcess.getDuccProcess().setReasonForStoppingProcess(
                     ReasonForStoppingProcess.InitializationTimeout.toString());
             deployedProcess.getDuccProcess().setProcessState(ProcessState.Stopping);
             deployedProcess.setStopping();

             // Mark process for death. Doesnt actually kill the process

             deployedProcess.kill();
             logger.info(methodName, null, ">>>> Agent Handling Process InitializationTimeout. PID:"
                     + duccEvent.getPid() + " Killing Process");

             undeployProcess(processEntry.getValue());

           } else if (duccEvent.getState().equals(ProcessState.Stopping)) {
             if (duccEvent.getMessage() != null && duccEvent.getMessage()
                     .equals(ReasonForStoppingProcess.ExceededErrorThreshold.toString())) {
               processEntry.getValue().setReasonForStoppingProcess(
                       ReasonForStoppingProcess.ExceededErrorThreshold.toString());
             }
             if (!deployedProcess.getDuccProcess().getProcessState().equals(ProcessState.Stopped)
                     && !deployedProcess.getDuccProcess().getProcessState()
                             .equals(ProcessState.Stopping)) {
               deployedProcess.getDuccProcess().setProcessState(ProcessState.Stopping);
               deployedProcess.setStopping();
             }
           }
           if (duccEvent.getUimaPipeline() != null) {
             StringBuffer buffer = new StringBuffer("\t\tUima Pipeline -");
             for (IUimaPipelineAEComponent uimaAeState : duccEvent.getUimaPipeline()) {
               buffer.append("\n\t\tAE:").append(uimaAeState.getAeName()).append(" state:")
                       .append(uimaAeState.getAeState()).append(" InitTime:")
                       .append(uimaAeState.getInitializationTime() / 1000).append(" secs. Thread:")
                       .append(uimaAeState.getAeThreadId());
             }
             logger.info(methodName, null, buffer.toString());
             ((DuccProcess) processEntry.getValue())
                     .setUimaPipelineComponents(duccEvent.getUimaPipeline());
           }
           return; // found it. Done
         }
       }
     } catch (InterruptedException e) {
     } finally {
       inventorySemaphore.release();
     }
   }

   /**
    * Deploys process using supplied command line
    *
    * @param process
    *          - Process with identity (DuccId)
    * @param commandLine
    *          - fully defined command line that will be used to exec the process.
    */
   private void deployProcess(IDuccProcess process, ICommandLine commandLine, IDuccStandardInfo info,
           DuccId workDuccId, ProcessMemoryAssignment processMemoryAssignment, boolean preemptable) {
     String methodName = "deployProcess";
     synchronized (monitor) {
       boolean deployProcess = true;
       Iterator<ManagedProcess> it = deployedProcesses.iterator();
       while (it.hasNext()) {
         // for (ManagedProcess deployedProcess : deployedProcesses) {
         ManagedProcess deployedProcess = it.next();
         // ignore duplicate start request for the same process
         if (deployedProcess.getDuccId().equals(process.getDuccId())) {
           deployProcess = false;
           break;
         }
       }
       if (deployProcess) {
         try {
           logger.info(methodName, workDuccId,
                   "Agent [" + getIdentity().getIp() + "] Deploying Process - DuccID:"
                           + process.getDuccId().getFriendly() + " Unique DuccID:"
                           + process.getDuccId().getUnique() + " Node Assignment:"
                           + process.getNodeIdentity().getIp() + " Process Memory Assignment:"
                           + processMemoryAssignment + " MBs");
           TimeWindow tw = new TimeWindow();
           tw.setStart(TimeStamp.getCurrentMillis());
           tw.setEnd(null);
           process.setTimeWindowInit(tw);
           ManagedProcess managedProcess = new ManagedProcess(process, commandLine, this, logger,
                   processMemoryAssignment, preemptable);
           managedProcess.setProcessInfo(info);
           managedProcess.setWorkDuccId(workDuccId);

           // enrich process spec with unique ducc id which will be
           // used to correlate message
           // exchanges
           // between the agent and launched process

           ManagedProcess deployedProcess = launcher.launchProcess(this, getIdentity(), process,
                   commandLine, this, managedProcess);
           processDeploy(deployedProcess);
         } catch (Exception e) {
           logger.error(methodName, null, e);
         }
       } else {
         logger.info(methodName, workDuccId,
                 "Ignoring duplicate request to start process - DuccID:"
                         + process.getDuccId().getFriendly() + " Unique DuccID:"
                         + process.getDuccId().getUnique());
       }
     }
   }

   class AgentStreamConsumer implements Runnable {
     private InputStream theStream;

     AgentStreamConsumer(InputStream is) {
       theStream = is;
     }

     public void run() {
       String methodName = "AgentStreamConsumer.run";

       BufferedReader bufferedReader = null;
       try {
         bufferedReader = new BufferedReader(new InputStreamReader(theStream));
         String line = null;
         while ((line = bufferedReader.readLine()) != null) {
           StringBuffer outputBuffer = new StringBuffer();
           outputBuffer.append(line + "\n");
         }
       } catch (Throwable t) {
         logger.warn(methodName, null, t);
         t.printStackTrace();
       } finally {
         try {
           bufferedReader.close();
         } catch (Exception e) {
         }
       }
     }
   }

   enum SIGNAL {
     SIGTERM("-15"), SIGKILL("-9");

     String signal = "";

     SIGNAL(String kind) {
       signal = kind;
     }

     public String get() {
       return signal;
     }
   };

   class ProcessRunner implements Runnable {
     ManagedProcess deployedProcess;

     public ProcessRunner(final ManagedProcess deployedProcess) {// final String pid, SIGNAL signal )
                                                                 // {
       this.deployedProcess = deployedProcess;
     }

     public void run() {
       stopProcess(deployedProcess.getDuccProcess());
     }
   }

   private boolean runnable(ManagedProcess process) {
     return (process.getDuccProcess().getProcessState().equals(ProcessState.Initializing)
             || process.getDuccProcess().getProcessState().equals(ProcessState.Starting)
             || process.getDuccProcess().getProcessState().equals(ProcessState.Started)
             || process.getDuccProcess().getProcessState().equals(ProcessState.Running));
   }

   /**
    * This method is called when an agent receives a STOP request. It sends SIGTERM to all
    * non-preemptable child processes and starts a timer. If the timer pops and child processes are
    * still running, the agent takes itself out via halt()
    */
   private boolean stopChildProcesses(boolean quiesceMode) {
     String methodName = "stopNow";
     boolean wait = false;
     try {
       Iterator<ManagedProcess> it = deployedProcesses.iterator();
       while (it.hasNext()) {
         ManagedProcess deployedProcess = it.next();
         String pid = deployedProcess.getDuccProcess().getPID();
         logger.info(methodName, null, "....Process:" + pid + " is JD=" + deployedProcess.isJd()
                 + " Preemptable:" + deployedProcess.isPreemptable());
         // dont send SIGTERM to non-preemptable processes in quiesce mode
         if ((quiesceMode && !deployedProcess.isPreemptable()) || deployedProcess.isStopping()
                 || pid == null || pid.trim().length() == 0 || !runnable(deployedProcess)) {
           continue;
         }

         logger.info(methodName, null,
                 "....Stopping Process - DuccId:" + deployedProcess.getDuccProcess().getDuccId()
                         + " PID:" + pid + " Sending SIGTERM Process State:"
                         + deployedProcess.getDuccProcess().getProcessState().toString()
                         + " Process Type:" + deployedProcess.getDuccProcess().getProcessType()
                         + " Uima AS:" + deployedProcess.isUimaAs() + " Preemtable:"
                         + deployedProcess.isPreemptable());
         wait = true;
         deployedProcess.setStopPriority(StopPriority.DONT_WAIT);
         // Stop each child process in its own thread to parallelize SIGTERM requests
         ExecutorService executor = Executors.newSingleThreadExecutor();
         executor.execute(new ProcessRunner(deployedProcess));

       }

     } catch (Exception e) {
       logger.warn(methodName, null, e);
     }
     return wait;
   }

   private void killChildProcesses(boolean killOnlyUimaAs, boolean quiesce) {
     String methodName = "killChildProcesses";

     try {
       if (useCgroups) {
         logger.info(methodName, null, "CgroupsManager.cleanup() before ");
         if (killOnlyUimaAs) {
           Set<String> pidsToKill = new HashSet<>();
           Iterator<ManagedProcess> it = deployedProcesses.iterator();
           while (it.hasNext()) {
             ManagedProcess p = it.next();
             if (!p.isPreemptable() && p.getPid() != null && p.isUimaAs()) {
               pidsToKill.add(p.getPid());
             }
           }
           if (!pidsToKill.isEmpty()) {
             logger.info(methodName, null, ">>>>>>>> Found " + pidsToKill.size()
                     + " UIMA-AS processes still running - killing all non-preemptables via kill -9");
             // Since SIGTERM may not be enough to take down a process, use cgroups to find
             // any UIMA-AS process still standing and do hard kill
             cgroupsManager.cleanupPids(pidsToKill);

           }

         } else {
           Set<String> pidsToKill = new HashSet<>();
           Iterator<ManagedProcess> it = deployedProcesses.iterator();
           // Since SIGTERM may not be enough to take down a process, use cgroups to find
           // any process still standing and do hard kill
           while (it.hasNext()) {
             ManagedProcess p = it.next();
             if ((!quiesce && !p.isPreemptable()) && p.getPid() != null && p.isUimaAs()) {
               pidsToKill.add(p.getPid());
             }
           }
           // cgroupsManager.cleanup();
           cgroupsManager.cleanupPids(pidsToKill);

         }
         logger.info(methodName, null, "CgroupsManager.cleanup() after ");
       } else {
         Iterator<ManagedProcess> it = deployedProcesses.iterator();
         while (it.hasNext()) {
           ManagedProcess deployedProcess = it.next();

           String pid = deployedProcess.getDuccProcess().getPID();
           if ((quiesce && !deployedProcess.isPreemptable()) || pid == null
                   || pid.trim().length() == 0 || !runnable(deployedProcess)) {
             continue;
           }
           logger.info(methodName, null,
                   "....Stopping Process - DuccId:" + deployedProcess.getDuccProcess().getDuccId()
                           + " PID:" + pid + " Sending SIGKILL Process State:"
                           + deployedProcess.getDuccProcess().getProcessState().toString());
           ICommandLine cmdLine;
           if (Utils.isWindows()) {
             cmdLine = new NonJavaCommandLine("taskkill");
             cmdLine.addArgument("/PID");
           } else {
             cmdLine = new NonJavaCommandLine("/bin/kill");
             cmdLine.addArgument("-9");
           }
           cmdLine.addArgument(pid);

           deployedProcess.setStopping();
           deployedProcess.setStopPriority(StopPriority.DONT_WAIT);

           launcher.launchProcess(this, getIdentity(), deployedProcess.getDuccProcess(), cmdLine,
                   this, deployedProcess);
         }

       }
     } catch (Exception e) {
       logger.warn(methodName, null, e);
     }

   }

   private void handleSigTermTimeout(ManagedProcess deployedProcess) {
     String methodName = "handleSigTermTimeout";
     if (!deployedProcess.getDuccProcess().getProcessState().equals(ProcessState.Stopped)) {
       logger.info(methodName, deployedProcess.getDuccId(),
               "------------ Agent Timed-out Waiting for Process with PID:"
                       + deployedProcess.getDuccProcess().getPID() + " to Stop. Process State:"
                       + deployedProcess.getDuccProcess().getProcessState()
                       + " .Process did not stop in allotted time of "
                       + maxTimeToWaitForProcessToStop + " millis");
       logger.info(methodName, deployedProcess.getDuccId(),
               ">>>>>>>>>>>>>>> Killing Process:" + deployedProcess.getDuccProcess().getPID()
                       + " .Process State:" + deployedProcess.getDuccProcess().getProcessState());
     }

     ICommand sigKillCommand = new SigKillCommand(deployedProcess, logger);
     launcher.launchOSCommand(sigKillCommand);
   }

   /**
    * Kills a given process
    *
    * @param process
    *          - process to kill
    */
   private void undeployProcess(IDuccProcess process) {
     String methodName = "undeployProcess";

     synchronized (monitor) {

       ManagedProcess deployedProcess = getDeployedProcess(process.getDuccId());

       // Given process does not exist in agent's inventory
       if (Objects.isNull(deployedProcess)) {
         logger.info(methodName, null,
                 ".... Process - DuccId:" + process.getDuccId() + " PID:" + process.getPID()
                         + " Not in Agent's inventory. Adding to the inventory with state=Stopped");
         process.setProcessState(ProcessState.Stopped);
         inventory.put(process.getDuccId(), process);
         processDeploy(new ManagedProcess(process, null, this, logger, new ProcessMemoryAssignment(),
                 true));
         return;
       }

       if (Objects.isNull(deployedProcess.getPid())) {
         if (!deployedProcess.getDuccProcess().getProcessState().equals(ProcessState.Stopped)) {
           // process not reported its PID yet when the process
           // reports its PID, check if it should be killed.
           logger.info(methodName, deployedProcess.getDuccId(), ".... Process - Ducc ID:"
                   + deployedProcess.getDuccId()
                   + " Has Not Started Yet. PID Not Available. Tagging Process For Kill When It Reports Status");
           deployedProcess.killAfterLaunch(true);
         }
         return;
       }

       ICommand sigTermCommand = new SigTermCommand(deployedProcess, logger);

       // Mark the process as stopping. When the process exits,
       // the agent can determine if the process died on its own
       // (due to say, user code problem) or if it died
       // due to Agent initiated stop.
       deployedProcess.setStopping();

       try {
         logger.info(methodName, deployedProcess.getDuccId(),
                 "------------ Agent Starting Killer Timer Task For Process with PID:"
                         + deployedProcess.getDuccProcess().getPID() + " Process State: "
                         + deployedProcess.getDuccProcess().getProcessState());
         // launch SIGTERM
         launcher.launchOSCommand(sigTermCommand);
         // fetch deployed process Future object which
         // will be used to determine if its still
         // running
         Future<?> future = deployedProcess.getFuture();
         // future.get() blocks until process stops or timeout occurs
         future.get(maxTimeToWaitForProcessToStop, TimeUnit.MILLISECONDS);
         logger.info(methodName, deployedProcess.getDuccId(), ">>>>>>>>>>>>> Process with PID:"
                 + deployedProcess.getDuccProcess().getPID() + " Terminated");
       } catch (TimeoutException e) {
         logger.info(methodName, deployedProcess.getDuccId(),
                 ">>>>>>>>> Timed Out Waiting For Process To Terminate After SIGTERM Was Sent");
         handleSigTermTimeout(deployedProcess);
       } catch (Exception e) {
         logger.error(methodName, deployedProcess.getDuccId(), e);
       }

       try {
         monitor.wait(500);
       } catch (InterruptedException ee) {

       }

       // check if defunct process
       if (isProcessRunning(deployedProcess.getDuccProcess())) {
         // wait for a short while before running DefunctProcessDetector
         // Sometimes when a process is killed via kill -9 it shows as
         // defunct in ps output. Not sure why this is so.

         // spin a thread where we check if the process is defunct. If true,
         // the process state is changed to Stopped and reason set to 'defunct'.
         // Next inventory publication will include this new state and the OR
         // can terminate a job.
         defunctDetectorExecutor.execute(new DefunctProcessDetector(deployedProcess, logger));
       }
       logger.info(methodName, deployedProcess.getDuccId(), "Inventory size:" + inventory.size()
               + " deployedProcesses size:" + deployedProcesses.size());
     }
   }

   public NodeIdentity getIdentity() {
     return nodeIdentity;
   }

   /**
    * Called when a process exits.
    */
   public void onProcessExit(IDuccProcess process) {
     String methodName = "onProcessExit";
     if (process == null) {
       return;
     }
     try {
       ProcessStateUpdate processStateUpdate = new ProcessStateUpdate(process.getProcessState(),
               process.getPID(), process.getDuccId().getUnique());
       ProcessStateUpdateDuccEvent event = new ProcessStateUpdateDuccEvent(processStateUpdate);
       // cleanup Camel route associated with a process that just stopped
       if (process.getPID() != null && super.getContext().getRoute(process.getPID()) != null) {
         try {
           // stop collecting process stats from /proc/<pid>/statm
           super.getContext().stopRoute(process.getPID());
         } catch (Exception e) {
           logger.error(methodName, null,
                   "....Unable to stop Camel route for PID:" + process.getPID());
         }
         // remove route from context, otherwise the routes accumulate over time causing memory leak
         super.getContext().removeRoute(process.getPID());
         StringBuilder sb = new StringBuilder("\n");
         logger.info(methodName, null,
                 "Removed Camel Route from Context for PID:" + process.getPID());

         for (Route route : super.getContext().getRoutes()) {
           sb.append("Camel Context - RouteId:" + route.getId() + "\n");
         }
         logger.info(methodName, null, sb.toString());
       }
       updateProcessStatus(event);
     } catch (Exception e) {
       logger.error(methodName, null, e);
     } finally {

     }
   }

   public void onJPInitTimeout(IDuccProcess process, long timeout) {
     String methodName = "onJPInitTimeout";
     try {
       System.out.println("--------- Agent Timedout While Waiting For JP (PID:" + process.getPID()
               + ") to initialize. The JP exceeded configured timeout of " + timeout / (60 * 1000)
               + " minutes");
       ProcessStateUpdate processStateUpdate = new ProcessStateUpdate(
               ProcessState.InitializationTimeout, process.getPID(),
               process.getDuccId().getUnique());
       ProcessStateUpdateDuccEvent event = new ProcessStateUpdateDuccEvent(processStateUpdate);
       updateProcessStatus(event);
     } catch (Exception e) {
       logger.error(methodName, null, e);
     }
   }

   public void shutdown(String reason) {
     String methodName = "shutdown";
     Iterator<ManagedProcess> it = deployedProcesses.iterator();
     while (it.hasNext()) {
       ManagedProcess deployedProcess = it.next();
       try {

         undeployProcess(deployedProcess.getDuccProcess());
       } catch (Exception e) {
         logger.error(methodName, null, e);
       }
     }
   }

   public static void lock() throws Exception {
     agentLock.acquire();
   }

   public static void unlock() throws Exception {
     agentLock.release();
   }

   public boolean isManagedProcess(Set<NodeUsersCollector.ProcessInfo> processList,
           NodeUsersCollector.ProcessInfo cpi) {
     synchronized (monitor) {
       Iterator<ManagedProcess> it = deployedProcesses.iterator();
       while (it.hasNext()) {
         ManagedProcess deployedProcess = it.next();
         if (deployedProcess.getDuccProcess() != null) {
           // Check if process has been deployed but has not yet
           // reported its PID.
           // This is normal. It takes a bit of time until the JP
           // reports
           // its PID to the Agent. If there is at least one process in
           // Agent
           // deploy list with no PID we assume it is the one.
           String dppid = deployedProcess.getDuccProcess().getPID();
           if (dppid == null || dppid.equals(String.valueOf(cpi.getPid()))) {
             return true;
           }
         }
       }
       for (NodeUsersCollector.ProcessInfo pi : processList) {
         if (pi.getPid() == cpi.getPPid() && pi.getChildren().size() > 0) { // is
           // the current process a child of another java
           // Process?
           return isManagedProcess(pi.getChildren(), pi);
         }
       }
     }
     return false;
   }

   public boolean isRogueProcess(String uid, Set<NodeUsersCollector.ProcessInfo> processList,
           NodeUsersCollector.ProcessInfo cpi) throws Exception {

     synchronized (monitor) {
       // if cgroups are enabled, check if a given PID (cpi) exists in any of
       // the containers. If so, the process is not rogue.
       if (useCgroups) {
         if (cgroupsManager.isPidInCGroup(String.valueOf(cpi.getPid()))) {
           return false;
         }
       }
       // Agent adds a process to its inventory before launching it. So it
       // is
       // possible that the inventory contains a process with no PID. If
       // there
       // is such process in the inventory we cannot determine that a given
       // pid is rogue yet. Eventually, the launched process reports its
       // PID
       boolean foundDeployedProcessWithNoPID = false;
       Iterator<ManagedProcess> it = deployedProcesses.iterator();
       while (it.hasNext()) {
         ManagedProcess deployedProcess = it.next();
         if (deployedProcess.getDuccProcess() != null) {
           // Check if process has been deployed but has not yet
           // reported its PID.
           // This is normal. It takes a bit of time until the JP
           // reports
           // its PID to the Agent. If there is at least one process in
           // Agent
           // deploy list with no PID we assume it is the one.
           if (deployedProcess.getDuccProcess().getPID() == null) {
             foundDeployedProcessWithNoPID = true;
             break;
           }
           String dppid = deployedProcess.getDuccProcess().getPID();
           // process in inventory, not rogue
           if (dppid != null && dppid.equals(String.valueOf(cpi.getPid()))) {
             return false;
           }
         }
       }
       // not found
       if (foundDeployedProcessWithNoPID) {
         return false;
       } else if (cpi.getPPid() == 1) { // Any process owned by init is rogue
         // interrupt agent's thread blocking in waitFor() awaiting process termination.
         // This process is a zombie and there is no need to waste the thread.
         interruptThreadInWaitFor(String.valueOf(cpi.getPid()));
         return true;
       } else {
         return isParentProcessRogue(processList, cpi);
       }
     }
   }

   private boolean isParentProcessRogue(Set<NodeUsersCollector.ProcessInfo> processList,
           NodeUsersCollector.ProcessInfo cpi) {
     // boolean found = false;
     for (NodeUsersCollector.ProcessInfo pi : processList) {
       if (pi.getPid() == cpi.getPPid()) { // is the current process a
         if (pi.isRogue()) { // if parent is rogue, a child is rogue as
           // well
           return true;
         }
         return false;
       } else {
         if (pi.getChildren().size() > 0) {
           return isParentProcessRogue(pi.getChildren(), cpi);
         }
       }
     }
     return true;

   }

   /**
    * Process resident memory collector routes. Collects resident memory at fixed interval from the
    * OS.
    *
    */
   public class ProcessMemoryUsageRoute extends RouteBuilder {
     private NodeAgent agent;

     private IDuccProcess process;

     private ManagedProcess managedProcess;

     public ProcessMemoryUsageRoute(NodeAgent agent, IDuccProcess process,
             ManagedProcess managedProcess) {
       this.process = process;
       this.managedProcess = managedProcess;
       this.agent = agent;
     }

     public void configure() throws Exception {
       Processor nmp = configurationFactory.processMetricsProcessor(agent, process, managedProcess);
       int fixedRate = configurationFactory.getNodeInventoryPublishDelay();
       from("timer:processMemPollingTimer?fixedRate=true&delay=100&period=" + fixedRate)
               .routeId(process.getPID()).autoStartup(true).process(nmp);
     }
   }

   @Override
   public boolean isStopping() {
     return stopping;
   }

   @Override
   public void quiesceAndStop() throws Exception {
     stop(true, -1);
   }

   private void stop(boolean quiesce, long waitTimeInSecs) throws Exception {
     synchronized (stopLock) {
       logger.info("stop", null, "Agent stop() - quiesce:" + quiesce);
       if (stopping) {
         return;
       }
       stopping = true;
       stateChange(EventType.SHUTDOWN);
       // Dispatch SIGTERM to all processes. If this is quiesce mode we dont try to stop
       // non-preemptable processes
       boolean wait = stopChildProcesses(quiesce);
       if (quiesce) {
         logger.info("stop", null, "Agent stopping managed processes");
         long waitTime = 60; // default
         try {
           waitTime = Long.valueOf(configurationFactory.processStopTimeout);
           // Normalize. The configurationFactory.processStopTimeout from
           // ducc.properties is in millis. The code below expects secs.
           waitTime = (waitTime / 1000);
         } catch (Exception e) {
         }
         // Version 2.10.2 of UIMA-AS is not supporting quiesce and stop
         // so we need to implement wait than kill -9 strategy.
         waitForChildProcessesToTerminateAndKill(wait, waitTime, true, quiesce);
         logger.info("stop", null,
                 ">>>>>>>>>>>> stop() waitForChildProcessesToTerminateAndKill() completed");
         // wait for JD processes to terminate. Return only when all non-preemptables
         // terminate.
         waitForChildProcessesToTerminate(false);
         logger.info("stop", null,
                 ">>>>>>>>>>>> stop() waitForChildProcessesToTerminate() completed");
       } else {
         logger.info("stop", null, "Agent stopping managed processes with reaper delay of "
                 + waitTimeInSecs + " secs");

         // wait for 60 secs and sends SIGKILL to any process still standing
         waitForChildProcessesToTerminateAndKill(wait, waitTimeInSecs, false, quiesce);
       }
       // Send an empty process map as the final inventory
       DuccEvent duccEvent = new NodeInventoryUpdateDuccEvent(inventory, getLastORSequence(),
               getIdentity());
       ORDispatcher.dispatch(duccEvent);
       logger.info("stop", null, "Agent published final inventory");

       logger.info("stop", null, "Stopping Publishing Metrics and Inventory");

       configurationFactory.stop();
       logger.info("stop", null, "Reaper thread finished - calling super.stop()");
       super.stop();
     }

   }

   @Override
   public void stop() throws Exception {
     // not quiesce, 60=default timeout before kill -9 is used
     stop(false, 60);
   }

   private void waitForChildProcessesToTerminateAndKill(boolean wait, long waitTimeInSecs,
           boolean killJustUimaAs, boolean quiesce) throws Exception {
     if (wait && !deployedProcesses.isEmpty()) {
       logger.info("waitForChildProcessesToTerminateAndKill", null,
               "Agent Sent SIGTERM to ALL Non-Preemptable Child Processes - Number of Deployed Processes:"
                       + deployedProcesses.size());

       Timer timer = new Timer(true);
       logger.info("waitForChildProcessesToTerminateAndKill", null, "Waiting", waitTimeInSecs,
               " secs before sending kill -9 to all ***non-preemptable*** child processes still running");
       CountDownLatch completionLatch = new CountDownLatch(1);
       // start a timer task which when triggered kills processes via kill -9
       timer.schedule(new KillTimerTask(completionLatch, killJustUimaAs, quiesce),
               waitTimeInSecs * 1000);

       // block this thread until killer task finishes its work
       completionLatch.await();

     }
     stopLock.wait(1000);
     logger.info("waitForChildProcessesToTerminateAndKill", null, "Done");

   }

   private void waitForChildProcessesToTerminate(boolean quiesceMode) throws Exception {
     logger.info("waitForChildProcessesToTerminate", null,
             "Agent Sent SIGTERM to Child Processes - Waiting for them to Quiesce - Number of Deployed Processes:"
                     + deployedProcesses.size());
     Iterator<ManagedProcess> it = deployedProcesses.iterator();
     while (it.hasNext()) {
       ManagedProcess p = it.next();
       // dont wait for JDs to stop. In quiesce mode we keep them
       // running until all JPs terminate and only than we stop them
       // if ( quiesceMode && ( p.isJd() || p.isUimaAs() ) ) {
       // in quiesce mode, skip UIMA-AS processes since currently
       // there is no support for quiesce there. So only wait for
       // POPs (JD), UIMA based JPs, and Services
       if (quiesceMode && p.isUimaAs()) {
         continue;
       }
       // block waiting for process to terminate
       p.getFuture().get();

     }
   }

   public Future<?> getDeployedJPFuture(IDuccId duccId) {
     Iterator<ManagedProcess> it = deployedProcesses.iterator();
     while (it.hasNext()) {
       // for (ManagedProcess deployedProcess : deployedProcesses) {
       ManagedProcess deployedProcess = it.next();
       // ignore duplicate start request for the same process
       if (deployedProcess.getDuccId().equals(duccId)) {
         return deployedProcess.getFuture();
       }
     }
     return null;
   }

   public ManagedProcess getDeployedProcess(IDuccId duccId) {
     Iterator<ManagedProcess> it = deployedProcesses.iterator();
     while (it.hasNext()) {
       ManagedProcess deployedProcess = it.next();
       // ignore duplicate start request for the same process
       if (deployedProcess.getDuccId().equals(duccId)) {
         return deployedProcess;
       }
     }
     return null;
   }

   /**
    * Copies reservations sent by the PM. It copies reservations associated with this node.
    *
    * @param reserves
    *          - list of ALL reservations
    * @throws Exception
    */
   public void setReservations(List<DuccUserReservation> reserves) throws Exception {
     try {
       reservationsSemaphore.acquire();
       if (reserves != null) {
         // clear old entries
         reservations.clear();
         // Only copy reservations for this node
         IDuccReservationMap reserveMap = new DuccReservationMap();
         for (DuccUserReservation r : reserves) {
           reserveMap.clear();
           for (Map.Entry<DuccId, IDuccReservation> entry : r.getUserReservations().entrySet()) {
             if (Utils.isThisNode(getIdentity().getIp(),
                     entry.getValue().getNodeIdentity().getIp())) {
               reserveMap.addReservation(entry.getValue());
             }
           }
           if (reserveMap.getMap().size() > 0) {
             DuccUserReservation reserve = new DuccUserReservation(r.getUserId(), r.getReserveID(),
                     reserveMap);
             reservations.add(reserve);
           }
         }
       }

       // this.reservations = reservations;
       logger.debug("setReservations", null,
               "+++++++++++ Copied User Reservations - List Size:" + reservations.size());
     } catch (InterruptedException e) {
     } finally {
       reservationsSemaphore.release();
     }
   }

   public List<DuccId> getUserReservations(String uid) {
     List<DuccId> reservationIds = new ArrayList<DuccId>();
     try {
       reservationsSemaphore.acquire();
       if (reservations != null) {
         for (DuccUserReservation r : reservations) {
           if (r.getUserId().equals(uid)) {
             for (Map.Entry<DuccId, IDuccReservation> entry : r.getUserReservations().entrySet()) {
               reservationIds.add(entry.getValue().getDuccId());
             }
             break;
           }
         }
       }
     } catch (InterruptedException e) {
     } finally {
       reservationsSemaphore.release();
     }
     return reservationIds;
   }

   public void copyAllUserReservations(TreeMap<String, NodeUsersInfo> map) {
     try {
       reservationsSemaphore.acquire();
       if (reservations != null) {
         logger.debug("copyAllUserReservations", null,
                 "+++++++++++ Copying User Reservations - List Size:" + reservations.size());
         for (DuccUserReservation r : reservations) {
           if ("System".equals(r.getUserId())) {
             continue;
           }
           NodeUsersInfo nui = null;
           if (map.containsKey(r.getUserId())) {
             nui = map.get(r.getUserId());
           } else {
             nui = new NodeUsersInfo(r.getUserId());
             map.put(r.getUserId(), nui);
           }
           nui.addReservation(r.getReserveID());
         }
       } else {
         logger.debug("copyAllUserReservations", null, " ***********  No Reservations");
       }
     } catch (InterruptedException e) {
     } finally {
       reservationsSemaphore.release();
     }

   }

   public boolean userHasReservation(String uid) throws Exception {
     try {
       reservationsSemaphore.acquire();

       for (DuccUserReservation r : reservations) {
         if (r.getUserId().equals(uid)) {
           return true;
         }
       }
     } catch (InterruptedException e) {
     } finally {
       reservationsSemaphore.release();
     }
     return false;
   }

   public Object deepCopy(Object original) throws Exception {
     ObjectInputStream ois = null;
     ObjectOutputStream oos;
     ByteArrayInputStream bis;
     ByteArrayOutputStream bos;
     Object copy;
     try {
       // serialize object to bytes
       bos = new ByteArrayOutputStream();
       oos = new ObjectOutputStream(bos);
       oos.writeObject(original);
       oos.close();

       // construct an object from the bytes
       bis = new ByteArrayInputStream(bos.toByteArray());
       ois = new ObjectInputStream(bis);
       copy = ois.readObject();
       return copy;
     } catch (Exception e) {
       throw e;
     } finally {
       if (ois != null) {
         ois.close();
       }
     }
   }

   public RogueProcessReaper getRogueProcessReaper() {
     return rogueProcessReaper;
   }

   /**
    * Called when an Agent receives self dispatched Ping message.
    */
   // public void ping(AgentPingEvent agentPing) {
   // nodeMonitor.nodeArrives(agentPing.getNode());
   // }
   /*
    * public boolean excludeUser(String userId ) { if ( configurationFactory.userExclusionList !=
    * null ) { // exclusion list contains comma separated user ids String[] excludedUsers =
    * configurationFactory.userExclusionList.split(","); for ( String excludedUser : excludedUsers )
    * { if ( excludedUser.equals(userId)) { return true; } } } return false; } public boolean
    * excludeProcess(String process ) { if ( configurationFactory.processExclusionList != null ) { //
    * exclusion list contains comma separated user ids String[] excludedProcesses =
    * configurationFactory.processExclusionList.split(","); for ( String excludedProcess :
    * excludedProcesses ) { if ( excludedProcess.equals(process)) { return true; } } } return false;
    * }
    */
   public static void main(String[] args) {
     try {
       NodeIdentity node = new NodeIdentity(InetAddress.getLocalHost().getHostAddress(),
               InetAddress.getLocalHost().getHostName());
       NodeAgent agent = new NodeAgent(node);

       List<DuccUserReservation> reserves = new ArrayList<DuccUserReservation>();

       IDuccReservationMap reserveMap = new DuccReservationMap();
       IDuccReservationMap reserveMap2 = new DuccReservationMap();

       NodeIdentity ni1 = node;
       // new NodeIdentity(, name);
       NodeIdentity ni2 = new NodeIdentity("111.111.111.111", "node100");
       NodeIdentity ni3 = node;
       NodeIdentity ni4 = new NodeIdentity("222.222.222.222", "node102");

       DuccId id1 = new DuccId(100);
       DuccId id2 = new DuccId(101);
       DuccId id3 = new DuccId(102);
       DuccId id4 = new DuccId(103);

       IDuccReservation reservation1 = new DuccReservation(id1, ni1, 1);
       reserveMap.addReservation(reservation1);
       IDuccReservation reservation2 = new DuccReservation(id2, ni2, 1);
       reserveMap.addReservation(reservation2);
       IDuccReservation reservation4 = new DuccReservation(id4, ni4, 1);
       reserveMap.addReservation(reservation4);

       IDuccReservation reservation3 = new DuccReservation(id3, ni3, 1);
       reserveMap2.addReservation(reservation3);

       DuccUserReservation reserve = new DuccUserReservation("joe", new DuccId(500), reserveMap);
       DuccUserReservation reserve2 = new DuccUserReservation("jane", new DuccId(500), reserveMap2);
       reserves.add(reserve);
       reserves.add(reserve2);

       agent.setReservations(reserves);
     } catch (Exception e) {
       e.printStackTrace();
     }
   }

   private class NodeExclusionParser {
     private boolean excludeNodeFromCGroups = false;

     private boolean excludeAP = false;

     public void parse(String exclFile) throws Exception {
       // <node>=cgroup,ap
       File exclusionFile = new File(exclFile);
       if (!exclusionFile.exists()) {
         return;
       }
       BufferedReader br = new BufferedReader(new FileReader(exclusionFile));
       String line;
       NodeIdentity node = getIdentity();
       String nodeName = node.getShortName();

       while ((line = br.readLine()) != null) {
         if (line.startsWith(nodeName)) {
           String exclusions = line.substring(line.indexOf("=") + 1);
           String[] parsedExclusions = exclusions.split(",");
           for (String exclusion : parsedExclusions) {

             if (exclusion.trim().equals("cgroup")) {
               excludeNodeFromCGroups = true;

             } else if (exclusion.trim().equals("ap")) {
               excludeAP = true;

             }
           }
           break;
         }
       }
       br.close();
     }

     public boolean apExcluded() {
       return excludeAP;
     }

     public boolean cgroupsExcluded() {
       return excludeNodeFromCGroups;
     }
   }

   public DuccLogger getLogger() {
     return logger;
   }

   private void handleQuiesceAndStopEvent(DuccAdminEventQuiesceAndStop event) {
     logger.info("handleQuiesceAndStopEvent", null, "... Agent Received an Admin Request to Stop");
     try {
       stop(true, -1);
     } catch (Exception e) {
       logger.info("handleQuiesceAndStopEvent", null, e);
     }

   }

   private void handleStopEvent(DuccAdminEventStop event) {
     logger.info("handleStopEvent", null, "... Agent Received an Admin Request to Stop");
     try {

       stop(false, event.getTimeout());
     } catch (Exception e) {
       logger.info("handleStopEvent", null, e);
     }

   }

   private void handleStopPublishingEvent(DuccAdminEventStopMetrics event) {
     if (isThisTargetNode(getTargetNodes(event.getTargets()))) {
       logger.info("handleStopPublishingEvent", null,
               "... Agent Received an Admin Request to Stop Metrics Collection and Publishing");
       // Stop Camel route responsible for driving collection and publishing of metrics
       configurationFactory.stopMetricsRoute();
       logger.info("handleStopPublishingEvent", null,
               "... Agent Stopped Metrics Collection and Publishing");
     }

   }

   private String[] getTargetNodes(String targets) {
     logger.info("getTargetNodes", null, " Targets for Admin Command:" + targets);

     return targets.split(",");
   }

   private boolean isThisTargetNode(String[] nodes) {
     for (String targetNode : nodes) {
       if (Utils.isMachineNameMatch(targetNode.trim(), getIdentity().getCanonicalName())) {
         return true;
       }
     }
     return false;
   }

   private boolean isTarget(String[] targets) {
     for (String target : targets) {
       String[] targetParts = target.trim().split("@");
       logger.info("isTarget", null,
               " Targets for Admin Command:" + target + " This agent canonical identity:"
                       + getIdentity().getCanonicalName() + " short name:"
                       + getIdentity().getShortName());
       if ("agent".equals(targetParts[0])) {
         if (Utils.isMachineNameMatch(targetParts[1].trim(), getIdentity().getShortName())) {
           return true;
         }
       }
     }
     return false;
   }

   @Override
   public void handleAdminEvent(DuccAdminEvent event) throws Exception {

     Thread t = new Thread() {
       public void run() {
         if (event instanceof DuccAdminEventStop) {
           if (isTarget(getTargetNodes(((DuccAdminEventStop) event).getTargets()))) {
             handleStopEvent((DuccAdminEventStop) event);
           }
         } else if (event instanceof DuccAdminEventQuiesceAndStop) {
           if (isTarget(getTargetNodes(((DuccAdminEventQuiesceAndStop) event).getTargets()))) {
             logger.info("handleAdminEvent", null, "Node a target for quiesce");
             handleQuiesceAndStopEvent((DuccAdminEventQuiesceAndStop) event);
           }
         } else if (event instanceof DuccAdminEventStopMetrics) {
           handleStopPublishingEvent((DuccAdminEventStopMetrics) event);
         } else {
           logger.info("handleAdminEvent", null,
                   "... Agent Received Unexpected Message of Type:" + event.getClass().getName());
         }
       }
     };
     t.start();

     /*
      * if (event instanceof DuccAdminEventStopMetrics) { // Get target machines from the message
      * String[] nodes = ((DuccAdminEventStopMetrics) event).getTargetNodes().split(","); // Check if
      * this message applies to this node for (String targetNode : nodes) { if
      * (Utils.isMachineNameMatch(targetNode.trim(), getIdentity().getCanonicalName())) {
      * logger.info("handleAdminEvent", null,
      * "... Agent Received an Admin Request to Stop Metrics Collection and Publishing"); // Stop
      * Camel route responsible for driving collection and publishing of metrics
      * configurationFactory.stopMetricsRoute(); logger.info("handleAdminEvent", null,
      * "... Agent Stopped Metrics Collection and Publishing"); break; } else {
      * logger.info("handleAdminEvent", null, "... Agent Not Target For Message:" +
      * event.getClass().getName()); } } } else { logger.info("handleAdminEvent", null,
      * "... Agent Received Unexpected Message of Type:" + event.getClass().getName());
      *
      * }
      */
   }

   private class KillTimerTask extends TimerTask {

     private CountDownLatch completionLatch;

     private boolean killJustUimaAs;

     private boolean quiesce;

     public KillTimerTask(CountDownLatch completionLatch, boolean killOnlyUimaAs, boolean quiesce) {
       this.completionLatch = completionLatch;
       this.killJustUimaAs = killOnlyUimaAs;
       this.quiesce = quiesce;
     }

     @Override
     public void run() {
       try {
         // send kill -9 to any child process still running
         killChildProcesses(killJustUimaAs, quiesce);
       } finally {
         completionLatch.countDown();
       }
     }

   }
 }