Merge branch 'v5-stable' into v6-stable

ChangeLog

@@ -282,12 +282,30 @@ Version 5.9.0  [V5-DEVEL] (rgerhards), 2011-03-??
   closes: http://bugzilla.adiscon.com/show_bug.cgi?id=236
 ---------------------------------------------------------------------------
 Version 5.8.12  [V5-stable] 2012-05-??
+- bugfix: delayble source could block action queue, even if there was
+  a disk queue associated with it. The root cause of this problem was
+  that it makes no sense to delay messages once they arrive in the 
+  action queue - the "input" that is being held in that case is the main
+  queue worker, what makes no sense.
+  Thanks to Marcin for alerting us on this problem and providing 
+  instructions to reproduce it.
 - bugfix: disk queue was not persisted on shutdown, regression of fix to
   http://bugzilla.adiscon.com/show_bug.cgi?id=299
   The new code also handles the case of shutdown of blocking light and 
   full delayable sources somewhat smarter and permits, assuming sufficient
   timouts, to persist message up to the max queue capacity. Also some nits
   in debug instrumentation have been fixed.
+- bugfix/omudpspoof: problems, including abort, happend when run on
+  multiple threads. Root cause is that libnet is not thread-safe. 
+  omudpspoof now guards libnet calls with their own mutex.
+- bugfix: if debug message could end up in log file when forking
+  if rsyslog was set to auto-background (thus fork, the default) and debug
+  mode to stdout was enabled, debug messages ended up in the first log file
+  opened. Currently, stdout logging is completely disabled in forking mode
+  (but writing to the debug log file is still possible). This is a change 
+  in behaviour, which is under review. If it causes problems to you,
+  please let us know.
+  Thanks to Tomas Heinrich for the patch.
 - bugfix/tcpflood: sending small test files did not work correctly
 ---------------------------------------------------------------------------
 Version 5.8.11  [V5-stable] 2012-05-03

action.c

@@ -1271,7 +1271,7 @@ doSubmitToActionQ(action_t *pAction, msg_t *pMsg)
 	if(pAction->pQueue->qType == QUEUETYPE_DIRECT)
 		iRet = qqueueEnqObjDirect(pAction->pQueue, (void*) MsgAddRef(pMsg));
 	else
-		iRet = qqueueEnqObj(pAction->pQueue, pMsg->flowCtlType, (void*) MsgAddRef(pMsg));
+		iRet = qqueueEnqObj(pAction->pQueue, eFLOWCTL_NO_DELAY, (void*) MsgAddRef(pMsg));
 
 	RETiRet;
 }

plugins/omudpspoof/omudpspoof.c

@@ -132,6 +132,7 @@ ENDinitConfVars
 /* add some variables needed for libnet */
 libnet_t *libnet_handle;
 char errbuf[LIBNET_ERRBUF_SIZE];
+pthread_mutex_t mutLibnet;
 
 /* forward definitions */
 static rsRetVal doTryResume(instanceData *pData);
@@ -194,6 +195,8 @@ ENDdbgPrintInstInfo
 
 
 /* Send a message via UDP
+ * Note: libnet is not thread-safe, so we need to ensure that only one
+ * instance ever is calling libnet code.
  * rgehards, 2007-12-20
  */
 static inline rsRetVal
@@ -207,6 +210,7 @@ UDPSend(instanceData *pData, uchar *pszSourcename, char *msg, size_t len)
 	struct sockaddr_in *tempaddr,source_ip;
 	libnet_ptag_t ip, ipo;
 	libnet_ptag_t udp;
+	sbool bNeedUnlock = 0;
 	DEFiRet;
 
 	if(pData->pSockArray == NULL) {
@@ -221,6 +225,8 @@ UDPSend(instanceData *pData, uchar *pszSourcename, char *msg, size_t len)
 	inet_pton(AF_INET, (char*)pszSourcename, &(source_ip.sin_addr));
 
 	bSendSuccess = FALSE;
+	d_pthread_mutex_lock(&mutLibnet);
+	bNeedUnlock = 1;
 	for (r = pData->f_addr; r; r = r->ai_next) {
 		tempaddr = (struct sockaddr_in *)r->ai_addr;
 		libnet_clear_packet(libnet_handle);
@@ -281,6 +287,9 @@ UDPSend(instanceData *pData, uchar *pszSourcename, char *msg, size_t len)
 	}
 
 finalize_it:
+	if(bNeedUnlock) {
+		d_pthread_mutex_unlock(&mutLibnet);
+	}
 	RETiRet;
 }
 
@@ -454,6 +463,7 @@ BEGINmodExit
 CODESTARTmodExit
 	/* destroy the libnet state needed for forged UDP sources */
 	libnet_destroy(libnet_handle);
+	pthread_mutex_destroy(&mutLibnet);
 	/* release what we no longer need */
 	objRelease(errmsg, CORE_COMPONENT);
 	objRelease(glbl, CORE_COMPONENT);
@@ -503,6 +513,7 @@ CODEmodInit_QueryRegCFSLineHdlr
 		errmsg.LogError(0, NO_ERRCODE, "Error initializing libnet, can not continue ");
 		ABORT_FINALIZE(RS_RET_ERR_LIBNET_INIT);
 	}
+	pthread_mutex_init(&mutLibnet, NULL);
 
 	CHKiRet(regCfSysLineHdlr((uchar *)"actionomudpspoofdefaulttemplate", 0, eCmdHdlrGetWord, NULL, &cs.pszTplName, NULL));
 	CHKiRet(regCfSysLineHdlr((uchar *)"actionomudpspoofsourcenametemplate", 0, eCmdHdlrGetWord, NULL, &cs.pszSourceNameTemplate, NULL));

runtime/debug.c

@@ -68,7 +68,7 @@ static int bPrintAllDebugOnExit = 0;
 static int bAbortTrace = 1;	/* print a trace after SIGABRT or SIGSEGV */
 static char *pszAltDbgFileName = NULL; /* if set, debug output is *also* sent to here */
 static int altdbg = -1;	/* and the handle for alternate debug output */
-static int stddbg;
+int stddbg = 1; /* the handle for regular debug output, set to stdout if not forking, -1 otherwise */
 
 /* list of files/objects that should be printed */
 typedef struct dbgPrintName_s {
@@ -1303,8 +1303,6 @@ dbgGetRuntimeOptions(void)
 	uchar *optname;
 
 	/* set some defaults */
-	stddbg = 1;
-
 	if((pszOpts = (uchar*) getenv("RSYSLOG_DEBUG")) != NULL) {
 		/* we have options set, so let's process them */
 		while(dbgGetRTOptNamVal(&pszOpts, &optname, &optval)) {

runtime/debug.h

@@ -35,6 +35,7 @@
 /* external static data elements (some time to be replaced) */
 extern int Debug;		/* debug flag  - read-only after startup */
 extern int debugging_on;	 /* read-only, except on sig USR1 */
+extern int stddbg; /* the handle for regular debug output, set to stdout if not forking, -1 otherwise */
 
 /* data types */
 

runtime/queue.c

@@ -2323,20 +2323,25 @@ doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr)
 	while(   (pThis->iMaxQueueSize > 0 && pThis->iQueueSize >= pThis->iMaxQueueSize)
 	      || (pThis->qType == QUEUETYPE_DISK && pThis->sizeOnDiskMax != 0
 	      	  && pThis->tVars.disk.sizeOnDisk > pThis->sizeOnDiskMax)) {
-		DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL - waiting to drain.\n");
-		if(glbl.GetGlobalInputTermState()) {
-			DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL, discard due to FORCE_TERM.\n");
-			ABORT_FINALIZE(RS_RET_FORCE_TERM);
-		}
-		timeoutComp(&t, pThis->toEnq);
 		STATSCOUNTER_INC(pThis->ctrFull, pThis->mutCtrFull);
-// TODO : handle enqOnly => discard!
-		if(pthread_cond_timedwait(&pThis->notFull, pThis->mut, &t) != 0) {
-			DBGOPRINT((obj_t*) pThis, "enqueueMsg: cond timeout, dropping message!\n");
+		if(pThis->toEnq == 0 || pThis->bEnqOnly) {
+			DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL - configured for immediate discarding.\n");
 			objDestruct(pUsr);
 			ABORT_FINALIZE(RS_RET_QUEUE_FULL);
-		}
+		} else {
+			DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL - waiting %dms to drain.\n", pThis->toEnq);
+			if(glbl.GetGlobalInputTermState()) {
+				DBGOPRINT((obj_t*) pThis, "enqueueMsg: queue FULL, discard due to FORCE_TERM.\n");
+				ABORT_FINALIZE(RS_RET_FORCE_TERM);
+			}
+			timeoutComp(&t, pThis->toEnq);
+			if(pthread_cond_timedwait(&pThis->notFull, pThis->mut, &t) != 0) {
+				DBGOPRINT((obj_t*) pThis, "enqueueMsg: cond timeout, dropping message!\n");
+				objDestruct(pUsr);
+				ABORT_FINALIZE(RS_RET_QUEUE_FULL);
+			}
 		dbgoprint((obj_t*) pThis, "enqueueMsg: wait solved queue full condition, enqueing\n");
+		}
 	}
 
 	/* and finally enqueue the message */

tools/syslogd.c

@@ -2501,7 +2501,7 @@ doGlblProcessInit(void)
 
 	if( !(Debug == DEBUG_FULL || NoFork) )
 	{
-		DBGPRINTF("Checking pidfile.\n");
+		DBGPRINTF("Checking pidfile '%s'.\n", PidFile);
 		if (!check_pid(PidFile))
 		{
 			memset(&sigAct, 0, sizeof (sigAct));
@@ -2509,6 +2509,9 @@ doGlblProcessInit(void)
 			sigAct.sa_handler = doexit;
 			sigaction(SIGTERM, &sigAct, NULL);
 
+			/* stop writing debug messages to stdout (if debugging is on) */
+			stddbg = -1;
+
 			if (fork()) {
 				/* Parent process
 				 */
@@ -2571,7 +2574,7 @@ doGlblProcessInit(void)
 	}
 
 	/* tuck my process id away */
-	DBGPRINTF("Writing pidfile %s.\n", PidFile);
+	DBGPRINTF("Writing pidfile '%s'.\n", PidFile);
 	if (!check_pid(PidFile))
 	{
 		if (!write_pid(PidFile))