Imported Upstream version 7.4.8upstream/7.4.8

19 files changed, 193 insertions, 54 deletions

diff --git a/runtime/conf.c b/runtime/conf.c
index c3c7e44..c01715c 100644
--- a/runtime/conf.c
+++ b/runtime/conf.c
@@ -573,6 +573,7 @@ rsRetVal DecodePropFilter(uchar *pline, struct cnfstmt *stmt)
 	} else {
 		errmsg.LogError(0, NO_ERRCODE, "error: invalid compare operation '%s' - ignoring selector",
 		           (char*) rsCStrGetSzStrNoNULL(pCSCompOp));
+		return(RS_RET_ERR);
 	}
 	rsCStrDestruct(&pCSCompOp); /* no longer needed */
 
diff --git a/runtime/cryprov.h b/runtime/cryprov.h
index 8496b74..005b33f 100644
--- a/runtime/cryprov.h
+++ b/runtime/cryprov.h
@@ -24,8 +24,6 @@
 #ifndef INCLUDED_CRYPROV_H
 #define INCLUDED_CRYPROV_H
 
-#include <gcrypt.h>
-
 /* interface */
 BEGINinterface(cryprov) /* name must also be changed in ENDinterface macro! */
 	rsRetVal (*Construct)(void *ppThis);
diff --git a/runtime/glbl.c b/runtime/glbl.c
index b3fe3a1..c57cedf 100644
--- a/runtime/glbl.c
+++ b/runtime/glbl.c
@@ -32,6 +32,7 @@
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <unistd.h>
+#include <pthread.h>
 #include <assert.h>
 
 #include "rsyslog.h"
@@ -71,6 +72,7 @@ static int option_DisallowWarning = 1;	/* complain if message from disallowed se
 static int bDisableDNS = 0; /* don't look up IP addresses of remote messages */
 static prop_t *propLocalIPIF = NULL;/* IP address to report for the local host (default is 127.0.0.1) */
 static prop_t *propLocalHostName = NULL;/* our hostname as FQDN - read-only after startup */
+static prop_t *propLocalHostNameToDelete = NULL;/* see GenerateLocalHostName function hdr comment! */
 static uchar *LocalHostName = NULL;/* our hostname  - read-only after startup, except HUP */
 static uchar *LocalHostNameOverride = NULL;/* user-overridden hostname - read-only after startup */
 static uchar *LocalFQDNName = NULL;/* our hostname as FQDN - read-only after startup, except HUP */
@@ -379,17 +381,31 @@ GetLocalDomain(void)
 /* generate the local hostname property. This must be done after the hostname info
  * has been set as well as PreserveFQDN.
  * rgerhards, 2009-06-30
+ * NOTE: This function tries to avoid locking by not destructing the previous value
+ * immediately. This is so that current readers can  continue to use the previous name.
+ * Otherwise, we would need to use read/write locks to protect the update process.
+ * In order to do so, we save the previous value and delete it when we are called again
+ * the next time. Note that this in theory is racy and can lead to a double-free.
+ * In practice, however, the window of exposure to trigger this is extremely short
+ * and as this functions is very infrequently being called (on HUP), the trigger
+ * condition for this bug is so highly unlikely that it never occurs in practice.
+ * Probably if you HUP rsyslog every few milliseconds, but who does that...
+ * To further reduce risk potential, we do only update the property when there
+ * actually is a hostname change, which makes it even less likely.
+ * rgerhards, 2013-10-28
  */
 static rsRetVal
 GenerateLocalHostNameProperty(void)
 {
-	DEFiRet;
+	uchar *pszPrev;
+	int lenPrev;
+	prop_t *hostnameNew;
 	uchar *pszName;
+	DEFiRet;
 
-	if(propLocalHostName != NULL)
-		prop.Destruct(&propLocalHostName);
+	if(propLocalHostNameToDelete != NULL)
+		prop.Destruct(&propLocalHostNameToDelete);
 
-	CHKiRet(prop.Construct(&propLocalHostName));
 	if(LocalHostNameOverride == NULL) {
 		if(LocalHostName == NULL)
 			pszName = (uchar*) "[localhost]";
@@ -403,8 +419,20 @@ GenerateLocalHostNameProperty(void)
 		pszName = LocalHostNameOverride;
 	}
 	DBGPRINTF("GenerateLocalHostName uses '%s'\n", pszName);
-	CHKiRet(prop.SetString(propLocalHostName, pszName, ustrlen(pszName)));
-	CHKiRet(prop.ConstructFinalize(propLocalHostName));
+
+	if(propLocalHostName == NULL)
+		pszPrev = (uchar*)""; /* make sure strcmp() below does not match */
+	else
+		prop.GetString(propLocalHostName, &pszPrev, &lenPrev);
+
+	if(ustrcmp(pszPrev, pszName)) {
+		/* we need to update */
+		CHKiRet(prop.Construct(&hostnameNew));
+		CHKiRet(prop.SetString(hostnameNew, pszName, ustrlen(pszName)));
+		CHKiRet(prop.ConstructFinalize(hostnameNew));
+		propLocalHostNameToDelete = propLocalHostName;
+		propLocalHostName = hostnameNew;
+	}
 
 finalize_it:
 	RETiRet;
@@ -445,6 +473,14 @@ GetWorkDir(void)
 	return(pszWorkDir == NULL ? (uchar*) "" : pszWorkDir);
 }
 
+/* return the "raw" working directory, which means
+ * NULL if unset.
+ */
+const uchar *
+glblGetWorkDirRaw(void)
+{
+	return pszWorkDir;
+}
 
 /* return the current default netstream driver */
 static uchar*
@@ -667,6 +703,8 @@ BEGINObjClassExit(glbl, OBJ_IS_CORE_MODULE) /* class, version */
 	free(LocalHostNameOverride);
 	free(LocalFQDNName);
 	objRelease(prop, CORE_COMPONENT);
+	if(propLocalHostNameToDelete != NULL)
+		prop.Destruct(&propLocalHostNameToDelete);
 	DESTROY_ATOMIC_HELPER_MUT(mutTerminateInputs);
 ENDObjClassExit(glbl)
 
diff --git a/runtime/glbl.h b/runtime/glbl.h
index e95e48f..44171f2 100644
--- a/runtime/glbl.h
+++ b/runtime/glbl.h
@@ -95,5 +95,6 @@ static inline void glblSetOurPid(pid_t pid) { glbl_ourpid = pid; }
 void glblPrepCnf(void);
 void glblProcessCnf(struct cnfobj *o);
 void glblDoneLoadCnf(void);
+const uchar * glblGetWorkDirRaw(void);
 
 #endif /* #ifndef GLBL_H_INCLUDED */
diff --git a/runtime/libgcry.h b/runtime/libgcry.h
index b77b0f9..692ce40 100644
--- a/runtime/libgcry.h
+++ b/runtime/libgcry.h
@@ -21,7 +21,7 @@
 #ifndef INCLUDED_LIBGCRY_H
 #define INCLUDED_LIBGCRY_H
 #include <stdint.h>
-
+#include <gcrypt.h>
 
 struct gcryctx_s {
 	uchar *key;
@@ -52,6 +52,7 @@ void rsgcryCtxDel(gcryctx ctx);
 int gcryfileDestruct(gcryfile gf, off64_t offsLogfile);
 rsRetVal rsgcryInitCrypt(gcryctx ctx, gcryfile *pgf, uchar *fname);
 int rsgcryEncrypt(gcryfile pF, uchar *buf, size_t *len);
+int gcryGetKeyFromProg(char *cmd, char **key, unsigned *keylen);
 
 /* error states */
 #define RSGCRYE_EI_OPEN 1 	/* error opening .encinfo file */
diff --git a/runtime/librsgt_read.c b/runtime/librsgt_read.c
index a9a5079..972b5a4 100644
--- a/runtime/librsgt_read.c
+++ b/runtime/librsgt_read.c
@@ -267,7 +267,7 @@ rsgt_tlvRecRead(FILE *fp, tlvrecord_t *rec)
 		rec->tlvlen = c;
 	}
 	if(fread(rec->data, (size_t) rec->tlvlen, 1, fp) != 1) {
-		r = RSGTE_IO;
+		r = feof(fp) ? RSGTE_EOF : RSGTE_IO;
 		goto done;
 	}
 
diff --git a/runtime/msg.c b/runtime/msg.c
index 36cbd26..10ecf48 100644
--- a/runtime/msg.c
+++ b/runtime/msg.c
@@ -43,7 +43,7 @@
 #include <libestr.h>
 #include <json.h>
 /* For struct json_object_iter, should not be necessary in future versions */
-#include <json/json_object_private.h>
+#include <json_object_private.h>
 #if HAVE_MALLOC_H
 #  include <malloc.h>
 #endif
@@ -3934,6 +3934,12 @@ msgAddJSON(msg_t *pM, uchar *name, struct json_object *json)
 		}
 		leaf = jsonPathGetLeaf(name, ustrlen(name));
 		CHKiRet(jsonPathFindParent(pM, name, leaf, &parent, 1));
+		if (json_object_get_type(parent) != json_type_object) {
+			DBGPRINTF("msgAddJSON: not a container in json path,"
+				"name is '%s'\n", name);
+			json_object_put(json);
+			ABORT_FINALIZE(RS_RET_INVLD_SETOP);
+		}
 		leafnode = json_object_object_get(parent, (char*)leaf);
 		if(leafnode == NULL) {
 			json_object_object_add(parent, (char*)leaf, json);
diff --git a/runtime/msg.h b/runtime/msg.h
index ac220b6..e7babdb 100644
--- a/runtime/msg.h
+++ b/runtime/msg.h
@@ -62,7 +62,6 @@ struct msg {
 				        once data has entered the queue, this property is no longer needed. */
 	pthread_mutex_t mut;
 	int	iRefCount;	/* reference counter (0 = unused) */
-	sbool	bAlreadyFreed;	/* aid to help detect a well-hidden bad bug -- TODO: remove when no longer needed */
 	sbool	bParseSuccess;	/* set to reflect state of last executed higher level parser */
 	short	iSeverity;	/* the severity 0..7 */
 	short	iFacility;	/* Facility code 0 .. 23*/
diff --git a/runtime/nsd_gtls.c b/runtime/nsd_gtls.c
index 6ef4feb..1110c7a 100644
--- a/runtime/nsd_gtls.c
+++ b/runtime/nsd_gtls.c
@@ -2,7 +2,7 @@
  *
  * An implementation of the nsd interface for GnuTLS.
  * 
- * Copyright (C) 2007, 2008 Rainer Gerhards and Adiscon GmbH.
+ * Copyright (C) 2007-2013 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of the rsyslog runtime library.
  *
@@ -547,10 +547,20 @@ gtlsAddOurCert(void)
 	keyFile = glbl.GetDfltNetstrmDrvrKeyFile();
 	dbgprintf("GTLS certificate file: '%s'\n", certFile);
 	dbgprintf("GTLS key file: '%s'\n", keyFile);
+	if(certFile == NULL) {
+		errmsg.LogError(0, RS_RET_CERT_MISSING, "error: certificate file is not set, cannot "
+				"continue");
+		ABORT_FINALIZE(RS_RET_CERT_MISSING);
+	}
+	if(keyFile == NULL) {
+		errmsg.LogError(0, RS_RET_CERTKEY_MISSING, "error: key file is not set, cannot "
+				"continue");
+		ABORT_FINALIZE(RS_RET_CERTKEY_MISSING);
+	}
 	CHKgnutls(gnutls_certificate_set_x509_key_file(xcred, (char*)certFile, (char*)keyFile, GNUTLS_X509_FMT_PEM));
 
 finalize_it:
-	if(iRet != RS_RET_OK) {
+	if(iRet != RS_RET_OK && iRet != RS_RET_CERT_MISSING && iRet != RS_RET_CERTKEY_MISSING) {
 		pGnuErr = gtlsStrerror(gnuRet);
 		errno = 0;
 		errmsg.LogError(0, iRet, "error adding our certificate. GnuTLS error %d, message: '%s', "
@@ -580,6 +590,11 @@ gtlsGlblInit(void)
 
 	/* sets the trusted cas file */
 	cafile = glbl.GetDfltNetstrmDrvrCAF();
+	if(cafile == NULL) {
+		errmsg.LogError(0, RS_RET_CA_CERT_MISSING, "error: ca certificate is not set, cannot "
+				"continue");
+		ABORT_FINALIZE(RS_RET_CA_CERT_MISSING);
+	}
 	dbgprintf("GTLS CA file: '%s'\n", cafile);
 	gnuRet = gnutls_certificate_set_x509_trust_file(xcred, (char*)cafile, GNUTLS_X509_FMT_PEM);
 	if(gnuRet < 0) {
diff --git a/runtime/queue.c b/runtime/queue.c
index 935a810..29549cd 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -12,7 +12,7 @@
  * function names - this makes it really hard to read and does not provide much
  * benefit, at least I (now) think so...
  *
- * Copyright 2008-2011 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2008-2013 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of the rsyslog runtime library.
  *
@@ -87,6 +87,7 @@ static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis);
 static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis);
 static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis);
 static rsRetVal qDestructDisk(qqueue_t *pThis);
+rsRetVal qqueueSetSpoolDir(qqueue_t *pThis, uchar *pszSpoolDir, int lenSpoolDir);
 
 /* some constants for queuePersist () */
 #define QUEUE_CHECKPOINT	1
@@ -256,7 +257,7 @@ qqueueDbgPrint(qqueue_t *pThis)
 		(pThis->pszFilePrefix == NULL) ? "[NONE]" : (char*)pThis->pszFilePrefix);
 	dbgoprint((obj_t*) pThis, "queue.size: %d\n", pThis->iMaxQueueSize);
 	dbgoprint((obj_t*) pThis, "queue.dequeuebatchsize: %d\n", pThis->iDeqBatchSize);
-	dbgoprint((obj_t*) pThis, "queue.maxdiskspace: %lld\n", pThis->iMaxFileSize);
+	dbgoprint((obj_t*) pThis, "queue.maxdiskspace: %lld\n", pThis->sizeOnDiskMax);
 	dbgoprint((obj_t*) pThis, "queue.highwatermark: %d\n", pThis->iHighWtrMrk);
 	dbgoprint((obj_t*) pThis, "queue.lowwatermark: %d\n", pThis->iLowWtrMrk);
 	dbgoprint((obj_t*) pThis, "queue.fulldelaymark: %d\n", pThis->iFullDlyMrk);
@@ -418,6 +419,7 @@ StartDA(qqueue_t *pThis)
 	CHKiRet(qqueueSetiDeqSlowdown(pThis->pqDA, pThis->iDeqSlowdown));
 	CHKiRet(qqueueSetMaxFileSize(pThis->pqDA, pThis->iMaxFileSize));
 	CHKiRet(qqueueSetFilePrefix(pThis->pqDA, pThis->pszFilePrefix, pThis->lenFilePrefix));
+	CHKiRet(qqueueSetSpoolDir(pThis->pqDA, pThis->pszSpoolDir, pThis->lenSpoolDir));
 	CHKiRet(qqueueSetiPersistUpdCnt(pThis->pqDA, pThis->iPersistUpdCnt));
 	CHKiRet(qqueueSetbSyncQueueFiles(pThis->pqDA, pThis->bSyncQueueFiles));
 	CHKiRet(qqueueSettoActShutdown(pThis->pqDA, pThis->toActShutdown));
@@ -731,7 +733,7 @@ qqueueLoadPersStrmInfoFixup(strm_t *pStrm, qqueue_t __attribute__((unused)) *pTh
 	DEFiRet;
 	ISOBJ_TYPE_assert(pStrm, strm);
 	ISOBJ_TYPE_assert(pThis, qqueue);
-	CHKiRet(strm.SetDir(pStrm, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir())));
+	CHKiRet(strm.SetDir(pStrm, pThis->pszSpoolDir, pThis->lenSpoolDir));
 finalize_it:
 	RETiRet;
 }
@@ -830,7 +832,7 @@ static rsRetVal qConstructDisk(qqueue_t *pThis)
 	} else {
 		CHKiRet(strm.Construct(&pThis->tVars.disk.pWrite));
 		CHKiRet(strm.SetbSync(pThis->tVars.disk.pWrite, pThis->bSyncQueueFiles));
-		CHKiRet(strm.SetDir(pThis->tVars.disk.pWrite, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir())));
+		CHKiRet(strm.SetDir(pThis->tVars.disk.pWrite, pThis->pszSpoolDir, pThis->lenSpoolDir));
 		CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pWrite, 10000000));
 		CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pWrite, STREAMMODE_WRITE));
 		CHKiRet(strm.SetsType(pThis->tVars.disk.pWrite, STREAMTYPE_FILE_CIRCULAR));
@@ -838,7 +840,7 @@ static rsRetVal qConstructDisk(qqueue_t *pThis)
 
 		CHKiRet(strm.Construct(&pThis->tVars.disk.pReadDeq));
 		CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDeq, 0));
-		CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDeq, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir())));
+		CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDeq, pThis->pszSpoolDir, pThis->lenSpoolDir));
 		CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pReadDeq, 10000000));
 		CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pReadDeq, STREAMMODE_READ));
 		CHKiRet(strm.SetsType(pThis->tVars.disk.pReadDeq, STREAMTYPE_FILE_CIRCULAR));
@@ -847,7 +849,7 @@ static rsRetVal qConstructDisk(qqueue_t *pThis)
 		CHKiRet(strm.Construct(&pThis->tVars.disk.pReadDel));
 		CHKiRet(strm.SetbSync(pThis->tVars.disk.pReadDel, pThis->bSyncQueueFiles));
 		CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDel, 1));
-		CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDel, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir())));
+		CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDel, pThis->pszSpoolDir, pThis->lenSpoolDir));
 		CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pReadDel, 10000000));
 		CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pReadDel, STREAMMODE_READ));
 		CHKiRet(strm.SetsType(pThis->tVars.disk.pReadDel, STREAMTYPE_FILE_CIRCULAR));
@@ -1302,6 +1304,7 @@ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThread
 {
 	DEFiRet;
 	qqueue_t *pThis;
+	const uchar *const workDir = glblGetWorkDirRaw();
 
 	ASSERT(ppThis != NULL);
 	ASSERT(pConsumer != NULL);
@@ -1311,13 +1314,15 @@ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThread
 
 	/* we have an object, so let's fill the properties */
 	objConstructSetObjInfo(pThis);
-	if((pThis->pszSpoolDir = (uchar*) strdup((char*)glbl.GetWorkDir())) == NULL)
-		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
 
+	if(workDir != NULL) {
+		if((pThis->pszSpoolDir = ustrdup(workDir)) == NULL)
+			ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+		pThis->lenSpoolDir = ustrlen(pThis->pszSpoolDir);
+	}
 	/* set some water marks so that we have useful defaults if none are set specifically */
 	pThis->iFullDlyMrk  = -1;
 	pThis->iLightDlyMrk = -1;
-	pThis->lenSpoolDir = ustrlen(pThis->pszSpoolDir);
 	pThis->iMaxFileSize = 1024 * 1024; /* default is 1 MiB */
 	pThis->iQueueSize = 0;
 	pThis->nLogDeq = 0;
@@ -2039,6 +2044,16 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 
 	ASSERT(pThis != NULL);
 
+	dbgoprint((obj_t*) pThis, "starting queue\n");
+
+	if(pThis->pszSpoolDir == NULL) {
+		/* note: we need to pick the path so late as we do not have
+		 *       the workdir during early config load
+		 */
+		if((pThis->pszSpoolDir = (uchar*) strdup((char*)glbl.GetWorkDir())) == NULL)
+			ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+		pThis->lenSpoolDir = ustrlen(pThis->pszSpoolDir);
+	}
 	/* set type-specific handlers and other very type-specific things
 	 * (we can not totally hide it...)
 	 */
@@ -2070,7 +2085,7 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 			pThis->iNumWorkerThreads = 1; /* we need exactly one worker */
 			/* pre-construct file name for .qi file */
 			pThis->lenQIFNam = snprintf((char*)pszQIFNam, sizeof(pszQIFNam) / sizeof(uchar),
-				"%s/%s.qi", (char*) glbl.GetWorkDir(), (char*)pThis->pszFilePrefix);
+				"%s/%s.qi", (char*) pThis->pszSpoolDir, (char*)pThis->pszFilePrefix);
 			pThis->pszQIFNam = ustrdup(pszQIFNam);
 			DBGOPRINT((obj_t*) pThis, ".qi file name is '%s', len %d\n", pThis->pszQIFNam,
 				(int) pThis->lenQIFNam);
@@ -2084,16 +2099,25 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 			break;
 	}
 
-	if(pThis->iFullDlyMrk == -1)
+	if(pThis->iMaxQueueSize < 100
+	   && (pThis->qType == QUEUETYPE_LINKEDLIST || pThis->qType == QUEUETYPE_FIXED_ARRAY)) {
+		errmsg.LogError(0, RS_RET_OK_WARN, "Note: queue.size=\"%d\" is very "
+			"low and can lead to unpredictable results. See also "
+			"http://www.rsyslog.com/lower-bound-for-queue-sizes/",
+			pThis->iMaxQueueSize);
+	}
+
+	/* we need to do a quick check if our water marks are set plausible. If not,
+	 * we correct the most important shortcomings.
+	 */
+	if(pThis->iFullDlyMrk == -1 || pThis->iFullDlyMrk > pThis->iMaxQueueSize)
 		pThis->iFullDlyMrk  = pThis->iMaxQueueSize
 			- (pThis->iMaxQueueSize / 100) *  3; /* default 97% */
-	if(pThis->iLightDlyMrk == -1)
+	if(pThis->iLightDlyMrk == -1 || pThis->iLightDlyMrk > pThis->iMaxQueueSize)
 		pThis->iLightDlyMrk = pThis->iMaxQueueSize
 			- (pThis->iMaxQueueSize / 100) * 30; /* default 70% */
-
-	/* we need to do a quick check if our water marks are set plausible. If not,
-	 * we correct the most important shortcomings. TODO: do that!!!! -- rgerhards, 2008-03-14
-	 */
+	if(pThis->iMaxQueueSize > 0 && pThis->iDeqBatchSize > pThis->iMaxQueueSize)
+		pThis->iDeqBatchSize = pThis->iMaxQueueSize;
 
 	/* finalize some initializations that could not yet be done because it is
 	 * influenced by properties which might have been set after queueConstruct ()
@@ -2126,9 +2150,9 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
 			pThis->iFullDlyMrk = wrk;
 	}
 
-	DBGOPRINT((obj_t*) pThis, "type %d, enq-only %d, disk assisted %d, maxFileSz %lld, maxQSize %d, lqsize %d, pqsize %d, child %d, "
+	DBGOPRINT((obj_t*) pThis, "type %d, enq-only %d, disk assisted %d, spoolDir '%s', maxFileSz %lld, maxQSize %d, lqsize %d, pqsize %d, child %d, "
 				  "full delay %d, light delay %d, deq batch size %d starting, high wtrrmrk %d, low wtrmrk %d\n",
-		  pThis->qType, pThis->bEnqOnly, pThis->bIsDA, pThis->iMaxFileSize, pThis->iMaxQueueSize,
+		  pThis->qType, pThis->bEnqOnly, pThis->bIsDA, pThis->pszSpoolDir, pThis->iMaxFileSize, pThis->iMaxQueueSize,
 		  getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis),
 		  pThis->pqParent == NULL ? 0 : 1, pThis->iFullDlyMrk, pThis->iLightDlyMrk,
 		  pThis->iDeqBatchSize, pThis->iHighWtrMrk, pThis->iLowWtrMrk);
@@ -2433,6 +2457,24 @@ CODESTARTobjDestruct(qqueue)
 ENDobjDestruct(qqueue)
 
 
+/* set the queue's spool directory. The directory MUST NOT be NULL.
+ * The passed-in string is duplicated. So if the caller does not need
+ * it any longer, it must free it.
+ */
+rsRetVal
+qqueueSetSpoolDir(qqueue_t *pThis, uchar *pszSpoolDir, int lenSpoolDir)
+{
+	DEFiRet;
+
+	free(pThis->pszSpoolDir);
+	CHKmalloc(pThis->pszSpoolDir = ustrdup(pszSpoolDir));
+	pThis->lenSpoolDir = lenSpoolDir;
+
+finalize_it:
+	RETiRet;
+}
+
+
 /* set the queue's file prefix
  * The passed-in string is duplicated. So if the caller does not need
  * it any longer, it must free it.
@@ -2564,7 +2606,7 @@ doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, msg_t *pMsg)
 	 * the queue to become ready or drop the new message. -- rgerhards, 2008-03-14
 	 */
 	while(   (pThis->iMaxQueueSize > 0 && pThis->iQueueSize >= pThis->iMaxQueueSize)
-	      || (pThis->qType == QUEUETYPE_DISK && pThis->sizeOnDiskMax != 0
+	      || ((pThis->qType == QUEUETYPE_DISK || pThis->bIsDA) && pThis->sizeOnDiskMax != 0
 	      	  && pThis->tVars.disk.sizeOnDisk > pThis->sizeOnDiskMax)) {
 		STATSCOUNTER_INC(pThis->ctrFull, pThis->mutCtrFull);
 		if(pThis->toEnq == 0 || pThis->bEnqOnly) {
@@ -2753,7 +2795,7 @@ qqueueApplyCnfParam(qqueue_t *pThis, struct cnfparamvals *pvals)
 		} else if(!strcmp(pblk.descr[i].name, "queue.dequeuebatchsize")) {
 			pThis->iDeqBatchSize = pvals[i].val.d.n;
 		} else if(!strcmp(pblk.descr[i].name, "queue.maxdiskspace")) {
-			pThis->iMaxFileSize = pvals[i].val.d.n;
+			pThis->sizeOnDiskMax = pvals[i].val.d.n;
 		} else if(!strcmp(pblk.descr[i].name, "queue.highwatermark")) {
 			pThis->iHighWtrMrk = pvals[i].val.d.n;
 		} else if(!strcmp(pblk.descr[i].name, "queue.lowwatermark")) {
@@ -2826,6 +2868,7 @@ DEFpropSetMeth(qqueue, iFullDlyMrk, int)
 DEFpropSetMeth(qqueue, iDiscardSeverity, int)
 DEFpropSetMeth(qqueue, iLightDlyMrk, int)
 DEFpropSetMeth(qqueue, bIsDA, int)
+DEFpropSetMeth(qqueue, iNumWorkerThreads, int)
 DEFpropSetMeth(qqueue, iMinMsgsPerWrkr, int)
 DEFpropSetMeth(qqueue, bSaveOnShutdown, int)
 DEFpropSetMeth(qqueue, pAction, action_t*)
diff --git a/runtime/queue.h b/runtime/queue.h
index 886fac8..7977108 100644
--- a/runtime/queue.h
+++ b/runtime/queue.h
@@ -219,6 +219,7 @@ PROTOTYPEpropSetMeth(qqueue, iLowWtrMrk, int);
 PROTOTYPEpropSetMeth(qqueue, iDiscardMrk, int);
 PROTOTYPEpropSetMeth(qqueue, iDiscardSeverity, int);
 PROTOTYPEpropSetMeth(qqueue, iMinMsgsPerWrkr, int);
+PROTOTYPEpropSetMeth(qqueue, iNumWorkerThreads, int);
 PROTOTYPEpropSetMeth(qqueue, bSaveOnShutdown, int);
 PROTOTYPEpropSetMeth(qqueue, pAction, action_t*);
 PROTOTYPEpropSetMeth(qqueue, iDeqSlowdown, int);
diff --git a/runtime/ratelimit.c b/runtime/ratelimit.c
index a808e04..016fd3c 100644
--- a/runtime/ratelimit.c
+++ b/runtime/ratelimit.c
@@ -73,16 +73,8 @@ static inline rsRetVal
 doLastMessageRepeatedNTimes(ratelimit_t *ratelimit, msg_t *pMsg, msg_t **ppRepMsg)
 {
 	int bNeedUnlockMutex = 0;
-	rsRetVal localRet;
 	DEFiRet;
 
-	if((pMsg->msgFlags & NEEDS_PARSING) != 0) {
-		if((localRet = parser.ParseMsg(pMsg)) != RS_RET_OK)  {
-			DBGPRINTF("Message discarded, parsing error %d\n", localRet);
-			ABORT_FINALIZE(RS_RET_DISCARDMSG);
-		}
-	}
-
 	if(ratelimit->bThreadSafe) {
 		pthread_mutex_lock(&ratelimit->mut);
 		bNeedUnlockMutex = 1;
@@ -209,6 +201,14 @@ rsRetVal
 ratelimitMsg(ratelimit_t *ratelimit, msg_t *pMsg, msg_t **ppRepMsg)
 {
 	DEFiRet;
+	rsRetVal localRet;
+
+	if((pMsg->msgFlags & NEEDS_PARSING) != 0) {
+		if((localRet = parser.ParseMsg(pMsg)) != RS_RET_OK)  {
+			DBGPRINTF("Message discarded, parsing error %d\n", localRet);
+			ABORT_FINALIZE(RS_RET_DISCARDMSG);
+		}
+	}
 
 	*ppRepMsg = NULL;
 	/* Only the messages having severity level at or below the
@@ -223,6 +223,10 @@ ratelimitMsg(ratelimit_t *ratelimit, msg_t *pMsg, msg_t **ppRepMsg)
 		CHKiRet(doLastMessageRepeatedNTimes(ratelimit, pMsg, ppRepMsg));
 	}
 finalize_it:
+	if(Debug) {
+		if(iRet == RS_RET_DISCARDMSG)
+			dbgprintf("message discarded by ratelimiting\n");
+	}
 	RETiRet;
 }
 
diff --git a/runtime/rsconf.c b/runtime/rsconf.c
index d8b81f1..960a34c 100644
--- a/runtime/rsconf.c
+++ b/runtime/rsconf.c
@@ -585,6 +585,7 @@ dropPrivileges(rsconf_t *cnf)
 static inline void
 tellCoreConfigLoadDone(void)
 {
+	DBGPRINTF("telling rsyslog core that config load for %p is done\n", loadConf);
 	glblDoneLoadCnf();
 }
 
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index 47b3478..e62ba86 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -3,7 +3,7 @@
  *
  * Begun 2005-09-15 RGerhards
  *
- * Copyright (C) 2005-2008 by Rainer Gerhards and Adiscon GmbH
+ * Copyright (C) 2005-2013 by Rainer Gerhards and Adiscon GmbH
  *
  * This file is part of the rsyslog runtime library.
  *
@@ -413,6 +413,9 @@ enum rsRetVal_				/** return value. All methods return this if not specified oth
 	RS_RET_CRY_INVLD_ALGO = -2326,/**< user specified invalid (unkonwn) crypto algorithm */
 	RS_RET_CRY_INVLD_MODE = -2327,/**< user specified invalid (unkonwn) crypto mode */
 	RS_RET_QUEUE_DISK_NO_FN = -2328,/**< disk queue configured, but filename not set */
+	RS_RET_CA_CERT_MISSING = -2329,/**< a CA cert is missing where one is required (e.g. TLS) */
+	RS_RET_CERT_MISSING = -2330,/**< a cert is missing where one is required (e.g. TLS) */
+	RS_RET_CERTKEY_MISSING = -2331,/**< a cert (private) key is missing where one is required (e.g. TLS) */
 
 	/* RainerScript error messages (range 1000.. 1999) */
 	RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */
diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index 5bf7ac0..1afb403 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -284,6 +284,30 @@ execStop(batch_t *pBatch, sbool *active)
 	}
 	RETiRet;
 }
+static rsRetVal
+execCall(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
+{
+	msg_t *pMsg;
+	int i;
+	DEFiRet;
+	if(stmt->d.s_call.ruleset == NULL) {
+		scriptExec(stmt->d.s_call.stmt, pBatch, active);
+	} else {
+		for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
+			CHKmalloc(pMsg = MsgDup((msg_t*) pBatch->pElem[i].pMsg));
+			DBGPRINTF("CALL: forwarding message %d to async ruleset %p\n",
+				  i, stmt->d.s_call.ruleset->pQueue);
+			MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY);
+			MsgSetRuleset(pMsg, stmt->d.s_call.ruleset);
+			/* Note: we intentionally use submitMsg2() here, as we process messages
+			 * that were already run through the rate-limiter.
+			 */
+			submitMsg2(pMsg);
+		}
+	}
+finalize_it:
+	RETiRet;
+}
 
 /* for details, see scriptExec() header comment! */
 // save current filter, evaluate new one
@@ -535,7 +559,7 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active)
 			execUnset(stmt, pBatch, active);
 			break;
 		case S_CALL:
-			scriptExec(stmt->d.s_call.stmt, pBatch, active);
+			execCall(stmt, pBatch, active);
 			break;
 		case S_IF:
 			execIf(stmt, pBatch, active);
diff --git a/runtime/ruleset.h b/runtime/ruleset.h
index cbf8243..64fe92f 100644
--- a/runtime/ruleset.h
+++ b/runtime/ruleset.h
@@ -90,6 +90,13 @@ rulesetGetName(ruleset_t *pRuleset)
 	return pRuleset->pszName;
 }
 
+/* returns 1 if the ruleset has a queue associtated, 0 if not */
+static inline int
+rulesetHasQueue(ruleset_t *pRuleset)
+{
+	return pRuleset->pQueue == NULL ? 0 : 1;
+}
+
 
 /* we will most probably convert this module back to traditional C
  * calling sequence, so here we go...
diff --git a/runtime/stringbuf.c b/runtime/stringbuf.c
index cb4f045..13f3871 100644
--- a/runtime/stringbuf.c
+++ b/runtime/stringbuf.c
@@ -107,7 +107,8 @@ finalize_it:
 
 /* a helper function for rsCStr*Strf()
  */
-static rsRetVal rsCStrConstructFromszStrv(cstr_t **ppThis, uchar *fmt, va_list ap)
+static rsRetVal rsCStrConstructFromszStrv(cstr_t **ppThis, char *fmt, va_list ap) __attribute__((format(gnu_printf,2, 0)));
+static rsRetVal rsCStrConstructFromszStrv(cstr_t **ppThis, char *fmt, va_list ap)
 {
 	DEFiRet;
 	cstr_t *pThis;
@@ -147,7 +148,7 @@ rsRetVal rsCStrConstructFromszStrf(cstr_t **ppThis, char *fmt, ...)
 	va_list ap;
 
 	va_start(ap, fmt);
-	iRet = rsCStrConstructFromszStrv(ppThis, (uchar*)fmt, ap);
+	iRet = rsCStrConstructFromszStrv(ppThis, fmt, ap);
 	va_end(ap);
 
 	RETiRet;
@@ -315,7 +316,7 @@ rsRetVal rsCStrAppendStrf(cstr_t *pThis, uchar *fmt, ...)
 	cstr_t *pStr = NULL;
 
 	va_start(ap, fmt);
-	iRet = rsCStrConstructFromszStrv(&pStr, fmt, ap);
+	iRet = rsCStrConstructFromszStrv(&pStr, (char*)fmt, ap);
 	va_end(ap);
 
 	CHKiRet(iRet);
@@ -563,7 +564,7 @@ rsRetVal cstrTrimTrailingWhiteSpace(cstr_t *pThis)
 	}
 	/* i now is the new string length! */
 	pThis->iStrLen = i;
-	pThis->pBuf[pThis->iStrLen] = '0'; /* we always have this space */
+	pThis->pBuf[pThis->iStrLen] = '\0'; /* we always have this space */
 
 done:	return RS_RET_OK;
 }
diff --git a/runtime/typedefs.h b/runtime/typedefs.h
index d3f68b4..2720109 100644
--- a/runtime/typedefs.h
+++ b/runtime/typedefs.h
@@ -25,14 +25,10 @@
  */
 #ifndef INCLUDED_TYPEDEFS_H
 #define INCLUDED_TYPEDEFS_H
-#if defined(__FreeBSD__)
+#if defined(__FreeBSD__) || !defined(HAVE_LSEEK64)
 #include <sys/types.h>
 #endif
 
-#ifndef HAVE_LSEEK64
-#include <unistd.h>
-#endif
-
 /* some universal fixed size integer defines ... */
 typedef long long int64;
 typedef long long unsigned uint64;
diff --git a/runtime/unicode-helper.h b/runtime/unicode-helper.h
index b7db276..db98ca3 100644
--- a/runtime/unicode-helper.h
+++ b/runtime/unicode-helper.h
@@ -53,7 +53,7 @@ static inline int ustrcmp(uchar *psz1, uchar *psz2)
 	return strcmp((char*) psz1, (char*) psz2);
 }
 
-static inline int ustrlen(uchar *psz)
+static inline int ustrlen(const uchar *psz)
 {
 	return strlen((char*) psz);
 }