imtcp: support for Linux-Type ratelimiting added

5 files changed, 41 insertions, 6 deletions

diff --git a/ChangeLog b/ChangeLog
index aeefee6c..973495b4 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,5 +1,6 @@
 ---------------------------------------------------------------------------
 Version 7.3.1  [devel] 2012-10-??
+- imtcp: support for Linux-Type ratelimiting added
 - imptcp: support for Linux-Type ratelimiting added
 - imudp enhancements:
   * support for input batching added (performance improvement)
diff --git a/doc/imtcp.html b/doc/imtcp.html
index 01ea2802..4bda46ba 100644
--- a/doc/imtcp.html
+++ b/doc/imtcp.html
@@ -17,10 +17,6 @@
 Encryption is natively provided by selecting the approprioate network stream driver and
 can also be provided by using <a href="rsyslog_stunnel.html">stunnel</a>
 (an alternative is the use the <a href="imgssapi.html">imgssapi</a> module).</p>
-<p>Multiple receivers may be configured by specifying
-$InputTCPServerRun multiple times. This is available since version 4.3.1, earlier
-versions do NOT support it.
-</p>
 
 <p><b>Configuration Directives</b>:</p>
 <p><b>Global Directives</b>:</p>
@@ -100,6 +96,13 @@ activated. This is the default and should be left unchanged until you know
 very well what you do. It may be useful to turn it off, if you know this framing
 is not used and some senders emit multi-line messages into the message stream.
 </li>
+<li><b>RateLimit.Interval</b> [number] - (available since 7.3.1) specifies the rate-limiting
+interval in seconds.  Default value is 0, which turns off rate limiting. Set it to a number
+of seconds (5 recommended) to activate rate-limiting.
+</li>
+<li><b>RateLimit.Burst</b> [number] - (available since 7.3.1) specifies the rate-limiting
+burst in number of messages. Default is 10,000.
+</li>
 </ul>
 <b>Caveats/Known Bugs:</b>
 <ul>
diff --git a/plugins/imtcp/imtcp.c b/plugins/imtcp/imtcp.c
index beb7d705..8d71d5f2 100644
--- a/plugins/imtcp/imtcp.c
+++ b/plugins/imtcp/imtcp.c
@@ -105,6 +105,8 @@ struct instanceConf_s {
 	uchar *pszBindRuleset;		/* name of ruleset to bind to */
 	ruleset_t *pBindRuleset;	/* ruleset to bind listener to (use system default if unspecified) */
 	uchar *pszInputName;		/* value for inputname property, NULL is OK and handled by core engine */
+	int ratelimitInterval;
+	int ratelimitBurst;
 	int bSuppOctetFram;
 	struct instanceConf_s *next;
 };
@@ -155,7 +157,9 @@ static struct cnfparamdescr inppdescr[] = {
 	{ "port", eCmdHdlrString, CNFPARAM_REQUIRED }, /* legacy: InputTCPServerRun */
 	{ "name", eCmdHdlrString, 0 },
 	{ "ruleset", eCmdHdlrString, 0 },
-	{ "supportOctetCountedFraming", eCmdHdlrBinary, 0 }
+	{ "supportOctetCountedFraming", eCmdHdlrBinary, 0 },
+	{ "ratelimit.interval", eCmdHdlrInt, 0 },
+	{ "ratelimit.burst", eCmdHdlrInt, 0 }
 };
 static struct cnfparamblk inppblk =
 	{ CNFPARAMBLK_VERSION,
@@ -251,6 +255,8 @@ createInstance(instanceConf_t **pinst)
 	inst->pszBindRuleset = NULL;
 	inst->pszInputName = NULL;
 	inst->bSuppOctetFram = 1;
+	inst->ratelimitInterval = 0;
+	inst->ratelimitBurst = 10000;
 
 	/* node created, let's add to config */
 	if(loadModConf->tail == NULL) {
@@ -334,6 +340,7 @@ addListner(modConfData_t *modConf, instanceConf_t *inst)
 	CHKiRet(tcpsrv.SetRuleset(pOurTcpsrv, inst->pBindRuleset));
 	CHKiRet(tcpsrv.SetInputName(pOurTcpsrv, inst->pszInputName == NULL ?
 						UCHAR_CONSTANT("imtcp") : inst->pszInputName));
+	CHKiRet(tcpsrv.SetLinuxLikeRatelimiters(pOurTcpsrv, inst->ratelimitInterval, inst->ratelimitBurst));
 	tcpsrv.configureTCPListen(pOurTcpsrv, inst->pszBindPort, inst->bSuppOctetFram);
 
 finalize_it:
@@ -376,6 +383,10 @@ CODESTARTnewInpInst
 			inst->pszBindRuleset = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
 		} else if(!strcmp(inppblk.descr[i].name, "supportOctetCountedFraming")) {
 			inst->bSuppOctetFram = (int) pvals[i].val.d.n;
+		} else if(!strcmp(inppblk.descr[i].name, "ratelimit.burst")) {
+			inst->ratelimitBurst = (int) pvals[i].val.d.n;
+		} else if(!strcmp(inppblk.descr[i].name, "ratelimit.interval")) {
+			inst->ratelimitInterval = (int) pvals[i].val.d.n;
 		} else {
 			dbgprintf("imtcp: program error, non-handled "
 			  "param '%s'\n", inppblk.descr[i].name);
diff --git a/tcpsrv.c b/tcpsrv.c
index 89ad7325..7ba557e0 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -153,6 +153,8 @@ addNewLstnPort(tcpsrv_t *pThis, uchar *pszPort, int bSuppOctetFram)
 	statname[sizeof(statname)-1] = '\0'; /* just to be on the save side... */
 	CHKiRet(statsobj.SetName(pEntry->stats, statname));
 	CHKiRet(ratelimitNew(&pEntry->ratelimiter, "tcperver", NULL));
+	ratelimitSetLinuxLike(pEntry->ratelimiter, pThis->ratelimitInterval, pThis->ratelimitBurst);
+	ratelimitSetThreadSafe(pEntry->ratelimiter);
 	STATSCOUNTER_INIT(pEntry->ctrSubmit, pEntry->mutCtrSubmit);
 	CHKiRet(statsobj.AddCounter(pEntry->stats, UCHAR_CONSTANT("submitted"),
 		ctrType_IntCtr, &(pEntry->ctrSubmit)));
@@ -916,6 +918,8 @@ BEGINobjConstruct(tcpsrv) /* be sure to specify the object type also in END macr
 	pThis->addtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER;
 	pThis->bDisableLFDelim = 0;
 	pThis->OnMsgReceive = NULL;
+	pThis->ratelimitInterval = 0;
+	pThis->ratelimitBurst = 10000;
 	pThis->bUseFlowControl = 1;
 ENDobjConstruct(tcpsrv)
 
@@ -1123,6 +1127,17 @@ finalize_it:
 }
 
 
+/* Set the linux-like ratelimiter settings */
+static rsRetVal
+SetLinuxLikeRatelimiters(tcpsrv_t *pThis, int ratelimitInterval, int ratelimitBurst)
+{
+	DEFiRet;
+	pThis->ratelimitInterval = ratelimitInterval;
+	pThis->ratelimitBurst = ratelimitBurst;
+	RETiRet;
+}
+
+
 /* Set the ruleset (ptr) to use */
 static rsRetVal
 SetRuleset(tcpsrv_t *pThis, ruleset_t *pRuleset)
@@ -1273,6 +1288,7 @@ CODESTARTobjQueryInterface(tcpsrv)
 	pIf->SetCBOnErrClose = SetCBOnErrClose;
 	pIf->SetOnMsgReceive = SetOnMsgReceive;
 	pIf->SetRuleset = SetRuleset;
+	pIf->SetLinuxLikeRatelimiters = SetLinuxLikeRatelimiters;
 	pIf->SetNotificationOnRemoteClose = SetNotificationOnRemoteClose;
 
 finalize_it:
diff --git a/tcpsrv.h b/tcpsrv.h
index 4884b34d..93e472c9 100644
--- a/tcpsrv.h
+++ b/tcpsrv.h
@@ -71,6 +71,8 @@ struct tcpsrv_s {
 
 	int addtlFrameDelim;	/**< additional frame delimiter for plain TCP syslog framing (e.g. to handle NetScreen) */
 	int bDisableLFDelim;	/**< if 1, standard LF frame delimiter is disabled (*very dangerous*) */
+	int ratelimitInterval;
+	int ratelimitBurst;
 	tcps_sess_t **pSessions;/**< array of all of our sessions */
 	void *pUsr;		/**< a user-settable pointer (provides extensibility for "derived classes")*/
 	/* callbacks */
@@ -143,8 +145,10 @@ BEGINinterface(tcpsrv) /* name must also be changed in ENDinterface macro! */
 	rsRetVal (*SetUseFlowControl)(tcpsrv_t*, int);
 	/* added v11 -- rgerhards, 2011-05-09 */
 	rsRetVal (*SetKeepAlive)(tcpsrv_t*, int);
+	/* added v13 -- rgerhards, 2012-10-15 */
+	rsRetVal (*SetLinuxLikeRatelimiters)(tcpsrv_t *pThis, int interval, int burst);
 ENDinterface(tcpsrv)
-#define tcpsrvCURR_IF_VERSION 12 /* increment whenever you change the interface structure! */
+#define tcpsrvCURR_IF_VERSION 13 /* increment whenever you change the interface structure! */
 /* change for v4:
  * - SetAddtlFrameDelim() added -- rgerhards, 2008-12-10
  * - SetInputName() added -- rgerhards, 2008-12-10