while ( nextElement != null )
{
//DEBUG
//System.out.println("FOUND: " + nextElement.toString());
TestLogstashExtractorPojo testInfo = TestLogstashExtractorPojo.fromDb(nextElement, TestLogstashExtractorPojo.class);
if ((null == testInfo.maxDocs) || (null == testInfo.logstash.config) || (null == testInfo.isAdmin) || (null == testInfo.sourceKey))
{
TestLogstashExtractorPojo testErr = new TestLogstashExtractorPojo();
testErr._id = testInfo._id;
testErr.error = "Internal Logic Error. Missing one of: maxDocs, isAdmin, sourceKey, logstash.config";
_logHarvesterQ.push(testErr.toDb());
return;
}//TESTED
// Validate/tranform the configuration:
StringBuffer errMessage = new StringBuffer();
String logstashConfig = LogstashConfigUtils.validateLogstashInput(testInfo.sourceKey, testInfo.logstash.config, errMessage, testInfo.isAdmin);
if (null == logstashConfig) { // Validation error...
TestLogstashExtractorPojo testErr = new TestLogstashExtractorPojo();
testErr._id = testInfo._id;
testErr.error = "Validation error: " + errMessage.toString();
_logHarvesterQ.push(testErr.toDb());
return;
}//TESTED
// Replacement for #LOGSTASH{host} - currently only replacement supported (+ #IKANOW{} in main code)
try {
logstashConfig = logstashConfig.replace("#LOGSTASH{host}", java.net.InetAddress.getLocalHost().getHostName());
}
catch (Exception e) {
logstashConfig = logstashConfig.replace("#LOGSTASH{host}", "localhost.localdomain");
}
//TESTED
String outputConf = _testOutputTemplate.replace("_XXX_COLLECTION_XXX_", testInfo._id.toString()); //TESTED
String sinceDbPath = LOGSTASH_WD + ".sincedb_" + testInfo._id.toString();
String conf = logstashConfig.replace("_XXX_DOTSINCEDB_XXX_", sinceDbPath) + outputConf.replace("_XXX_SOURCEKEY_XXX_", testInfo.sourceKey);
boolean allWorked = false;
Process logstashProcess = null;
try {
// 1] Create the process
ArrayList<String> args = new ArrayList<String>(4);
args.addAll(Arrays.asList(LOGSTASH_BINARY, "-e", conf));
if (0 == testInfo.maxDocs) {
args.add("-t"); // test mode, must faster
}//TESTED
if ((null != testInfo.logstash.testDebugOutput) && testInfo.logstash.testDebugOutput) {
args.add("--debug");
}
else {
args.add("--verbose");
}
ProcessBuilder logstashProcessBuilder = new ProcessBuilder(args);
logstashProcessBuilder = logstashProcessBuilder.directory(new File(LOGSTASH_WD)).redirectErrorStream(true);
logstashProcessBuilder.environment().put("JAVA_OPTS", "");
//DEBUG
//System.out.println("STARTING: " + ArrayUtils.toString(logstashProcessBuilder.command().toArray()));
// 2] Kick off the process
logstashProcess = logstashProcessBuilder.start();
StringWriter outputAndError = new StringWriter();
OutputCollector outAndErrorStream = new OutputCollector(logstashProcess.getInputStream(), new PrintWriter(outputAndError));
outAndErrorStream.start();
final int toWait_s = 240;
boolean exited = false;
// 3] Check the output collection for records
int errorVal = 0;
long priorCount = 0L;
int priorLogCount = 0;
int timeOfLastLoggingChange = 0;
int timeOfLastDocCountChange = 0;
String reasonForExit = "";
int inactivityTimeout_s = 10; // (default)
if (null != testInfo.logstash.testInactivityTimeout_secs) {
inactivityTimeout_s = testInfo.logstash.testInactivityTimeout_secs;
}
for (int i = 0; i < toWait_s; i += 5) {
try {
Thread.sleep(5000);
}
catch (Exception e) {}
long count = DbManager.getCollection("ingest", testInfo._id.toString()).count();
// 3.1] Do we have all the records (or is the number staying static)
//DEBUG
//System.out.println("FOUND: " + count + " VS " + priorCount + " , " + priorPriorCount);
// 3.1a] All done?
if ((count >= testInfo.maxDocs) && (count > 0))
{
allWorked = true;
break;
}//TESTED
// 3.1b] If not, has anything changes?
if (priorCount != count) {
timeOfLastDocCountChange = i;
}
if (priorLogCount != outAndErrorStream.getLines()) {
timeOfLastLoggingChange = i;
}
// 3.1c] Check for inactivity
if ((timeOfLastDocCountChange > 0) &&
(i - timeOfLastDocCountChange) >= inactivityTimeout_s)
{
// Delay between events: treat as success
allWorked = true;
break;
}//TESTED
if ((0 == count) && outAndErrorStream.getPipelineStarted() &&
((timeOfLastLoggingChange > 0) &&
(i - timeOfLastLoggingChange) >= inactivityTimeout_s))
{
// Delay between log messages after pipeline started, no documents, treat as failure
//DEBUG
//System.out.println("LOG LINES! " + i + " NUM = " + outAndErrorStream.getLines());
errorVal = 1;
reasonForExit = "No records received and logging inactive.\n";
break;
}//TESTED
// 3.2] Has the process exited unexpectedly?
try {
errorVal = logstashProcess.exitValue();
reasonForExit = "Logstash process exited with error: " + errorVal + ".\n";
exited = true;
//DEBUG
//System.out.println("GOT EXIT VALUE: " + errorVal);
break;
}//TESTED
catch (Exception e) {} // that's OK we're just still going is all...
priorCount = count;
priorLogCount = outAndErrorStream.getLines();
} //(end loop while waiting for job to complete)
// 4] If the process is still running then kill it
if (!exited) {
//DEBUG
//System.out.println("EXITED WITHOUT FINISHING");
logstashProcess.destroy();
}//TESTED
// 5] Things to do when the job is done: (worked or not)
// Send a message to the harvester
outAndErrorStream.join(); // (if we're here then must have closed the process, wait for it to die)
TestLogstashExtractorPojo testErr = new TestLogstashExtractorPojo();
testErr._id = testInfo._id;
if ((testInfo.maxDocs > 0) || (0 != errorVal)) {
testErr.error = reasonForExit + outputAndError.toString();
// (note this is capped at well below the BSON limit in the thread below)
}
else { // maxDocs==0 (ie pre-publish test) AND no error returned
testErr.error = null;
}
_logHarvesterQ.push(testErr.toDb());
//TESTED
}
catch (Exception e) {
//DEBUG
//e.printStackTrace();
TestLogstashExtractorPojo testErr = new TestLogstashExtractorPojo();
testErr._id = testInfo._id;
testErr.error = "Internal Logic Error: " + e.getMessage();
_logHarvesterQ.push(testErr.toDb());
}//TOTEST
finally {
// If we created a sincedb path then remove it:
try {