Merge branch 'v5-stable' into v5-beta

Conflicts:
	runtime/ruleset.c

6 files changed, 119 insertions, 18 deletions

diff --git a/ChangeLog b/ChangeLog
index 6b7db905..01b55037 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -178,6 +178,7 @@ Version 5.9.0  [V5-DEVEL] (rgerhards), 2011-06-08
   closes: http://bugzilla.adiscon.com/show_bug.cgi?id=236
 ---------------------------------------------------------------------------
 Version 5.8.13  [V5-stable] 2012-06-??
+- bugfix: DA queue could cause abort
 - bugfix: "last message repeated n times" message was missing hostname
   Thanks to Zdenek Salvet for finding this bug and to Bodik for reporting
 - bugfix "$PreserveFQDN on" was not honored in some modules
@@ -185,6 +186,15 @@ Version 5.8.13  [V5-stable] 2012-06-??
 - bugfix: randomized IP option header in omudpspoof caused problems
   closes: http://bugzilla.adiscon.com/show_bug.cgi?id=327
   Thanks to Rick Brown for helping to test out the patch.
+- bugfix: potential abort if output plugin logged message during shutdown
+  note that none of the rsyslog-provided plugins does this
+  Thanks to bodik and Rohit Prasad for alerting us on this bug and
+  analyzing it.
+  fixes: http://bugzilla.adiscon.com/show_bug.cgi?id=347
+- bugfix: multiple main queues with same queue file name was not detected
+  This lead to queue file corruption. While the root cause is a config
+  error, it is a bug that this important and hard to find config error
+  was not detected by rsyslog.
 ---------------------------------------------------------------------------
 Version 5.8.12  [V5-stable] 2012-06-06
 - add small delay (50ms) after sending shutdown message
diff --git a/runtime/obj.c b/runtime/obj.c
index f9afecd8..666dd6b0 100644
--- a/runtime/obj.c
+++ b/runtime/obj.c
@@ -611,6 +611,8 @@ static rsRetVal objDeserializeProperty(var_t *pProp, strm_t *pStrm)
 	number_t i;
 	number_t iLen;
 	uchar c;
+	int step = 0; /* which step was successful? */
+	int64 offs;
 
 	assert(pProp != NULL);
 
@@ -631,13 +633,16 @@ static rsRetVal objDeserializeProperty(var_t *pProp, strm_t *pStrm)
 		NEXTC;
 	}
 	CHKiRet(cstrFinalize(pProp->pcsName));
+	step = 1;
 
 	/* property type */
 	CHKiRet(objDeserializeNumber(&i, pStrm));
 	pProp->varType = i;
+	step = 2;
 
 	/* size (needed for strings) */
 	CHKiRet(objDeserializeNumber(&iLen, pStrm));
+	step = 3;
 
 	/* we now need to deserialize the value */
 	switch(pProp->varType) {
@@ -654,12 +659,44 @@ static rsRetVal objDeserializeProperty(var_t *pProp, strm_t *pStrm)
 			dbgprintf("invalid VARTYPE %d\n", pProp->varType);
 			break;
 	}
+	step = 4;
 
 	/* we should now be at the end of the line. So the next char must be \n */
 	NEXTC;
 	if(c != '\n') ABORT_FINALIZE(RS_RET_INVALID_PROPFRAME);
 
 finalize_it:
+	if(Debug && iRet != RS_RET_OK) {
+		strm.GetCurrOffset(pStrm, &offs);
+		dbgprintf("error %d deserializing property name, offset %lld, step %d\n",
+			  iRet, offs, step);
+		if(step >= 1) {
+			dbgprintf("error property name: '%s'\n", rsCStrGetSzStrNoNULL(pProp->pcsName));
+		}
+		if(step >= 2) {
+			dbgprintf("error var type: '%d'\n", pProp->varType);
+		}
+		if(step >= 3) {
+			dbgprintf("error len: '%d'\n", (int) iLen);
+		}
+		if(step >= 4) {
+			switch(pProp->varType) {
+				case VARTYPE_STR:
+					dbgprintf("error data string: '%s'\n",
+						  rsCStrGetSzStrNoNULL(pProp->val.pStr));
+					break;
+				case VARTYPE_NUMBER:
+					dbgprintf("error number: %d\n", (int) pProp->val.num);
+					break;
+				case VARTYPE_SYSLOGTIME:
+					dbgprintf("syslog time was successfully parsed (but "
+					          "is not displayed\n");
+					break;
+				default:
+					break;
+			}
+		}
+	}
 	RETiRet;
 }
 
diff --git a/runtime/queue.c b/runtime/queue.c
index 94cc1fcd..bdd57f99 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -1798,6 +1798,7 @@ ConsumerDA(qqueue_t *pThis, wti_t *pWti)
 {
 	int i;
 	int iCancelStateSave;
+	int bNeedReLock = 0;	/**< do we need to lock the mutex again? */
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, qqueue);
@@ -1807,6 +1808,7 @@ ConsumerDA(qqueue_t *pThis, wti_t *pWti)
 
 	/* we now have a non-idle batch of work, so we can release the queue mutex and process it */
 	d_pthread_mutex_unlock(pThis->mut);
+	bNeedReLock = 1;
 
 	/* at this spot, we may be cancelled */
 	pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave);
@@ -1825,10 +1827,10 @@ ConsumerDA(qqueue_t *pThis, wti_t *pWti)
 	/* but now cancellation is no longer permitted */
 	pthread_setcancelstate(iCancelStateSave, NULL);
 
-	/* now we are done, but need to re-aquire the mutex */
-	d_pthread_mutex_lock(pThis->mut);
-
 finalize_it:
+	/* now we are done, but potentially need to re-aquire the mutex */
+	if(bNeedReLock)
+		d_pthread_mutex_lock(pThis->mut);
 	DBGOPRINT((obj_t*) pThis, "DAConsumer returns with iRet %d\n", iRet);
 	RETiRet;
 }
diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index 2f307f05..14dc4c74 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -500,7 +500,7 @@ debugPrintAll(void)
 static rsRetVal
 rulesetCreateQueue(void __attribute__((unused)) *pVal, int *pNewVal)
 {
-	uchar *rulesetMainQName;
+	uchar *rsname;
 	DEFiRet;
 
 	if(pCurrRuleset == NULL) {
@@ -518,10 +518,9 @@ rulesetCreateQueue(void __attribute__((unused)) *pVal, int *pNewVal)
 	if(pNewVal == 0)
 		FINALIZE; /* if it is turned off, we do not need to change anything ;) */
 
-	dbgprintf("adding a ruleset-specific \"main\" queue");
-	rulesetMainQName = (pCurrRuleset->pszName == NULL)? UCHAR_CONSTANT("ruleset") :
-							    pCurrRuleset->pszName;
-	CHKiRet(createMainQueue(&pCurrRuleset->pQueue, rulesetMainQName));
+	rsname = (pCurrRuleset->pszName == NULL) ? (uchar*) "[ruleset]" : pCurrRuleset->pszName;
+	DBGPRINTF("adding a ruleset-specific \"main\" queue for ruleset '%s'\n", rsname);
+	CHKiRet(createMainQueue(&pCurrRuleset->pQueue, rsname));
 
 finalize_it:
 	RETiRet;
diff --git a/runtime/stream.c b/runtime/stream.c
index 9645a3fe..d6ee1e39 100644
--- a/runtime/stream.c
+++ b/runtime/stream.c
@@ -16,7 +16,7 @@
  * it turns out to be problematic. Then, we need to quasi-refcount the number of accesses
  * to the object.
  *
- * Copyright 2008, 2009 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2008-2012 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of the rsyslog runtime library.
  *
@@ -361,7 +361,7 @@ static rsRetVal strmCloseFile(strm_t *pThis)
 		pThis->fdDir = -1;
 	}
 
-	if(pThis->bDeleteOnClose) {
+	if(pThis->bDeleteOnClose && pThis->pszCurrFName != NULL) {
 		if(unlink((char*) pThis->pszCurrFName) == -1) {
 			char errStr[1024];
 			int err = errno;
@@ -369,14 +369,12 @@ static rsRetVal strmCloseFile(strm_t *pThis)
 			DBGPRINTF("error %d unlinking '%s' - ignored: %s\n",
 				   errno, pThis->pszCurrFName, errStr);
 		}
-	}
-
-	pThis->iCurrOffs = 0;	/* we are back at begin of file */
-	if(pThis->pszCurrFName != NULL) {
 		free(pThis->pszCurrFName);	/* no longer needed in any case (just for open) */
 		pThis->pszCurrFName = NULL;
 	}
 
+	pThis->iCurrOffs = 0;	/* we are back at begin of file */
+
 	RETiRet;
 }
 
diff --git a/tools/syslogd.c b/tools/syslogd.c
index 63fffbfa..75ee96eb 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -230,6 +230,11 @@ typedef struct legacyOptsLL_s {
 } legacyOptsLL_t;
 legacyOptsLL_t *pLegacyOptsLL = NULL;
 
+struct queuefilenames_s {
+	struct queuefilenames_s *next;
+	uchar *name;
+} *queuefilenames = NULL;
+
 /* global variables for config file state */
 int	iCompatibilityMode = 0;		/* version we should be compatible with; 0 means sysklogd. It is
 					   the default, so if no -c<n> option is given, we make ourselvs
@@ -736,11 +741,19 @@ submitMsg(msg_t *pMsg)
 	ISOBJ_TYPE_assert(pMsg, msg);
 
 	pRuleset = MsgGetRuleset(pMsg);
-
 	pQueue = (pRuleset == NULL) ? pMsgQueue : ruleset.GetRulesetQueue(pRuleset);
+
+	/* if a plugin logs a message during shutdown, the queue may no longer exist */
+	if(pQueue == NULL) {
+		DBGPRINTF("submitMsg() could not submit message - "
+			  "queue does (no longer?) exist - ignored\n");
+		FINALIZE;
+	}
+
 	MsgPrepareEnqueue(pMsg);
 	qqueueEnqObj(pQueue, pMsg->flowCtlType, (void*) pMsg);
 
+finalize_it:
 	RETiRet;
 }
 
@@ -761,12 +774,20 @@ multiSubmitMsg(multi_submit_t *pMultiSub)
 	if(pMultiSub->nElem == 0)
 		FINALIZE;
 
+	pRuleset = MsgGetRuleset(pMultiSub->ppMsgs[0]);
+	pQueue = (pRuleset == NULL) ? pMsgQueue : ruleset.GetRulesetQueue(pRuleset);
+
+	/* if a plugin logs a message during shutdown, the queue may no longer exist */
+	if(pQueue == NULL) {
+		DBGPRINTF("multiSubmitMsg() could not submit message - "
+			  "queue does (no longer?) exist - ignored\n");
+		FINALIZE;
+	}
+
 	for(i = 0 ; i < pMultiSub->nElem ; ++i) {
 		MsgPrepareEnqueue(pMultiSub->ppMsgs[i]);
 	}
 
-	pRuleset = MsgGetRuleset(pMultiSub->ppMsgs[0]);
-	pQueue = (pRuleset == NULL) ? pMsgQueue : ruleset.GetRulesetQueue(pRuleset);
 	iRet = pQueue->MultiEnq(pQueue, pMultiSub);
 	pMultiSub->nElem = 0;
 
@@ -1026,6 +1047,14 @@ static void doDie(int sig)
 static void
 freeAllDynMemForTermination(void)
 {
+	struct queuefilenames_s *qfn, *qfnDel;
+
+	for(qfn = queuefilenames ; qfn != NULL ; ) {
+		qfnDel = qfn;
+		qfn = qfn->next;
+		free(qfnDel->name);
+		free(qfnDel);
+	}
 	free(pszMainMsgQFName);
 	free(pModDir);
 	free(pszConfDAGFile);
@@ -1552,6 +1581,10 @@ startInputModules(void)
  */
 rsRetVal createMainQueue(qqueue_t **ppQueue, uchar *pszQueueName)
 {
+	struct queuefilenames_s *qfn;
+	uchar *qfname = NULL;
+	static int qfn_renamenum = 0;
+	uchar qfrenamebuf[1024];
 	DEFiRet;
 
 	/* switch the message object to threaded operation, if necessary */
@@ -1577,10 +1610,32 @@ rsRetVal createMainQueue(qqueue_t **ppQueue, uchar *pszQueueName)
 		errmsg.LogError(0, NO_ERRCODE, "Invalid " #directive ", error %d. Ignored, running with default setting", iRet); \
 	}
 
+	if(pszMainMsgQFName != NULL) {
+		/* check if the queue file name is unique, else emit an error */
+		for(qfn = queuefilenames ; qfn != NULL ; qfn = qfn->next) {
+			dbgprintf("check queue file name '%s' vs '%s'\n", qfn->name, pszMainMsgQFName);
+			if(!ustrcmp(qfn->name, pszMainMsgQFName)) {
+				snprintf((char*)qfrenamebuf, sizeof(qfrenamebuf), "%d-%s-%s",
+					 ++qfn_renamenum, pszMainMsgQFName, 
+					 (pszQueueName == NULL) ? "NONAME" : (char*)pszQueueName);
+				qfname = ustrdup(qfrenamebuf);
+				errmsg.LogError(0, NO_ERRCODE, "Error: queue file name '%s' already in use "
+					" - using '%s' instead", pszMainMsgQFName, qfname);
+				break;
+			}
+		}
+		if(qfname == NULL)
+			qfname = ustrdup(pszMainMsgQFName);
+		qfn = malloc(sizeof(struct queuefilenames_s));
+		qfn->name = qfname;
+		qfn->next = queuefilenames;
+		queuefilenames = qfn;
+	}
+
 	setQPROP(qqueueSetMaxFileSize, "$MainMsgQueueFileSize", iMainMsgQueMaxFileSize);
 	setQPROP(qqueueSetsizeOnDiskMax, "$MainMsgQueueMaxDiskSpace", iMainMsgQueMaxDiskSpace);
 	setQPROP(qqueueSetiDeqBatchSize, "$MainMsgQueueDequeueBatchSize", iMainMsgQueDeqBatchSize);
-	setQPROPstr(qqueueSetFilePrefix, "$MainMsgQueueFileName", pszMainMsgQFName);
+	setQPROPstr(qqueueSetFilePrefix, "$MainMsgQueueFileName", qfname);
 	setQPROP(qqueueSetiPersistUpdCnt, "$MainMsgQueueCheckpointInterval", iMainMsgQPersistUpdCnt);
 	setQPROP(qqueueSetbSyncQueueFiles, "$MainMsgQueueSyncQueueFiles", bMainMsgQSyncQeueFiles);
 	setQPROP(qqueueSettoQShutdown, "$MainMsgQueueTimeoutShutdown", iMainMsgQtoQShutdown );