summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--ChangeLog3
-rw-r--r--dirty.h9
-rw-r--r--doc/v7compatibility.html37
-rw-r--r--plugins/imfile/imfile.c22
-rw-r--r--plugins/imkmsg/imkmsg.c2
-rw-r--r--plugins/imptcp/imptcp.c25
-rw-r--r--plugins/imsolaris/imsolaris.c2
-rw-r--r--plugins/imtcp/imtcp.c7
-rw-r--r--plugins/imudp/imudp.c18
-rw-r--r--plugins/imuxsock/imuxsock.c17
-rw-r--r--runtime/Makefile.am2
-rw-r--r--runtime/ratelimit.c234
-rw-r--r--runtime/ratelimit.h40
-rw-r--r--runtime/rsconf.c4
-rw-r--r--runtime/rsyslog.h1
-rw-r--r--runtime/typedefs.h1
-rw-r--r--tcps_sess.c16
-rw-r--r--tcpsrv.c3
-rw-r--r--tcpsrv.h1
-rw-r--r--tools/syslogd.c33
20 files changed, 412 insertions, 65 deletions
diff --git a/ChangeLog b/ChangeLog
index bf1c37e8..307abceb 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,6 +1,7 @@
---------------------------------------------------------------------------
Version 7.3.1 [devel] 2012-10-??
-- change lumberjack cookie to "@cee" from "@cee "
+- imudp: support for input batching added (performance improvement)
+- change lumberjack cookie to "@cee:" from "@cee: "
CEE originally specified the cookie with SP, whereas other lumberjack
tools used it without space. In order to keep interop with lumberjack,
we now use the cookie without space as well. I hope this can be changed
diff --git a/dirty.h b/dirty.h
index a3940cb9..30b30bec 100644
--- a/dirty.h
+++ b/dirty.h
@@ -27,10 +27,13 @@
#ifndef DIRTY_H_INCLUDED
#define DIRTY_H_INCLUDED 1
-rsRetVal multiSubmitMsg(multi_submit_t *pMultiSub);
-rsRetVal submitMsg(msg_t *pMsg);
+rsRetVal __attribute__((deprecated)) multiSubmitMsg(multi_submit_t *pMultiSub);
+rsRetVal multiSubmitMsg2(multi_submit_t *pMultiSub); /* friends only! */
+rsRetVal submitMsg2(msg_t *pMsg);
+rsRetVal __attribute__((deprecated)) submitMsg(msg_t *pMsg);
+rsRetVal multiSubmitFlush(multi_submit_t *pMultiSub);
rsRetVal logmsgInternal(int iErr, int pri, uchar *msg, int flags);
-rsRetVal parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int flags, flowControl_t flowCtlTypeu, prop_t *pInputName, struct syslogTime *stTime, time_t ttGenTime, ruleset_t *pRuleset);
+rsRetVal __attribute__((deprecated)) parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int flags, flowControl_t flowCtlTypeu, prop_t *pInputName, struct syslogTime *stTime, time_t ttGenTime, ruleset_t *pRuleset);
rsRetVal diagGetMainMsgQSize(int *piSize); /* for imdiag */
rsRetVal createMainQueue(qqueue_t **ppQueue, uchar *pszQueueName);
diff --git a/doc/v7compatibility.html b/doc/v7compatibility.html
index 692a4fe1..932e2076 100644
--- a/doc/v7compatibility.html
+++ b/doc/v7compatibility.html
@@ -42,6 +42,43 @@ They tell that the construct is deprecated and which statement is to be used
as replacement. This does <b>not</b> affect operations: both modules are still
fully operational and will not be removed in the v7 timeframe.
+<h2>"last message repeated n times" Processing</h2>
+<p>This processing has been optimized and moved to the input side. This results
+in usually far better performance and also de-couples different sources
+from the same
+processing. It is now also integrated in to the more generic rate-limiting
+processing.
+<h3>User-Noticable Changes</h3>
+The code works almost as before, with two exceptions:
+<ul>
+<li>The supression amount can be different, as the new algorithm
+ precisely check's a single source, and while that source is being
+ read. The previous algorithm worked on a set of mixed messages
+ from multiple sources.
+<li>The previous algorithm wrote a "last message repeated n times" message
+ at least every 60 seconds. For performance reasons, we do no longer do
+ this but write this message only when a new message arrives or rsyslog
+ is shut down.
+</ul>
+<p>Note that the new algorithms needs support from input modules. If old
+modules which do not have the necessary support are used, duplicate
+messages will most probably not be detected. Upgrading the module code is
+simple, and all rsyslog-provided plugins support the new method, so this
+should not be a real problem (crafting a solution would result in rather
+complex code - for a case that most probably would never happen).
+<h3>Performance Implications</h3>
+<p>In general, the new method enables far faster output procesing. However, it
+needs to be noted that the "last message repeated n" processing needs parsed
+messages in order to detect duplicated. Consequently, if it is enabled the
+parser step cannot be deferred to the main queue processing thread and
+thus must be done during input processing. The changes workload distribution
+and may have (good or bad) effect on the overall performance. If you have
+a very high performance installation, it is suggested to check the performance
+profile before deploying the new version. Note: for high-performance
+environments it is highly recommended NOT to use "last message repeated n times"
+processing but rather the other (more efficient) rate-limiting methods. These
+also do NOT require the parsing step to be done during input processing.
+
<p><font size="2">This documentation is part of the
<a href="http://www.rsyslog.com/">rsyslog</a> project.<br>
Copyright &copy; 2011-2012 by <a href="http://www.gerhards.net/rainer">Rainer Gerhards</a> and
diff --git a/plugins/imfile/imfile.c b/plugins/imfile/imfile.c
index 453b6b05..99210b76 100644
--- a/plugins/imfile/imfile.c
+++ b/plugins/imfile/imfile.c
@@ -48,6 +48,7 @@
#include "prop.h"
#include "stringbuf.h"
#include "ruleset.h"
+#include "ratelimit.h"
MODULE_TYPE_INPUT /* must be present for input modules, do not remove */
MODULE_TYPE_NOKEEP
@@ -82,6 +83,7 @@ typedef struct fileInfo_s {
strm_t *pStrm; /* its stream (NULL if not assigned) */
int readMode; /* which mode to use in ReadMulteLine call? */
ruleset_t *pRuleset; /* ruleset to bind listener to (use system default if unspecified) */
+ ratelimit_t *ratelimiter;
multi_submit_t multiSub;
} fileInfo_t;
@@ -189,9 +191,7 @@ static rsRetVal enqLine(fileInfo_t *pInfo, cstr_t *cstrLine)
pMsg->iFacility = LOG_FAC(pInfo->iFacility);
pMsg->iSeverity = LOG_PRI(pInfo->iSeverity);
MsgSetRuleset(pMsg, pInfo->pRuleset);
- pInfo->multiSub.ppMsgs[pInfo->multiSub.nElem++] = pMsg;
- if(pInfo->multiSub.nElem == pInfo->multiSub.maxElem)
- CHKiRet(multiSubmitMsg(&pInfo->multiSub));
+ ratelimitAddMsg(pInfo->ratelimiter, &pInfo->multiSub, pMsg);
finalize_it:
RETiRet;
}
@@ -304,18 +304,7 @@ static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData)
}
finalize_it:
- if(pThis->multiSub.nElem > 0) {
- /* submit everything that was not yet submitted */
- CHKiRet(multiSubmitMsg(&pThis->multiSub));
- }
- ; /*EMPTY STATEMENT - needed to keep compiler happy - see below! */
- /* Note: the problem above is that pthread:cleanup_pop() is a macro which
- * evaluates to something like "} while(0);". So the code would become
- * "finalize_it: }", that is a label without a statement. The C standard does
- * not permit this. So we add an empty statement "finalize_it: ; }" and
- * everybody is happy. Note that without the ;, an error is reported only
- * on some platforms/compiler versions. -- rgerhards, 2008-08-15
- */
+ multiSubmitFlush(&pThis->multiSub);
pthread_cleanup_pop(0);
if(pCStr != NULL) {
@@ -423,6 +412,7 @@ addListner(instanceConf_t *inst)
pThis->lenTag = ustrlen(pThis->pszTag);
pThis->pszStateFile = (uchar*) strdup((char*) inst->pszStateFile);
+ CHKiRet(ratelimitNew(&pThis->ratelimiter));
CHKmalloc(pThis->multiSub.ppMsgs = MALLOC(inst->nMultiSub * sizeof(msg_t*)));
pThis->multiSub.maxElem = inst->nMultiSub;
pThis->multiSub.nElem = 0;
@@ -773,6 +763,8 @@ CODESTARTafterRun
persistStrmState(&files[i]);
strm.Destruct(&(files[i].pStrm));
}
+ ratelimitDestruct(files[i].ratelimiter);
+ free(files[i].multiSub.ppMsgs);
free(files[i].pszFileName);
free(files[i].pszTag);
free(files[i].pszStateFile);
diff --git a/plugins/imkmsg/imkmsg.c b/plugins/imkmsg/imkmsg.c
index 2a97f82d..d1a83879 100644
--- a/plugins/imkmsg/imkmsg.c
+++ b/plugins/imkmsg/imkmsg.c
@@ -113,7 +113,7 @@ enqMsg(uchar *msg, uchar* pszTag, int iFacility, int iSeverity, struct timeval *
pMsg->iFacility = iFacility;
pMsg->iSeverity = iSeverity;
pMsg->json = json;
- CHKiRet(submitMsg(pMsg));
+ CHKiRet(submitMsg(pMsg, NULL));
finalize_it:
RETiRet;
diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c
index 8150fc33..8495b293 100644
--- a/plugins/imptcp/imptcp.c
+++ b/plugins/imptcp/imptcp.c
@@ -67,6 +67,7 @@
#include "ruleset.h"
#include "msg.h"
#include "statsobj.h"
+#include "ratelimit.h"
#include "net.h" /* for permittedPeers, may be removed when this is removed */
/* the define is from tcpsrv.h, we need to find a new (but easier!!!) abstraction layer some time ... */
@@ -195,6 +196,7 @@ struct ptcpsrv_s {
sbool bKeepAlive; /* support keep-alive packets */
sbool bEmitMsgOnClose;
sbool bSuppOctetFram;
+ ratelimit_t *ratelimiter;
};
/* the ptcp session object. Describes a single active session.
@@ -295,6 +297,7 @@ destructSess(ptcpsess_t *pSess)
static void
destructSrv(ptcpsrv_t *pSrv)
{
+ ratelimitDestruct(pSrv->ratelimiter);
prop.Destruct(&pSrv->pInputName);
pthread_mutex_destroy(&pSrv->mutSessLst);
free(pSrv->pszInputName);
@@ -679,14 +682,7 @@ doSubmitMsg(ptcpsess_t *pThis, struct syslogTime *stTime, time_t ttGenTime, mult
MsgSetRuleset(pMsg, pSrv->pRuleset);
STATSCOUNTER_INC(pThis->pLstn->ctrSubmit, pThis->pLstn->mutCtrSubmit);
- if(pMultiSub == NULL) {
- CHKiRet(submitMsg(pMsg));
- } else {
- pMultiSub->ppMsgs[pMultiSub->nElem++] = pMsg;
- if(pMultiSub->nElem == pMultiSub->maxElem)
- CHKiRet(multiSubmitMsg(pMultiSub));
- }
-
+ ratelimitAddMsg(pSrv->ratelimiter, pMultiSub, pMsg);
finalize_it:
/* reset status variables */
@@ -805,12 +801,11 @@ processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttG
* we have just received a bunch of data! -- rgerhards, 2009-06-16
* EXTRACT from tcps_sess.c
*/
-#define NUM_MULTISUB 1024
static rsRetVal
DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen)
{
multi_submit_t multiSub;
- msg_t *pMsgs[NUM_MULTISUB];
+ msg_t *pMsgs[CONF_NUM_MULTISUB];
struct syslogTime stTime;
time_t ttGenTime;
char *pEnd;
@@ -821,7 +816,7 @@ DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen)
datetime.getCurrTime(&stTime, &ttGenTime);
multiSub.ppMsgs = pMsgs;
- multiSub.maxElem = NUM_MULTISUB;
+ multiSub.maxElem = CONF_NUM_MULTISUB;
multiSub.nElem = 0;
/* We now copy the message to the session buffer. */
@@ -831,15 +826,11 @@ DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen)
CHKiRet(processDataRcvd(pThis, *pData++, &stTime, ttGenTime, &multiSub));
}
- if(multiSub.nElem > 0) {
- /* submit anything that was not yet submitted */
- CHKiRet(multiSubmitMsg(&multiSub));
- }
+ iRet = multiSubmitFlush(&multiSub);
finalize_it:
RETiRet;
}
-#undef NUM_MULTISUB
/****************************************** --END-- TCP SUPPORT FUNCTIONS ***********************************/
@@ -1130,6 +1121,8 @@ addListner(modConfData_t __attribute__((unused)) *modConf, instanceConf_t *inst)
pSrv->iKeepAliveProbes = inst->iKeepAliveProbes;
pSrv->iKeepAliveTime = inst->iKeepAliveTime;
pSrv->bEmitMsgOnClose = inst->bEmitMsgOnClose;
+ CHKiRet(ratelimitNew(&pSrv->ratelimiter));
+ ratelimitSetThreadSafe(pSrv->ratelimiter);
CHKmalloc(pSrv->port = ustrdup(inst->pszBindPort));
pSrv->iAddtlFrameDelim = inst->iAddtlFrameDelim;
if(inst->pszBindAddr == NULL)
diff --git a/plugins/imsolaris/imsolaris.c b/plugins/imsolaris/imsolaris.c
index a220e72a..1e7d9b0f 100644
--- a/plugins/imsolaris/imsolaris.c
+++ b/plugins/imsolaris/imsolaris.c
@@ -212,7 +212,7 @@ readLog(int fd, uchar *pRcv, int iMaxLine)
pMsg->iFacility = LOG_FAC(hdr.pri);
pMsg->iSeverity = LOG_PRI(hdr.pri);
pMsg->msgFlags = NEEDS_PARSING | NO_PRI_IN_RAW;
- CHKiRet(submitMsg(pMsg));
+ CHKiRet(submitMsg(pMsg, NULL));
}
finalize_it:
diff --git a/plugins/imtcp/imtcp.c b/plugins/imtcp/imtcp.c
index eaf9a213..beb7d705 100644
--- a/plugins/imtcp/imtcp.c
+++ b/plugins/imtcp/imtcp.c
@@ -580,7 +580,9 @@ ENDwillRun
BEGINafterRun
CODESTARTafterRun
- /* do cleanup here */
+ if(pOurTcpsrv != NULL)
+ iRet = tcpsrv.Destruct(&pOurTcpsrv);
+
net.clearAllowedSenders(UCHAR_CONSTANT("TCP"));
ENDafterRun
@@ -594,9 +596,6 @@ ENDisCompatibleWithFeature
BEGINmodExit
CODESTARTmodExit
- if(pOurTcpsrv != NULL)
- iRet = tcpsrv.Destruct(&pOurTcpsrv);
-
if(pPermPeersRoot != NULL) {
net.DestructPermittedPeers(&pPermPeersRoot);
}
diff --git a/plugins/imudp/imudp.c b/plugins/imudp/imudp.c
index 0dda30ec..7f9e00dc 100644
--- a/plugins/imudp/imudp.c
+++ b/plugins/imudp/imudp.c
@@ -4,8 +4,6 @@
* NOTE: read comments in module-template.h to understand how this file
* works!
*
- * File begun on 2007-12-21 by RGerhards (extracted from syslogd.c)
- *
* Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of rsyslog.
@@ -53,6 +51,7 @@
#include "prop.h"
#include "ruleset.h"
#include "statsobj.h"
+#include "ratelimit.h"
#include "unicode-helper.h"
MODULE_TYPE_INPUT
@@ -77,6 +76,7 @@ static struct lstn_s {
int sock; /* socket */
ruleset_t *pRuleset; /* bound ruleset */
statsobj_t *stats; /* listener stats */
+ ratelimit_t *ratelimiter;
STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit)
} *lcnfRoot = NULL, *lcnfLast = NULL;
@@ -248,6 +248,7 @@ addListner(instanceConf_t *inst)
newlcnfinfo->next = NULL;
newlcnfinfo->sock = newSocks[iSrc];
newlcnfinfo->pRuleset = inst->pBindRuleset;
+ CHKiRet(ratelimitNew(&newlcnfinfo->ratelimiter));
/* support statistics gathering */
CHKiRet(statsobj.Construct(&(newlcnfinfo->stats)));
snprintf((char*)statname, sizeof(statname), "imudp(%s:%s)", bindName, port);
@@ -304,7 +305,6 @@ std_checkRuleset_genErrMsg(__attribute__((unused)) modConfData_t *modConf, insta
static inline rsRetVal
processSocket(thrdInfo_t *pThrd, struct lstn_s *lstn, struct sockaddr_storage *frominetPrev, int *pbIsPermitted)
{
- DEFiRet;
int iNbrTimeUsed;
time_t ttGenTime;
struct syslogTime stTime;
@@ -314,9 +314,15 @@ processSocket(thrdInfo_t *pThrd, struct lstn_s *lstn, struct sockaddr_storage *f
msg_t *pMsg;
prop_t *propFromHost = NULL;
prop_t *propFromHostIP = NULL;
+ multi_submit_t multiSub;
+ msg_t *pMsgs[CONF_NUM_MULTISUB];
char errStr[1024];
+ DEFiRet;
assert(pThrd != NULL);
+ multiSub.ppMsgs = pMsgs;
+ multiSub.maxElem = CONF_NUM_MULTISUB;
+ multiSub.nElem = 0;
iNbrTimeUsed = 0;
while(1) { /* loop is terminated if we have a bad receive, done below in the body */
if(pThrd->bShallStop == RSTRUE)
@@ -383,12 +389,15 @@ processSocket(thrdInfo_t *pThrd, struct lstn_s *lstn, struct sockaddr_storage *f
if(*pbIsPermitted == 2)
pMsg->msgFlags |= NEEDS_ACLCHK_U; /* request ACL check after resolution */
CHKiRet(msgSetFromSockinfo(pMsg, &frominet));
- CHKiRet(submitMsg(pMsg));
+ CHKiRet(ratelimitAddMsg(lstn->ratelimiter, &multiSub, pMsg));
STATSCOUNTER_INC(lstn->ctrSubmit, lstn->mutCtrSubmit);
}
}
+
finalize_it:
+ multiSubmitFlush(&multiSub);
+
if(propFromHost != NULL)
prop.Destruct(&propFromHost);
if(propFromHostIP != NULL)
@@ -884,6 +893,7 @@ CODESTARTafterRun
net.clearAllowedSenders((uchar*)"UDP");
for(lstn = lcnfRoot ; lstn != NULL ; ) {
statsobj.Destruct(&(lstn->stats));
+ ratelimitDestruct(lstn->ratelimiter);
close(lstn->sock);
lstnDel = lstn;
lstn = lstn->next;
diff --git a/plugins/imuxsock/imuxsock.c b/plugins/imuxsock/imuxsock.c
index a4933115..3421863d 100644
--- a/plugins/imuxsock/imuxsock.c
+++ b/plugins/imuxsock/imuxsock.c
@@ -55,6 +55,7 @@
#include "statsobj.h"
#include "datetime.h"
#include "hashtable.h"
+#include "ratelimit.h"
MODULE_TYPE_INPUT
MODULE_TYPE_NOKEEP
@@ -810,7 +811,17 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct tim
int toffs; /* offset for trusted properties */
struct syslogTime dummyTS;
struct json_object *json = NULL, *jval;
+ msg_t *repMsg;
DEFiRet;
+#warning experimental code needs to be made production-ready!
+/* we need to decide how many ratelimiters we use --> hashtable
+ also remove current homegrown ratelimiting functionality and
+ replace it with the new one.
+ */
+rsRetVal localRet;
+static ratelimit_t *ratelimit = NULL;
+if(ratelimit == NULL)
+ ratelimitNew(&ratelimit);
/* TODO: handle format errors?? */
/* we need to parse the pri first, because we need the severity for
@@ -982,7 +993,11 @@ 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));
+ localRet = ratelimitMsg(ratelimit, pMsg, &repMsg);
+ if(repMsg != NULL)
+ CHKiRet(submitMsg(repMsg));
+ if(localRet == RS_RET_OK)
+ CHKiRet(submitMsg(pMsg));
STATSCOUNTER_INC(ctrSubmit, mutCtrSubmit);
finalize_it:
diff --git a/runtime/Makefile.am b/runtime/Makefile.am
index 7af26d2b..7abbc258 100644
--- a/runtime/Makefile.am
+++ b/runtime/Makefile.am
@@ -65,6 +65,8 @@ librsyslog_la_SOURCES = \
ruleset.h \
prop.c \
prop.h \
+ ratelimit.c \
+ ratelimit.h \
cfsysline.c \
cfsysline.h \
sd-daemon.c \
diff --git a/runtime/ratelimit.c b/runtime/ratelimit.c
new file mode 100644
index 00000000..2df5f78a
--- /dev/null
+++ b/runtime/ratelimit.c
@@ -0,0 +1,234 @@
+/* ratelimit.c
+ * support for rate-limiting sources, including "last message
+ * repeated n times" processing.
+ *
+ * Copyright 2012 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * -or-
+ * see COPYING.ASL20 in the source distribution
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "config.h"
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+
+#include "rsyslog.h"
+#include "errmsg.h"
+#include "ratelimit.h"
+#include "datetime.h"
+#include "parser.h"
+#include "unicode-helper.h"
+#include "msg.h"
+#include "dirty.h"
+
+/* definitions for objects we access */
+DEFobjStaticHelpers
+DEFobjCurrIf(errmsg)
+DEFobjCurrIf(glbl)
+DEFobjCurrIf(datetime)
+DEFobjCurrIf(parser)
+
+/* static data */
+
+/* generate a "repeated n times" message */
+static inline msg_t *
+ratelimitGenRepMsg(ratelimit_t *ratelimit)
+{
+ msg_t *repMsg;
+ size_t lenRepMsg;
+ uchar szRepMsg[1024];
+
+ if(ratelimit->nsupp == 1) { /* we simply use the original message! */
+ repMsg = MsgAddRef(ratelimit->pMsg);
+ } else {/* we need to duplicate, original message may still be in use in other
+ * parts of the system! */
+ if((repMsg = MsgDup(ratelimit->pMsg)) == NULL) {
+ DBGPRINTF("Message duplication failed, dropping repeat message.\n");
+ goto done;
+ }
+ lenRepMsg = snprintf((char*)szRepMsg, sizeof(szRepMsg),
+ " message repeated %d times: [%.800s]",
+ ratelimit->nsupp, getMSG(ratelimit->pMsg));
+ MsgReplaceMSG(repMsg, szRepMsg, lenRepMsg);
+ }
+
+done: return repMsg;
+}
+
+/* ratelimit a message, that means:
+ * - handle "last message repeated n times" logic
+ * - handle actual (discarding) rate-limiting
+ * This function returns RS_RET_OK, if the caller shall process
+ * the message regularly and RS_RET_DISCARD if the caller must
+ * discard the message. The caller should also discard the message
+ * if another return status occurs. This places some burden on the
+ * caller logic, but provides best performance. Demanding this
+ * cooperative mode can enable a faulty caller to thrash up part
+ * of the system, but we accept that risk (a faulty caller can
+ * always do all sorts of evil, so...)
+ * If *ppRepMsg != NULL on return, the caller must enqueue that
+ * message before the original message.
+ */
+rsRetVal
+ratelimitMsg(ratelimit_t *ratelimit, msg_t *pMsg, msg_t **ppRepMsg)
+{
+ rsRetVal localRet;
+ int bNeedUnlockMutex = 0;
+ 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;
+ }
+
+ *ppRepMsg = NULL;
+ /* suppress duplicate messages */
+ if( ratelimit->pMsg != NULL &&
+ getMSGLen(pMsg) == getMSGLen(ratelimit->pMsg) &&
+ !ustrcmp(getMSG(pMsg), getMSG(ratelimit->pMsg)) &&
+ !strcmp(getHOSTNAME(pMsg), getHOSTNAME(ratelimit->pMsg)) &&
+ !strcmp(getPROCID(pMsg, LOCK_MUTEX), getPROCID(ratelimit->pMsg, LOCK_MUTEX)) &&
+ !strcmp(getAPPNAME(pMsg, LOCK_MUTEX), getAPPNAME(ratelimit->pMsg, LOCK_MUTEX))) {
+ ratelimit->nsupp++;
+ DBGPRINTF("msg repeated %d times\n", ratelimit->nsupp);
+ /* use current message, so we have the new timestamp
+ * (means we need to discard previous one) */
+ msgDestruct(&ratelimit->pMsg);
+ ratelimit->pMsg = pMsg;
+ ABORT_FINALIZE(RS_RET_DISCARDMSG);
+ } else {/* new message, do "repeat processing" & save it */
+ if(ratelimit->pMsg != NULL) {
+ if(ratelimit->nsupp > 0) {
+ *ppRepMsg = ratelimitGenRepMsg(ratelimit);
+ ratelimit->nsupp = 0;
+ }
+ msgDestruct(&ratelimit->pMsg);
+ }
+ ratelimit->pMsg = MsgAddRef(pMsg);
+ }
+
+finalize_it:
+ if(bNeedUnlockMutex)
+ pthread_mutex_unlock(&ratelimit->mut);
+ RETiRet;
+}
+
+
+/* add a message to a ratelimiter/multisubmit structure.
+ * ratelimiting is automatically handled according to the ratelimit
+ * settings.
+ * if pMultiSub == NULL, a single-message enqueue happens (under reconsideration)
+ */
+rsRetVal
+ratelimitAddMsg(ratelimit_t *ratelimit, multi_submit_t *pMultiSub, msg_t *pMsg)
+{
+ rsRetVal localRet;
+ msg_t *repMsg;
+ DEFiRet;
+
+ if(pMultiSub == NULL) {
+dbgprintf("DDDDD: multiSubmitAddMsg, not checking ratelimiter for single submit!\n");
+#warning missing multisub Implementation?
+ CHKiRet(submitMsg(pMsg));
+ } else {
+ localRet = ratelimitMsg(ratelimit, pMsg, &repMsg);
+ if(repMsg != NULL) {
+ pMultiSub->ppMsgs[pMultiSub->nElem++] = repMsg;
+ if(pMultiSub->nElem == pMultiSub->maxElem)
+ CHKiRet(multiSubmitMsg2(pMultiSub));
+ }
+ if(localRet == RS_RET_OK) {
+ pMultiSub->ppMsgs[pMultiSub->nElem++] = pMsg;
+ if(pMultiSub->nElem == pMultiSub->maxElem)
+ CHKiRet(multiSubmitMsg2(pMultiSub));
+ }
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+rsRetVal
+ratelimitNew(ratelimit_t **ppThis)
+{
+ ratelimit_t *pThis;
+ DEFiRet;
+
+ CHKmalloc(pThis = calloc(1, sizeof(ratelimit_t)));
+ *ppThis = pThis;
+finalize_it:
+ RETiRet;
+}
+
+/* enable thread-safe operations mode. This make sure that
+ * a single ratelimiter can be called from multiple threads. As
+ * this causes some overhead and is not always required, it needs
+ * to be explicitely enabled. This operation cannot be undone
+ * (think: why should one do that???)
+ */
+void
+ratelimitSetThreadSafe(ratelimit_t *ratelimit)
+{
+ ratelimit->bThreadSafe = 1;
+ pthread_mutex_init(&ratelimit->mut, NULL);
+}
+
+void
+ratelimitDestruct(ratelimit_t *ratelimit)
+{
+ msg_t *pMsg;
+ if(ratelimit->pMsg != NULL) {
+ if(ratelimit->nsupp > 0) {
+ pMsg = ratelimitGenRepMsg(ratelimit);
+ if(pMsg != NULL)
+ submitMsg2(pMsg);
+ }
+ msgDestruct(&ratelimit->pMsg);
+ }
+ if(ratelimit->bThreadSafe)
+ pthread_mutex_destroy(&ratelimit->mut);
+ free(ratelimit);
+}
+
+void
+ratelimitModExit(void)
+{
+ objRelease(datetime, CORE_COMPONENT);
+ objRelease(glbl, CORE_COMPONENT);
+ objRelease(errmsg, CORE_COMPONENT);
+ objRelease(parser, CORE_COMPONENT);
+}
+
+rsRetVal
+ratelimitModInit(void)
+{
+ DEFiRet;
+ CHKiRet(objGetObjInterface(&obj));
+ CHKiRet(objUse(glbl, CORE_COMPONENT));
+ CHKiRet(objUse(datetime, CORE_COMPONENT));
+ CHKiRet(objUse(errmsg, CORE_COMPONENT));
+ CHKiRet(objUse(parser, CORE_COMPONENT));
+finalize_it:
+ RETiRet;
+}
+
diff --git a/runtime/ratelimit.h b/runtime/ratelimit.h
new file mode 100644
index 00000000..37dad900
--- /dev/null
+++ b/runtime/ratelimit.h
@@ -0,0 +1,40 @@
+/* header for ratelimit.c
+ *
+ * Copyright 2012 Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * -or-
+ * see COPYING.ASL20 in the source distribution
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef INCLUDED_RATELIMIT_H
+#define INCLUDED_RATELIMIT_H
+
+struct ratelimit_s {
+ unsigned nsupp; /**< nbr of msgs suppressed */
+ msg_t *pMsg;
+ sbool bThreadSafe; /**< do we need to operate in Thread-Safe mode? */
+ pthread_mutex_t mut; /**< mutex if thread-safe operation desired */
+};
+
+/* prototypes */
+rsRetVal ratelimitNew(ratelimit_t **ppThis);
+void ratelimitSetThreadSafe(ratelimit_t *ratelimit);
+rsRetVal ratelimitMsg(ratelimit_t *ratelimit, msg_t *pMsg, msg_t **ppRep);
+rsRetVal ratelimitAddMsg(ratelimit_t *ratelimit, multi_submit_t *pMultiSub, msg_t *pMsg);
+void ratelimitDestruct(ratelimit_t *pThis);
+rsRetVal ratelimitModInit(void);
+void ratelimitModExit(void);
+
+#endif /* #ifndef INCLUDED_RATELIMIT_H */
diff --git a/runtime/rsconf.c b/runtime/rsconf.c
index ad588832..cf29c720 100644
--- a/runtime/rsconf.c
+++ b/runtime/rsconf.c
@@ -114,8 +114,8 @@ void cnfDoCfsysline(char *ln);
*/
BEGINobjConstruct(rsconf) /* be sure to specify the object type also in END macro! */
pThis->globals.bDebugPrintTemplateList = 1;
- pThis->globals.bDebugPrintModuleList = 1;
- pThis->globals.bDebugPrintCfSysLineHandlerList = 1;
+ pThis->globals.bDebugPrintModuleList = 0;
+ pThis->globals.bDebugPrintCfSysLineHandlerList = 0;
pThis->globals.bLogStatusMsgs = DFLT_bLogStatusMsgs;
pThis->globals.bErrMsgToStderr = 1;
pThis->globals.umask = -1;
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index 07d58d68..c02db53c 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -69,6 +69,7 @@
* approach taken here is considered appropriate.
* rgerhards, 2010-06-24
*/
+#define CONF_NUM_MULTISUB 1024 /* default number of messages per multisub structure */
/* ############################################################# *
* # End Config Settings # *
diff --git a/runtime/typedefs.h b/runtime/typedefs.h
index ccae08b2..1e0cb466 100644
--- a/runtime/typedefs.h
+++ b/runtime/typedefs.h
@@ -92,6 +92,7 @@ typedef struct cfgmodules_etry_s cfgmodules_etry_t;
typedef struct outchannels_s outchannels_t;
typedef struct modConfData_s modConfData_t;
typedef struct instanceConf_s instanceConf_t;
+typedef struct ratelimit_s ratelimit_t;
typedef int rs_size_t; /* we do never need more than 2Gig strings, signed permits to
* use -1 as a special flag. */
typedef rsRetVal (*prsf_t)(struct vmstk_s*, int); /* pointer to a RainerScript function */
diff --git a/tcps_sess.c b/tcps_sess.c
index e7149cb7..16fd94f5 100644
--- a/tcps_sess.c
+++ b/tcps_sess.c
@@ -47,6 +47,7 @@
#include "msg.h"
#include "datetime.h"
#include "prop.h"
+#include "ratelimit.h"
#include "debug.h"
@@ -264,14 +265,7 @@ defaultDoSubmitMessage(tcps_sess_t *pThis, struct syslogTime *stTime, time_t ttG
MsgSetRuleset(pMsg, pThis->pLstnInfo->pRuleset);
STATSCOUNTER_INC(pThis->pLstnInfo->ctrSubmit, pThis->pLstnInfo->mutCtrSubmit);
- if(pMultiSub == NULL) {
- CHKiRet(submitMsg(pMsg));
- } else {
- pMultiSub->ppMsgs[pMultiSub->nElem++] = pMsg;
- if(pMultiSub->nElem == pMultiSub->maxElem)
- CHKiRet(multiSubmitMsg(pMultiSub));
- }
-
+ ratelimitAddMsg(pThis->pLstnInfo->ratelimiter, pMultiSub, pMsg);
finalize_it:
/* reset status variables */
@@ -487,11 +481,7 @@ DataRcvd(tcps_sess_t *pThis, char *pData, size_t iLen)
while(pData < pEnd) {
CHKiRet(processDataRcvd(pThis, *pData++, &stTime, ttGenTime, &multiSub));
}
-
- if(multiSub.nElem > 0) {
- /* submit anything that was not yet submitted */
- CHKiRet(multiSubmitMsg(&multiSub));
- }
+ iRet = multiSubmitFlush(&multiSub);
finalize_it:
RETiRet;
diff --git a/tcpsrv.c b/tcpsrv.c
index bf12f1fa..6b134017 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -72,6 +72,7 @@
#include "nspoll.h"
#include "errmsg.h"
#include "ruleset.h"
+#include "ratelimit.h"
#include "unicode-helper.h"
@@ -151,6 +152,7 @@ addNewLstnPort(tcpsrv_t *pThis, uchar *pszPort, int bSuppOctetFram)
snprintf((char*)statname, sizeof(statname), "%s(%s)", pThis->pszInputName, pszPort);
statname[sizeof(statname)-1] = '\0'; /* just to be on the save side... */
CHKiRet(statsobj.SetName(pEntry->stats, statname));
+ CHKiRet(ratelimitNew(&pEntry->ratelimiter));
STATSCOUNTER_INIT(pEntry->ctrSubmit, pEntry->mutCtrSubmit);
CHKiRet(statsobj.AddCounter(pEntry->stats, UCHAR_CONSTANT("submitted"),
ctrType_IntCtr, &(pEntry->ctrSubmit)));
@@ -295,6 +297,7 @@ static void deinit_tcp_listener(tcpsrv_t *pThis)
while(pEntry != NULL) {
free(pEntry->pszPort);
prop.Destruct(&pEntry->pInputName);
+ ratelimitDestruct(pEntry->ratelimiter);
pDel = pEntry;
pEntry = pEntry->pNext;
free(pDel);
diff --git a/tcpsrv.h b/tcpsrv.h
index d66f682c..4884b34d 100644
--- a/tcpsrv.h
+++ b/tcpsrv.h
@@ -42,6 +42,7 @@ struct tcpLstnPortList_s {
ruleset_t *pRuleset; /**< associated ruleset */
statsobj_t *stats; /**< associated stats object */
sbool bSuppOctetFram; /**< do we support octect-counted framing? (if no->legay only!)*/
+ ratelimit_t *ratelimiter;
STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit)
tcpLstnPortList_t *pNext; /**< next port or NULL */
};
diff --git a/tools/syslogd.c b/tools/syslogd.c
index 8ba8edd3..a56cea94 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -125,6 +125,7 @@
#include "dnscache.h"
#include "sd-daemon.h"
#include "rainerscript.h"
+#include "ratelimit.h"
/* definitions for objects we access */
DEFobjCurrIf(obj)
@@ -417,7 +418,7 @@ parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int fla
CHKiRet(prop.Destruct(&pProp));
CHKiRet(MsgSetRcvFromIPStr(pMsg, hnameIP, ustrlen(hnameIP), &pProp));
CHKiRet(prop.Destruct(&pProp));
- CHKiRet(submitMsg(pMsg));
+ CHKiRet(submitMsg2(pMsg));
finalize_it:
RETiRet;
@@ -488,7 +489,7 @@ logmsgInternal(int iErr, int pri, uchar *msg, int flags)
/* we have the queue, so we can simply provide the
* message to the queue engine.
*/
- submitMsg(pMsg);
+ submitMsg2(pMsg);
}
finalize_it:
RETiRet;
@@ -624,7 +625,7 @@ int i;
* rgerhards, 2008-02-13
*/
rsRetVal
-submitMsg(msg_t *pMsg)
+submitMsg2(msg_t *pMsg)
{
qqueue_t *pQueue;
ruleset_t *pRuleset;
@@ -649,13 +650,19 @@ finalize_it:
RETiRet;
}
+rsRetVal
+submitMsg(msg_t *pMsg)
+{
+ return submitMsg2(pMsg);
+}
+
/* submit multiple messages at once, very similar to submitMsg, just
* for multi_submit_t. All messages need to go into the SAME queue!
* rgerhards, 2009-06-16
*/
rsRetVal
-multiSubmitMsg(multi_submit_t *pMultiSub)
+multiSubmitMsg2(multi_submit_t *pMultiSub)
{
int i;
qqueue_t *pQueue;
@@ -686,8 +693,24 @@ multiSubmitMsg(multi_submit_t *pMultiSub)
finalize_it:
RETiRet;
}
+rsRetVal
+multiSubmitMsg(multi_submit_t *pMultiSub) /* backward compat. level */
+{
+ return multiSubmitMsg2(pMultiSub);
+}
+/* flush multiSubmit, e.g. at end of read records */
+rsRetVal
+multiSubmitFlush(multi_submit_t *pMultiSub)
+{
+ DEFiRet;
+dbgprintf("DDDD: multiSubmitFlish, nElem %d\n", pMultiSub->nElem);
+ if(pMultiSub->nElem > 0) {
+ iRet = multiSubmitMsg2(pMultiSub);
+ }
+ RETiRet;
+}
static void
@@ -1473,6 +1496,7 @@ InitGlobalClasses(void)
CHKiRet(objUse(net, LM_NET_FILENAME));
dnscacheInit();
initRainerscript();
+ ratelimitModInit();
finalize_it:
if(iRet != RS_RET_OK) {
@@ -1511,6 +1535,7 @@ GlobalClassExit(void)
/* TODO: implement the rest of the deinit */
/* dummy "classes */
strExit();
+ ratelimitModExit();
#if 0
CHKiRet(objGetObjInterface(&obj)); /* this provides the root pointer for all other queries */
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-16 03:40:37 +0000