1 files changed, 89 insertions, 116 deletions

diff --git a/plugins/imuxsock/imuxsock.c b/plugins/imuxsock/imuxsock.c
index 1a2b696e..c503852c 100644
--- a/plugins/imuxsock/imuxsock.c
+++ b/plugins/imuxsock/imuxsock.c
@@ -6,7 +6,7 @@
  *
  * File begun on 2007-12-20 by RGerhards (extracted from syslogd.c)
  *
- * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2013 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of rsyslog.
  *
@@ -55,6 +55,7 @@
 #include "statsobj.h"
 #include "datetime.h"
 #include "hashtable.h"
+#include "ratelimit.h"
 
 MODULE_TYPE_INPUT
 MODULE_TYPE_NOKEEP
@@ -105,15 +106,6 @@ STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit)
 STATSCOUNTER_DEF(ctrLostRatelimit, mutCtrLostRatelimit)
 STATSCOUNTER_DEF(ctrNumRatelimiters, mutCtrNumRatelimiters)
 
-struct rs_ratelimit_state {
-	unsigned short interval;
-	unsigned short burst;
-	unsigned done;
-	unsigned missed;
-	time_t begin;
-};
-typedef struct rs_ratelimit_state rs_ratelimit_state_t;
-
 
 /* a very simple "hash function" for process IDs - we simply use the
  * pid itself: it is quite expected that all pids may log some time, but
@@ -143,6 +135,7 @@ typedef struct lstn_s {
 	int flowCtl;		/* flow control settings for this socket */
 	int ratelimitInterval;
 	int ratelimitBurst;
+	ratelimit_t *dflt_ratelimiter;/*ratelimiter to apply if none else is to be used */
 	intTiny ratelimitSev;	/* severity level (and below) for which rate-limiting shall apply */
 	struct hashtable *ht;	/* our hashtable for rate-limiting */
 	sbool bParseHost;	/* should parser parse host name?  read-only after startup */
@@ -151,7 +144,9 @@ typedef struct lstn_s {
 	sbool bAnnotate;	/* annotate events with trusted properties */
 	sbool bParseTrusted;	/* parse trusted properties */
 	sbool bWritePid;	/* write original PID into tag */
+	sbool bDiscardOwnMsgs;	/* discard messages that originated from ourselves */
 	sbool bUseSysTimeStamp;	/* use timestamp from system (instead of from message) */
+	sbool bUnlink;		/* unlink&re-create socket at start and end of processing */
 } lstn_t;
 static lstn_t listeners[MAXFUNIX];
 
@@ -206,6 +201,8 @@ struct instanceConf_s {
 	int ratelimitSeverity;
 	int bAnnotate;			/* annotate trusted properties */
 	int bParseTrusted;		/* parse trusted properties */
+	sbool bDiscardOwnMsgs;		/* discard messages that originated from our own pid? */
+	sbool bUnlink;
 	struct instanceConf_s *next;
 };
 
@@ -223,7 +220,9 @@ struct modConfData_s {
 	sbool bOmitLocalLogging;
 	sbool bWritePidSysSock;
 	sbool bUseSysTimeStamp;
+	sbool bDiscardOwnMsgs;
 	sbool configSetViaV2Method;
+	sbool bUnlink;
 };
 static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */
 static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current load process */
@@ -232,7 +231,9 @@ static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current lo
 static struct cnfparamdescr modpdescr[] = {
 	{ "syssock.use", eCmdHdlrBinary, 0 },
 	{ "syssock.name", eCmdHdlrGetWord, 0 },
+	{ "syssock.unlink", eCmdHdlrBinary, 0 },
 	{ "syssock.ignoretimestamp", eCmdHdlrBinary, 0 },
+	{ "syssock.ignoreownmessages", eCmdHdlrBinary, 0 },
 	{ "syssock.flowcontrol", eCmdHdlrBinary, 0 },
 	{ "syssock.usesystimestamp", eCmdHdlrBinary, 0 },
 	{ "syssock.annotate", eCmdHdlrBinary, 0 },
@@ -251,8 +252,10 @@ static struct cnfparamblk modpblk =
 /* input instance parameters */
 static struct cnfparamdescr inppdescr[] = {
 	{ "socket", eCmdHdlrString, CNFPARAM_REQUIRED }, /* legacy: addunixlistensocket */
+	{ "unlink", eCmdHdlrBinary, 0 },
 	{ "createpath", eCmdHdlrBinary, 0 },
 	{ "parsetrusted", eCmdHdlrBinary, 0 },
+	{ "ignoreownmessages", eCmdHdlrBinary, 0 },
 	{ "hostname", eCmdHdlrString, 0 },
 	{ "ignoretimestamp", eCmdHdlrBinary, 0 },
 	{ "flowcontrol", eCmdHdlrBinary, 0 },
@@ -272,74 +275,9 @@ static struct cnfparamblk inppblk =
 /* we do not use this, because we do not bind to a ruleset so far
  * enable when this is changed: #include "im-helper.h" */ /* must be included AFTER the type definitions! */
 
-static void 
-initRatelimitState(struct rs_ratelimit_state *rs, unsigned short interval, unsigned short burst)
-{
-	rs->interval = interval;
-	rs->burst = burst;
-	rs->done = 0;
-	rs->missed = 0;
-	rs->begin = 0;
-}
-
 static int bLegacyCnfModGlobalsPermitted;/* are legacy module-global config parameters permitted? */
 
 
-/* ratelimiting support, modelled after the linux kernel
- * returns 1 if message is within rate limit and shall be 
- * processed, 0 otherwise.
- * This implementation is NOT THREAD-SAFE and must not 
- * be called concurrently.
- */
-static inline int
-withinRatelimit(struct rs_ratelimit_state *rs, time_t tt, pid_t pid)
-{
-	int ret;
-	uchar msgbuf[1024];
-
-	if(rs->interval == 0) {
-		ret = 1;
-		goto finalize_it;
-	}
-
-	assert(rs->burst != 0);
-
-	if(rs->begin == 0)
-		rs->begin = tt;
-
-	/* resume if we go out of out time window */
-	if(tt > rs->begin + rs->interval) {
-		if(rs->missed) {
-			snprintf((char*)msgbuf, sizeof(msgbuf),
-			         "imuxsock lost %u messages from pid %lu due to rate-limiting",
-				 rs->missed, (unsigned long) pid);
-			logmsgInternal(RS_RET_RATE_LIMITED, LOG_SYSLOG|LOG_INFO, msgbuf, 0);
-			rs->missed = 0;
-		}
-		rs->begin = 0;
-		rs->done = 0;
-	}
-
-	/* do actual limit check */
-	if(rs->burst > rs->done) {
-		rs->done++;
-		ret = 1;
-	} else {
-		if(rs->missed == 0) {
-			snprintf((char*)msgbuf, sizeof(msgbuf),
-			         "imuxsock begins to drop messages from pid %lu due to rate-limiting",
-				 (unsigned long) pid);
-			logmsgInternal(RS_RET_RATE_LIMITED, LOG_SYSLOG|LOG_INFO, msgbuf, 0);
-		}
-		rs->missed++;
-		ret = 0;
-	}
-
-finalize_it:
-	return ret;
-}
-
-
 /* create input instance, set default paramters, and
  * add it to the list of instances.
  */
@@ -352,7 +290,7 @@ createInstance(instanceConf_t **pinst)
 	inst->sockName = NULL;
 	inst->pLogHostName = NULL;
 	inst->ratelimitInterval = DFLT_ratelimitInterval;
-	inst->ratelimitBurst = DFLT_ratelimitSeverity;
+	inst->ratelimitBurst = DFLT_ratelimitBurst;
 	inst->ratelimitSeverity = DFLT_ratelimitSeverity;
 	inst->bUseFlowCtl = 0;
 	inst->bIgnoreTimestamp = 1;
@@ -361,6 +299,8 @@ createInstance(instanceConf_t **pinst)
 	inst->bWritePid = 0;
 	inst->bAnnotate = 0;
 	inst->bParseTrusted = 0;
+	inst->bDiscardOwnMsgs = 1;
+	inst->bUnlink = 1;
 	inst->next = NULL;
 
 	/* node created, let's add to config */
@@ -446,7 +386,8 @@ addListner(instanceConf_t *inst)
 			CHKiRet(prop.ConstructFinalize(listeners[nfd].hostName));
 		}
 		if(inst->ratelimitInterval > 0) {
-			if((listeners[nfd].ht = create_hashtable(100, hash_from_key_fn, key_equals_fn, NULL)) == NULL) {
+			if((listeners[nfd].ht = create_hashtable(100, hash_from_key_fn, key_equals_fn,
+				(void(*)(void*))ratelimitDestruct)) == NULL) {
 				/* in this case, we simply turn off rate-limiting */
 				DBGPRINTF("imuxsock: turning off rate limiting because we could not "
 					  "create hash table\n");
@@ -460,11 +401,19 @@ addListner(instanceConf_t *inst)
 		listeners[nfd].flags = inst->bIgnoreTimestamp ? IGNDATE : NOFLAG;
 		listeners[nfd].bCreatePath = inst->bCreatePath;
 		listeners[nfd].sockName = ustrdup(inst->sockName);
-		listeners[nfd].bUseCreds = (inst->bWritePid || inst->ratelimitInterval || inst->bAnnotate) ? 1 : 0;
+		listeners[nfd].bUseCreds = (inst->bDiscardOwnMsgs || inst->bWritePid || inst->ratelimitInterval || inst->bAnnotate) ? 1 : 0;
 		listeners[nfd].bAnnotate = inst->bAnnotate;
 		listeners[nfd].bParseTrusted = inst->bParseTrusted;
+		listeners[nfd].bDiscardOwnMsgs = inst->bDiscardOwnMsgs;
+		listeners[nfd].bUnlink = inst->bUnlink;
 		listeners[nfd].bWritePid = inst->bWritePid;
 		listeners[nfd].bUseSysTimeStamp = inst->bUseSysTimeStamp;
+		CHKiRet(ratelimitNew(&listeners[nfd].dflt_ratelimiter, "imuxsock", NULL));
+		ratelimitSetLinuxLike(listeners[nfd].dflt_ratelimiter,
+				      listeners[nfd].ratelimitInterval,
+				      listeners[nfd].ratelimitBurst);
+		ratelimitSetSeverity(listeners[nfd].dflt_ratelimiter,
+				     listeners[nfd].ratelimitSev);
 		nfd++;
 	} else {
 		errmsg.LogError(0, NO_ERRCODE, "Out of unix socket name descriptors, ignoring %s\n",
@@ -476,7 +425,7 @@ finalize_it:
 }
 
 
-/* discard all log sockets except for "socket" 0. Data for it comes from
+/* discard/Destruct all log sockets except for "socket" 0. Data for it comes from
  * the constant memory pool - and if not, it is freeed via some other pointer.
  */
 static rsRetVal discardLogSockets(void)
@@ -494,6 +443,7 @@ static rsRetVal discardLogSockets(void)
 		if(listeners[i].ht != NULL) {
 			hashtable_destroy(listeners[i].ht, 1); /* 1 => free all values automatically */
 		}
+		ratelimitDestruct(listeners[i].dflt_ratelimiter);
 	}
 
 	return RS_RET_OK;
@@ -508,7 +458,8 @@ createLogSocket(lstn_t *pLstn)
 	struct sockaddr_un sunx;
 	DEFiRet;
 
-	unlink((char*)pLstn->sockName);
+	if(pLstn->bUnlink)
+		unlink((char*)pLstn->sockName);
 	memset(&sunx, 0, sizeof(sunx));
 	sunx.sun_family = AF_UNIX;
 	if(pLstn->bCreatePath) {
@@ -605,19 +556,26 @@ finalize_it:
  * listener (the latter being a performance enhancement).
  */
 static inline rsRetVal
-findRatelimiter(lstn_t *pLstn, struct ucred *cred, rs_ratelimit_state_t **prl)
+findRatelimiter(lstn_t *pLstn, struct ucred *cred, ratelimit_t **prl)
 {
-	rs_ratelimit_state_t *rl;
+	ratelimit_t *rl;
 	int r;
 	pid_t *keybuf;
+	char pidbuf[256];
 	DEFiRet;
 
 	if(cred == NULL)
 		FINALIZE;
+#if 0 // TODO: check deactivated?
 	if(pLstn->ratelimitInterval == 0) {
 		*prl = NULL;
 		FINALIZE;
 	}
+#endif
+	if(pLstn->ht == NULL) {
+		*prl = NULL;
+		FINALIZE;
+	}
 
 	rl = hashtable_search(pLstn->ht, &cred->pid);
 	if(rl == NULL) {
@@ -625,10 +583,14 @@ findRatelimiter(lstn_t *pLstn, struct ucred *cred, rs_ratelimit_state_t **prl)
 		DBGPRINTF("imuxsock: no ratelimiter for pid %lu, creating one\n",
 			  (unsigned long) cred->pid);
 		STATSCOUNTER_INC(ctrNumRatelimiters, mutCtrNumRatelimiters);
-		CHKmalloc(rl = malloc(sizeof(rs_ratelimit_state_t)));
+		snprintf(pidbuf, sizeof(pidbuf), "pid %lu",
+			(unsigned long) cred->pid);
+		pidbuf[sizeof(pidbuf)-1] = '\0'; /* to be on safe side */
+		CHKiRet(ratelimitNew(&rl, "imuxsock", pidbuf));
+		ratelimitSetLinuxLike(rl, pLstn->ratelimitInterval, pLstn->ratelimitBurst);
+		ratelimitSetSeverity(rl, pLstn->ratelimitSev);
 		CHKmalloc(keybuf = malloc(sizeof(pid_t)));
 		*keybuf = cred->pid;
-		initRatelimitState(rl, pLstn->ratelimitInterval, pLstn->ratelimitBurst);
 		r = hashtable_insert(pLstn->ht, keybuf, rl);
 		if(r == 0)
 			ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
@@ -637,6 +599,8 @@ findRatelimiter(lstn_t *pLstn, struct ucred *cred, rs_ratelimit_state_t **prl)
 	*prl = rl;
 
 finalize_it:
+	if(*prl == NULL)
+		*prl = pLstn->dflt_ratelimiter;
 	RETiRet;
 }
 
@@ -763,28 +727,6 @@ copyescaped(uchar *dstbuf, uchar *inbuf, int inlen)
 }
 
 
-#if 0
-/* Creates new field to be added to event
- * used for SystemLogParseTrusted parsing
- */
-struct ee_field *
-createNewField(char *fieldname, char *value, int lenValue) {
-	es_str_t *newStr;
-	struct ee_value *newVal;
-	struct ee_field *newField;
-
-	newStr = es_newStrFromBuf(value, (es_size_t) lenValue);
-
-	newVal = ee_newValue(ctxee);
-	ee_setStrValue(newVal, newStr);
-
-	newField = ee_newFieldFromNV(ctxee, fieldname, newVal);
-
-	return newField;
-}
-#endif
-
-
 /* submit received message to the queue engine
  * We now parse the message according to expected format so that we
  * can also mangle it if necessary.
@@ -803,8 +745,8 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct tim
 	uchar bufParseTAG[CONF_TAG_MAXSIZE];
 	struct syslogTime st;
 	time_t tt;
-	rs_ratelimit_state_t *ratelimiter = NULL;
 	int lenProp;
+	ratelimit_t *ratelimiter = NULL;
 	uchar propBuf[1024];
 	uchar msgbuf[8192];
 	uchar *pmsgbuf;
@@ -813,6 +755,11 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct tim
 	struct json_object *json = NULL, *jval;
 	DEFiRet;
 
+	if(pLstn->bDiscardOwnMsgs && cred != NULL && cred->pid == glblGetOurPid()) {
+		DBGPRINTF("imuxsock: discarding message from our own pid\n");
+		FINALIZE;
+	}
+
 	/* TODO: handle format errors?? */
 	/* we need to parse the pri first, because we need the severity for
 	 * rate-limiting as well.
@@ -831,10 +778,7 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct tim
 	facil = LOG_FAC(pri);
 	sever = LOG_PRI(pri);
 
-	if(sever >= pLstn->ratelimitSev) {
-		/* note: if cred == NULL, then ratelimiter == NULL as well! */
-		findRatelimiter(pLstn, cred, &ratelimiter); /* ignore error, better so than others... */
-	}
+	findRatelimiter(pLstn, cred, &ratelimiter); /* ignore error, better so than others... */
 
 	if(ts == NULL) {
 		datetime.getCurrTime(&st, &tt);
@@ -843,10 +787,12 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct tim
 		tt = ts->tv_sec;
 	}
 
+#if 0 // TODO: think about stats counters (or wait for request...?)
 	if(ratelimiter != NULL && !withinRatelimit(ratelimiter, tt, cred->pid)) {
 		STATSCOUNTER_INC(ctrLostRatelimit, mutCtrLostRatelimit);
 		FINALIZE;
 	}
+#endif
 
 	/* created trusted properties */
 	if(cred != NULL && pLstn->bAnnotate) {
@@ -981,8 +927,7 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct tim
 
 	MsgSetRcvFrom(pMsg, pLstn->hostName == NULL ? glbl.GetLocalHostNameProp() : pLstn->hostName);
 	CHKiRet(MsgSetRcvFromIP(pMsg, pLocalHostIP));
-	CHKiRet(submitMsg(pMsg));
-
+	ratelimitAddMsg(ratelimiter, NULL, pMsg);
 	STATSCOUNTER_INC(ctrSubmit, mutCtrSubmit);
 finalize_it:
 	RETiRet;
@@ -1117,13 +1062,20 @@ activateListeners()
 	listeners[0].ratelimitInterval = runModConf->ratelimitIntervalSysSock;
 	listeners[0].ratelimitBurst = runModConf->ratelimitBurstSysSock;
 	listeners[0].ratelimitSev = runModConf->ratelimitSeveritySysSock;
-	listeners[0].bUseCreds = (runModConf->bWritePidSysSock || runModConf->ratelimitIntervalSysSock || runModConf->bAnnotateSysSock) ? 1 : 0;
+	listeners[0].bUseCreds = (runModConf->bWritePidSysSock || runModConf->ratelimitIntervalSysSock || runModConf->bAnnotateSysSock || runModConf->bDiscardOwnMsgs) ? 1 : 0;
 	listeners[0].bWritePid = runModConf->bWritePidSysSock;
 	listeners[0].bAnnotate = runModConf->bAnnotateSysSock;
 	listeners[0].bParseTrusted = runModConf->bParseTrusted;
+	listeners[0].bDiscardOwnMsgs = runModConf->bDiscardOwnMsgs;
+	listeners[0].bUnlink = runModConf->bUnlink;
 	listeners[0].bUseSysTimeStamp = runModConf->bUseSysTimeStamp;
 	listeners[0].flags = runModConf->bIgnoreTimestamp ? IGNDATE : NOFLAG;
 	listeners[0].flowCtl = runModConf->bUseFlowCtl ? eFLOWCTL_LIGHT_DELAY : eFLOWCTL_NO_DELAY;
+	CHKiRet(ratelimitNew(&listeners[0].dflt_ratelimiter, "imuxsock", NULL));
+		ratelimitSetLinuxLike(listeners[0].dflt_ratelimiter,
+		listeners[0].ratelimitInterval,
+		listeners[0].ratelimitBurst);
+	ratelimitSetSeverity(listeners[0].dflt_ratelimiter,listeners[0].ratelimitSev);
 
 	sd_fds = sd_listen_fds(0);
 	if(sd_fds < 0) {
@@ -1165,6 +1117,8 @@ CODESTARTbeginCnfLoad
 	pModConf->bWritePidSysSock = 0;
 	pModConf->bAnnotateSysSock = 0;
 	pModConf->bParseTrusted = 0;
+	pModConf->bDiscardOwnMsgs = 1;
+	pModConf->bUnlink = 1;
 	pModConf->ratelimitIntervalSysSock = DFLT_ratelimitInterval;
 	pModConf->ratelimitBurstSysSock = DFLT_ratelimitBurst;
 	pModConf->ratelimitSeveritySysSock = DFLT_ratelimitSeverity;
@@ -1199,6 +1153,10 @@ CODESTARTsetModCnf
 			loadModConf->pLogSockName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
 		} else if(!strcmp(modpblk.descr[i].name, "syssock.ignoretimestamp")) {
 			loadModConf->bIgnoreTimestamp = (int) pvals[i].val.d.n;
+		} else if(!strcmp(modpblk.descr[i].name, "syssock.ignoreownmessages")) {
+			loadModConf->bDiscardOwnMsgs = (int) pvals[i].val.d.n;
+		} else if(!strcmp(modpblk.descr[i].name, "syssock.unlink")) {
+			loadModConf->bUnlink = (int) pvals[i].val.d.n;
 		} else if(!strcmp(modpblk.descr[i].name, "syssock.flowcontrol")) {
 			loadModConf->bUseFlowCtl = (int) pvals[i].val.d.n;
 		} else if(!strcmp(modpblk.descr[i].name, "syssock.usesystimestamp")) {
@@ -1261,6 +1219,10 @@ CODESTARTnewInpInst
 			inst->bCreatePath = (int) pvals[i].val.d.n;
 		} else if(!strcmp(modpblk.descr[i].name, "parsetrusted")) {
 			inst->bParseTrusted = (int) pvals[i].val.d.n;
+		} else if(!strcmp(modpblk.descr[i].name, "ignoreownmessages")) {
+			inst->bDiscardOwnMsgs = (int) pvals[i].val.d.n;
+		} else if(!strcmp(modpblk.descr[i].name, "unlink")) {
+			inst->bUnlink = (int) pvals[i].val.d.n;
 		} else if(!strcmp(modpblk.descr[i].name, "hostname")) {
 			inst->pLogHostName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
 		} else if(!strcmp(modpblk.descr[i].name, "ignoretimestamp")) {
@@ -1440,8 +1402,10 @@ CODESTARTafterRun
 			    listeners[i].fd <  SD_LISTEN_FDS_START + sd_fds)
 				continue;
 
-			DBGPRINTF("imuxsock: unlinking unix socket file[%d] %s\n", i, listeners[i].sockName);
-			unlink((char*) listeners[i].sockName);
+			if(listeners[i].bUnlink) {
+				DBGPRINTF("imuxsock: unlinking unix socket file[%d] %s\n", i, listeners[i].sockName);
+				unlink((char*) listeners[i].sockName);
+			}
 		}
 
 	discardLogSockets();
@@ -1553,8 +1517,17 @@ CODEmodInit_QueryRegCFSLineHdlr
 	listeners[0].bUseCreds = 0;
 	listeners[0].bAnnotate = 0;
 	listeners[0].bParseTrusted = 0;
+	listeners[0].bDiscardOwnMsgs = 1;
+	listeners[0].bUnlink = 1;
 	listeners[0].bCreatePath = 0;
 	listeners[0].bUseSysTimeStamp = 1;
+	if((listeners[0].ht = create_hashtable(100, hash_from_key_fn, key_equals_fn,
+		(void(*)(void*))ratelimitDestruct)) == NULL) {
+		/* in this case, we simply turn off rate-limiting */
+		DBGPRINTF("imuxsock: turning off rate limiting for system socket "
+			  "because we could not create hash table\n");
+		listeners[0].ratelimitInterval = 0;
+	}
 
 	/* initialize socket names */
 	for(i = 1 ; i < MAXFUNIX ; ++i) {