Added:
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServiceInstance.java
URL:
http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServiceInstance.java?rev=1564022&view=auto
==============================================================================
---
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServiceInstance.java
(added)
+++
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServiceInstance.java
Mon Feb 3 19:35:48 2014
@@ -0,0 +1,341 @@
+/*
+ * 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.sm;
+
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Map;
+
+import org.apache.uima.ducc.common.utils.DuccLogger;
+import org.apache.uima.ducc.common.utils.DuccProperties;
+import org.apache.uima.ducc.transport.event.common.IDuccState.JobState;
+
+/**
+* Represent a single instance.
+*
+* This is a simple class, mostly just a container for the state machine.
+*/
+@SuppressWarnings("serial")
+class ServiceInstance
+ implements SmConstants
+{
+ private DuccLogger logger =
DuccLogger.getLogger(this.getClass().getName(), COMPONENT_NAME);
+
+ long numeric_id; // unique numeric
ducc-assigned id
+ ServiceSet sset; // handle to the service
definitiopn
+
+ JobState state = JobState.Undefined; // orchestartor state
+ String user = null;
+
+ boolean stopped; // careful .. this means it
was stopped by a stop order from somewhere,
+ // NOT that it's
terminating
+
+ String ducc_home = System.getProperty("DUCC_HOME");
+ String api_classpath = ducc_home + "/lib/uima-ducc-cli.jar" + ":" +
System.getProperty("java.class.path");
+
+ ServiceInstance(ServiceSet sset)
+ {
+ this.numeric_id = -1;
+ this.sset = sset;
+ this.stopped = true;
+ }
+
+ public long getId() {
+ return this.numeric_id;
+ }
+
+ void setId(long id) {
+ this.numeric_id = id;
+ }
+
+ void setUser(String user)
+ {
+ this.user = user;
+ }
+
+ public void setState(JobState state)
+ {
+ this.state = state;
+ }
+
+ public JobState getState()
+ {
+ return this.state;
+ }
+
+ /**
+ * Stopped by some stop order?
+ */
+ public synchronized boolean isStopped()
+ {
+ return this.stopped;
+ }
+
+ /**
+ * On it's way up, or already up, and not stopped for any reason.
+ */
+ public synchronized boolean isRunning()
+ {
+ // String methodName = "setState";
+ // Received, // Job has been vetted,
persisted, and assigned unique Id
+ // WaitingForDriver, // Process Manager is launching
Job Driver
+ // WaitingForServices, // Service Manager is
checking/starting services for Job
+ // WaitingForResources, // Scheduler is assigning resources to
Job
+ // Initializing, // Process Agents are
initializing pipelines
+ // Running, // At least one Process
Agent has reported process initialization complete
+ // Completing, // Job processing is
completing
+ // Completed, // Job processing is
completed
+ // Undefined // None of the above
+
+ switch ( state ) {
+ case Completing:
+ case Completed:
+ return false;
+ default:
+ return !isStopped();
+ }
+ }
+
+ synchronized void setStopped(boolean s)
+ {
+ this.stopped = s;
+ }
+
+
+// void setState(DuccWorkJob dwj)
+// {
+// this.state = dwj.getJobState();
+// }
+
+ long start(String spec, DuccProperties meta_props)
+ {
+ String methodName = "start";
+
+ logger.info(methodName, sset.getId(), "START INSTANCE");
+ setStopped(false);
+ this.user = meta_props.getProperty("user");
+
+ // Simple use of ducc_ling, just submit as the user. The
specification will have the working directory
+ // and classpath needed for the service, handled by the Orchestrator
and Job Driver.
+ String[] args = {
+ System.getProperty("ducc.agent.launcher.ducc_spawn_path"),
+ "-u",
+ user,
+ "--",
+ System.getProperty("ducc.jvm"),
+ "-cp",
+ api_classpath,
+ "org.apache.uima.ducc.cli.DuccServiceSubmit",
+ "--specification",
+ spec
+ };
+
+ for ( int i = 0; i < args.length; i++ ) {
+ if ( i > 0 && (args[i-1].equals("-cp") ) ) {
+ // The classpaths can be just awful filling the logs with
junk. It will end up in the agent log
+ // anyway so let's inhibit it here.
+ logger.debug(methodName, sset.getId(), "Args[", i, "]:
<CLASSPATH>");
+ } else {
+ logger.debug(methodName, sset.getId(), "Args[", i, "]:",
args[i]);
+ }
+ }
+
+ ProcessBuilder pb = new ProcessBuilder(args);
+ Map<String, String> env = pb.environment();
+ env.put("DUCC_HOME", System.getProperty("DUCC_HOME"));
+
+ ArrayList<String> stdout_lines = new ArrayList<String>();
+ ArrayList<String> stderr_lines = new ArrayList<String>();
+ try {
+ Process p = pb.start();
+
+ int rc = p.waitFor();
+ logger.debug(methodName, null, "DuccServiceSubmit returns with
rc", rc);
+
+ // TODO: we should attach these streams to readers in threads
because too much output
+ // can cause blocking, deadlock, ugliness.
+ InputStream stdout = p.getInputStream();
+ InputStream stderr = p.getErrorStream();
+ BufferedReader stdout_reader = new BufferedReader(new
InputStreamReader(stdout));
+ BufferedReader stderr_reader = new BufferedReader(new
InputStreamReader(stderr));
+ String line = null;
+ while ( (line = stdout_reader.readLine()) != null ) {
+ stdout_lines.add(line);
+ }
+
+ line = null;
+ while ( (line = stderr_reader.readLine()) != null ) {
+ stderr_lines.add(line);
+ }
+
+ } catch (Throwable t) {
+ // TODO Auto-generated catch block
+ logger.error(methodName, null, t);
+ }
+
+ for ( String s : stderr_lines ) {
+ logger.info(methodName, sset.getId(), "Start stderr:", s);
+ }
+
+ // That was annoying. Now search the lines for some hint of the id.
+ boolean inhibit_cp = false;
+ boolean started = false;
+ StringBuffer submit_buffer = new StringBuffer();
+ boolean recording = false;
+ for ( String s : stdout_lines ) {
+
+ // simple logic to inhibit printing the danged classpath
+ if ( inhibit_cp ) {
+ inhibit_cp = false;
+ logger.info(methodName, sset.getId(), "<INHIBITED CP>");
+ } else {
+ logger.info(methodName, sset.getId(), "Start stdout:", s);
+ }
+
+ if ( s.indexOf("-cp") >= 0 ) {
+ inhibit_cp = true;
+ }
+
+ if ( recording ) {
+ submit_buffer.append(s.trim());
+ submit_buffer.append(";");
+ }
+ if ( s.startsWith("1001 Command launching...") ) {
+ recording = true;
+ continue;
+ }
+
+ if ( s.startsWith("Service") && s.endsWith("submitted") ) {
+ String[] toks = s.split("\\s");
+ try {
+ numeric_id = Long.parseLong(toks[1]);
+ started = true;
+ logger.info(methodName, null, "Request to start service "
+ sset.getId().toString() + " accepted as service instance ", numeric_id);
+ } catch ( NumberFormatException e ) {
+ logger.warn(methodName, null, "Request to start service "
+ sset.getId().toString() + " failed, can't interpret response.: " + s);
+ }
+
+ }
+ }
+
+ if ( ! started ) {
+ logger.warn(methodName, null, "Request to start service " +
sset.getId().toString() + " failed.");
+ meta_props.put("submit_error", submit_buffer.toString());
+ sset.log_errors(stdout_lines, stderr_lines);
+ } else {
+ meta_props.remove("submit_error");
+ state = JobState.Received;
+ }
+ logger.info(methodName, sset.getId(), "START INSTANCE COMPLETE");
+
+ return numeric_id;
+ }
+
+
+ /**
+ * This assumes the caller has already verified that I'm a registered
service.
+ */
+ void stop()
+ {
+ String methodName = "stop";
+
+ setStopped(true);
+ String[] args = {
+ System.getProperty("ducc.agent.launcher.ducc_spawn_path"),
+ "-u",
+ user,
+ "--",
+ System.getProperty("ducc.jvm"),
+ "-cp",
+ api_classpath,
+ "org.apache.uima.ducc.cli.DuccServiceCancel",
+ "--id",
+ Long.toString(numeric_id),
+ };
+
+ for ( int i = 0; i < args.length; i++ ) {
+ if ( i > 0 && (args[i-1].equals("-cp") ) ) {
+ // The classpaths can be just awful filling the logs with
junk. It will end up in the agent log
+ // anyway so let's inhibit it here.
+ logger.debug(methodName, sset.getId(), "Instance", numeric_id,
"Args[", i, "]: <CLASSPATH>");
+ } else {
+ logger.debug(methodName, sset.getId(), "Instance", numeric_id,
"Args[", i, "]:", args[i]);
+ }
+ }
+
+ ProcessBuilder pb = new ProcessBuilder(args);
+ Map<String, String> env = pb.environment();
+ env.put("DUCC_HOME", System.getProperty("DUCC_HOME"));
+
+ ArrayList<String> stdout_lines = new ArrayList<String>();
+ ArrayList<String> stderr_lines = new ArrayList<String>();
+ int rc = 0;
+ try {
+ Process p = pb.start();
+
+ rc = p.waitFor();
+ logger.info(methodName, sset.getId(), "DuccServiceCancel returns
with rc", rc);
+
+ if (logger.isTrace() || (rc != 0)) {
+ InputStream stdout = p.getInputStream();
+ InputStream stderr = p.getErrorStream();
+ BufferedReader stdout_reader = new BufferedReader(new
InputStreamReader(stdout));
+ BufferedReader stderr_reader = new BufferedReader(new
InputStreamReader(stderr));
+
+ String line = null;
+ while ( (line = stdout_reader.readLine()) != null ) {
+ stdout_lines.add(line);
+ }
+
+ line = null;
+ while ( (line = stderr_reader.readLine()) != null ) {
+ stderr_lines.add(line);
+ }
+ }
+
+ } catch (Throwable t) {
+ // TODO Auto-generated catch block
+ logger.error(methodName, null, t);
+ }
+
+ if ( logger.isTrace() || ( rc != 0) ) {
+ boolean inhibit_cp = false;
+ for ( String s : stdout_lines ) {
+ // simple logic to inhibit printing the danged classpath
+ if ( inhibit_cp ) {
+ inhibit_cp = false;
+ logger.info(methodName, sset.getId(), "Instance",
numeric_id, "<INHIBITED CP>");
+ } else {
+ logger.info(methodName, sset.getId(), "Instance",
numeric_id, "Stop stdout:", s);
+ }
+
+ if ( s.indexOf("-cp") >= 0 ) {
+ inhibit_cp = true;
+ }
+ }
+
+ for ( String s : stderr_lines ) {
+ logger.info(methodName, sset.getId(), "Instance", numeric_id,
"Stop stderr:", s);
+ }
+ }
+ }
+
+}
Propchange:
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServiceInstance.java
------------------------------------------------------------------------------
svn:executable = *
Modified:
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServiceManagerComponent.java
URL:
http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServiceManagerComponent.java?rev=1564022&r1=1564021&r2=1564022&view=diff
==============================================================================
---
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServiceManagerComponent.java
(original)
+++
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServiceManagerComponent.java
Mon Feb 3 19:35:48 2014
@@ -26,8 +26,8 @@ import java.util.Map;
import java.util.Properties;
import java.util.UUID;
-import org.apache.uima.ducc.cli.IUiOptions.UiOption;
import org.apache.camel.CamelContext;
+import org.apache.uima.ducc.cli.IUiOptions.UiOption;
import org.apache.uima.ducc.cli.UimaAsPing;
import org.apache.uima.ducc.common.boot.DuccDaemonRuntimeProperties;
import org.apache.uima.ducc.common.boot.DuccDaemonRuntimeProperties.DaemonName;
@@ -59,7 +59,6 @@ import org.apache.uima.ducc.transport.ev
import org.apache.uima.ducc.transport.event.SmStateDuccEvent;
import org.apache.uima.ducc.transport.event.common.DuccWorkJob;
import org.apache.uima.ducc.transport.event.common.DuccWorkMap;
-import org.apache.uima.ducc.transport.event.common.IDuccState.JobState;
import org.apache.uima.ducc.transport.event.common.IDuccTypes.DuccType;
import org.apache.uima.ducc.transport.event.common.IDuccWork;
import org.apache.uima.ducc.transport.event.common.IDuccWorkService;
@@ -112,6 +111,7 @@ public class ServiceManagerComponent
private boolean testmode = false;
Map<String, String> administrators = new HashMap<String, String>();
+ String version = "1.1.0";
public ServiceManagerComponent(CamelContext context)
{
@@ -120,9 +120,7 @@ public class ServiceManagerComponent
handler = new ServiceHandler(this);
}
- public DuccLogger getLogger() {
- return logger;
- }
+
/**
* Initialization tasks:
* - read all the service descriptors
@@ -154,14 +152,7 @@ public class ServiceManagerComponent
DuccProperties metaprops = new DuccProperties();
String meta_filename = serviceFileKey(stem + ".meta");
- metaprops.load(meta_filename);
-
- String sc = metaprops.getProperty("service-class");
- if ( (sc != null) && ( sc.equals("Implicit") ||
sc.equals("Submitted") ) ) {
- logger.info(methodName, null, "Scrubbing", sc,
"service", stem);
- deleteProperties(stem, meta_filename, metaprops,
props_filename, props);
- continue;
- }
+ metaprops.load(meta_filename);
int friendly = 0;
String uuid = "";
@@ -261,11 +252,11 @@ public class ServiceManagerComponent
super.start(service, args);
DuccDaemonRuntimeProperties.getInstance().boot(DaemonName.ServiceManager,getProcessJmxUrl());
- // (UIMA-3397) failure_max =
SystemPropertyResolver.getIntProperty("ducc.sm.instance.failure.max",
failure_max);
+ failure_max =
SystemPropertyResolver.getIntProperty("ducc.sm.instance.failure.max",
failure_max);
meta_ping_rate =
SystemPropertyResolver.getIntProperty("ducc.sm.meta.ping.rate", meta_ping_rate);
meta_ping_timeout =
SystemPropertyResolver.getIntProperty("ducc.sm.meta.ping.timeout",
meta_ping_timeout);
meta_ping_stability =
SystemPropertyResolver.getIntProperty("ducc.sm.meta.ping.stability",
meta_ping_stability);
- default_ping_class =
SystemPropertyResolver.getStringProperty("ducc.sm.default.uima-as.ping.class",
UimaAsPing.class.getName());
+ default_ping_class =
SystemPropertyResolver.getStringProperty("ducc.sm.default.monitor.class",
UimaAsPing.class.getName());
String rm = SystemPropertyResolver.getStringProperty("ducc.runmode",
"");
if ( rm.equals("Test") ) testmode = true;
@@ -278,7 +269,7 @@ public class ServiceManagerComponent
logger.warn(methodName, null, "Incorrect value for property
ducc.signature.required: " + sig + ". Setting to default of \"on\"");
}
- logger.info(methodName, null,
"------------------------------------------------------------------------------------");
+ logger.info(methodName, null, "---------------------------- NEW
-----------------------------------------------------");
logger.info(methodName, null, "Service Manager starting:");
logger.info(methodName, null, " DUCC home : ",
System.getProperty("DUCC_HOME"));
logger.info(methodName, null, " ActiveMQ URL : ",
System.getProperty("ducc.broker.url"));
@@ -295,6 +286,7 @@ public class ServiceManagerComponent
logger.info(methodName, null, " Service ping stability : ",
meta_ping_stability);
logger.info(methodName, null, " Instance Failure Max : ",
failure_max);
logger.info(methodName, null, " DUCC Version : ",
Version.version());
+ logger.info(methodName, null, " SM Version : ",
version);
logger.info(methodName, null,
"------------------------------------------------------------------------------------");
readAdministrators();
@@ -334,16 +326,16 @@ public class ServiceManagerComponent
* At boot only ... pass in the set of all known active services to each
service so it can update
* internal state with current published state.
*/
- public synchronized void synchronizeHandler(DuccWorkMap work)
+ public synchronized void bootHandler(DuccWorkMap work)
{
- Map<DuccId, JobState> ids = new HashMap<DuccId, JobState>();
+ Map<DuccId, DuccWorkJob> services = new HashMap<DuccId, DuccWorkJob>();
for ( IDuccWork w : work.values() ) {
if ( w.getDuccType() != DuccType.Service ) continue;
DuccWorkJob j = (DuccWorkJob) w;
if ( !j.isActive() ) continue;
- ids.put(j.getDuccId(), j.getJobState());
+ services.put(j.getDuccId(), j);
}
- handler.synchronizeImplementors(ids);
+ handler.bootImplementors(services);
}
/**
@@ -368,7 +360,7 @@ public class ServiceManagerComponent
HashMap<DuccId, IDuccWork> modifiedJobs = new HashMap<DuccId,
IDuccWork>();
HashMap<DuccId, IDuccWork> modifiedServices = new HashMap<DuccId,
IDuccWork>();
- logger.debug(methodName, null, "---Processing Orchestrator
State---");
+ logger.info(methodName, null, "===== Orchestrator State Arrives
=====");
if ( workMap.size() == 0 ) {
logger.debug(methodName, null, "OR state is empty");
@@ -379,8 +371,7 @@ public class ServiceManagerComponent
DuccMapDifference<DuccId, IDuccWork> diffmap =
DuccCollectionUtils.difference(workMap, localMap);
for ( IDuccWork w : workMap.values() ) {
- //IDuccWork j = (IDuccWork) w;
- logger.debug(methodName, w.getDuccId(), w.getDuccType(), "Arrives
in state =", w.getStateObject());
+ logger.trace(methodName, w.getDuccId(), w.getDuccType(), "Arrives
in state =", w.getStateObject());
}
// Stuff on the left is new
@@ -426,7 +417,7 @@ public class ServiceManagerComponent
for ( IDuccWork w : work.values() ) {
if ( w.getDuccType() == DuccType.Reservation ) continue;
- logger.debug(methodName, w.getDuccId(), "Reconciling, deleting",
w.getDuccType());
+ logger.debug(methodName, w.getDuccId(), "Reconciling, deleting
instance of type ", w.getDuccType());
switch(w.getDuccType()) {
case Job:
localMap.removeDuccWork(w.getDuccId());
@@ -460,7 +451,25 @@ public class ServiceManagerComponent
if ( l.getDuccType() == DuccType.Reservation ) continue;
- logger.debug(methodName, l.getDuccId(), "Reconciling, incoming
state = ", l.getStateObject(), " my state = ", r.getStateObject());
+ if ( l.getStateObject() != r.getStateObject() ) {
+ String serviceType = "/ Job";
+ switch ( l.getDuccType() ) {
+ case Service:
+ switch (
((IDuccWorkService)l).getServiceDeploymentType() )
+ {
+ case uima:
+ case custom:
+ serviceType = "/ Service";
+ break;
+ case other:
+ serviceType = "/ ManagedReservation";
+ break;
+ }
+ break;
+
+ }
+ logger.debug(methodName, l.getDuccId(), "Reconciling",
l.getDuccType(), serviceType, "incoming state = ", l.getStateObject(), " my
state = ", r.getStateObject());
+ }
// Update our own state by replacing the old (right) object with
the new (left)
switch(l.getDuccType()) {
@@ -506,9 +515,9 @@ public class ServiceManagerComponent
String methodName = "publish";
try {
SmStateDuccEvent ev = new SmStateDuccEvent();
- logger.info(methodName, null, "Publishing State, active job count
=", map.size());
+ logger.trace(methodName, null, "Publishing State, active job count
=", map.size());
if (logger.isDebug()) {
- logger.debug(methodName, null, map.toPrint());
+ logger.info(methodName, null, map.toPrint());
}
ev.setServiceMap(map);
eventDispatcher.dispatch(stateEndpoint, ev, ""); // tell the
world what is scheduled (note empty string)
@@ -540,7 +549,7 @@ public class ServiceManagerComponent
try {
if ( first_update ) {
- synchronizeHandler(incomingMap);
+ bootHandler(incomingMap);
first_update = false;
}
processIncoming(incomingMap);
@@ -594,7 +603,7 @@ public class ServiceManagerComponent
userHome = LinuxUtils.getUserHome(user);
}
- Crypto crypto = new Crypto(user,userHome,AccessType.READER);
+ Crypto crypto = new Crypto(user, userHome,AccessType.READER);
String signature = (String)crypto.decrypt(auth_block);
return user.equals(signature);
Modified:
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServicePingMain.java
URL:
http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServicePingMain.java?rev=1564022&r1=1564021&r2=1564022&view=diff
==============================================================================
---
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServicePingMain.java
(original)
+++
uima/sandbox/uima-ducc/trunk/uima-ducc-sm/src/main/java/org/apache/uima/ducc/sm/ServicePingMain.java
Mon Feb 3 19:35:48 2014
@@ -20,20 +20,16 @@ package org.apache.uima.ducc.sm;
import java.io.IOException;
import java.io.InputStream;
+import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.UnknownHostException;
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.OptionBuilder;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.apache.commons.cli.PosixParser;
import org.apache.uima.ducc.cli.AServicePing;
import org.apache.uima.ducc.cli.ServiceStatistics;
import org.apache.uima.ducc.common.IServiceStatistics;
+import org.apache.uima.ducc.common.utils.DuccProperties;
/**
@@ -56,49 +52,28 @@ public class ServicePingMain
int error_max = 10;
int error_count = 0;
+ DuccProperties clioptions = new DuccProperties();
+
public ServicePingMain()
{
-
+ clioptions.put("--class", clioptions);
+ clioptions.put("--endpoint", clioptions);
+ clioptions.put("--port", clioptions);
+ clioptions.put("--arguments", clioptions);
}
- @SuppressWarnings("static-access")
- private void addOptions(Options options)
+ static void usage()
{
- //
- // Verbs here
- //
- options.addOption(OptionBuilder
- .withLongOpt (ServicePing.Class.decode())
- .withDescription(ServicePing.Class.description())
- .withArgName (ServicePing.Class.argname())
- .hasOptionalArg ()
- .create ()
- );
-
- options.addOption(OptionBuilder
- .withLongOpt (ServicePing.Endpoint.decode())
- .withDescription(ServicePing.Endpoint.description())
- .withArgName (ServicePing.Endpoint.argname())
- .hasArg (true)
- .create ()
- );
-
- options.addOption(OptionBuilder
- .withLongOpt (ServicePing.Port.decode())
- .withDescription(ServicePing.Port.description())
- .withArgName (ServicePing.Port.argname())
- .hasArg (true)
- .create ()
- );
-
- options.addOption(OptionBuilder
- .withLongOpt (ServicePing.Port.decode())
- .withDescription(ServicePing.Port.description())
- .withArgName (ServicePing.Port.argname())
- .hasArg (true)
- .create ()
- );
+ System.out.println("Usage:");
+ System.out.println(" java org.apache.uima.ducc.smnew.ServicePingMain
<args>");
+ System.out.println("Where args are:");
+ System.out.println(" --class classname This is the class
implementing the pinger.");
+ System.out.println(" --endpoint ep This is the
endpoint specified in the registration.");
+ System.out.println(" --port port This is the listen
port the SM is listening on.");
+ System.out.println(" --arguments classname These are the
arguments for the pinger, supplied in the registration.");
+
+ System.exit(1);
}
static void appendStackTrace(StringBuffer s, Throwable t)
@@ -151,15 +126,55 @@ public class ServicePingMain
return pinger;
}
- void handleError(Throwable t)
+ void handleError(AServicePing custom, Throwable t)
{
t.printStackTrace();
if ( ++error_count >= error_max ) {
+ custom.stop();
System.out.println("Exceeded error count. Exiting.");
System.exit(1);
}
}
+ /**
+ * Simple argument parser for this class. It is spawned only by SM so
even though we do
+ * validity checking, we assume the args are correct and complete, and
just crash hard if not as
+ * it's an internal error that should not occur.
+ */
+ void parseOptions(String[] args)
+ {
+ // First read them all in
+ if ( debug ) {
+ for ( int i = 0; i < args.length; i++ ) {
+ System.out.println("Args[" + i + "] = " + args[i]);
+ }
+ }
+
+ for ( int i = 0; i < args.length; ) {
+ if ( clioptions.containsKey(args[i]) ) {
+ if ( clioptions.get(args[i]) != clioptions ) {
+ System.out.println("Duplicate argument, not allowed: " +
args[i]);
+ System.exit(1);
+ }
+ System.out.println("Put " + args[i] + ", " + args[i+1]);
+ clioptions.put(args[i], args[i+1]);
+ i += 2;
+ } else {
+ System.out.println("Invalid argument: " + args[i]);
+ System.exit(1);
+ }
+ }
+
+ // Now make sure they all exist
+ for ( Object o : clioptions.keySet() ) {
+ String k = (String) o;
+ if ( clioptions.get(k) == clioptions ) {
+ System.out.println("Missing argument: " + k);
+ System.exit(1);
+ }
+ }
+ }
+
//
// 1. Instantiate the pinger if possible.
// 2. Read ducc.proeprties to find the ping interval
@@ -168,42 +183,34 @@ public class ServicePingMain
// The ServiceManager must start this process as the user. It monitors
stdout for success
// or failute of the ping and reacts accordingly.
//
- protected void start(String[] args)
+ protected int start(String[] args)
{
- Options options = new Options();
- addOptions(options);
-
- CommandLineParser parser = new PosixParser();
- CommandLine commandLine = null;
IServiceStatistics default_statistics = new ServiceStatistics(false,
false, "<N/A>");
- try {
- commandLine = parser.parse(options, args);
- } catch (ParseException e) {
- print("Cannot parse command line:", e);
- return;
- }
-
- String arguments =
commandLine.getOptionValue(ServicePing.Arguments.decode());
- String pingClass =
commandLine.getOptionValue(ServicePing.Class.decode());
- String endpoint =
commandLine.getOptionValue(ServicePing.Endpoint.decode());
- String port =
commandLine.getOptionValue(ServicePing.Port.decode());
+ parseOptions(args);
+ String arguments = clioptions.getStringProperty("--arguments");
+ String pingClass = clioptions.getStringProperty("--class");
+ String endpoint = clioptions.getStringProperty("--endpoint");
+ int port = clioptions.getIntProperty("--port");
Socket sock = null;
+
try {
- sock = new Socket("localhost", Integer.parseInt(port));
+ sock = new Socket("localhost", port);
} catch (NumberFormatException e2) {
e2.printStackTrace();
- return;
+ return 1;
} catch (UnknownHostException e2) {
e2.printStackTrace();
- return;
+ return 1;
} catch (IOException e2) {
e2.printStackTrace();
- return;
- }
+ return 1;
+ } finally {
+ try {sock.close();} catch ( Throwable t) {}
+ }
print ("ServicePingMain listens on port", sock.getLocalPort());
InputStream sock_in = null;
OutputStream sock_out = null;
@@ -212,7 +219,7 @@ public class ServicePingMain
sock_out = sock.getOutputStream();
} catch (IOException e2) {
e2.printStackTrace();
- return;
+ return 1;
}
ObjectOutputStream oos;
@@ -221,42 +228,66 @@ public class ServicePingMain
oos.flush();
} catch (IOException e1) {
e1.printStackTrace();
- return;
+ return 1;
+ }
+
+ ObjectInputStream ois;
+ try {
+ ois = new ObjectInputStream(sock_in);
+ } catch (IOException e1) {
+ e1.printStackTrace();
+ return 1;
}
AServicePing custom = resolve(pingClass, arguments, endpoint);
if ( custom == null ) {
print("bad_pinger:", pingClass, endpoint);
- return;
+ return 1;
}
while ( true ) {
if ( debug ) print("ServicePingMeta starts ping.");
-
- byte[] cmd = new byte[1];
- cmd[0] = 0;
- int eof = 0;
+
+ Ping ping = null;
try {
- eof = sock_in.read(cmd);
+ ping = (Ping) ois.readObject();
+ if ( debug ) {
+ print("Implementors:", ping.getInstances());
+ print("References:" , ping.getReferences());
+ }
} catch (IOException e) {
- handleError(e);
+ handleError(custom, e);
+ } catch ( ClassNotFoundException e) {
+ handleError(custom, e);
}
- if ( debug ) print("Read cmd", new String(cmd), "eof", eof);
-
- if ( eof == -1 ) {
- print("EOF on input pipe. Exiting");
- custom.stop();
- return;
- }
+
+ boolean quit = ping.isQuit();
+ if ( debug ) print("Read ping: ", quit);
try {
- if ( cmd[0] == 'P' ) {
+ if ( quit ) {
+ if ( debug ) System.out.println("Calling custom.stop");
+ custom.stop();
+ oos.close();
+ ois.close();
+ sock.close();
+ if ( debug ) System.out.println("Custom.stop returns");
+ return 0;
+ } else {
+ Pong pr = new Pong();
+ custom.setSmState(ping.getSmState());
IServiceStatistics ss = custom.getStatistics();
if ( ss == null ) {
ss = default_statistics;
}
- // print("Is alive: " + ss.isAlive());
- oos.writeObject(ss);
+
+ pr.setStatistics (ss);
+ pr.setAdditions (custom.getAdditions());
+ pr.setDeletions (custom.getDeletions());
+ pr.setExcessiveFailures(custom.isExcessiveFailures());
+
+
+ oos.writeObject(pr);
oos.flush();
// The ObjectOutputStream will cache instances and if all
you do is change a
@@ -265,14 +296,9 @@ public class ServicePingMain
// clone() here also if you want, but this is simplest and
safest since we have
// no control over what the external pinger gives us.
oos.reset();
- } else if ( cmd[0] == 'Q' ) {
- custom.stop();
- return;
- } else {
- System.err.println("Invalid command
recieved: " + Byte.toString(cmd[0]));
- }
- } catch (Throwable e) {
- handleError(e);
+ }
+ } catch (Throwable e) {
+ handleError(custom, e);
}
}
}
@@ -280,7 +306,9 @@ public class ServicePingMain
public static void main(String[] args)
{
ServicePingMain wrapper = new ServicePingMain();
- wrapper.start(args);
+ int rc = wrapper.start(args);
+ System.exit(rc);
}
}
+
