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); } } +