reordered imudp processing.

Message parsing is now done as part of main message queue worker
processing (was part of the input thread) This should also improve
performance, as potentially more work is done in parallel.

11 files changed, 396 insertions, 20 deletions

diff --git a/ChangeLog b/ChangeLog
index 0f3d4591..2f9f34a6 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -9,6 +9,12 @@ the output driver. It is recommended to drain queues with the previous
 version before switching to this one.
 ********************************* WARNING *********************************
 
+- reordered imudp processing. Message parsing is now done as part of main
+  message queue worker processing (was part of the input thread)
+  This should also improve performance, as potentially more work is
+  done in parallel.
+- bugfix: compressed syslog messages could be slightly mis-uncompressed
+  if the last byte of the compressed record was a NUL
 - added $UDPServerTimeRequery option which enables to work with
   less acurate timestamps in favor of performance. This enables querying
   of the time only every n-th time if imudp is running in the tight
diff --git a/dirty.h b/dirty.h
index 8e7ffd84..f5d0415e 100644
--- a/dirty.h
+++ b/dirty.h
@@ -61,6 +61,9 @@ extern int  bReduceRepeatMsgs;
 #define	BACKOFF(f)	{ if (++(f)->f_repeatcount > MAXREPEAT) \
 				 (f)->f_repeatcount = MAXREPEAT; \
 			}
+extern int bDropTrailingLF;
+extern uchar cCCEscapeChar;
+extern int  bEscapeCCOnRcv;
 #ifdef USE_NETZIP
 /* config param: minimum message size to try compression. The smaller
  * the message, the less likely is any compression gain. We check for
diff --git a/plugins/imudp/imudp.c b/plugins/imudp/imudp.c
index 7f9afb68..aab201a9 100644
--- a/plugins/imudp/imudp.c
+++ b/plugins/imudp/imudp.c
@@ -40,6 +40,8 @@
 #include "srUtils.h"
 #include "errmsg.h"
 #include "glbl.h"
+#include "msg.h"
+#include "parser.h"
 #include "datetime.h"
 
 MODULE_TYPE_INPUT
@@ -153,15 +155,16 @@ processSocket(int fd, struct sockaddr_storage *frominetPrev, int *pbIsPermitted,
 	time_t ttGenTime;
 	struct syslogTime stTime;
 	socklen_t socklen;
-	ssize_t l;
+	ssize_t lenRcvBuf;
 	struct sockaddr_storage frominet;
+	msg_t *pMsg;
 	char errStr[1024];
 
 	iNbrTimeUsed = 0;
 	while(1) { /* loop is terminated if we have a bad receive, done below in the body */
 		socklen = sizeof(struct sockaddr_storage);
-		l = recvfrom(fd, (char*) pRcvBuf, iMaxLine, 0, (struct sockaddr *)&frominet, &socklen);
-		if(l < 0) {
+		lenRcvBuf = recvfrom(fd, (char*) pRcvBuf, iMaxLine, 0, (struct sockaddr *)&frominet, &socklen);
+		if(lenRcvBuf < 0) {
 			if(errno != EINTR && errno != EAGAIN) {
 				rs_strerror_r(errno, errStr, sizeof(errStr));
 				DBGPRINTF("INET socket error: %d = %s.\n", errno, errStr);
@@ -193,13 +196,25 @@ processSocket(int fd, struct sockaddr_storage *frominetPrev, int *pbIsPermitted,
 			}	
 		}
 
-		DBGPRINTF("Message from inetd socket: #%d, host: %s, isPermitted: %d\n", fd, fromHost, *pbIsPermitted);
+		DBGPRINTF("recv(%d,%d)/%s,acl:%d,msg:%.80s\n", fd, (int) lenRcvBuf, fromHost, *pbIsPermitted, pRcvBuf);
+
 		if(*pbIsPermitted)  {
 			if((iTimeRequery == 0) || (iNbrTimeUsed++ % iTimeRequery) == 0) {
 				datetime.getCurrTime(&stTime, &ttGenTime);
 			}
-			parseAndSubmitMessage(fromHost, fromHostIP, pRcvBuf, l,
-				MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_NO_DELAY, (uchar*)"imudp", &stTime, ttGenTime);
+			/* we now create our own message object and submit it to the queue */
+			CHKiRet(msgConstructWithTime(&pMsg, &stTime, ttGenTime));
+			/* first trim the buffer to what we have actually received */
+			CHKmalloc(pMsg->pszRawMsg = malloc(sizeof(uchar)* lenRcvBuf));
+			memcpy(pMsg->pszRawMsg, pRcvBuf, lenRcvBuf);
+			pMsg->iLenRawMsg = lenRcvBuf;
+			MsgSetInputName(pMsg, "imudp");
+			MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY);
+			pMsg->bParseHOSTNAME  = MSG_PARSE_HOSTNAME;
+			pMsg->msgFlags  = NOFLAG;
+			MsgSetRcvFrom(pMsg, (char*)fromHost);
+			CHKiRet(MsgSetRcvFromIP(pMsg, fromHostIP));
+			CHKiRet(submitMsg(pMsg));
 		}
 	}
 
diff --git a/runtime/Makefile.am b/runtime/Makefile.am
index 81a9d5bd..7c70dd8d 100644
--- a/runtime/Makefile.am
+++ b/runtime/Makefile.am
@@ -16,6 +16,8 @@ librsyslog_la_SOURCES = \
 	glbl.c \
 	conf.c \
 	conf.h \
+	parser.h \
+	parser.c \
 	msg.c \
 	msg.h \
 	linkedlist.c \
diff --git a/runtime/msg.h b/runtime/msg.h
index d2fc2f30..98635f85 100644
--- a/runtime/msg.h
+++ b/runtime/msg.h
@@ -52,7 +52,10 @@ struct msg {
 	BEGINobjInstance;	/* Data to implement generic object - MUST be the first data element! */
 	pthread_mutexattr_t mutAttr;
 	pthread_mutex_t mut;
-	int	iRefCount;	/* reference counter (0 = unused) */
+	flowControl_t flowCtlType; /**< type of flow control we can apply, for enqueueing, needs not to be persisted because
+				        once data has entered the queue, this property is no longer needed. */
+	short	iRefCount;	/* reference counter (0 = unused) */
+	short	bIsParsed;	/* is message parsed? (0=no, 1=yes), 0 means parser needs to be called */
 	short	bParseHOSTNAME;	/* should the hostname be parsed from the message? */
 	   /* background: the hostname is not present on "regular" messages
 	    * received via UNIX domain sockets from the same machine. However,
@@ -60,8 +63,6 @@ struct msg {
 	    * sockets. All in all, the parser would need parse templates, that would
 	    * resolve all these issues... rgerhards, 2005-10-06
 	    */
-	flowControl_t flowCtlType; /**< type of flow control we can apply, for enqueueing, needs not to be persisted because
-				        once data has entered the queue, this property is no longer needed. */
 	short	iSeverity;	/* the severity 0..7 */
 	uchar *pszSeverity;	/* severity as string... */
 	int iLenSeverity;	/* ... and its length. */
diff --git a/runtime/parser.c b/runtime/parser.c
new file mode 100644
index 00000000..8c4272a0
--- /dev/null
+++ b/runtime/parser.c
@@ -0,0 +1,314 @@
+/* parser.c
+ * This module contains functions for message parsers. It still needs to be
+ * converted into an object (and much extended).
+ *
+ * Module begun 2008-10-09 by Rainer Gerhards (based on previous code from syslogd.c)
+ *
+ * Copyright 2008 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+#include "config.h"
+#include <stdlib.h>
+#include <ctype.h>
+#include <assert.h>
+#ifdef USE_NETZIP
+#include <zlib.h>
+#endif
+
+#include "rsyslog.h"
+#include "dirty.h"
+#include "msg.h"
+#include "obj.h"
+#include "errmsg.h"
+
+/* some defines */
+#define DEFUPRI		(LOG_USER|LOG_NOTICE)
+
+#warning "msg object must be updated with new property for persisting the queue!"
+/* definitions for objects we access */
+DEFobjStaticHelpers
+DEFobjCurrIf(glbl)
+DEFobjCurrIf(errmsg)
+
+/* static data */
+
+
+/* this is a dummy class init
+ */
+rsRetVal parserClassInit(void)
+{
+	DEFiRet;
+
+	/* request objects we use */
+	CHKiRet(objGetObjInterface(&obj)); /* this provides the root pointer for all other queries */
+	CHKiRet(objUse(glbl, CORE_COMPONENT));
+	CHKiRet(objUse(errmsg, CORE_COMPONENT));
+// TODO: free components! see action.c
+finalize_it:
+	RETiRet;
+}
+
+
+/* uncompress a received message if it is compressed.
+ * pMsg->pszRawMsg buffer is updated.
+ * rgerhards, 2008-10-09
+ */
+static inline rsRetVal uncompressMessage(msg_t *pMsg)
+{
+	DEFiRet;
+#	ifdef USE_NETZIP
+	uchar *deflateBuf = NULL;
+	uLongf iLenDefBuf;
+	uchar *pszMsg;
+	size_t lenMsg;
+	
+	assert(pMsg != NULL);
+	pszMsg = pMsg->pszRawMsg;
+	lenMsg = pMsg->iLenRawMsg;
+
+	/* we first need to check if we have a compressed record. If so,
+	 * we must decompress it.
+	 */
+	if(lenMsg > 0 && *pszMsg == 'z') { /* compressed data present? (do NOT change order if conditions!) */
+		/* we have compressed data, so let's deflate it. We support a maximum
+		 * message size of iMaxLine. If it is larger, an error message is logged
+		 * and the message is dropped. We do NOT try to decompress larger messages
+		 * as such might be used for denial of service. It might happen to later
+		 * builds that such functionality be added as an optional, operator-configurable
+		 * feature.
+		 */
+		int ret;
+		iLenDefBuf = glbl.GetMaxLine();
+		CHKmalloc(deflateBuf = malloc(sizeof(uchar) * (iLenDefBuf + 1)));
+		ret = uncompress((uchar *) deflateBuf, &iLenDefBuf, (uchar *) pszMsg+1, lenMsg-1);
+		DBGPRINTF("Compressed message uncompressed with status %d, length: new %ld, old %d.\n",
+		        ret, (long) iLenDefBuf, (int) (lenMsg-1));
+		/* Now check if the uncompression worked. If not, there is not much we can do. In
+		 * that case, we log an error message but ignore the message itself. Storing the
+		 * compressed text is dangerous, as it contains control characters. So we do
+		 * not do this. If someone would like to have a copy, this code here could be
+		 * modified to do a hex-dump of the buffer in question. We do not include
+		 * this functionality right now.
+		 * rgerhards, 2006-12-07
+		 */
+		if(ret != Z_OK) {
+			errmsg.LogError(0, NO_ERRCODE, "Uncompression of a message failed with return code %d "
+			            "- enable debug logging if you need further information. "
+				    "Message ignored.", ret);
+			FINALIZE; /* unconditional exit, nothing left to do... */
+		}
+		free(pMsg->pszRawMsg);
+		pMsg->pszRawMsg = deflateBuf;
+		pMsg->iLenRawMsg = iLenDefBuf;
+		deflateBuf = NULL; /* logically "freed" - caller is now responsible */
+	}
+finalize_it:
+	if(deflateBuf != NULL)
+		free(deflateBuf);
+
+#	else /* ifdef USE_NETZIP */
+
+	/* in this case, we still need to check if the message is compressed. If so, we must
+	 * tell the user we can not accept it.
+	 */
+	if(len > 0 && *msg == 'z') {
+		errmsg.LogError(0, NO_ERRCODE, "Received a compressed message, but rsyslogd does not have compression "
+		         "support enabled. The message will be ignored.");
+		ABORT_FINALIZE(RS_RET_NO_ZIP);
+	}	
+
+#	endif /* ifdef USE_NETZIP */
+
+	RETiRet;
+}
+
+
+/* sanitize a received message
+ * if a message gets to large during sanitization, it is truncated. This is
+ * as specified in the upcoming syslog RFC series.
+ * rgerhards, 2008-10-09
+ * We check if we have a NUL character at the very end of the
+ * message. This seems to be a frequent problem with a number of senders.
+ * So I have now decided to drop these NULs. However, if they are intentional,
+ * that may cause us some problems, e.g. with syslog-sign. On the other hand,
+ * current code always has problems with intentional NULs (as it needs to escape
+ * them to prevent problems with the C string libraries), so that does not
+ * really matter. Just to be on the save side, we'll log destruction of such
+ * NULs in the debug log.
+ * rgerhards, 2007-09-14
+ */
+static inline rsRetVal
+sanitizeMessage(msg_t *pMsg)
+{
+	DEFiRet;
+	uchar *pszMsg;
+	uchar *pDst; /* destination for copy job */
+	size_t lenMsg;
+	size_t iSrc;
+	size_t iDst;
+	size_t iMaxLine;
+
+	assert(pMsg != NULL);
+
+#	ifdef USE_NETZIP
+	CHKiRet(uncompressMessage(pMsg));
+#	endif
+
+	pszMsg = pMsg->pszRawMsg;
+	lenMsg = pMsg->iLenRawMsg;
+
+	/* remove NUL character at end of message (see comment in function header) */
+	if(pszMsg[lenMsg-1] == '\0') {
+		DBGPRINTF("dropped NUL at very end of message\n");
+		lenMsg--;
+	}
+
+	/* then we check if we need to drop trailing LFs, which often make
+	 * their way into syslog messages unintentionally. In order to remain
+	 * compatible to recent IETF developments, we allow the user to
+	 * turn on/off this handling.  rgerhards, 2007-07-23
+	 */
+	if(bDropTrailingLF && pszMsg[lenMsg-1] == '\n') {
+		DBGPRINTF("dropped LF at very end of message (DropTrailingLF is set)\n");
+		lenMsg--;
+	}
+
+	/* now copy over the message and sanitize it */
+	/* TODO: can we get cheaper memory alloc? {alloca()?}*/
+	iMaxLine = glbl.GetMaxLine();
+	CHKmalloc(pDst = malloc(sizeof(uchar) * (iMaxLine + 1)));
+	iSrc = iDst = 0;
+	while(iSrc < lenMsg && iDst < iMaxLine) {
+		if(pszMsg[iSrc] == '\0') { /* guard against \0 characters... */
+			/* changed to the sequence (somewhat) proposed in
+			 * draft-ietf-syslog-protocol-19. rgerhards, 2006-11-30
+			 */
+			if(iDst + 3 < iMaxLine) { /* do we have space? */
+				pDst[iDst++] =  cCCEscapeChar;
+				pDst[iDst++] = '0';
+				pDst[iDst++] = '0';
+				pDst[iDst++] = '0';
+			} /* if we do not have space, we simply ignore the '\0'... */
+			  /* log an error? Very questionable... rgerhards, 2006-11-30 */
+			  /* decided: we do not log an error, it won't help... rger, 2007-06-21 */
+		} else if(bEscapeCCOnRcv && iscntrl((int) pszMsg[iSrc])) {
+			/* we are configured to escape control characters. Please note
+			 * that this most probably break non-western character sets like
+			 * Japanese, Korean or Chinese. rgerhards, 2007-07-17
+			 * Note: sysklogd logs octal values only for DEL and CCs above 127.
+			 * For others, it logs ^n where n is the control char converted to an
+			 * alphabet character. We like consistency and thus escape it to octal
+			 * in all cases. If someone complains, we may change the mode. At least
+			 * we known now what's going on.
+			 * rgerhards, 2007-07-17
+			 */
+			if(iDst + 3 < iMaxLine) { /* do we have space? */
+				pDst[iDst++] = cCCEscapeChar;
+				pDst[iDst++] = '0' + ((pszMsg[iSrc] & 0300) >> 6);
+				pDst[iDst++] = '0' + ((pszMsg[iSrc] & 0070) >> 3);
+				pDst[iDst++] = '0' + ((pszMsg[iSrc] & 0007));
+			} /* again, if we do not have space, we ignore the char - see comment at '\0' */
+		} else {
+			pDst[iDst++] = pszMsg[iSrc];
+		}
+		++iSrc;
+	}
+	pDst[iDst] = '\0'; /* space *is* reserved for this! */
+
+	/* we have a sanitized string. Let's save it now */
+	free(pMsg->pszRawMsg);
+	if((pMsg->pszRawMsg = malloc((iDst+1) * sizeof(uchar))) == NULL) {
+		/* when we get no new buffer, we use what we already have ;) */
+		pMsg->pszRawMsg = pDst;
+	} else {
+		/* trim buffer */
+		memcpy(pMsg->pszRawMsg, pDst, iDst+1);
+		free(pDst); /* too big! */
+		pMsg->iLenRawMsg = iDst;
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+/* Parse a received message. The object's rawmsg property is taken and
+ * parsed according to the relevant standards. This can later be
+ * extended to support configured parsers.
+ * rgerhards, 2008-10-09
+ */
+rsRetVal parseMsg(msg_t *pMsg)
+{
+	DEFiRet;
+	uchar *msg;
+	int pri;
+
+	CHKiRet(sanitizeMessage(pMsg));
+
+	/* we needed to sanitize first, because we otherwise do not have a C-string we can print... */
+	DBGPRINTF("msg parser: flags %x, from '%s', msg %s\n", pMsg->msgFlags, pMsg->pszRcvFrom, pMsg->pszRawMsg);
+
+	/* pull PRI */
+	pri = DEFUPRI;
+	msg = pMsg->pszRawMsg;
+	if(*msg == '<') {
+		pri = 0;
+		while(isdigit((int) *++msg)) {
+			pri = 10 * pri + (*msg - '0');
+		}
+		if(*msg == '>')
+			++msg;
+		if(pri & ~(LOG_FACMASK|LOG_PRIMASK))
+			pri = DEFUPRI;
+	}
+	pMsg->iFacility = LOG_FAC(pri);
+	pMsg->iSeverity = LOG_PRI(pri);
+	MsgSetUxTradMsg(pMsg, (char*) msg);
+
+	if(pMsg->bParseHOSTNAME == 0)
+		MsgSetHOSTNAME(pMsg, (char*) pMsg->pszRcvFrom);
+
+	/* rger 2005-11-24 (happy thanksgiving!): we now need to check if we have
+	 * a traditional syslog message or one formatted according to syslog-protocol.
+	 * We need to apply different parsers depending on that. We use the
+	 * -protocol VERSION field for the detection.
+	 */
+	if(msg[0] == '1' && msg[1] == ' ') {
+		dbgprintf("Message has syslog-protocol format.\n");
+		setProtocolVersion(pMsg, 1);
+		if(parseRFCSyslogMsg(pMsg, pMsg->msgFlags) == 1) { // TODO: parseRFC... should pull flags from pMsg
+			msgDestruct(&pMsg);
+			ABORT_FINALIZE(RS_RET_ERR); // TODO: we need to handle these cases!
+		}
+	} else { /* we have legacy syslog */
+		dbgprintf("Message has legacy syslog format.\n");
+		setProtocolVersion(pMsg, 0);
+		if(parseLegacySyslogMsg(pMsg, pMsg->msgFlags) == 1) {
+			msgDestruct(&pMsg);
+			ABORT_FINALIZE(RS_RET_ERR); // TODO: we need to handle these cases!
+		}
+	}
+
+	/* finalize message object */
+	pMsg->bIsParsed = 1; /* this message is now parsed */
+	MsgPrepareEnqueue(pMsg); /* "historical" name - preparese for multi-threading */
+
+finalize_it:
+	RETiRet;
+}
diff --git a/runtime/parser.h b/runtime/parser.h
new file mode 100644
index 00000000..cec9c083
--- /dev/null
+++ b/runtime/parser.h
@@ -0,0 +1,30 @@
+/* header for parser.c
+ * This is not yet an object, but contains all those code necessary to
+ * parse syslog messages.
+ *
+ * Copyright 2008 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+#ifndef INCLUDED_PARSE_H
+#define INCLUDED_PARSE_H
+
+extern rsRetVal parserClassInit(void);
+extern rsRetVal parseMsg(msg_t*);
+
+#endif /* #ifndef INCLUDED_PARSE_H */
diff --git a/runtime/queue.c b/runtime/queue.c
index f5f770b3..25c0bd5f 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -1513,8 +1513,6 @@ queueRateLimiter(queue_t *pThis)
 
 	ISOBJ_TYPE_assert(pThis, queue);
 
-	dbgoprint((obj_t*) pThis, "entering rate limiter\n");
-
 	iDelay = 0;
 	if(pThis->iDeqtWinToHr != 25) { /* 25 means disabled */
 		/* time calls are expensive, so only do them when needed */
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index 361bfb47..619343bd 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -252,6 +252,7 @@ enum rsRetVal_				/** return value. All methods return this if not specified oth
 	RS_RET_QUEUE_FULL = -2105, /**< queue is full, operation could not be completed */
 	RS_RET_ACCEPT_ERR = -2106, /**< error during accept() system call */
 	RS_RET_INVLD_TIME = -2107, /**< invalid timestamp (e.g. could not be parsed) */
+	RS_RET_NO_ZIP = -2108, /**< ZIP functionality is not present */
 
 	/* RainerScript error messages (range 1000.. 1999) */
 	RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */
diff --git a/runtime/wti.c b/runtime/wti.c
index 365b25d5..4abca090 100644
--- a/runtime/wti.c
+++ b/runtime/wti.c
@@ -299,7 +299,7 @@ wtiWorkerCancelCleanup(void *arg)
 	pWtp = pThis->pWtp;
 	ISOBJ_TYPE_assert(pWtp, wtp);
 
-	dbgprintf("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis));
+	DBGPRINTF("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis));
 	
 	/* call user supplied handler (that one e.g. requeues the element) */
 	pWtp->pfOnWorkerCancel(pThis->pWtp->pUsr, pThis->pUsrp);
@@ -391,7 +391,7 @@ wtiWorker(wti_t *pThis)
 		/* if we reach this point, we are still protected by the mutex */
 
 		if(pWtp->pfIsIdle(pWtp->pUsr, MUTEX_ALREADY_LOCKED)) {
-			dbgprintf("%s: worker IDLE, waiting for work.\n", wtiGetDbgHdr(pThis));
+			DBGPRINTF("%s: worker IDLE, waiting for work.\n", wtiGetDbgHdr(pThis));
 			pWtp->pfOnIdle(pWtp->pUsr, MUTEX_ALREADY_LOCKED);
 
 			if(pWtp->toWrkShutdown == -1) {
@@ -400,7 +400,7 @@ wtiWorker(wti_t *pThis)
 			} else {
 				timeoutComp(&t, pWtp->toWrkShutdown);/* get absolute timeout */
 				if(d_pthread_cond_timedwait(pWtp->pcondBusy, pWtp->pmutUsr, &t) != 0) {
-					dbgprintf("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis));
+					DBGPRINTF("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis));
 					bInactivityTOOccured = 1; /* indicate we had a timeout */
 				}
 			}
diff --git a/tools/syslogd.c b/tools/syslogd.c
index e794e2d1..1a26333d 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -128,6 +128,7 @@
 #include "vm.h"
 #include "errmsg.h"
 #include "datetime.h"
+#include "parser.h"
 #include "sysvar.h"
 
 /* definitions for objects we access */
@@ -249,15 +250,15 @@ typedef struct legacyOptsLL_s {
 legacyOptsLL_t *pLegacyOptsLL = NULL;
 
 /* global variables for config file state */
-static int	bDropTrailingLF = 1; /* drop trailing LF's on reception? */
+int	bDropTrailingLF = 1; /* drop trailing LF's on reception? */
 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
 					   as compatible to sysklogd as possible. */
 static int	bDebugPrintTemplateList = 1;/* output template list in debug mode? */
 static int	bDebugPrintCfSysLineHandlerList = 1;/* output cfsyslinehandler list in debug mode? */
 static int	bDebugPrintModuleList = 1;/* output module list in debug mode? */
-static uchar	cCCEscapeChar = '\\';/* character to be used to start an escape sequence for control chars */
-static int 	bEscapeCCOnRcv = 1; /* escape control characters on reception: 0 - no, 1 - yes */
+uchar	cCCEscapeChar = '\\';/* character to be used to start an escape sequence for control chars */
+int 	bEscapeCCOnRcv = 1; /* escape control characters on reception: 0 - no, 1 - yes */
 static int	bErrMsgToStderr = 1; /* print error messages to stderr (in addition to everything else)? */
 int 	bReduceRepeatMsgs; /* reduce repeated message - 0 - no, 1 - yes */
 int	bActExecWhenPrevSusp; /* execute action only when previous one was suspended? */
@@ -596,6 +597,7 @@ static inline rsRetVal printline(uchar *hname, uchar *hnameIP, uchar *msg, int b
 	int pri;
 	msg_t *pMsg;
 
+	pMsg->bIsParsed = 1; /* this is a hack until this function can be removed TODO: do it soon (rgerhards, 2008-10-09)! */
 	/* Now it is time to create the message object (rgerhards) */
 	if(stTime == NULL) {
 		CHKiRet(msgConstruct(&pMsg));
@@ -1190,6 +1192,9 @@ msgConsumer(void __attribute__((unused)) *notNeeded, void *pUsr)
 
 	assert(pMsg != NULL);
 
+	if(pMsg->bIsParsed == 0) {
+		parseMsg(pMsg);
+	}
 	processMsg(pMsg);
 	msgDestruct(&pMsg);
 
@@ -1311,7 +1316,7 @@ static int parseRFCStructuredData(char **pp2parse, char *pResult)
  *
  * rger, 2005-11-24
  */
-static int parseRFCSyslogMsg(msg_t *pMsg, int flags)
+int parseRFCSyslogMsg(msg_t *pMsg, int flags)
 {
 	char *p2parse;
 	char *pBuf;
@@ -1407,7 +1412,7 @@ static int parseRFCSyslogMsg(msg_t *pMsg, int flags)
  * but I thought I log it in this comment.
  * rgerhards, 2006-01-10
  */
-static int parseLegacySyslogMsg(msg_t *pMsg, int flags)
+int parseLegacySyslogMsg(msg_t *pMsg, int flags)
 {
 	char *p2parse;
 	char *pBuf;
@@ -2908,6 +2913,8 @@ InitGlobalClasses(void)
 	CHKiRet(actionClassInit());
 	pErrObj = "template";
 	CHKiRet(templateInit());
+	pErrObj = "parser";
+	CHKiRet(parserClassInit());
 
 	/* TODO: the dependency on net shall go away! -- rgerhards, 2008-03-07 */
 	pErrObj = "net";
@@ -3526,6 +3533,5 @@ int main(int argc, char **argv)
 	dbgClassInit();
 	return realMain(argc, argv);
 }
-
 /* vim:set ai:
  */