| /* |
| * 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.processors; |
| |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Map; |
| |
| import org.apache.camel.Exchange; |
| import org.apache.uima.ducc.agent.NodeAgent; |
| import org.apache.uima.ducc.common.utils.DuccLogger; |
| import org.apache.uima.ducc.common.utils.id.DuccId; |
| import org.apache.uima.ducc.transport.agent.IUimaPipelineAEComponent; |
| import org.apache.uima.ducc.transport.dispatcher.DuccEventDispatcher; |
| import org.apache.uima.ducc.transport.event.NodeInventoryUpdateDuccEvent; |
| import org.apache.uima.ducc.transport.event.common.IDuccProcess; |
| import org.apache.uima.ducc.transport.event.common.IProcessState.ProcessState; |
| |
| /** |
| * |
| * |
| */ |
| public class DefaultNodeInventoryProcessor implements NodeInventoryProcessor { |
| DuccLogger logger = new DuccLogger(this.getClass(), "AGENT"); |
| |
| boolean inventoryChanged = true; |
| |
| private NodeAgent agent; |
| |
| private Map<DuccId, IDuccProcess> previousInventory; |
| |
| private int forceInventoryUpdateMaxThreshold = 0; |
| |
| private long counter = 0; |
| |
| public DefaultNodeInventoryProcessor(NodeAgent agent, String inventoryPublishRateSkipCount) { |
| this.agent = agent; |
| try { |
| forceInventoryUpdateMaxThreshold = Integer.parseInt(inventoryPublishRateSkipCount); |
| } catch (Exception e) { |
| } |
| // Dont allow 0 |
| if (forceInventoryUpdateMaxThreshold == 0) { |
| forceInventoryUpdateMaxThreshold = 1; |
| } |
| } |
| |
| /** |
| * Get a copy of agent {@code Process} inventory |
| */ |
| public Map<DuccId, IDuccProcess> getInventory() { |
| return agent.getInventoryCopy(); |
| } |
| |
| public void dispatchInventoryUpdate(DuccEventDispatcher dispatcher, String targetEndpoint, |
| Map<DuccId, IDuccProcess> inventory) throws Exception { |
| NodeInventoryUpdateDuccEvent duccEvent = new NodeInventoryUpdateDuccEvent(inventory, |
| agent.getLastORSequence(), agent.getIdentity()); |
| dispatcher.dispatch(targetEndpoint, duccEvent); |
| logger.info("dispatchInventoryUpdate", null, |
| "Agent dispatched inventory update event to endpoint:" + targetEndpoint); |
| } |
| |
| /** |
| * |
| */ |
| public void process(Exchange outgoingMessage) throws Exception { |
| String methodName = "process"; |
| // Get a deep copy of agent's inventory |
| Map<DuccId, IDuccProcess> inventory = getInventory(); |
| // Determine if the inventory changed since the last publishing was done |
| // First check if the inventory expanded or shrunk. If the same in size, |
| // compare process states and PID. If either of the two changed for any |
| // of the processes trigger immediate publish. If no changes found, |
| // publish |
| // according to skip counter |
| // (ducc.agent.node.inventory.publish.rate.skip) |
| // configured in ducc.properties. |
| if (previousInventory != null) { |
| if (agent.getEventListener().forceInvotoryUpdate()) { |
| inventoryChanged = true; |
| agent.getEventListener().resetForceInventoryUpdateFlag(); |
| } |
| if (inventory.size() != previousInventory.size()) { |
| inventoryChanged = true; |
| } else { |
| // Inventory maps are equal in size, check if all processes in |
| // the current |
| // inventory exist in the previous inventory snapshot. If not, |
| // it means that |
| // that perhaps a new process was added and one was removed. In |
| // this case, |
| // force the publish, since there was a change. |
| for (Map.Entry<DuccId, IDuccProcess> currentProcess : inventory.entrySet()) { |
| // Check if a process in the current inventory exists in a |
| // previous |
| // inventory snapshot |
| if (previousInventory.containsKey(currentProcess.getKey())) { |
| IDuccProcess previousProcess = previousInventory.get(currentProcess.getKey()); |
| // check if either PID or process state has changed |
| if (currentProcess.getValue().getPID() != null && previousProcess.getPID() == null) { |
| inventoryChanged = true; |
| break; |
| } else if (!currentProcess.getValue().getProcessState() |
| .equals(previousProcess.getProcessState())) { |
| inventoryChanged = true; |
| break; |
| } else { |
| List<IUimaPipelineAEComponent> breakdown = currentProcess.getValue() |
| .getUimaPipelineComponents(); |
| if (breakdown != null && breakdown.size() > 0) { |
| List<IUimaPipelineAEComponent> previousBreakdown = previousProcess |
| .getUimaPipelineComponents(); |
| if (previousBreakdown == null || previousBreakdown.size() == 0 |
| || breakdown.size() != previousBreakdown.size()) { |
| inventoryChanged = true; |
| } else { |
| for (IUimaPipelineAEComponent uimaAeState : breakdown) { |
| boolean found = false; |
| for (IUimaPipelineAEComponent previousUimaAeState : previousBreakdown) { |
| if (uimaAeState.getAeName().equals(previousUimaAeState.getAeName())) { |
| found = true; |
| if (!uimaAeState.getAeState().equals(previousUimaAeState.getAeState()) |
| || uimaAeState.getInitializationTime() != previousUimaAeState |
| .getInitializationTime()) { |
| inventoryChanged = true; |
| break; |
| } |
| } |
| } |
| if (!found) { |
| inventoryChanged = true; |
| } |
| |
| if (inventoryChanged) { |
| break; |
| } |
| |
| } |
| } |
| |
| } |
| } |
| } else { |
| // New inventory contains a process not in the previous |
| // snapshot |
| inventoryChanged = true; |
| break; |
| } |
| } |
| } |
| } |
| |
| // Get this inventory snapshot |
| previousInventory = inventory; |
| // Broadcast inventory if there is a change or configured number of |
| // epochs |
| // passed since the last broadcast. This is configured in |
| // ducc.properties with |
| // property ducc.agent.node.inventory.publish.rate.skip |
| try { |
| if (inventory.size() > 0 && (inventoryChanged || // if there is |
| // inventory |
| // change, |
| // publish |
| forceInventoryUpdateMaxThreshold == 0 || // skip rate in |
| // ducc.properties |
| // is zero, |
| // publish |
| (counter > 0 && (counter % forceInventoryUpdateMaxThreshold) == 0))) { // if |
| // reached |
| // skip |
| // rate, |
| // publish |
| |
| StringBuffer sb = new StringBuffer("Node Inventory (" + inventory.size() + ")"); |
| for (Map.Entry<DuccId, IDuccProcess> p : inventory.entrySet()) { |
| /* |
| * long endInitLong = 0; String endInit = ""; ITimeWindow wInit = |
| * p.getValue().getTimeWindowInit(); if(wInit != null) { endInit = wInit.getEnd(); |
| * endInitLong = wInit.getEndLong(); } long startRunLong = 0; String startRun = ""; |
| * ITimeWindow wRun = p.getValue().getTimeWindowRun(); if(wRun != null) { startRun = |
| * wRun.getStart(); startRunLong = wRun.getStartLong(); } if(endInitLong > startRunLong) { |
| * logger.warn(methodName, null, "endInit:"+endInitLong+" "+"startRun:"+startRunLong); } |
| */ |
| if (p.getValue().getUimaPipelineComponents() == null) { |
| p.getValue().setUimaPipelineComponents(new ArrayList<IUimaPipelineAEComponent>()); |
| } |
| if (!p.getValue().getProcessState().equals(ProcessState.Initializing)) { |
| p.getValue().getUimaPipelineComponents().clear(); |
| } |
| int pipelineInitStats = (p.getValue().getUimaPipelineComponents() == null) ? 0 |
| : p.getValue().getUimaPipelineComponents().size(); |
| StringBuffer gcInfo = new StringBuffer(); |
| if (p.getValue().getGarbageCollectionStats() != null) { |
| gcInfo.append(" GC Total=") |
| .append(p.getValue().getGarbageCollectionStats().getCollectionCount()) |
| .append(" GC Time=") |
| .append(p.getValue().getGarbageCollectionStats().getCollectionTime()) |
| .append(" "); |
| |
| } |
| sb.append("\n\t[Process Type=").append(p.getValue().getProcessType()).append(" DUCC ID=") |
| .append(p.getValue().getDuccId()).append(" PID=").append(p.getValue().getPID()) |
| .append(" State=").append(p.getValue().getProcessState()) |
| .append(" Resident Memory=").append(p.getValue().getResidentMemory()) |
| .append(gcInfo.toString()).append(" Init Stats List Size:" + pipelineInitStats) |
| .append(" Reason: " + p.getValue().getReasonForStoppingProcess()).append("] "); |
| if (p.getValue().getProcessState().equals(ProcessState.Stopped) |
| || p.getValue().getProcessState().equals(ProcessState.Failed) |
| || p.getValue().getProcessState().equals(ProcessState.Killed)) { |
| sb.append(" Reason:" + p.getValue().getReasonForStoppingProcess()); |
| sb.append(" Extended Reason:" + p.getValue().getExtendedReasonForStoppingProcess()); |
| } |
| |
| if (!p.getValue().getProcessState().equals(ProcessState.Running) |
| && !p.getValue().getProcessState().equals(ProcessState.Initializing)) { |
| sb.append(" Exit Code=" + p.getValue().getProcessExitCode()); |
| } |
| |
| } |
| logger.info(methodName, null, "Agent " + agent.getIdentity().getCanonicalName() |
| + " Posting Inventory:" + sb.toString()); |
| outgoingMessage.getIn().setBody(new NodeInventoryUpdateDuccEvent(inventory, |
| agent.getLastORSequence(), agent.getIdentity())); |
| |
| } else { |
| // Add null to the body of the message. A filter |
| // defined in the Camel route (AgentConfiguration.java) |
| // has a predicate to check for null body and throws |
| // away such a message. |
| outgoingMessage.getIn().setBody(null); |
| } |
| } catch (Exception e) { |
| logger.error(methodName, null, e); |
| } finally { |
| if (inventoryChanged) { |
| counter = 0; |
| } else { |
| counter++; |
| } |
| inventoryChanged = false; |
| } |
| |
| } |
| |
| } |
