Author: challngr
Date: Tue Jun 17 14:38:50 2014
New Revision: 1603187
URL: http://svn.apache.org/r1603187
Log:
UIMA-3727 Updates
Modified:
uima/sandbox/uima-ducc/trunk/uima-ducc-examples/src/main/java/org/apache/uima/ducc/ping/SamplePing.java
Modified:
uima/sandbox/uima-ducc/trunk/uima-ducc-examples/src/main/java/org/apache/uima/ducc/ping/SamplePing.java
URL:
http://svn.apache.org/viewvc/uima/sandbox/uima-ducc/trunk/uima-ducc-examples/src/main/java/org/apache/uima/ducc/ping/SamplePing.java?rev=1603187&r1=1603186&r2=1603187&view=diff
==============================================================================
---
uima/sandbox/uima-ducc/trunk/uima-ducc-examples/src/main/java/org/apache/uima/ducc/ping/SamplePing.java
(original)
+++
uima/sandbox/uima-ducc/trunk/uima-ducc-examples/src/main/java/org/apache/uima/ducc/ping/SamplePing.java
Tue Jun 17 14:38:50 2014
@@ -16,9 +16,12 @@
* specific language governing permissions and limitations
* under the License.
*/
-
package org.apache.uima.ducc.ping;
+/*
+ * IMPORTANT: This is a sample custom pinger for illustration purposes only.
It is not
+ * supported in any way.
+ */
import java.io.IOException;
import java.io.StringReader;
@@ -47,34 +50,37 @@ import org.apache.uima.ducc.common.utils
import org.apache.uima.resource.ResourceInitializationException;
import org.apache.uima.util.Level;
+
+
// 'q_thresh=nn,window=mm,broker_jmx=1100,meta_timeout=10000'
public class SamplePing
extends AServicePing
{
- String ep;
+ String ep; // Endpoint, passed in during
initialization
+
+ String qname; // Service queue name, parsed from ep
+ String broker; // Service broker, parsed from ep
+ int meta_timeout = 5000; // default
+
+ String broker_host; // AMQ hostname, parsed from 'broker'
URL
+ int broker_jmx_port = 1099; // AMQ jmx port, default
- String endpoint;
- String broker;
- int meta_timeout = 5000;
-
- String broker_host;
- int broker_jmx_port = 1099;
- boolean connected;
- UimaAsServiceMonitor monitor;
- DuccLogger logger = null;
-
- String nodeIp;
- String pid;
- boolean gmfail = false;
+ UimaAsServiceMonitor monitor; // Part of ping API, knows how to
query AMQ for broker stats
- int additions = 0; // n additions to signal to SM
- Long[] deletions = null; // which instances to shrink
+ String nodeIp; // For UIMA-AS get-meta callback, IP
of node answering get-meta
+ String pid; // " " " " ,
process answering get-meta
+ boolean gmfail = false; // Did get=meta work?
+ boolean fast_shrink = true; // If false, don't shrink instances
if there are producers
+ // still connected to the Q.
- int min_instances = 0; // smallest instances to maintain
+ int additions = 0; // number of additions to signal to SM
+ Long[] deletions = null; // which specific instances to shrink
+
+ int min_instances = 0; // minimum instances to maintain
int max_instances = 10; // max instances to allow
int max_growth = 5; // max processes to grow at any time
- double goal = 2.00; // want to get enqueue time towith
this factor of individual service time
+ double goal = 2.00; // want to get enqueue time to within
this factor of individual service time
int cursor = 0; // growth and shrinkage window cursor
int expansion_period = 5; // size of window in minutes
@@ -84,31 +90,26 @@ public class SamplePing
public SamplePing()
{
- this.logger = null;
- }
- public SamplePing(DuccLogger logger)
- {
- this.logger = logger;
}
public void init(String args, String ep)
throws Exception
{
String methodName = "init";
- this.ep = ep;
+ this.ep = ep;
doLog(methodName, "Ping.init(" + args + ", " + ep + " start.");
- // Ep is of the form UIMA-AS:queuename:broker
+
+ // Ep is of the form UIMA-AS:queuename:broker. Parse out queue name
and broker URL
int ndx = ep.indexOf(":");
ep = ep.substring(ndx+1);
ndx = ep.indexOf(":");
- this.endpoint = ep.substring(0, ndx).trim();
+ this.qname = ep.substring(0, ndx).trim();
this.broker = ep.substring(ndx+1).trim();
- // broker is a URL that we need to parse in order to get the actual
host and port
- // for jmx
+ // broker is a URL that we need to parse in order to get the actual
host for jmx calls
URL url = null;
try {
url = new URL(null, broker, new TcpStreamHandler());
@@ -116,18 +117,19 @@ public class SamplePing
throw new IllegalArgumentException("Invalid broker URL: " +
broker);
}
broker_host = url.getHost();
- // not needed here fyi broker_port = url.getPort();
-
-
+
+
+ // inhibit noisy default UIMAAs logging
UIMAFramework.getLogger(BaseUIMAAsynchronousEngineCommon_impl.class).setLevel(Level.OFF);
UIMAFramework.getLogger(BaseUIMAAsynchronousEngine_impl.class).setLevel(Level.OFF);
- //UIMAFramework.getLogger().setLevel(Level.OFF);
- // there are a couple junky messages that slip by the above
configurations. turn the whole danged thing off.
UIMAFramework.getLogger().setLevel(Level.INFO);
+ // Parse out the pinger arguments which are comma-delimeted in the
argument string
+ // We do this by splitting on ',', then writing as strings to a
StringReader, and
+ // finally loading a PropertiesFile from them. DuccProperties is used
because it
+ // smart typed extraction of the properties.
if ( args != null ) {
- // 'q_thresh=nn,window=mm,broker_jmx_port=1100,meta_timeout=10000'
- // turn the argument string into properties
+
String[] as = args.split(",");
StringWriter sw = new StringWriter();
for ( String s : as ) sw.write(s + "\n");
@@ -142,23 +144,24 @@ public class SamplePing
meta_timeout = props.getIntProperty("meta-timeout", meta_timeout);
broker_jmx_port = props.getIntProperty("broker-jmx-port",
broker_jmx_port);
expansion_period = props.getIntProperty("window",
expansion_period);
- this.log_enabled = props.getBooleanProperty("enable-log",
this.log_enabled);
min_instances = props.getIntProperty("min", min_instances);
max_instances = props.getIntProperty("max", max_instances);
max_growth = props.getIntProperty("max-growth", max_growth);
+ fast_shrink = props.getBooleanProperty("fast-shrink", true);
goal = props.getDoubleProperty("goal", goal);
}
- // insure window is in minutes, with one pane per ping
+ // Set up expansion/deletion windows with window size always 1 minute
regardless
+ // of ping frequency.
double calls_per_minute = 60000.00 / monitor_rate;
window_size = (int) ( ((double)expansion_period) * calls_per_minute);
expansion_window = new int[window_size];
deletion_window = new int[window_size];
- System.out.println("log_enabled is " + log_enabled);
doLog("<ctr>",
"INIT: meta-timeout", meta_timeout,
+ "broker-host", broker_host,
"broker-jmx-port", broker_jmx_port,
"monitor-window", expansion_period,
"window-size", window_size,
@@ -167,12 +170,15 @@ public class SamplePing
"min-instances", min_instances,
"max-growth", max_growth,
"goal", goal,
- "enable-log", log_enabled);
+ "fast-shrink", fast_shrink);
- this.monitor = new UimaAsServiceMonitor(endpoint, broker_host,
broker_jmx_port);
+ // Set up the jmx queue monitor and reset statistics to insure each
calculation is
+ // based on current data, not old data.
+ this.monitor = new UimaAsServiceMonitor(qname, broker_host,
broker_jmx_port);
monitor.resetStatistics();
}
+ // Release resources when terminated.
public void stop()
{
String methodName = "stop";
@@ -180,6 +186,9 @@ public class SamplePing
doLog(methodName, "------------ Stop signal arrives, pinger exits.");
}
+ // This is called by the Service Manager to collect the pinger data. This
pinger
+ // issues a get-meta to the service to insure it's alive, then collects
queue statistics via JMX
+ // and determines expansion and deletion for return when SM asks about
them.
public IServiceStatistics getStatistics()
{
String methodName = "getStatistics";
@@ -189,30 +198,30 @@ public class SamplePing
// "health" has no real meaning. maybe we can get rid of it one day?
try {
- monitor.collect();
+ monitor.collect(); // Get jmx stuff
statistics.setHealthy(true);
} catch ( Throwable t ) {
statistics.setHealthy(false);
monitor.setJmxFailure(t.getMessage());
}
- // Instantiate Uima AS Client
+ // Instantiate Uima AS Client and issue get-meta
BaseUIMAAsynchronousEngine_impl uimaAsEngine = new
BaseUIMAAsynchronousEngine_impl();
UimaCbListener listener = new UimaCbListener();
uimaAsEngine.addStatusCallbackListener(listener);
Map<String, Object> appCtx = new HashMap<String, Object>();
appCtx.put(UimaAsynchronousEngine.ServerUri, broker);
- appCtx.put(UimaAsynchronousEngine.ENDPOINT, endpoint);
+ appCtx.put(UimaAsynchronousEngine.ENDPOINT, qname);
appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, meta_timeout);
try {
- uimaAsEngine.initialize(appCtx);
+ uimaAsEngine.initialize(appCtx); // This
performs the get-meta
statistics.setAlive(true);
statistics.setHealthy(true && statistics.isHealthy());
listener.ok();
} catch( ResourceInitializationException e) {
- listener.timeout();
- doLog(methodName, "Cannot issue getMeta to: " + endpoint + ":" +
broker);
+ listener.timeout(); // Service not
responding
+ doLog(methodName, "Cannot issue getMeta to: " + qname + ":" +
broker);
statistics.setHealthy(false);
statistics.setAlive(false);
} finally {
@@ -223,20 +232,20 @@ public class SamplePing
}
}
- monitor.setSource(nodeIp, pid, gmfail);
- statistics.setInfo(monitor.format());
+ monitor.setSource(nodeIp, pid, gmfail); // remember
who responded
+ statistics.setInfo(monitor.format()); // set string
for web server
- calculateNewDeployment(statistics);
+ calculateNewDeployment(statistics); // Decide on
instance expansion or deletion
return statistics;
}
/**
- * Override from AServicePing
+ * Override from AServicePing, set in calculateNewDeployment while
analyzing queue statistics
*/
public long getLastUse()
{
- return last_use;
+ return last_use;
}
//
================================================================================
@@ -293,7 +302,7 @@ public class SamplePing
int new_ni = 0;
// what we're looking for it Tt: the average time it takes on thread
to compute one unit of work.
if ( Q == 0 ) {
- if ( pc != 0 ) {
+ if ( (pc != 0) && ( !fast_shrink) ) {
doLog(methodName, "Inhibit shrinkage because pc =", pc, "Q =",
Q);
deletion_window[cursor] = 0;
} else {
@@ -311,24 +320,20 @@ public class SamplePing
double g = Tt * goal; // we want to get eT close to this
double r = eT / g; // the ratio of current queue to to
desired, which is the amount we need to
// change the instances by
- doLog(methodName, "eT", eT, "Q", Q, "cc", cc, "Ti", Ti, "Tt", Tt,
"g", g, "r", r, "active", active, "ninstances", ninst, "max_instances",
max_instances);
+ doLog(methodName, "eT", eT, "Q", Q, "cc", cc, "Ti", Ti, "Tt", Tt,
"g", g, "r", r,
+ "active", active, "ninstances", ninst, "max_instances",
max_instances);
if ( r > 1 ) { // we want moref
// We must smooth it out. First get the delta instances.
+ // Then cap it accounting for instances started but not yet
connected to the q
// Then cap it on the max for the service.
// Then cap it on (arbitrarily) 5 so we don't blast too many
new instances at once
- new_ni = (int) Math.ceil(active * r);
- doLog(methodName, "1 new_ni", new_ni);
-
+ new_ni = (int) Math.ceil(active * r); // delta
new_ni = Math.max(new_ni - ninst, 0); // here we account for
intances that aren't yet started
- doLog(methodName, "2 new_ni", new_ni);
-
- new_ni = Math.min(new_ni, max_instances);
- doLog(methodName, "3 new_ni", new_ni);
+ new_ni = Math.min(new_ni, max_instances); // cap on configured
max
+ new_ni = Math.min(new_ni, max_growth); // cap on growth rate
- new_ni = Math.min(new_ni, max_growth);
- doLog(methodName, "4 new_ni", new_ni);
if ( new_ni > 0 ) { // do we actually
expand in the end?
doLog(methodName, "Expand, new_n1:", new_ni);
@@ -343,7 +348,7 @@ public class SamplePing
}
if ( r < .5 ) { // we're
over-provisioned to the goal
- if ( pc != 0 ) {
+ if ( (pc != 0) && (!fast_shrink) ) {
// never shrink if there are producers
doLog(methodName, "Inhibit shrinkage because pc =", pc,
"r=", r);
deletion_window[cursor] = 0;
@@ -397,7 +402,8 @@ public class SamplePing
// NEW INTERFACES
//
================================================================================
/**
- * Implement this to indicate how many new instances to start.
+ * Implement this to indicate how many new instances to start. The value
here
+ * is calculted above in calculateNewDeployment.
*/
public int getAdditions()
{
@@ -405,7 +411,8 @@ public class SamplePing
}
/**
- * Implement this to indicate how many instances to stop
+ * Implement this to indicate how many instances to stop. The value here
+ * is calculted above in calculateNewDeployment.
*/
public Long[] getDeletions()
{
@@ -416,6 +423,10 @@ public class SamplePing
// END NEW INTERFACES
//
================================================================================
+ /**
+ * This is a callback class for the UIMA-AS get-meta, that tells us which
process on
+ * which host responded to the get-meta.
+ */
class UimaCbListener extends UimaAsBaseCallbackListener
{
public UimaCbListener()
