You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by cw...@apache.org on 2013/05/07 19:24:23 UTC
svn commit: r1479988 [2/2] - in /uima/sandbox/uima-ducc/trunk:
uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/
uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/config/
uima-ducc-common/src/main/java/org/apache/uima/ducc/common/admin/e...
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/config/AgentConfiguration.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/config/AgentConfiguration.java?rev=1479988&r1=1479987&r2=1479988&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/config/AgentConfiguration.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-agent/src/main/java/org/apache/uima/ducc/agent/config/AgentConfiguration.java Tue May 7 17:24:23 2013
@@ -15,7 +15,7 @@
* 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.config;
import java.util.Map;
@@ -28,6 +28,8 @@ import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.dataformat.xstream.XStreamDataFormat;
import org.apache.camel.impl.DefaultClassResolver;
+import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.model.RoutesDefinition;
import org.apache.uima.ducc.agent.NodeAgent;
import org.apache.uima.ducc.agent.event.AgentEventListener;
import org.apache.uima.ducc.agent.launcher.Launcher;
@@ -60,386 +62,438 @@ import org.springframework.context.annot
import com.thoughtworks.xstream.XStream;
@Configuration
-
-@Import({ DuccTransportConfiguration.class,
- CommonConfiguration.class, NodeMetricsConfiguration.class })
+@Import({ DuccTransportConfiguration.class, CommonConfiguration.class,
+ NodeMetricsConfiguration.class })
public class AgentConfiguration {
- DuccLogger logger = new DuccLogger(this.getClass(), "Agent");
- // fetch the name of an endpoint where the JM expects incoming requests
-// @Value("#{ systemProperties['IP'] }")
- public String ip = System.getenv("IP");
- public String nodeName = System.getenv("NodeName");
-
-// public static String agentPingEnpoint = "activemq:topic:agent.ping.endpoint";
-// public static String agentPingSelectorName="agent_ip";
-// public static String agentPingSelector;
-
- //private DuccEventDispatcher agentPingDispatcher;
-
- @Value("#{ systemProperties['ducc.agent.launcher.thread.pool.size'] }")
- String launcherThreadPoolSize;
-
- @Value("#{ systemProperties['ducc.agent.launcher.process.stop.timeout'] }")
- public String processStopTimeout;
-
- @Value("#{ systemProperties['ducc.agent.node.inventory.publish.rate.skip'] }")
- public String inventoryPublishRateSkipCount;
-
- // Get comma separated list of processes to ignore while detecting rogue processes
+ DuccLogger logger = new DuccLogger(this.getClass(), "Agent");
+
+ // fetch the name of an endpoint where the JM expects incoming requests
+ // @Value("#{ systemProperties['IP'] }")
+ public String ip = System.getenv("IP");
+
+ public String nodeName = System.getenv("NodeName");
+
+ private CamelContext camelContext;
+
+ private RouteBuilder metricsRouteBuilder;
+
+ // public static String agentPingEnpoint = "activemq:topic:agent.ping.endpoint";
+ // public static String agentPingSelectorName="agent_ip";
+ // public static String agentPingSelector;
+
+ // private DuccEventDispatcher agentPingDispatcher;
+
+ @Value("#{ systemProperties['ducc.agent.launcher.thread.pool.size'] }")
+ String launcherThreadPoolSize;
+
+ @Value("#{ systemProperties['ducc.agent.launcher.process.stop.timeout'] }")
+ public String processStopTimeout;
+
+ @Value("#{ systemProperties['ducc.agent.node.inventory.publish.rate.skip'] }")
+ public String inventoryPublishRateSkipCount;
+
+ // Get comma separated list of processes to ignore while detecting rogue processes
@Value("#{ systemProperties['ducc.agent.rogue.process.exclusion.filter'] }")
- public String processExclusionList;
-
+ public String processExclusionList;
+
// Get comma separated list of users to ignore while detecting rogue processes
@Value("#{ systemProperties['ducc.agent.rogue.process.user.exclusion.filter'] }")
- public String userExclusionList;
- @Autowired
- DuccTransportConfiguration agentTransport;
- @Autowired
- NodeMetricsConfiguration nodeMetrics;
- @Autowired
- CommonConfiguration common;
- /**
- * Creates {@code AgentEventListener} that will handle incoming messages.
- *
- * @param agent - {@code NodeAgent} instance to initialize the listener
- *
- * @return {@code AgentEventListener} instance
- */
- public AgentEventListener agentDelegateListener(NodeAgent agent) {
- return new AgentEventListener(agent);
- }
- /**
- * Creates Camel Router to generate Node Metrics at regular intervals.
- *
- * @param targetEndpointToReceiveNodeMetricsUpdate - where to post NodeMetrics
- * @param nodeMetricsPublishRate - how to publish NodeMetrics
- * @return - {@code RouteBuilder} instance
- *
- * @throws Exception
- */
- private RouteBuilder routeBuilderForNodeMetricsPost(final NodeAgent agent, final String targetEndpointToReceiveNodeMetricsUpdate, final int nodeMetricsPublishRate) throws Exception {
- final Processor nmp = nodeMetricsProcessor(agent);
- final Predicate blastFilter = new DuccBlastGuardPredicate(agent.logger);
- final Processor cp = new ConfirmProcessor();
- return new RouteBuilder() {
- public void configure() {
- onException(Exception.class).handled(true).process(new ErrorProcessor());
- from("timer:nodeMetricsTimer?fixedRate=true&period=" + nodeMetricsPublishRate)
- .routeId("NodeMetricsPostRoute")
-
- // This route uses a filter to prevent sudden bursts of messages which
- // may flood DUCC daemons causing chaos. The filter disposes any message
- // that appears in a window of 1 sec or less.
- .filter(blastFilter)
- .process(nmp)
- .to(targetEndpointToReceiveNodeMetricsUpdate)
- .process(cp);
- }
- };
- }
-
- /**
- * Creates Camel Router to generate Node Metrics at regular intervals.
- *
- * @param targetEndpointToReceiveNodeMetricsUpdate - where to post NodeMetrics
- * @param nodeMetricsPublishRate - how to publish NodeMetrics
- * @return - {@code RouteBuilder} instance
- *
- * @throws Exception
- */
- private RouteBuilder routeBuilderForNodeInventoryPost(final NodeAgent agent, final String targetEndpointToReceiveNodeInventoryUpdate, final int nodeInventoryPublishRate) throws Exception {
- final Processor nmp = nodeInventoryProcessor(agent);
- return new RouteBuilder() {
- public void configure() {
- final Predicate bodyNotNull = body().isNotNull();
-
- final Predicate blastGuard = new DuccBlastGuardPredicate(agent.logger);
- onException(Exception.class).maximumRedeliveries(0).handled(true).process(new ErrorProcessor());
-
- from("timer:nodeInventoryTimer?fixedRate=true&period=" + nodeInventoryPublishRate)
- .routeId("NodeInventoryPostRoute")
- // This route uses a filter to prevent sudden bursts of messages which
- // may flood DUCC daemons causing chaos. The filter disposes any message
- // that appears in a window of 1 sec or less.
- .filter(blastGuard)
- // add inventory to the body of the message
- .process(nmp)
- // filter out messages with no body. Since this route is on a timer
- // it keeps generating flow of messages. However, the agent only
- // publishes inventory if there is a change or configured number of
- // epochs has passed. Otherwise, the agent puts null in the body of
- // the message and this route should just throw it away.
- .filter(bodyNotNull)
- .to(targetEndpointToReceiveNodeInventoryUpdate);
- }
- };
- }
- /**
- * Creates Camel Router to handle incoming messages
- *
- * @param delegate - {@code AgentEventListener} to delegate messages to
- *
- * @return {@code RouteBuilder} instance
- */
- public synchronized RouteBuilder routeBuilderForIncomingRequests(final NodeAgent agent, final AgentEventListener delegate) {
- return new RouteBuilder() {
- public void configure() {
- onException(Throwable.class).maximumRedeliveries(0).handled(false).process(new ErrorProcessor());
- from(common.agentRequestEndpoint)
- .routeId("IncomingRequestsRoute")
- //.process(new DebugProcessor())
- .bean(delegate);
- }
- };
- }
- /**
- * Creates Camel Router to handle incoming messages
- *
- * @param delegate - {@code AgentEventListener} to delegate messages to
- *
- * @return {@code RouteBuilder} instance
- */
- public synchronized RouteBuilder routeBuilderForManagedProcessStateUpdate(final NodeAgent agent, final AgentEventListener delegate) {
- return new RouteBuilder() {
-
- // Custom filter to select messages that are targeted for this agent
- // Checks the node list in a message to determine if this agent is
- // the target.
- Predicate filter = new DuccNodeFilter(agent);
- public void configure() {
- onException(Throwable.class).
- maximumRedeliveries(0).
- handled(true).
- process(new ErrorProcessor())
- .stop();
-
- from(common.managedProcessStateUpdateEndpoint)
- .routeId("ManageProcessStateUpdateRoute")
- //.process(new StateUpdateDebugProcessor(logger))
- .choice().when(filter)
- .bean(delegate)
- .end();
- }
- };
- }
-
-// private RouteBuilder routeBuilderForNodePing(final NodeAgent agent, final String targetEndpoint) throws Exception {
-// return new RouteBuilder() {
-// PingProcessor pingProcessor = new PingProcessor(agent);
-// public void configure() {
-// System.out.println("Agent Listening on Ping Endpoint:"+targetEndpoint);
-// onException(Exception.class).handled(true).process(new ErrorProcessor());
-// from(targetEndpoint)
-// .routeId("NodePingRoute")
-// .process( pingProcessor);
-// }
-// };
-// }
-
-
- public class DebugProcessor implements Processor {
-
- public void process(Exchange exchange) throws Exception {
- String methodName="process";
-// if ( logger.isLevelEnabled(Level.TRACE) ) {
- XStreamDataFormat xStreamDataFormat = new XStreamDataFormat();
- XStream xStream = xStreamDataFormat.getXStream(new DefaultClassResolver());
- String marshalledEvent = xStream.toXML(exchange.getIn().getBody());
- logger.info(methodName, null, marshalledEvent);
-// }
- }
-
- }
-
- public static class ConfirmProcessor implements Processor {
- boolean first = true;
- public void process(Exchange exchange) throws Exception {
-// if ( first ) {
-// synchronized(this) {
-// this.wait(20000);
-// }
-// first = false;
-// }
-
-// XStreamDataFormat xStreamDataFormat = new XStreamDataFormat();
-// XStream xStream = xStreamDataFormat.getXStream(new DefaultClassResolver());
-// String marshalledEvent = xStream.toXML(exchange.getIn().getBody());
-//
-// System.out.println("Agent Published Metrics:\n"+
-// marshalledEvent);
-
- }
- }
- public static class StateUpdateDebugProcessor implements Processor {
- DuccLogger logger;
-
- StateUpdateDebugProcessor(DuccLogger logger ) {
- this.logger = logger;
- }
- public void process(Exchange exchange) throws Exception {
- Map<String, Object> map = exchange.getIn().getHeaders();
- StringBuffer sb = new StringBuffer();
- for( Entry<String,Object> entry : map.entrySet()) {
- sb.append(entry.getKey()).append("=").append(entry.getValue()).append("\n");
- }
- logger.info("StateUpdateDebugProcessor.process", null, "Headers:\n\t"+sb.toString());
- //System.out.println("\t\tAgent received state update from managed process");
-
- }
- }
-
- public class ErrorProcessor implements Processor {
-
- public void process(Exchange exchange) throws Exception {
- // the caused by exception is stored in a property on the exchange
- Throwable caused = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Throwable.class);
- logger.error("ErrorProcessor.process", null, caused);
- //assertNotNull(caused);
- // here you can do what you want, but Camel regard this exception as handled, and
- // this processor as a failurehandler, so it wont do redeliveries. So this is the
- // end of this route. But if we want to route it somewhere we can just get a
- // producer template and send it.
-
- // send it to our mock endpoint
- //exchange.getContext().createProducerTemplate().send("mock:myerror", exchange);
- }
- }
- public static class TransportProcessor implements Processor {
-
- public void process(Exchange exchange) throws Exception {
- try {
- System.out.println(">>> Agent Received Message of type:"+exchange.getIn().getBody().getClass().getName());
- } catch( Exception e ) {
- e.printStackTrace();
- }
-// Destination replyTo = exchange.getIn().getHeader("JMSReplyTo",
-// Destination.class);
- // System.out.println("... transport - value of replyTo:" +
- // replyTo);
- }
-
- }
-// public static class PingProcessor implements Processor {
-// private NodeAgent agent;
-//
-// PingProcessor(NodeAgent agent ) {
-// this.agent = agent;
-// }
-// public void process(Exchange exchange) throws Exception {
-// try {
-// agent.ping((AgentPingEvent)exchange.getIn().getBody());
-// } catch( Exception e ) {
-// e.printStackTrace();
-// }
-// }
-//
-// }
- private NodeIdentity nodeIdentity() throws Exception {
- NodeIdentity ni = null;
-
- if ( ip != null ) {
- // Inject IP to enable deployment of multiple Agents on the same node with
- // different identity
- ni = new NodeIdentity(ip,nodeName); // this should only be used for testing
- } else {
- ni = new NodeIdentity();
- }
- return ni;
- }
-
- private Launcher launcher() {
- return new Launcher();
- }
- public DuccEventDispatcher getCommonProcessDispatcher(CamelContext camelContext) throws Exception {
- return agentTransport.duccEventDispatcher(common.managedServiceEndpoint, camelContext);
- }
- @Bean
- public NodeAgent nodeAgent() throws Exception {
- try {
- CamelContext camelContext = common.camelContext();
- camelContext.disableJMX();
-
- NodeAgent agent = new NodeAgent(nodeIdentity(), launcher(), camelContext, this);
-// agentPingSelector = agentPingSelectorName+"='"+agent.getIdentity().getIp()+"'";
- // optionally configures Camel Context for JMS. Checks the 'agentRequestEndpoint' to
- // to determine type of transport. If the the endpoint starts with "activemq:", a
- // special ActiveMQ component will be activated to enable JMS transport
- agentTransport.configureJMSTransport(common.agentRequestEndpoint,camelContext);
- AgentEventListener delegateListener = agentDelegateListener(agent);
-// agentPingDispatcher =
-// agentTransport.duccEventDispatcher(agentPingEnpoint, camelContext);
-
- if ( common.managedProcessStateUpdateEndpointType != null && common.managedProcessStateUpdateEndpointType.equalsIgnoreCase("socket") ) {
+ public String userExclusionList;
+
+ @Autowired
+ DuccTransportConfiguration agentTransport;
+
+ @Autowired
+ NodeMetricsConfiguration nodeMetrics;
+
+ @Autowired
+ CommonConfiguration common;
+
+ /**
+ * Creates {@code AgentEventListener} that will handle incoming messages.
+ *
+ * @param agent
+ * - {@code NodeAgent} instance to initialize the listener
+ *
+ * @return {@code AgentEventListener} instance
+ */
+ public AgentEventListener agentDelegateListener(NodeAgent agent) {
+ return new AgentEventListener(agent);
+ }
+
+ /**
+ * Creates Camel Router to generate Node Metrics at regular intervals.
+ *
+ * @param targetEndpointToReceiveNodeMetricsUpdate
+ * - where to post NodeMetrics
+ * @param nodeMetricsPublishRate
+ * - how to publish NodeMetrics
+ * @return - {@code RouteBuilder} instance
+ *
+ * @throws Exception
+ */
+ private RouteBuilder routeBuilderForNodeMetricsPost(final NodeAgent agent,
+ final String targetEndpointToReceiveNodeMetricsUpdate, final int nodeMetricsPublishRate)
+ throws Exception {
+ final Processor nmp = nodeMetricsProcessor(agent);
+ final Predicate blastFilter = new DuccBlastGuardPredicate(agent.logger);
+ final Processor cp = new ConfirmProcessor();
+ return new RouteBuilder() {
+ public void configure() {
+ onException(Exception.class).handled(true).process(new ErrorProcessor());
+ from("timer:nodeMetricsTimer?fixedRate=true&period=" + nodeMetricsPublishRate)
+ .routeId("NodeMetricsPostRoute")
+
+ // This route uses a filter to prevent sudden bursts of messages which
+ // may flood DUCC daemons causing chaos. The filter disposes any message
+ // that appears in a window of 1 sec or less.
+ .filter(blastFilter).process(nmp).to(targetEndpointToReceiveNodeMetricsUpdate)
+ .process(cp);
+ }
+ };
+ }
+
+ /**
+ * Creates Camel Router to generate Node Metrics at regular intervals.
+ *
+ * @param targetEndpointToReceiveNodeMetricsUpdate
+ * - where to post NodeMetrics
+ * @param nodeMetricsPublishRate
+ * - how to publish NodeMetrics
+ * @return - {@code RouteBuilder} instance
+ *
+ * @throws Exception
+ */
+ private RouteBuilder routeBuilderForNodeInventoryPost(final NodeAgent agent,
+ final String targetEndpointToReceiveNodeInventoryUpdate,
+ final int nodeInventoryPublishRate) throws Exception {
+ final Processor nmp = nodeInventoryProcessor(agent);
+ return new RouteBuilder() {
+ public void configure() {
+ final Predicate bodyNotNull = body().isNotNull();
+
+ final Predicate blastGuard = new DuccBlastGuardPredicate(agent.logger);
+ onException(Exception.class).maximumRedeliveries(0).handled(true)
+ .process(new ErrorProcessor());
+
+ from("timer:nodeInventoryTimer?fixedRate=true&period=" + nodeInventoryPublishRate)
+ .routeId("NodeInventoryPostRoute")
+ // This route uses a filter to prevent sudden bursts of messages which
+ // may flood DUCC daemons causing chaos. The filter disposes any message
+ // that appears in a window of 1 sec or less.
+ .filter(blastGuard)
+ // add inventory to the body of the message
+ .process(nmp)
+ // filter out messages with no body. Since this route is on a timer
+ // it keeps generating flow of messages. However, the agent only
+ // publishes inventory if there is a change or configured number of
+ // epochs has passed. Otherwise, the agent puts null in the body of
+ // the message and this route should just throw it away.
+ .filter(bodyNotNull).to(targetEndpointToReceiveNodeInventoryUpdate);
+ }
+ };
+ }
+
+ /**
+ * Creates Camel Router to handle incoming messages
+ *
+ * @param delegate
+ * - {@code AgentEventListener} to delegate messages to
+ *
+ * @return {@code RouteBuilder} instance
+ */
+ public synchronized RouteBuilder routeBuilderForIncomingRequests(final NodeAgent agent,
+ final AgentEventListener delegate) {
+ return new RouteBuilder() {
+ public void configure() {
+ onException(Throwable.class).maximumRedeliveries(0).handled(false)
+ .process(new ErrorProcessor());
+ from(common.agentRequestEndpoint).routeId("IncomingRequestsRoute")
+ // .process(new DebugProcessor())
+ .bean(delegate);
+ }
+ };
+ }
+
+ /**
+ * Creates Camel Router to handle incoming messages
+ *
+ * @param delegate
+ * - {@code AgentEventListener} to delegate messages to
+ *
+ * @return {@code RouteBuilder} instance
+ */
+ public synchronized RouteBuilder routeBuilderForManagedProcessStateUpdate(final NodeAgent agent,
+ final AgentEventListener delegate) {
+ return new RouteBuilder() {
+
+ // Custom filter to select messages that are targeted for this agent
+ // Checks the node list in a message to determine if this agent is
+ // the target.
+ Predicate filter = new DuccNodeFilter(agent);
+
+ public void configure() {
+ onException(Throwable.class).maximumRedeliveries(0).handled(true)
+ .process(new ErrorProcessor()).stop();
+
+ from(common.managedProcessStateUpdateEndpoint).routeId("ManageProcessStateUpdateRoute")
+ // .process(new StateUpdateDebugProcessor(logger))
+ .choice().when(filter).bean(delegate).end();
+ }
+ };
+ }
+
+ // private RouteBuilder routeBuilderForNodePing(final NodeAgent agent, final String
+ // targetEndpoint) throws Exception {
+ // return new RouteBuilder() {
+ // PingProcessor pingProcessor = new PingProcessor(agent);
+ // public void configure() {
+ // System.out.println("Agent Listening on Ping Endpoint:"+targetEndpoint);
+ // onException(Exception.class).handled(true).process(new ErrorProcessor());
+ // from(targetEndpoint)
+ // .routeId("NodePingRoute")
+ // .process( pingProcessor);
+ // }
+ // };
+ // }
+
+ public class DebugProcessor implements Processor {
+
+ public void process(Exchange exchange) throws Exception {
+ String methodName = "process";
+ // if ( logger.isLevelEnabled(Level.TRACE) ) {
+ XStreamDataFormat xStreamDataFormat = new XStreamDataFormat();
+ XStream xStream = xStreamDataFormat.getXStream(new DefaultClassResolver());
+ String marshalledEvent = xStream.toXML(exchange.getIn().getBody());
+ logger.info(methodName, null, marshalledEvent);
+ // }
+ }
+
+ }
+
+ public static class ConfirmProcessor implements Processor {
+ boolean first = true;
+
+ public void process(Exchange exchange) throws Exception {
+ // if ( first ) {
+ // synchronized(this) {
+ // this.wait(20000);
+ // }
+ // first = false;
+ // }
+
+ // XStreamDataFormat xStreamDataFormat = new XStreamDataFormat();
+ // XStream xStream = xStreamDataFormat.getXStream(new DefaultClassResolver());
+ // String marshalledEvent = xStream.toXML(exchange.getIn().getBody());
+ //
+ // System.out.println("Agent Published Metrics:\n"+
+ // marshalledEvent);
+
+ }
+ }
+
+ public static class StateUpdateDebugProcessor implements Processor {
+ DuccLogger logger;
+
+ StateUpdateDebugProcessor(DuccLogger logger) {
+ this.logger = logger;
+ }
+
+ public void process(Exchange exchange) throws Exception {
+ Map<String, Object> map = exchange.getIn().getHeaders();
+ StringBuffer sb = new StringBuffer();
+ for (Entry<String, Object> entry : map.entrySet()) {
+ sb.append(entry.getKey()).append("=").append(entry.getValue()).append("\n");
+ }
+ logger.info("StateUpdateDebugProcessor.process", null, "Headers:\n\t" + sb.toString());
+
+ }
+ }
+
+ public class ErrorProcessor implements Processor {
+
+ public void process(Exchange exchange) throws Exception {
+ // the caused by exception is stored in a property on the exchange
+ Throwable caused = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Throwable.class);
+ logger.error("ErrorProcessor.process", null, caused);
+ // assertNotNull(caused);
+ // here you can do what you want, but Camel regard this exception as handled, and
+ // this processor as a failurehandler, so it wont do redeliveries. So this is the
+ // end of this route. But if we want to route it somewhere we can just get a
+ // producer template and send it.
+
+ // send it to our mock endpoint
+ // exchange.getContext().createProducerTemplate().send("mock:myerror", exchange);
+ }
+ }
+
+ public static class TransportProcessor implements Processor {
+
+ public void process(Exchange exchange) throws Exception {
+ try {
+ System.out.println(">>> Agent Received Message of type:"
+ + exchange.getIn().getBody().getClass().getName());
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ // Destination replyTo = exchange.getIn().getHeader("JMSReplyTo",
+ // Destination.class);
+ // System.out.println("... transport - value of replyTo:" +
+ // replyTo);
+ }
+
+ }
+
+ // public static class PingProcessor implements Processor {
+ // private NodeAgent agent;
+ //
+ // PingProcessor(NodeAgent agent ) {
+ // this.agent = agent;
+ // }
+ // public void process(Exchange exchange) throws Exception {
+ // try {
+ // agent.ping((AgentPingEvent)exchange.getIn().getBody());
+ // } catch( Exception e ) {
+ // e.printStackTrace();
+ // }
+ // }
+ //
+ // }
+ private NodeIdentity nodeIdentity() throws Exception {
+ NodeIdentity ni = null;
+
+ if (ip != null) {
+ // Inject IP to enable deployment of multiple Agents on the same node with
+ // different identity
+ ni = new NodeIdentity(ip, nodeName); // this should only be used for testing
+ } else {
+ ni = new NodeIdentity();
+ }
+ return ni;
+ }
+
+ private Launcher launcher() {
+ return new Launcher();
+ }
+
+ public DuccEventDispatcher getCommonProcessDispatcher(CamelContext camelContext) throws Exception {
+ return agentTransport.duccEventDispatcher(common.managedServiceEndpoint, camelContext);
+ }
+
+ @Bean
+ public NodeAgent nodeAgent() throws Exception {
+ try {
+ camelContext = common.camelContext();
+ camelContext.disableJMX();
+
+ NodeAgent agent = new NodeAgent(nodeIdentity(), launcher(), camelContext, this);
+ // optionally configures Camel Context for JMS. Checks the 'agentRequestEndpoint' to
+ // to determine type of transport. If the the endpoint starts with "activemq:", a
+ // special ActiveMQ component will be activated to enable JMS transport
+ agentTransport.configureJMSTransport(common.agentRequestEndpoint, camelContext);
+ AgentEventListener delegateListener = agentDelegateListener(agent);
+
+ if (common.managedProcessStateUpdateEndpointType != null
+ && common.managedProcessStateUpdateEndpointType.equalsIgnoreCase("socket")) {
String agentSocketParams = "";
- if ( common.managedProcessStateUpdateEndpointParams != null ) {
- agentSocketParams = "?"+common.managedProcessStateUpdateEndpointParams;
+ if (common.managedProcessStateUpdateEndpointParams != null) {
+ agentSocketParams = "?" + common.managedProcessStateUpdateEndpointParams;
}
int agentPort = Utils.findFreePort();
- common.managedProcessStateUpdateEndpoint = "mina:tcp://localhost:"+agentPort+agentSocketParams;
- // Remember the agent port since we need to tell JPs where to send their state updates
+ common.managedProcessStateUpdateEndpoint = "mina:tcp://localhost:" + agentPort
+ + agentSocketParams;
+ // Remember the agent port since we need to tell JPs where to send their state updates
System.setProperty(NodeAgent.ProcessStateUpdatePort, String.valueOf(agentPort));
}
- camelContext.addRoutes(this.routeBuilderForManagedProcessStateUpdate(agent,delegateListener));
- camelContext.addRoutes(this.routeBuilderForIncomingRequests(agent,delegateListener));
- camelContext.addRoutes(this.routeBuilderForNodeInventoryPost(agent, common.nodeInventoryEndpoint, Integer.parseInt(common.nodeInventoryPublishRate)));
- camelContext.addRoutes(this.routeBuilderForNodeMetricsPost(agent, common.nodeMetricsEndpoint, Integer.parseInt(common.nodeMetricsPublishRate)));
-// camelContext.addRoutes(this.routeBuilderForNodePing(agent,agentPingEnpoint+"?selector="+agentPingSelector) );
-
- logger.info("nodeAgent", null,"------- Agent Initialized - Identity Name:"+agent.getIdentity().getName()+" IP:"+agent.getIdentity().getIp()+" JP State Update Endpoint:"+common.managedProcessStateUpdateEndpoint);
- return agent;
-
- } catch( Exception e) {
- e.printStackTrace();
- }
- return null;
- }
-
- @Bean
- public NodeMetricsProcessor nodeMetricsProcessor(NodeAgent agent) throws Exception {
- if (Utils.isLinux()) {
- return new LinuxNodeMetricsProcessor(agent, "/proc/meminfo", "/proc/loadavg");
- } else {
- return new DefaultNodeMetricsProcessor(agent);
- }
- }
- public ProcessMetricsProcessor processMetricsProcessor(NodeAgent agent, IDuccProcess process, ManagedProcess managedProcess) throws Exception {
- if (Utils.isLinux()) {
- return new LinuxProcessMetricsProcessor(logger, process, agent, "/proc/"+process.getPID()+"/statm", "/proc/stat", "/proc/"+process.getPID()+"/stat", managedProcess);
- } else {
- return new DefaultProcessMetricsProcessor(process, agent);
- }
-
- }
- public NodeInventoryProcessor nodeInventoryProcessor(NodeAgent agent) {
- return new DefaultNodeInventoryProcessor(agent, inventoryPublishRateSkipCount);
- }
- private class DuccNodeFilter implements Predicate {
- private NodeAgent agent = null;
- public DuccNodeFilter(NodeAgent agent) {
- this.agent = agent;
- }
- public synchronized boolean matches(Exchange exchange) {
- String methodName="DuccNodeFilter.matches";
- boolean result = false;
- if ( common.managedProcessStateUpdateEndpoint.startsWith("mina")) {
- // mina is a socket component with point-to-point semantics thus
- // the client always sends a message to the correct agent. No reason
- // to determine if this is a target agent.
- result = true;
- } else {
- try {
- String nodes = (String)exchange.getIn().getHeader(DuccExchange.TARGET_NODES_HEADER_NAME);
- logger.trace(methodName, null, ">>>>>>>>> Agent: ["+agent.getIdentity().getIp()+"] Received a Message. Is Agent target for message:"+result+". Target Agents:"+nodes);
- result = Utils.isTargetNodeForMessage(nodes, agent.getIdentity().getNodeIdentities());
- } catch( Throwable e) {
- e.printStackTrace();
- logger.error(methodName, null, e, new Object[] {});
- }
- }
- return result;
- }
- }
-// public DuccEventDispatcher getAgentPingDispatcher() {
-// return agentPingDispatcher;
-// }
+ camelContext
+ .addRoutes(this.routeBuilderForManagedProcessStateUpdate(agent, delegateListener));
+ camelContext.addRoutes(this.routeBuilderForIncomingRequests(agent, delegateListener));
+ camelContext.addRoutes(this.routeBuilderForNodeInventoryPost(agent,
+ common.nodeInventoryEndpoint, Integer.parseInt(common.nodeInventoryPublishRate)));
+ metricsRouteBuilder = this.routeBuilderForNodeMetricsPost(agent, common.nodeMetricsEndpoint,
+ Integer.parseInt(common.nodeMetricsPublishRate));
+ camelContext.addRoutes(metricsRouteBuilder);
+
+ logger.info("nodeAgent", null, "------- Agent Initialized - Identity Name:"
+ + agent.getIdentity().getName() + " IP:" + agent.getIdentity().getIp()
+ + " JP State Update Endpoint:" + common.managedProcessStateUpdateEndpoint);
+ return agent;
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ return null;
+ }
+
+ @Bean
+ public NodeMetricsProcessor nodeMetricsProcessor(NodeAgent agent) throws Exception {
+ if (Utils.isLinux()) {
+ return new LinuxNodeMetricsProcessor(agent, "/proc/meminfo", "/proc/loadavg");
+ } else {
+ return new DefaultNodeMetricsProcessor(agent);
+ }
+ }
+
+ public ProcessMetricsProcessor processMetricsProcessor(NodeAgent agent, IDuccProcess process,
+ ManagedProcess managedProcess) throws Exception {
+ if (Utils.isLinux()) {
+ return new LinuxProcessMetricsProcessor(logger, process, agent, "/proc/" + process.getPID()
+ + "/statm", "/proc/stat", "/proc/" + process.getPID() + "/stat", managedProcess);
+ } else {
+ return new DefaultProcessMetricsProcessor(process, agent);
+ }
+
+ }
+
+ public NodeInventoryProcessor nodeInventoryProcessor(NodeAgent agent) {
+ return new DefaultNodeInventoryProcessor(agent, inventoryPublishRateSkipCount);
+ }
+
+ public void stopMetricsRoute() {
+ String methodName = "stopMetricsRoute";
+ try {
+ RoutesDefinition rsd = metricsRouteBuilder.getRouteCollection();
+ for (RouteDefinition rd : rsd.getRoutes()) {
+ camelContext.stopRoute(rd.getId());
+ logger.error(methodName, null, ">>>> Agent Stopped Metrics Publishing");
+ }
+
+ } catch (Exception e) {
+ logger.error(methodName, null, e);
+ }
+ }
+
+ private class DuccNodeFilter implements Predicate {
+ private NodeAgent agent = null;
+
+ public DuccNodeFilter(NodeAgent agent) {
+ this.agent = agent;
+ }
+
+ public synchronized boolean matches(Exchange exchange) {
+ String methodName = "DuccNodeFilter.matches";
+ boolean result = false;
+ if (common.managedProcessStateUpdateEndpoint.startsWith("mina")) {
+ // mina is a socket component with point-to-point semantics thus
+ // the client always sends a message to the correct agent. No reason
+ // to determine if this is a target agent.
+ result = true;
+ } else {
+ try {
+ String nodes = (String) exchange.getIn().getHeader(DuccExchange.TARGET_NODES_HEADER_NAME);
+ logger.trace(methodName, null, ">>>>>>>>> Agent: [" + agent.getIdentity().getIp()
+ + "] Received a Message. Is Agent target for message:" + result
+ + ". Target Agents:" + nodes);
+ result = Utils.isTargetNodeForMessage(nodes, agent.getIdentity().getNodeIdentities());
+ } catch (Throwable e) {
+ e.printStackTrace();
+ logger.error(methodName, null, e, new Object[] {});
+ }
+ }
+ return result;
+ }
+ }
+ // public DuccEventDispatcher getAgentPingDispatcher() {
+ // return agentPingDispatcher;
+ // }
}
Added: uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/admin/event/DuccAdminEventStopMetrics.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/admin/event/DuccAdminEventStopMetrics.java?rev=1479988&view=auto
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/admin/event/DuccAdminEventStopMetrics.java (added)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/admin/event/DuccAdminEventStopMetrics.java Tue May 7 17:24:23 2013
@@ -0,0 +1,38 @@
+/*
+ * 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.common.admin.event;
+
+public class DuccAdminEventStopMetrics extends DuccAdminEvent {
+ private static final long serialVersionUID = 6499822168988392919L;
+ // comma separated list of nodes that are target for this message
+ private String targetNodes = new String();
+
+
+ public DuccAdminEventStopMetrics(String nodes) {
+ this.targetNodes = nodes;
+ }
+ /**
+ * Returns comma separated list of target nodes for this message
+ * @return
+ */
+ public String getTargetNodes() {
+ return this.targetNodes;
+ }
+
+}
Propchange: uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/admin/event/DuccAdminEventStopMetrics.java
------------------------------------------------------------------------------
svn:eol-style = LF
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/component/AbstractDuccComponent.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/component/AbstractDuccComponent.java?rev=1479988&r1=1479987&r2=1479988&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/component/AbstractDuccComponent.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/component/AbstractDuccComponent.java Tue May 7 17:24:23 2013
@@ -15,7 +15,7 @@
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
-*/
+ */
package org.apache.uima.ducc.common.component;
import java.lang.management.ManagementFactory;
@@ -49,7 +49,6 @@ import org.apache.uima.ducc.common.utils
import org.apache.uima.ducc.common.utils.DuccProperties;
import org.apache.uima.ducc.common.utils.Utils;
-
/**
* Abstract class which every Ducc component should extend from. Provides support for loading
* property files and resolving any placeholders in property values. Creates a special purpose Camel
@@ -72,7 +71,7 @@ public abstract class AbstractDuccCompon
private Object monitor = new Object();
private DuccLogger logger;
-
+
public AbstractDuccComponent(String componentName) {
this(componentName, null);
}
@@ -81,6 +80,7 @@ public abstract class AbstractDuccCompon
this.componentName = componentName;
setContext(context);
}
+
/**
* Creates Camel Router for Ducc admin events. Any event arriving on this channel will be handled
* by this abstract class.
@@ -98,21 +98,32 @@ public abstract class AbstractDuccCompon
System.out.println("Configuring Admin Channel on Endpoint:" + endpoint);
onException(Exception.class).handled(true).process(new ErrorProcessor());
- from(endpoint)
- .routeId("AdminRoute")
- .unmarshal().xstream().process(new AdminEventProcessor(delegate));
+ from(endpoint).routeId("AdminRoute").unmarshal().xstream()
+ .process(new AdminEventProcessor(delegate));
}
});
+ DuccLogger dl = getLogger();
+ if (dl != null) {
+ logger = dl;
+ logger.info("startAdminChannel", null, "Admin Channel Activated on endpoint:" + endpoint);
+ }
}
+
public DuccLogger getLogger() {
- return null; // ducc components can override this
+ return null; // ducc components can override this
}
+
public class ErrorProcessor implements Processor {
public void process(Exchange exchange) throws Exception {
// the caused by exception is stored in a property on the exchange
Throwable caused = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Throwable.class);
caused.printStackTrace();
+ DuccLogger dl = getLogger();
+ if (dl != null) {
+ dl.error("ErrorProcessor.process()", null, caused);
+
+ }
}
}
@@ -252,7 +263,9 @@ public abstract class AbstractDuccCompon
public void start(DuccService service, String[] args) throws Exception {
String endpoint = null;
this.service = service;
- if (System.getProperty("ducc.deploy.components") != null && !System.getProperty("ducc.deploy.components").equals("uima-as") && (endpoint = System.getProperty("ducc.admin.endpoint")) != null) {
+ if (System.getProperty("ducc.deploy.components") != null
+ && !System.getProperty("ducc.deploy.components").equals("uima-as")
+ && (endpoint = System.getProperty("ducc.admin.endpoint")) != null) {
System.out.println(".....Starting Admin Channel on endpoint:" + endpoint);
startAdminChannel(endpoint, this);
}
@@ -277,8 +290,9 @@ public abstract class AbstractDuccCompon
// Register Ducc Component MBean with JMX.
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
- ObjectName name = new ObjectName("org.apache.uima.ducc.service.admin.jmx:type=DuccComponentMBean,name="
- + getClass().getSimpleName());
+ ObjectName name = new ObjectName(
+ "org.apache.uima.ducc.service.admin.jmx:type=DuccComponentMBean,name="
+ + getClass().getSimpleName());
mbs.registerMBean(this, name);
}
@@ -287,13 +301,13 @@ public abstract class AbstractDuccCompon
}
public void stop() throws Exception {
- if ( getLogger() != null ) {
+ if (getLogger() != null) {
logger = getLogger();
}
- if ( logger == null ) {
+ if (logger == null) {
System.out.println("----------stop() called");
} else {
- logger.info("stop", null,"----------stop() called");
+ logger.info("stop", null, "----------stop() called");
}
synchronized (monitor) {
if (stopping) {
@@ -304,41 +318,41 @@ public abstract class AbstractDuccCompon
Thread th = new Thread(new Runnable() {
public void run() {
try {
- if ( logger == null ) {
+ if (logger == null) {
System.out.println("Stopping Camel Routes");
} else {
- logger.info("stop", null,"Stopping Camel Routes");
+ logger.info("stop", null, "Stopping Camel Routes");
}
-
+
List<Route> routes = context.getRoutes();
for (Route route : routes) {
- System.out.println("--- Stopping Route:"+route.getId());
+ System.out.println("--- Stopping Route:" + route.getId());
route.getConsumer().stop();
-// List<Service> services = route.getServices();
-// for (Service service : services) {
-// System.out.println("Stopping Route:"+route.getId());
-// service.stop();
-// }
- System.out.println("--- Stopping Endpoint Route:"+route.getId());
+ // List<Service> services = route.getServices();
+ // for (Service service : services) {
+ // System.out.println("Stopping Route:"+route.getId());
+ // service.stop();
+ // }
+ System.out.println("--- Stopping Endpoint Route:" + route.getId());
route.getEndpoint().stop();
}
-
+
System.out.println("--- Stopping AMQC");
ActiveMQComponent amqc = (ActiveMQComponent) context.getComponent("activemq");
amqc.stop();
System.out.println("Stopping AMQC - shutdown");
amqc.shutdown();
- if ( logger == null ) {
+ if (logger == null) {
System.out.println("Stopping Camel Context");
} else {
- logger.info("stop", null,"Stopping Camel Context");
+ logger.info("stop", null, "Stopping Camel Context");
}
context.stop();
- if ( logger == null ) {
+ if (logger == null) {
System.out.println("Camel Context Stopped");
} else {
- logger.info("stop", null,"Camel Context Stopped");
+ logger.info("stop", null, "Camel Context Stopped");
}
ObjectName name = new ObjectName(
@@ -357,10 +371,10 @@ public abstract class AbstractDuccCompon
if (service != null) {
service.stop();
}
- if ( logger == null ) {
+ if (logger == null) {
System.out.println("Component cleanup completed - terminating process");
} else {
- logger.info("stop", null,"Component cleanup completed - terminating process");
+ logger.info("stop", null, "Component cleanup completed - terminating process");
}
} catch (Exception e) {
@@ -434,28 +448,38 @@ public abstract class AbstractDuccCompon
}
public void process(final Exchange exchange) throws Exception {
+ logger = getLogger();
+ if (logger != null) {
+
+ logger.info("AdminEventProcessor.process()", null, "Received Admin Message of Type:"
+ + exchange.getIn().getBody().getClass().getName());
+ }
if (exchange.getIn().getBody() instanceof DuccAdminEventKill) {
- // start a new thread to process the admin kill event. Need to do this
- // so that Camel thread associated with admin channel can go back to
- // its pool. Otherwise, we will not be able to stop the admin channel.
+ // start a new thread to process the admin kill event. Need to do this
+ // so that Camel thread associated with admin channel can go back to
+ // its pool. Otherwise, we will not be able to stop the admin channel.
Thread th = new Thread(new Runnable() {
public void run() {
try {
delegate.onDuccAdminKillEvent((DuccAdminEventKill) exchange.getIn().getBody());
- } catch(Exception e) {
-
+ } catch (Exception e) {
+
}
}
});
th.start();
} else {
- System.out
- .println("---------- Ducc Process Received Unknown Message on Admin Channel. Message Type:"
- + exchange.getIn().getBody().getClass().getName());
+ handleAdminEvent((DuccAdminEvent) exchange.getIn().getBody());
}
}
}
+ /**
+ * Components interested in receiving DuccAdminEvents should override this method
+ */
+ public void handleAdminEvent(DuccAdminEvent event) throws Exception {
+ }
+
static class ServiceShutdownHook extends Thread {
private AbstractDuccComponent duccProcess;
@@ -464,52 +488,44 @@ public abstract class AbstractDuccCompon
}
public void run() {
- try {
- System.out.println("DUCC Service Caught Kill Signal - Registering Killer Task and Stopping ...");
-
- // schedule a kill task which will kill this process after 1 minute
- Timer killTimer = new Timer();
- killTimer.schedule( new KillerThreadTask(), 60*1000);
-
- // try to stop the process cleanly
- duccProcess.stop();
- } catch (Exception e) {
- e.printStackTrace();
- }
+ try {
+ System.out
+ .println("DUCC Service Caught Kill Signal - Registering Killer Task and Stopping ...");
+
+ // schedule a kill task which will kill this process after 1 minute
+ Timer killTimer = new Timer();
+ killTimer.schedule(new KillerThreadTask(), 60 * 1000);
+
+ // try to stop the process cleanly
+ duccProcess.stop();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
}
}
// This task will run if stop() fails to stop the process within 1 minute
static class KillerThreadTask extends TimerTask {
public void run() {
- try {
- System.out.println("Process is about to kill itself via Runtime.getRuntime().halt()");
- // Take the jvm down hard. This call will not
- // invoke registered ShutdownHooks and just
- // terminates the jvm.
- Runtime.getRuntime().halt(-1);
-/*
- String[] killcmd = new String[3];
- if ( Utils.isWindows() ) {
- killcmd[0] = "taskkill";
- killcmd[1] = "/PID";
- } else {
- killcmd[0] = "/bin/kill";
- killcmd[1] = "-9";
- }
- // get this process PID
- killcmd[2] = Utils.getPID();
- ProcessBuilder pb = new ProcessBuilder(killcmd);
- pb.redirectErrorStream(true);
- // kill itself via hard OS specific kill cmd
- pb.start();
- // should be the last thing this process does
-*/
- } catch (Exception e) {
- e.printStackTrace();
- }
+ try {
+ System.out.println("Process is about to kill itself via Runtime.getRuntime().halt()");
+ // Take the jvm down hard. This call will not
+ // invoke registered ShutdownHooks and just
+ // terminates the jvm.
+ Runtime.getRuntime().halt(-1);
+ /*
+ * String[] killcmd = new String[3]; if ( Utils.isWindows() ) { killcmd[0] = "taskkill";
+ * killcmd[1] = "/PID"; } else { killcmd[0] = "/bin/kill"; killcmd[1] = "-9"; } // get this
+ * process PID killcmd[2] = Utils.getPID(); ProcessBuilder pb = new ProcessBuilder(killcmd);
+ * pb.redirectErrorStream(true); // kill itself via hard OS specific kill cmd pb.start(); //
+ * should be the last thing this process does
+ */
+ } catch (Exception e) {
+ e.printStackTrace();
}
+ }
}
+
public void setLogLevel(String clz, String level) {
service.setLogLevel(clz, level);
}
Modified: uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/component/DuccLifecycle.java
URL: http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/component/DuccLifecycle.java?rev=1479988&r1=1479987&r2=1479988&view=diff
==============================================================================
--- uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/component/DuccLifecycle.java (original)
+++ uima/sandbox/uima-ducc/trunk/uima-ducc-common/src/main/java/org/apache/uima/ducc/common/component/DuccLifecycle.java Tue May 7 17:24:23 2013
@@ -18,9 +18,11 @@
*/
package org.apache.uima.ducc.common.component;
+import org.apache.uima.ducc.common.admin.event.DuccAdminEvent;
import org.apache.uima.ducc.common.main.DuccService;
public interface DuccLifecycle {
public void start(DuccService service) throws Exception;
public void stop() throws Exception;
+ public void handleAdminEvent(DuccAdminEvent event) throws Exception;
}