Merge branch 'v5-devel'

Conflicts:
	ChangeLog
	runtime/datetime.c

4 files changed, 174 insertions, 17 deletions

diff --git a/plugins/imdiag/imdiag.c b/plugins/imdiag/imdiag.c
index bf972191..7cacc44a 100644
--- a/plugins/imdiag/imdiag.c
+++ b/plugins/imdiag/imdiag.c
@@ -246,7 +246,7 @@ injectMsg(uchar *pszCmd, tcps_sess_t *pSess)
 		doInjectMsg(i + iFrom);
 	}
 
-	CHKiRet(sendResponse(pSess, "messages injected\n"));
+	CHKiRet(sendResponse(pSess, "%d messages injected\n", nMsgs));
 
 finalize_it:
 	RETiRet;
@@ -259,10 +259,13 @@ static rsRetVal
 waitMainQEmpty(tcps_sess_t *pSess)
 {
 	int iMsgQueueSize;
+	int iPrint = 0;
 	DEFiRet;
 
 	CHKiRet(diagGetMainMsgQSize(&iMsgQueueSize));
 	while(iMsgQueueSize > 0) {
+		if(iPrint++ % 500 == 0) 
+			dbgprintf("imdiag sleeping, wait mainq drain, curr size %d\n", iMsgQueueSize);
 		srSleep(0,2);	/* wait a little bit */
 		CHKiRet(diagGetMainMsgQSize(&iMsgQueueSize));
 	}
@@ -297,6 +300,7 @@ OnMsgReceived(tcps_sess_t *pSess, uchar *pRcv, int iLenMsg)
 
 	getFirstWord(&pszMsg, cmdBuf, sizeof(cmdBuf)/sizeof(uchar), TO_LOWERCASE);
 
+	dbgprintf("imdiag received command '%s'\n", cmdBuf);
 	if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("getmainmsgqueuesize"))) {
 		CHKiRet(diagGetMainMsgQSize(&iMsgQueueSize));
 		CHKiRet(sendResponse(pSess, "%d\n", iMsgQueueSize));
diff --git a/plugins/imuxsock/imuxsock.c b/plugins/imuxsock/imuxsock.c
index 424d0904..6d60efaf 100644
--- a/plugins/imuxsock/imuxsock.c
+++ b/plugins/imuxsock/imuxsock.c
@@ -333,11 +333,8 @@ CODESTARTafterRun
 		if (funixn[i] && funix[i] != -1)
 			unlink((char*) funixn[i]);
 	/* free no longer needed string */
-	if(pLogSockName != NULL)
-		free(pLogSockName);
-	if(pLogHostName != NULL) {
-		free(pLogHostName);
-	}
+	free(pLogSockName);
+	free(pLogHostName);
 
 	discardFunixn();
 	nfunix = 1;
diff --git a/plugins/ompgsql/ompgsql.c b/plugins/ompgsql/ompgsql.c
index eb774835..cb6b6a4d 100644
--- a/plugins/ompgsql/ompgsql.c
+++ b/plugins/ompgsql/ompgsql.c
@@ -170,6 +170,9 @@ tryExec(uchar *pszCmd, instanceData *pData)
 	int bHadError = 0;
 
 	/* try insert */
+BEGINfunc
+RUNLOG_VAR("%p", pData->f_hpgsql);
+RUNLOG_VAR("%s", pszCmd);
 	pgRet = PQexec(pData->f_hpgsql, (char*)pszCmd);
 	execState = PQresultStatus(pgRet);
 	if(execState != PGRES_COMMAND_OK && execState != PGRES_TUPLES_OK) {
@@ -178,6 +181,7 @@ tryExec(uchar *pszCmd, instanceData *pData)
 	}
 	PQclear(pgRet);
 
+ENDfunc
 	return(bHadError);
 }
 
@@ -230,6 +234,14 @@ CODESTARTtryResume
 	}
 ENDtryResume
 
+
+BEGINbeginTransaction
+CODESTARTbeginTransaction
+dbgprintf("ompgsql: beginTransaction\n");
+	iRet = writePgSQL((uchar*) "begin", pData); /* TODO: make user-configurable */
+ENDbeginTransaction
+
+
 BEGINdoAction
 CODESTARTdoAction
 	dbgprintf("\n");
@@ -237,6 +249,13 @@ CODESTARTdoAction
 ENDdoAction
 
 
+BEGINendTransaction
+CODESTARTendTransaction
+	iRet = writePgSQL((uchar*) "commit;", pData); /* TODO: make user-configurable */
+dbgprintf("ompgsql: endTransaction\n");
+ENDendTransaction
+
+
 BEGINparseSelectorAct
 	int iPgSQLPropErr = 0;
 CODESTARTparseSelectorAct
@@ -314,6 +333,7 @@ ENDmodExit
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
 CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_TXIF_OMOD_QUERIES /* we support the transactional interface! */
 ENDqueryEtryPt
 
 
@@ -322,6 +342,8 @@ CODESTARTmodInit
 	*ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
 CODEmodInit_QueryRegCFSLineHdlr
 	CHKiRet(objUse(errmsg, CORE_COMPONENT));
+	INITChkCoreFeature(bCoreSupportsBatching, CORE_FEATURE_BATCHING);
+	DBGPRINTF("ompgsql: %susing transactional output interface.\n", bCoreSupportsBatching ? "" : "not ");
 ENDmodInit
 /* vi:set ai:
  */
diff --git a/plugins/omtesting/omtesting.c b/plugins/omtesting/omtesting.c
index 411bcf88..8f6cdbe5 100644
--- a/plugins/omtesting/omtesting.c
+++ b/plugins/omtesting/omtesting.c
@@ -22,7 +22,7 @@
  * NOTE: read comments in module-template.h to understand how this file
  *       works!
  *
- * Copyright 2007 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007, 2009 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of rsyslog.
  *
@@ -46,12 +46,14 @@
 #include <stdio.h>
 #include <stdarg.h>
 #include <stdlib.h>
+#include <time.h>
 #include <string.h>
 #include <ctype.h>
 #include <assert.h>
 #include "dirty.h"
 #include "syslogd-types.h"
 #include "module-template.h"
+#include "cfsysline.h"
 
 MODULE_TYPE_OUTPUT
 
@@ -59,9 +61,18 @@ MODULE_TYPE_OUTPUT
  */
 DEF_OMOD_STATIC_DATA
 
+static int bEchoStdout = 0;	/* echo non-failed messages to stdout */
+
 typedef struct _instanceData {
+	enum { MD_SLEEP, MD_FAIL, MD_RANDFAIL, MD_ALWAYS_SUSPEND }
+		mode;
+	int	bEchoStdout;
 	int	iWaitSeconds;
 	int	iWaitUSeconds;	/* milli-seconds (one million of a second, just to make sure...) */
+	int 	iCurrCallNbr;
+	int	iFailFrequency;
+	int	iResumeAfter;
+	int	iCurrRetries;
 } instanceData;
 
 BEGINcreateInstance
@@ -85,19 +96,106 @@ CODESTARTisCompatibleWithFeature
 ENDisCompatibleWithFeature
 
 
-BEGINtryResume
-CODESTARTtryResume
-ENDtryResume
+/* implement "fail" command in retry processing */
+static rsRetVal doFailOnResume(instanceData *pData)
+{
+	DEFiRet;
 
-BEGINdoAction
-CODESTARTdoAction
+	dbgprintf("fail retry curr %d, max %d\n", pData->iCurrRetries, pData->iResumeAfter);
+	if(++pData->iCurrRetries == pData->iResumeAfter) {
+		iRet = RS_RET_OK;
+	} else {
+		iRet = RS_RET_SUSPENDED;
+	}
+
+	RETiRet;
+}
+
+
+/* implement "fail" command */
+static rsRetVal doFail(instanceData *pData)
+{
+	DEFiRet;
+
+	dbgprintf("fail curr %d, frquency %d\n", pData->iCurrCallNbr, pData->iFailFrequency);
+	if(pData->iCurrCallNbr++ % pData->iFailFrequency == 0) {
+		pData->iCurrRetries = 0;
+		iRet = RS_RET_SUSPENDED;
+	}
+
+	RETiRet;
+}
+
+
+/* implement "sleep" command */
+static rsRetVal doSleep(instanceData *pData)
+{
+	DEFiRet;
 	struct timeval tvSelectTimeout;
 
 	dbgprintf("sleep(%d, %d)\n", pData->iWaitSeconds, pData->iWaitUSeconds);
 	tvSelectTimeout.tv_sec = pData->iWaitSeconds;
 	tvSelectTimeout.tv_usec = pData->iWaitUSeconds; /* milli seconds */
 	select(0, NULL, NULL, NULL, &tvSelectTimeout);
-	//dbgprintf(":omtesting: end doAction(), iRet %d\n", iRet);
+	RETiRet;
+}
+
+
+/* implement "randomfail" command */
+static rsRetVal doRandFail(void)
+{
+	DEFiRet;
+	if((rand() >> 4) < (RAND_MAX >> 5)) { /* rougly same probability */
+		iRet = RS_RET_OK;
+		dbgprintf("omtesting randfail: succeeded this time\n");
+	} else {
+		iRet = RS_RET_SUSPENDED;
+		dbgprintf("omtesting randfail: failed this time\n");
+	}
+	RETiRet;
+}
+
+
+BEGINtryResume
+CODESTARTtryResume
+	dbgprintf("omtesting tryResume() called\n");
+	switch(pData->mode) {
+		case MD_SLEEP:
+			break;
+		case MD_FAIL:
+			iRet = doFailOnResume(pData);
+			break;
+		case MD_RANDFAIL:
+			iRet = doRandFail();
+			break;
+		case MD_ALWAYS_SUSPEND:
+			iRet = RS_RET_SUSPENDED;
+	}
+	dbgprintf("omtesting tryResume() returns iRet %d\n", iRet);
+ENDtryResume
+
+
+BEGINdoAction
+CODESTARTdoAction
+	dbgprintf("omtesting received msg '%s'\n", ppString[0]);
+	switch(pData->mode) {
+		case MD_SLEEP:
+			iRet = doSleep(pData);
+			break;
+		case MD_FAIL:
+			iRet = doFail(pData);
+			break;
+		case MD_RANDFAIL:
+			iRet = doRandFail();
+		case MD_ALWAYS_SUSPEND:
+			iRet = RS_RET_SUSPENDED;
+	}
+
+	if(iRet == RS_RET_OK && pData->bEchoStdout) {
+		fprintf(stdout, "%s", ppString[0]);
+		fflush(stdout);
+	}
+	dbgprintf(":omtesting: end doAction(), iRet %d\n", iRet);
 ENDdoAction
 
 
@@ -113,7 +211,7 @@ BEGINparseSelectorAct
 	int i;
 	uchar szBuf[1024];
 CODESTARTparseSelectorAct
-CODE_STD_STRING_REQUESTparseSelectorAct(0)
+CODE_STD_STRING_REQUESTparseSelectorAct(1)
 	/* code here is quick and dirty - if you like, clean it up. But keep
 	 * in mind it is just a testing aid ;) -- rgerhards, 2007-12-31
 	 */
@@ -135,6 +233,7 @@ CODE_STD_STRING_REQUESTparseSelectorAct(0)
 	if(isspace(*p))
 		++p;
 
+	dbgprintf("omtesting command: '%s'\n", szBuf);
 	if(!strcmp((char*) szBuf, "sleep")) {
 		/* parse seconds */
 		for(i = 0 ; *p && !isspace(*p) && ((unsigned) i < sizeof(szBuf) - 1) ; ++i) {
@@ -152,12 +251,43 @@ CODE_STD_STRING_REQUESTparseSelectorAct(0)
 		if(isspace(*p))
 			++p;
 		pData->iWaitUSeconds = atoi((char*) szBuf);
-	}
-	/* once there are other modes, here is the spot to add it! */
-	else {
+		pData->mode = MD_SLEEP;
+	} else if(!strcmp((char*) szBuf, "fail")) {
+		/* "fail fail-freqency resume-after"
+		 * fail-frequency specifies how often doAction() fails
+		 * resume-after speicifes how fast tryResume() should come back with success
+		 * all numbers being "times called"
+		 */
+		/* parse fail-frequence */
+		for(i = 0 ; *p && !isspace(*p) && ((unsigned) i < sizeof(szBuf) - 1) ; ++i) {
+			szBuf[i] = *p++;
+		}
+		szBuf[i] = '\0';
+		if(isspace(*p))
+			++p;
+		pData->iFailFrequency = atoi((char*) szBuf);
+		/* parse resume-after */
+		for(i = 0 ; *p && !isspace(*p) && ((unsigned) i < sizeof(szBuf) - 1) ; ++i) {
+			szBuf[i] = *p++;
+		}
+		szBuf[i] = '\0';
+		if(isspace(*p))
+			++p;
+		pData->iResumeAfter = atoi((char*) szBuf);
+		pData->iCurrCallNbr = 1;
+		pData->mode = MD_FAIL;
+	} else if(!strcmp((char*) szBuf, "randfail")) {
+		pData->mode = MD_RANDFAIL;
+	} else if(!strcmp((char*) szBuf, "always_suspend")) {
+		pData->mode = MD_ALWAYS_SUSPEND;
+	} else {
 		dbgprintf("invalid mode '%s', doing 'sleep 1 0' - fix your config\n", szBuf);
 	}
 
+	pData->bEchoStdout = bEchoStdout;
+	CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, OMSR_NO_RQD_TPL_OPTS,
+				         (uchar*)"RSYSLOG_TraditionalForwardFormat"));
+
 CODE_STD_FINALIZERparseSelectorAct
 ENDparseSelectorAct
 
@@ -177,6 +307,10 @@ BEGINmodInit()
 CODESTARTmodInit
 	*ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
 CODEmodInit_QueryRegCFSLineHdlr
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"actionomtestingechostdout", 0, eCmdHdlrBinary, NULL,
+				   &bEchoStdout, STD_LOADABLE_MODULE_ID));
+	/* we seed the random-number generator in any case... */
+	srand(time(NULL));
 ENDmodInit
 /*
  * vi:set ai: