summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--ChangeLog6
-rw-r--r--action.c74
-rw-r--r--action.h4
-rw-r--r--configure.ac2
-rw-r--r--doc/imklog.html5
-rw-r--r--doc/impstats.html4
-rw-r--r--doc/imuxsock.html7
-rw-r--r--plugins/imklog/imklog.c27
-rw-r--r--plugins/imptcp/imptcp.c58
-rw-r--r--plugins/imudp/imudp.c173
-rw-r--r--plugins/imuxsock/imuxsock.c35
-rw-r--r--runtime/msg.c4
-rw-r--r--runtime/net.c54
-rw-r--r--runtime/net.h6
-rw-r--r--runtime/queue.c12
-rw-r--r--runtime/queue.h3
-rw-r--r--runtime/ruleset.c5
-rw-r--r--tcps_sess.c1
-rw-r--r--tcpsrv.c15
-rw-r--r--tcpsrv.h3
-rw-r--r--tools/syslogd.c11
21 files changed, 386 insertions, 123 deletions
diff --git a/ChangeLog b/ChangeLog
index ef9d3735..48ef8884 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,9 @@
+- added capability to use a local interface IP address as fromhost-ip for
+ imuxsock imklog
+ new config directives: $IMUXSockLocalIPIF, $klogLocalIPIF
+- added configuration directives to customize queue light delay marks
+ $MainMsgQueueLightDelayMark, $ActionQueueLightDelayMark; both
+ specify number of messages starting at which a delay happens.
---------------------------------------------------------------------------
Version 5.8.9 [V5-stable] 2012-03-??
- bugfix: stopped DA queue was never processed after a restart due to a
diff --git a/action.c b/action.c
index 278625ce..7de10d49 100644
--- a/action.c
+++ b/action.c
@@ -112,6 +112,7 @@
#include "datetime.h"
#include "unicode-helper.h"
#include "atomic.h"
+#include "statsobj.h"
#define NO_TIME_PROVIDED 0 /* indicate we do not provide any cached time */
@@ -127,6 +128,7 @@ DEFobjCurrIf(obj)
DEFobjCurrIf(datetime)
DEFobjCurrIf(module)
DEFobjCurrIf(errmsg)
+DEFobjCurrIf(statsobj)
static int iActExecOnceInterval = 0; /* execute action once every nn seconds */
static int iActExecEveryNthOccur = 0; /* execute action every n-th occurence (0,1=always) */
@@ -142,6 +144,7 @@ static queueType_t ActionQueType = QUEUETYPE_DIRECT; /* type of the main messag
static int iActionQueueSize = 1000; /* size of the main message queue above */
static int iActionQueueDeqBatchSize = 16; /* batch size for action queues */
static int iActionQHighWtrMark = 800; /* high water mark for disk-assisted queues */
+static int iActionQLightDlyMrk = -1; /* light delay mark for disk-assisted queues */
static int iActionQLowWtrMark = 200; /* low water mark for disk-assisted queues */
static int iActionQDiscardMark = 9800; /* begin to discard messages */
static int iActionQDiscardSeverity = 8; /* by default, discard nothing to prevent unintentional loss */
@@ -224,6 +227,7 @@ actionResetQueueParams(void)
iActionQueueSize = 1000; /* size of the main message queue above */
iActionQueueDeqBatchSize = 16; /* default batch size */
iActionQHighWtrMark = 800; /* high water mark for disk-assisted queues */
+ iActionQLightDlyMrk = -1;
iActionQLowWtrMark = 200; /* low water mark for disk-assisted queues */
iActionQDiscardMark = 9800; /* begin to discard messages */
iActionQDiscardSeverity = 8; /* discard warning and above */
@@ -263,6 +267,12 @@ rsRetVal actionDestruct(action_t *pThis)
qqueueDestruct(&pThis->pQueue);
}
+ /* destroy stats object, if we have one (may not always be
+ * be the case, e.g. if turned off)
+ */
+ if(pThis->statsobj != NULL)
+ statsobj.Destruct(&pThis->statsobj);
+
if(pThis->pMod != NULL)
pThis->pMod->freeInstance(pThis->pModData);
@@ -313,12 +323,40 @@ rsRetVal
actionConstructFinalize(action_t *pThis)
{
DEFiRet;
- uchar pszQName[64]; /* friendly name of our queue */
+ uchar pszAName[64]; /* friendly name of our queue */
ASSERT(pThis != NULL);
- /* find a name for our queue */
- snprintf((char*) pszQName, sizeof(pszQName)/sizeof(uchar), "action %d queue", iActionNbr);
+ /* generate a friendly name for us action stats */
+ if(pThis->pszName == NULL) {
+ snprintf((char*) pszAName, sizeof(pszAName)/sizeof(uchar), "action %d", iActionNbr);
+ } else {
+ ustrncpy(pszAName, pThis->pszName, sizeof(pszAName));
+ pszAName[63] = '\0'; /* to be on the save side */
+ }
+
+ /* support statistics gathering */
+ CHKiRet(statsobj.Construct(&pThis->statsobj));
+ CHKiRet(statsobj.SetName(pThis->statsobj, pszAName));
+
+ STATSCOUNTER_INIT(pThis->ctrProcessed, pThis->mutCtrProcessed);
+ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("processed"),
+ ctrType_IntCtr, &pThis->ctrProcessed));
+
+ STATSCOUNTER_INIT(pThis->ctrFail, pThis->mutCtrFail);
+ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("failed"),
+ ctrType_IntCtr, &pThis->ctrFail));
+
+ CHKiRet(statsobj.ConstructFinalize(pThis->statsobj));
+
+ /* create our queue */
+ /* find a (friendly) name for our queue */
+ if(pThis->pszName == NULL) {
+ snprintf((char*) pszAName, sizeof(pszAName)/sizeof(uchar), "action %d queue", iActionNbr);
+ } else {
+ snprintf((char*) pszAName, sizeof(pszAName)/sizeof(uchar), "%s queue", pThis->pszName);
+ }
+ pszAName[63] = '\0'; /* to be on the save side */
/* now check if we can run the action in "firehose mode" during stage one of
* its processing (that is before messages are enqueued into the action q).
@@ -362,7 +400,7 @@ actionConstructFinalize(action_t *pThis)
*/
CHKiRet(qqueueConstruct(&pThis->pQueue, ActionQueType, 1, iActionQueueSize,
(rsRetVal (*)(void*, batch_t*, int*))processBatchMain));
- obj.SetName((obj_t*) pThis->pQueue, pszQName);
+ obj.SetName((obj_t*) pThis->pQueue, pszAName);
/* ... set some properties ... */
# define setQPROP(func, directive, data) \
@@ -386,6 +424,9 @@ actionConstructFinalize(action_t *pThis)
setQPROP(qqueueSettoWrkShutdown, "$ActionQueueWorkerTimeoutThreadShutdown", iActionQtoWrkShutdown);
setQPROP(qqueueSettoEnq, "$ActionQueueTimeoutEnqueue", iActionQtoEnq);
setQPROP(qqueueSetiHighWtrMrk, "$ActionQueueHighWaterMark", iActionQHighWtrMark);
+ if(iActionQLightDlyMrk > 0) {
+ setQPROP(qqueueSetiLightDlyMrk, "$ActionQueueLightDelayMark", iActionQLightDlyMrk);
+ }
setQPROP(qqueueSetiLowWtrMrk, "$ActionQueueLowWaterMark", iActionQLowWtrMark);
setQPROP(qqueueSetiDiscardMrk, "$ActionQueueDiscardMark", iActionQDiscardMark);
setQPROP(qqueueSetiDiscardSeverity, "$ActionQueueDiscardSeverity", iActionQDiscardSeverity);
@@ -1072,6 +1113,7 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem)
&& pBatch->pElem[i].state != BATCH_STATE_COMM ) {
pBatch->pElem[i].state = BATCH_STATE_BAD;
pBatch->pElem[i].bPrevWasSuspended = 1;
+ STATSCOUNTER_INC(pAction->ctrFail, pAction->mutCtrFail);
}
}
bDone = 1;
@@ -1261,6 +1303,7 @@ doSubmitToActionQ(action_t *pAction, msg_t *pMsg)
{
DEFiRet;
+ STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed);
if(pAction->pQueue->qType == QUEUETYPE_DIRECT)
iRet = qqueueEnqObjDirect(pAction->pQueue, (void*) MsgAddRef(pMsg));
else
@@ -1537,6 +1580,18 @@ finalize_it:
RETiRet;
}
+static inline void
+countStatsBatchEnq(action_t *pAction, batch_t *pBatch)
+{
+ int i;
+ for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
+ if( pBatch->pElem[i].bFilterOK
+ && pBatch->pElem[i].state != BATCH_STATE_DISC) {
+ STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed);
+ }
+ }
+}
+
/* enqueue a batch in direct mode. We have put this into its own function just to avoid
* cluttering the actual submit function.
@@ -1573,16 +1628,19 @@ doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch)
pBatch->pElem[i].bFilterOK = 0;
bModifiedFilter = 1;
}
- if(pBatch->pElem[i].bFilterOK)
+ if(pBatch->pElem[i].bFilterOK && pBatch->pElem[i].state != BATCH_STATE_DISC) {
+ STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed);
bNeedSubmit = 1;
+ }
DBGPRINTF("action %p[%d]: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n",
pAction, i, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state,
pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
}
if(bNeedSubmit) {
+ /* note: stats were already computed above */
iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch);
} else {
- DBGPRINTF("no need to submit batch, all bFilterOK==0\n");
+ DBGPRINTF("no need to submit batch, all bFilterOK==0 or discarded\n");
}
if(bModifiedFilter) {
for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
@@ -1594,6 +1652,8 @@ doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch)
}
}
} else {
+ if(GatherStats)
+ countStatsBatchEnq(pAction, pBatch);
iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch);
}
@@ -1817,6 +1877,7 @@ rsRetVal actionClassInit(void)
CHKiRet(objUse(datetime, CORE_COMPONENT));
CHKiRet(objUse(module, CORE_COMPONENT));
CHKiRet(objUse(errmsg, CORE_COMPONENT));
+ CHKiRet(objUse(statsobj, CORE_COMPONENT));
CHKiRet(regCfSysLineHdlr((uchar *)"actionname", 0, eCmdHdlrGetWord, NULL, &pszActionName, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuefilename", 0, eCmdHdlrGetWord, NULL, &pszActionQFName, NULL));
@@ -1827,6 +1888,7 @@ rsRetVal actionClassInit(void)
CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuehighwatermark", 0, eCmdHdlrInt, NULL, &iActionQHighWtrMark, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuelowwatermark", 0, eCmdHdlrInt, NULL, &iActionQLowWtrMark, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardmark", 0, eCmdHdlrInt, NULL, &iActionQDiscardMark, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuelightdelaymark", 0, eCmdHdlrInt, NULL, &iActionQLightDlyMrk, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardseverity", 0, eCmdHdlrInt, NULL, &iActionQDiscardSeverity, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuecheckpointinterval", 0, eCmdHdlrInt, NULL, &iActionQPersistUpdCnt, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesyncqueuefiles", 0, eCmdHdlrBinary, NULL, &bActionQSyncQeueFiles, NULL));
diff --git a/action.h b/action.h
index 0c08d3d6..add55a5e 100644
--- a/action.h
+++ b/action.h
@@ -89,6 +89,10 @@ struct action_s {
pthread_mutex_t mutActExec; /* mutex to guard actual execution of doAction for single-threaded modules */
uchar *pszName; /* action name (for documentation) */
DEF_ATOMIC_HELPER_MUT(mutCAS);
+ /* for statistics subsystem */
+ statsobj_t *statsobj;
+ STATSCOUNTER_DEF(ctrProcessed, mutCtrProcessed);
+ STATSCOUNTER_DEF(ctrFail, mutCtrFail);
};
diff --git a/configure.ac b/configure.ac
index 3cb623d4..b38a1955 100644
--- a/configure.ac
+++ b/configure.ac
@@ -2,7 +2,7 @@
# Process this file with autoconf to produce a configure script.
AC_PREREQ(2.61)
-AC_INIT([rsyslog],[5.8.8],[rsyslog@lists.adiscon.com])
+AC_INIT([rsyslog],[5.8.8-newstats10],[rsyslog@lists.adiscon.com])
AM_INIT_AUTOMAKE
m4_ifdef([AM_SILENT_RULES], [AM_SILENT_RULES([yes])])
diff --git a/doc/imklog.html b/doc/imklog.html
index f273753f..2e3b3bc2 100644
--- a/doc/imklog.html
+++ b/doc/imklog.html
@@ -36,6 +36,11 @@ processing.<span style="font-weight: bold;"></span></li>
<li><span style="font-weight: bold;"></span>$DebugPrintKernelSymbols
[on/<b>off</b>]<br>
Linux only, ignored on other platforms (but may be specified)</li>
+<li><b>$klogLocalIPIF</b> [interface name] - (available since 5.9.6) - if provided, the IP of the specified
+interface (e.g. "eth0") shall be used as fromhost-ip for imklog-originating messages.
+If this directive is not given OR the interface cannot be found (or has no IP address),
+the default of "127.0.0.1" is used.
+</li>
<li>$klogSymbolLookup [on/<b>off</b>] --
disables imklog kernel symbol translation (former klogd -x option). NOTE that
this option is counter-productive on recent kernels (>= 2.6) because the
diff --git a/doc/impstats.html b/doc/impstats.html
index cede4874..260c1aa4 100644
--- a/doc/impstats.html
+++ b/doc/impstats.html
@@ -18,7 +18,9 @@ prepared to change your trending scripts when you upgrade to a newer rsyslog ver
output is periodic, with the interval being configurable (default is 5 minutes).
Be sure that your configuration records the counter messages (default is syslog.info).
<p>Note that loading this module has impact on rsyslog performance. Depending on
-settings, this impact may be severe (for high-load environments).
+settings, this impact may be noticable (for high-load environments).
+<p>The rsyslog website has an updated overview of available
+<a href="http://rsyslog.com/rsyslog-statistic-counter/">rsyslog statistic counters</a>.
</p>
<p><b>Configuration Directives</b>:</p>
<ul>
diff --git a/doc/imuxsock.html b/doc/imuxsock.html
index ee5db22d..f54b5564 100644
--- a/doc/imuxsock.html
+++ b/doc/imuxsock.html
@@ -64,6 +64,11 @@ burst in number of messages. Default is 200.
<li><b>$IMUXSockRateLimitSeverity</b> [numerical severity] - specifies the severity of
messages that shall be rate-limited.
</li>
+<li><b>$IMUXSockLocalIPIF</b> [interface name] - (available since 5.9.6) - if provided, the IP of the specified
+interface (e.g. "eth0") shall be used as fromhost-ip for imuxsock-originating messages.
+If this directive is not given OR the interface cannot be found (or has no IP address),
+the default of "127.0.0.1" is used.
+</li>
<li><b>$InputUnixListenSocketUsePIDFromSystem</b> [on/<b>off</b>] - specifies if the pid being logged shall
be obtained from the log socket itself. If so, the TAG part of the message is rewritten.
It is recommended to turn this option on, but the default is "off" to keep compatible
@@ -150,7 +155,7 @@ $SystemLogRateLimitInterval 0 # turn off rate limiting
<p><font size="2">This documentation is part of the
<a href="http://www.rsyslog.com/">rsyslog</a>
project.<br>
-Copyright &copy; 2008-2010 by <a href="http://www.gerhards.net/rainer">Rainer
+Copyright &copy; 2008-2012 by <a href="http://www.gerhards.net/rainer">Rainer
Gerhards</a> and
<a href="http://www.adiscon.com/">Adiscon</a>.
Released under the GNU GPL version 3 or higher.</font></p>
diff --git a/plugins/imklog/imklog.c b/plugins/imklog/imklog.c
index 16adbc21..66d7d143 100644
--- a/plugins/imklog/imklog.c
+++ b/plugins/imklog/imklog.c
@@ -44,6 +44,7 @@
#include <stdarg.h>
#include <ctype.h>
#include <stdlib.h>
+#include <sys/socket.h>
#include "dirty.h"
#include "cfsysline.h"
@@ -52,6 +53,7 @@
#include "module-template.h"
#include "datetime.h"
#include "imklog.h"
+#include "net.h"
#include "glbl.h"
#include "prop.h"
#include "unicode-helper.h"
@@ -64,6 +66,7 @@ DEF_IMOD_STATIC_DATA
DEFobjCurrIf(datetime)
DEFobjCurrIf(glbl)
DEFobjCurrIf(prop)
+DEFobjCurrIf(net)
/* configuration settings */
int dbgPrintSymbols = 0; /* this one is extern so the helpers can access it! */
@@ -71,6 +74,7 @@ int symbols_twice = 0;
int use_syscall = 0;
int symbol_lookup = 0; /* on recent kernels > 2.6, the kernel does this */
int bPermitNonKernel = 0; /* permit logging of messages not having LOG_KERN facility */
+static uchar *pLocalIPIF = NULL;
int iFacilIntMsg; /* the facility to use for internal messages (set by driver) */
uchar *pszPath = NULL;
int console_log_level = -1;
@@ -233,10 +237,22 @@ ENDrunInput
BEGINwillRun
+ uchar myIP[128];
+ rsRetVal localRet;
CODESTARTwillRun
/* we need to create the inputName property (only once during our lifetime) */
CHKiRet(prop.CreateStringProp(&pInputName, UCHAR_CONSTANT("imklog"), sizeof("imklog") - 1));
- CHKiRet(prop.CreateStringProp(&pLocalHostIP, UCHAR_CONSTANT("127.0.0.1"), sizeof("127.0.0.1") - 1));
+ if(pLocalIPIF == NULL) {
+ strcpy((char*)myIP, "127.0.0.1");
+ } else {
+ localRet = net.GetIFIPAddr(pLocalIPIF, AF_UNSPEC, myIP, (int) sizeof(myIP));
+ if(localRet != RS_RET_OK) {
+ DBGPRINTF("imuxsock: could not obtain my IP, using 127.0.0.1 instead\n");
+ strcpy((char*)myIP, "127.0.0.1");
+ }
+ }
+ DBGPRINTF("imklog: using '%s' as localhost IP\n", myIP);
+ CHKiRet(prop.CreateStringProp(&pLocalHostIP, myIP, ustrlen(myIP)));
iRet = klogWillRun();
finalize_it:
@@ -251,6 +267,7 @@ CODESTARTafterRun
prop.Destruct(&pInputName);
if(pLocalHostIP != NULL)
prop.Destruct(&pLocalHostIP);
+ free(pLocalIPIF);
ENDafterRun
@@ -258,6 +275,7 @@ BEGINmodExit
CODESTARTmodExit
/* release objects we used */
objRelease(glbl, CORE_COMPONENT);
+ objRelease(net, CORE_COMPONENT);
objRelease(datetime, CORE_COMPONENT);
objRelease(prop, CORE_COMPONENT);
if(pszPath != NULL)
@@ -282,6 +300,10 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
free(pszPath);
pszPath = NULL;
}
+ if(pLocalIPIF != NULL) {
+ free(pLocalIPIF);
+ pLocalIPIF = NULL;
+ }
iFacilIntMsg = klogFacilIntMsg();
return RS_RET_OK;
}
@@ -293,6 +315,7 @@ CODEmodInit_QueryRegCFSLineHdlr
CHKiRet(objUse(datetime, CORE_COMPONENT));
CHKiRet(objUse(glbl, CORE_COMPONENT));
CHKiRet(objUse(prop, CORE_COMPONENT));
+ CHKiRet(objUse(net, CORE_COMPONENT));
iFacilIntMsg = klogFacilIntMsg();
@@ -304,6 +327,8 @@ CODEmodInit_QueryRegCFSLineHdlr
CHKiRet(omsdRegCFSLineHdlr((uchar *)"klogpermitnonkernelfacility", 0, eCmdHdlrBinary, NULL, &bPermitNonKernel, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"klogconsoleloglevel", 0, eCmdHdlrInt, NULL, &console_log_level, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"kloginternalmsgfacility", 0, eCmdHdlrFacility, NULL, &iFacilIntMsg, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr((uchar *)"kloglocalipif", 0, eCmdHdlrGetWord,
+ NULL, &pLocalIPIF, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID));
ENDmodInit
/* vim:set ai:
diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c
index 65fe703c..e191c64f 100644
--- a/plugins/imptcp/imptcp.c
+++ b/plugins/imptcp/imptcp.c
@@ -65,6 +65,7 @@
#include "datetime.h"
#include "ruleset.h"
#include "msg.h"
+#include "statsobj.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 ... */
@@ -82,7 +83,7 @@ DEFobjCurrIf(prop)
DEFobjCurrIf(datetime)
DEFobjCurrIf(errmsg)
DEFobjCurrIf(ruleset)
-
+DEFobjCurrIf(statsobj)
/* config settings */
@@ -123,7 +124,8 @@ struct ptcpsrv_s {
* includes support for doubly-linked list.
*/
struct ptcpsess_s {
- ptcpsrv_t *pSrv; /* our server */
+// ptcpsrv_t *pSrv; /* our server TODO: check remove! */
+ ptcplstn_t *pLstn; /* our listener */
ptcpsess_t *prev, *next;
int sock;
epolld_t *epd;
@@ -151,6 +153,8 @@ struct ptcplstn_s {
ptcplstn_t *prev, *next;
int sock;
epolld_t *epd;
+ statsobj_t *stats; /* listener stats */
+ STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit)
};
@@ -188,7 +192,7 @@ static char rcvBuf[128*1024];
/* forward definitions */
static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal);
-static rsRetVal addLstn(ptcpsrv_t *pSrv, int sock);
+static rsRetVal addLstn(ptcpsrv_t *pSrv, int sock, int isIPv6);
/* some simple constructors/destructors */
@@ -233,6 +237,7 @@ startupSrv(ptcpsrv_t *pSrv)
int sockflags;
struct addrinfo hints, *res = NULL, *r;
uchar *lstnIP;
+ int isIPv6 = 0;
lstnIP = pSrv->lstnIP == NULL ? UCHAR_CONSTANT("") : pSrv->lstnIP;
@@ -265,8 +270,9 @@ startupSrv(ptcpsrv_t *pSrv)
continue;
}
-#ifdef IPV6_V6ONLY
if(r->ai_family == AF_INET6) {
+ isIPv6 = 1;
+#ifdef IPV6_V6ONLY
int iOn = 1;
if(setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY,
(char *)&iOn, sizeof (iOn)) < 0) {
@@ -274,8 +280,10 @@ startupSrv(ptcpsrv_t *pSrv)
sock = -1;
continue;
}
- }
#endif
+ } else {
+ isIPv6 = 0;
+ }
if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)) < 0 ) {
DBGPRINTF("error %d setting tcp socket option\n", errno);
close(sock);
@@ -337,7 +345,7 @@ startupSrv(ptcpsrv_t *pSrv)
/* if we reach this point, we were able to obtain a valid socket, so we can
* create our listener object. -- rgerhards, 2010-08-10
*/
- CHKiRet(addLstn(pSrv, sock));
+ CHKiRet(addLstn(pSrv, sock, isIPv6));
++numSocks;
}
@@ -492,22 +500,25 @@ static rsRetVal
doSubmitMsg(ptcpsess_t *pThis, struct syslogTime *stTime, time_t ttGenTime, multi_submit_t *pMultiSub)
{
msg_t *pMsg;
+ ptcpsrv_t *pSrv;
DEFiRet;
if(pThis->iMsg == 0) {
DBGPRINTF("discarding zero-sized message\n");
FINALIZE;
}
+ pSrv = pThis->pLstn->pSrv;
/* we now create our own message object and submit it to the queue */
CHKiRet(msgConstructWithTime(&pMsg, stTime, ttGenTime));
MsgSetRawMsg(pMsg, (char*)pThis->pMsg, pThis->iMsg);
- MsgSetInputName(pMsg, pThis->pSrv->pInputName);
+ MsgSetInputName(pMsg, pSrv->pInputName);
MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME;
MsgSetRcvFrom(pMsg, pThis->peerName);
CHKiRet(MsgSetRcvFromIP(pMsg, pThis->peerIP));
- MsgSetRuleset(pMsg, pThis->pSrv->pRuleset);
+ MsgSetRuleset(pMsg, pSrv->pRuleset);
+ STATSCOUNTER_INC(pThis->pLstn->ctrSubmit, pThis->pLstn->mutCtrSubmit);
if(pMultiSub == NULL) {
CHKiRet(submitMsg(pMsg));
@@ -589,7 +600,8 @@ processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttG
}
if(( (c == '\n')
- || ((pThis->pSrv->iAddtlFrameDelim != TCPSRV_NO_ADDTL_DELIMITER) && (c == pThis->pSrv->iAddtlFrameDelim))
+ || ((pThis->pLstn->pSrv->iAddtlFrameDelim != TCPSRV_NO_ADDTL_DELIMITER)
+ && (c == pThis->pLstn->pSrv->iAddtlFrameDelim))
) && pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delimiter? */
doSubmitMsg(pThis, stTime, ttGenTime, pMultiSub);
pThis->inputState = eAtStrtFram;
@@ -747,14 +759,25 @@ finalize_it:
/* add a listener to the server
*/
static rsRetVal
-addLstn(ptcpsrv_t *pSrv, int sock)
+addLstn(ptcpsrv_t *pSrv, int sock, int isIPv6)
{
DEFiRet;
ptcplstn_t *pLstn;
+ uchar statname[64];
CHKmalloc(pLstn = malloc(sizeof(ptcplstn_t)));
pLstn->pSrv = pSrv;
pLstn->sock = sock;
+ /* support statistics gathering */
+ CHKiRet(statsobj.Construct(&(pLstn->stats)));
+ snprintf((char*)statname, sizeof(statname), "imptcp(%s/%s/%s)",
+ (pSrv->lstnIP == NULL) ? "*" : (char*)pSrv->lstnIP, pSrv->port,
+ isIPv6 ? "IPv6" : "IPv4");
+ statname[sizeof(statname)-1] = '\0'; /* just to be on the save side... */
+ CHKiRet(statsobj.SetName(pLstn->stats, statname));
+ CHKiRet(statsobj.AddCounter(pLstn->stats, UCHAR_CONSTANT("submitted"),
+ ctrType_IntCtr, &(pLstn->ctrSubmit)));
+ CHKiRet(statsobj.ConstructFinalize(pLstn->stats));
/* add to start of server's listener list */
pLstn->prev = NULL;
@@ -773,14 +796,15 @@ finalize_it:
/* add a session to the server
*/
static rsRetVal
-addSess(ptcpsrv_t *pSrv, int sock, prop_t *peerName, prop_t *peerIP)
+addSess(ptcplstn_t *pLstn, int sock, prop_t *peerName, prop_t *peerIP)
{
DEFiRet;
ptcpsess_t *pSess = NULL;
+ ptcpsrv_t *pSrv = pLstn->pSrv;
CHKmalloc(pSess = malloc(sizeof(ptcpsess_t)));
CHKmalloc(pSess->pMsg = malloc(iMaxLine * sizeof(uchar)));
- pSess->pSrv = pSrv;
+ pSess->pLstn = pLstn;
pSess->sock = sock;
pSess->inputState = eAtStrtFram;
pSess->iMsg = 0;
@@ -824,7 +848,7 @@ closeSess(ptcpsess_t *pSess)
pSess->next->prev = pSess->prev;
if(pSess->prev == NULL) {
/* need to update root! */
- pSess->pSrv->pSess = pSess->next;
+ pSess->pLstn->pSrv->pSess = pSess->next;
} else {
pSess->prev->next = pSess->next;
}
@@ -949,7 +973,7 @@ lstnActivity(ptcplstn_t *pLstn)
if(localRet == RS_RET_NO_MORE_DATA)
break;
CHKiRet(localRet);
- CHKiRet(addSess(pLstn->pSrv, newSock, peerName, peerIP));
+ CHKiRet(addSess(pLstn, newSock, peerName, peerIP));
}
finalize_it:
@@ -979,7 +1003,7 @@ sessActivity(ptcpsess_t *pSess)
CHKiRet(DataRcvd(pSess, rcvBuf, lenRcv));
} else if (lenRcv == 0) {
/* session was closed, do clean-up */
- if(pSess->pSrv->bEmitMsgOnClose) {
+ if(pSess->pLstn->pSrv->bEmitMsgOnClose) {
uchar *peerName;
int lenPeer;
prop.GetString(pSess->peerName, &peerName, &lenPeer);
@@ -1085,6 +1109,8 @@ shutdownSrv(ptcpsrv_t *pSrv)
pLstn = pSrv->pLstn;
while(pLstn != NULL) {
close(pLstn->sock);
+ statsobj.Destruct(&(pLstn->stats));
+ /* now unlink listner */
lstnDel = pLstn;
pLstn = pLstn->next;
DBGPRINTF("imptcp shutdown listen socket %d\n", lstnDel->sock);
@@ -1132,6 +1158,7 @@ CODESTARTmodExit
/* release objects we used */
objRelease(glbl, CORE_COMPONENT);
+ objRelease(statsobj, CORE_COMPONENT);
objRelease(prop, CORE_COMPONENT);
objRelease(net, LM_NET_FILENAME);
objRelease(datetime, CORE_COMPONENT);
@@ -1167,6 +1194,7 @@ CODEmodInit_QueryRegCFSLineHdlr
initConfigSettings();
/* request objects we use */
CHKiRet(objUse(glbl, CORE_COMPONENT));
+ CHKiRet(objUse(statsobj, CORE_COMPONENT));
CHKiRet(objUse(prop, CORE_COMPONENT));
CHKiRet(objUse(net, LM_NET_FILENAME));
CHKiRet(objUse(errmsg, CORE_COMPONENT));
diff --git a/plugins/imudp/imudp.c b/plugins/imudp/imudp.c
index a5002591..112738b4 100644
--- a/plugins/imudp/imudp.c
+++ b/plugins/imudp/imudp.c
@@ -6,7 +6,7 @@
*
* File begun on 2007-12-21 by RGerhards (extracted from syslogd.c)
*
- * Copyright 2007-2009 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2011 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -26,6 +26,7 @@
* A copy of the GPL can be found in the file "COPYING" in this distribution.
*/
#include "config.h"
+#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
@@ -51,6 +52,7 @@
#include "datetime.h"
#include "prop.h"
#include "ruleset.h"
+#include "statsobj.h"
#include "unicode-helper.h"
MODULE_TYPE_INPUT
@@ -66,6 +68,16 @@ DEFobjCurrIf(net)
DEFobjCurrIf(datetime)
DEFobjCurrIf(prop)
DEFobjCurrIf(ruleset)
+DEFobjCurrIf(statsobj)
+
+
+static struct lstn_s {
+ struct lstn_s *next;
+ int sock; /* socket */
+ ruleset_t *pRuleset; /* bound ruleset */
+ statsobj_t *stats; /* listener stats */
+ STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit)
+} *lcnfRoot = NULL, *lcnfLast = NULL;
static int bDoACLCheck; /* are ACL checks neeed? Cached once immediately before listener startup */
static int iMaxLine; /* maximum UDP message size supported */
@@ -73,9 +85,7 @@ static time_t ttLastDiscard = 0; /* timestamp when a message from a non-permitte
* This shall prevent remote DoS when the "discard on disallowed sender"
* message is configured to be logged on occurance of such a case.
*/
-static int *udpLstnSocks = NULL; /* Internet datagram sockets, first element is nbr of elements
- * read-only after init(), but beware of restart! */
-static ruleset_t **udpRulesets = NULL; /* ruleset to be used with sockets in question (entry 0 is empty) */
+
static uchar *pszBindAddr = NULL; /* IP to bind socket to */
static uchar *pRcvBuf = NULL; /* receive buffer (for a single packet). We use a global and alloc
* it so that we can check available memory in willRun() and request
@@ -190,9 +200,11 @@ static rsRetVal addListner(void __attribute__((unused)) *pVal, uchar *pNewVal)
DEFiRet;
uchar *bindAddr;
int *newSocks;
- int *tmpSocks;
- int iSrc, iDst;
- ruleset_t **tmpRulesets;
+ int iSrc;
+ struct lstn_s *newlcnfinfo;
+ uchar *bindName;
+ uchar *port;
+ uchar statname[64];
/* check which address to bind to. We could do this more compact, but have not
* done so in order to make the code more readable. -- rgerhards, 2007-12-27
@@ -203,55 +215,45 @@ static rsRetVal addListner(void __attribute__((unused)) *pVal, uchar *pNewVal)
bindAddr = NULL;
else
bindAddr = pszBindAddr;
+ bindName = (bindAddr == NULL) ? (uchar*)"*" : bindAddr;
- DBGPRINTF("Trying to open syslog UDP ports at %s:%s.\n",
- (bindAddr == NULL) ? (uchar*)"*" : bindAddr, pNewVal);
+ DBGPRINTF("Trying to open syslog UDP ports at %s:%s.\n", bindName, pNewVal);
- newSocks = net.create_udp_socket(bindAddr, (pNewVal == NULL || *pNewVal == '\0') ? (uchar*) "514" : pNewVal, 1);
+ port = (pNewVal == NULL || *pNewVal == '\0') ? (uchar*) "514" : pNewVal;
+ newSocks = net.create_udp_socket(bindAddr, port, 1);
if(newSocks != NULL) {
/* we now need to add the new sockets to the existing set */
- if(udpLstnSocks == NULL) {
- /* esay, we can just replace it */
- udpLstnSocks = newSocks;
- CHKmalloc(udpRulesets = (ruleset_t**) MALLOC(sizeof(ruleset_t*) * (newSocks[0] + 1)));
- for(iDst = 1 ; iDst <= newSocks[0] ; ++iDst)
- udpRulesets[iDst] = pBindRuleset;
- } else {
- /* we need to add them */
- tmpSocks = (int*) MALLOC(sizeof(int) * (1 + newSocks[0] + udpLstnSocks[0]));
- tmpRulesets = (ruleset_t**) MALLOC(sizeof(ruleset_t*) * (1 + newSocks[0] + udpLstnSocks[0]));
- if(tmpSocks == NULL || tmpRulesets == NULL) {
- DBGPRINTF("out of memory trying to allocate udp listen socket array\n");
- /* in this case, we discard the new sockets but continue with what we
- * already have
- */
- free(newSocks);
- free(tmpSocks);
- free(tmpRulesets);
- ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
- } else {
- /* ready to copy */
- iDst = 1;
- for(iSrc = 1 ; iSrc <= udpLstnSocks[0] ; ++iSrc, ++iDst) {
- tmpSocks[iDst] = udpLstnSocks[iSrc];
- tmpRulesets[iDst] = udpRulesets[iSrc];
- }
- for(iSrc = 1 ; iSrc <= newSocks[0] ; ++iSrc, ++iDst) {
- tmpSocks[iDst] = newSocks[iSrc];
- tmpRulesets[iDst] = pBindRuleset;
- }
- tmpSocks[0] = udpLstnSocks[0] + newSocks[0];
- free(newSocks);
- free(udpLstnSocks);
- udpLstnSocks = tmpSocks;
- free(udpRulesets);
- udpRulesets = tmpRulesets;
+ /* ready to copy */
+ for(iSrc = 1 ; iSrc <= newSocks[0] ; ++iSrc) {
+ CHKmalloc(newlcnfinfo = (struct lstn_s*) MALLOC(sizeof(struct lstn_s)));
+ newlcnfinfo->next = NULL;
+ newlcnfinfo->sock = newSocks[iSrc];
+ newlcnfinfo->pRuleset = pBindRuleset;
+ /* support statistics gathering */
+ CHKiRet(statsobj.Construct(&(newlcnfinfo->stats)));
+ snprintf((char*)statname, sizeof(statname), "imudp(%s:%s)", bindName, port);
+ statname[sizeof(statname)-1] = '\0'; /* just to be on the save side... */
+ CHKiRet(statsobj.SetName(newlcnfinfo->stats, statname));
+ CHKiRet(statsobj.AddCounter(newlcnfinfo->stats, UCHAR_CONSTANT("submitted"),
+ ctrType_IntCtr, &(newlcnfinfo->ctrSubmit)));
+ CHKiRet(statsobj.ConstructFinalize(newlcnfinfo->stats));
+ /* link to list. Order must be preserved to take care for
+ * conflicting matches.
+ */
+ if(lcnfRoot == NULL)
+ lcnfRoot = newlcnfinfo;
+ if(lcnfLast == NULL)
+ lcnfLast = newlcnfinfo;
+ else {
+ lcnfLast->next = newlcnfinfo;
+ lcnfLast = newlcnfinfo;
}
}
}
finalize_it:
free(pNewVal); /* in any case, this is no longer needed */
+ free(newSocks);
RETiRet;
}
@@ -294,8 +296,7 @@ finalize_it:
* on scheduling order. -- rgerhards, 2008-10-02
*/
static inline rsRetVal
-processSocket(thrdInfo_t *pThrd, int fd, struct sockaddr_storage *frominetPrev, int *pbIsPermitted,
- ruleset_t *pRuleset)
+processSocket(thrdInfo_t *pThrd, struct lstn_s *lstn, struct sockaddr_storage *frominetPrev, int *pbIsPermitted)
{
DEFiRet;
int iNbrTimeUsed;
@@ -315,7 +316,7 @@ processSocket(thrdInfo_t *pThrd, int fd, struct sockaddr_storage *frominetPrev,
if(pThrd->bShallStop == TRUE)
ABORT_FINALIZE(RS_RET_FORCE_TERM);
socklen = sizeof(struct sockaddr_storage);
- lenRcvBuf = recvfrom(fd, (char*) pRcvBuf, iMaxLine, 0, (struct sockaddr *)&frominet, &socklen);
+ lenRcvBuf = recvfrom(lstn->sock, (char*) pRcvBuf, iMaxLine, 0, (struct sockaddr *)&frominet, &socklen);
if(lenRcvBuf < 0) {
if(errno != EINTR && errno != EAGAIN) {
rs_strerror_r(errno, errStr, sizeof(errStr));
@@ -360,7 +361,7 @@ processSocket(thrdInfo_t *pThrd, int fd, struct sockaddr_storage *frominetPrev,
*pbIsPermitted = 1; /* no check -> everything permitted */
}
- DBGPRINTF("recv(%d,%d),acl:%d,msg:%s\n", fd, (int) lenRcvBuf, *pbIsPermitted, pRcvBuf);
+ DBGPRINTF("recv(%d,%d),acl:%d,msg:%s\n", lstn->sock, (int) lenRcvBuf, *pbIsPermitted, pRcvBuf);
if(*pbIsPermitted != 0) {
if((iTimeRequery == 0) || (iNbrTimeUsed++ % iTimeRequery) == 0) {
@@ -370,13 +371,14 @@ processSocket(thrdInfo_t *pThrd, int fd, struct sockaddr_storage *frominetPrev,
CHKiRet(msgConstructWithTime(&pMsg, &stTime, ttGenTime));
MsgSetRawMsg(pMsg, (char*)pRcvBuf, lenRcvBuf);
MsgSetInputName(pMsg, pInputName);
- MsgSetRuleset(pMsg, pRuleset);
+ MsgSetRuleset(pMsg, lstn->pRuleset);
MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY);
pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME | NEEDS_DNSRESOL;
if(*pbIsPermitted == 2)
pMsg->msgFlags |= NEEDS_ACLCHK_U; /* request ACL check after resolution */
CHKiRet(msgSetFromSockinfo(pMsg, &frominet));
CHKiRet(submitMsg(pMsg));
+ STATSCOUNTER_INC(lstn->ctrSubmit, lstn->mutCtrSubmit);
}
}
@@ -443,6 +445,8 @@ rsRetVal rcvMainLoop(thrdInfo_t *pThrd)
struct epoll_event *udpEPollEvt = NULL;
struct epoll_event currEvt[NUM_EPOLL_EVENTS];
char errStr[1024];
+ struct lstn_s *lstn;
+ int nLstn;
/* start "name caching" algo by making sure the previous system indicator
* is invalidated.
@@ -451,7 +455,11 @@ rsRetVal rcvMainLoop(thrdInfo_t *pThrd)
bIsPermitted = 0;
memset(&frominetPrev, 0, sizeof(frominetPrev));
- CHKmalloc(udpEPollEvt = calloc(udpLstnSocks[0], sizeof(struct epoll_event)));
+ /* count num listeners -- do it here in order to avoid inconsistency */
+ nLstn = 0;
+ for(lstn = lcnfRoot ; lstn != NULL ; lstn = lstn->next)
+ ++nLstn;
+ CHKmalloc(udpEPollEvt = calloc(nLstn, sizeof(struct epoll_event)));
#if defined(EPOLL_CLOEXEC) && defined(HAVE_EPOLL_CREATE1)
DBGPRINTF("imudp uses epoll_create1()\n");
@@ -471,16 +479,18 @@ rsRetVal rcvMainLoop(thrdInfo_t *pThrd)
/* fill the epoll set - we need to do this only once, as the set
* can not change dyamically.
*/
- for (i = 0; i < *udpLstnSocks; i++) {
- if (udpLstnSocks[i+1] != -1) {
+ i = 0;
+ for(lstn = lcnfRoot ; lstn != NULL ; lstn = lstn->next) {
+ if(lstn->sock != -1) {
udpEPollEvt[i].events = EPOLLIN | EPOLLET;
- udpEPollEvt[i].data.u64 = i+1;
- if(epoll_ctl(efd, EPOLL_CTL_ADD, udpLstnSocks[i+1], &(udpEPollEvt[i])) < 0) {
+ udpEPollEvt[i].data.u64 = (long long unsigned) lstn;
+ if(epoll_ctl(efd, EPOLL_CTL_ADD, lstn->sock, &(udpEPollEvt[i])) < 0) {
rs_strerror_r(errno, errStr, sizeof(errStr));
errmsg.LogError(errno, NO_ERRCODE, "epoll_ctrl failed on fd %d with %s\n",
- udpLstnSocks[i+1], errStr);
+ lstn->sock, errStr);
}
}
+ i++;
}
while(1) {
@@ -492,8 +502,7 @@ rsRetVal rcvMainLoop(thrdInfo_t *pThrd)
break; /* terminate input! */
for(i = 0 ; i < nfds ; ++i) {
- processSocket(pThrd, udpLstnSocks[currEvt[i].data.u64], &frominetPrev, &bIsPermitted,
- udpRulesets[currEvt[i].data.u64]);
+ processSocket(pThrd, (struct lstn_s*)currEvt[i].data.u64, &frominetPrev, &bIsPermitted);
}
}
@@ -510,10 +519,10 @@ rsRetVal rcvMainLoop(thrdInfo_t *pThrd)
DEFiRet;
int maxfds;
int nfds;
- int i;
fd_set readfds;
struct sockaddr_storage frominetPrev;
int bIsPermitted;
+ struct lstn_s *lstn;
/* start "name caching" algo by making sure the previous system indicator
* is invalidated.
@@ -524,22 +533,18 @@ rsRetVal rcvMainLoop(thrdInfo_t *pThrd)
DBGPRINTF("imudp uses select()\n");
while(1) {
- /* Add the Unix Domain Sockets to the list of read
- * descriptors.
- * rgerhards 2005-08-01: we must now check if there are
- * any local sockets to listen to at all. If the -o option
- * is given without -a, we do not need to listen at all..
+ /* Add the Unix Domain Sockets to the list of read descriptors.
*/
maxfds = 0;
FD_ZERO(&readfds);
/* Add the UDP listen sockets to the list of read descriptors. */
- for (i = 0; i < *udpLstnSocks; i++) {
- if (udpLstnSocks[i+1] != -1) {
+ for(lstn = lcnfRoot ; lstn != NULL ; lstn = lstn->next) {
+ if (lstn->sock != -1) {
if(Debug)
- net.debugListenInfo(udpLstnSocks[i+1], "UDP");
- FD_SET(udpLstnSocks[i+1], &readfds);
- if(udpLstnSocks[i+1]>maxfds) maxfds=udpLstnSocks[i+1];
+ net.debugListenInfo(lstn->sock, "UDP");
+ FD_SET(lstn->sock, &readfds);
+ if(lstn->sock>maxfds) maxfds=lstn->sock;
}
}
if(Debug) {
@@ -555,10 +560,9 @@ rsRetVal rcvMainLoop(thrdInfo_t *pThrd)
if(glbl.GetGlobalInputTermState() == 1)
break; /* terminate input! */
- for(i = 0; nfds && i < *udpLstnSocks; i++) {
- if(FD_ISSET(udpLstnSocks[i+1], &readfds)) {
- processSocket(pThrd, udpLstnSocks[i+1], &frominetPrev, &bIsPermitted,
- udpRulesets[i+1]);
+ for(lstn = lcnfRoot ; nfds && lstn != NULL ; lstn = lstn->next) {
+ if(FD_ISSET(lstn->sock, &readfds)) {
+ processSocket(pThrd, lstn, &frominetPrev, &bIsPermitted);
--nfds; /* indicate we have processed one descriptor */
}
}
@@ -570,7 +574,7 @@ rsRetVal rcvMainLoop(thrdInfo_t *pThrd)
#endif /* #if HAVE_EPOLL_CREATE1 */
/* This function is called to gather input.
- * Note that udpLstnSocks must be non-NULL because otherwise we would not have
+ * Note that sock must be non-NULL because otherwise we would not have
* indicated that we want to run (or we have a programming error ;)). -- rgerhards, 2008-10-02
*/
BEGINrunInput
@@ -591,8 +595,10 @@ CODESTARTwillRun
net.HasRestrictions(UCHAR_CONSTANT("UDP"), &bDoACLCheck); /* UDP */
/* if we could not set up any listners, there is no point in running... */
- if(udpLstnSocks == NULL)
+ if(lcnfRoot == NULL) {
+ DBGPRINTF("imudp: no listeners configured, will not run\n");
ABORT_FINALIZE(RS_RET_NO_RUN);
+ }
iMaxLine = glbl.GetMaxLine();
@@ -602,15 +608,18 @@ ENDwillRun
BEGINafterRun
+ struct lstn_s *lstn, *lstnDel;
CODESTARTafterRun
/* do cleanup here */
net.clearAllowedSenders((uchar*)"UDP");
- if(udpLstnSocks != NULL) {
- net.closeUDPListenSockets(udpLstnSocks);
- udpLstnSocks = NULL;
- free(udpRulesets);
- udpRulesets = NULL;
+ for(lstn = lcnfRoot ; lstn != NULL ; ) {
+ statsobj.Destruct(&(lstn->stats));
+ close(lstn->sock);
+ lstnDel = lstn;
+ lstn = lstn->next;
+ free(lstnDel);
}
+ lcnfRoot = lcnfLast = NULL;
if(pRcvBuf != NULL) {
free(pRcvBuf);
pRcvBuf = NULL;
@@ -625,6 +634,7 @@ CODESTARTmodExit
/* release what we no longer need */
objRelease(errmsg, CORE_COMPONENT);
objRelease(glbl, CORE_COMPONENT);
+ objRelease(statsobj, CORE_COMPONENT);
objRelease(datetime, CORE_COMPONENT);
objRelease(prop, CORE_COMPONENT);
objRelease(ruleset, CORE_COMPONENT);
@@ -662,6 +672,7 @@ CODESTARTmodInit
CODEmodInit_QueryRegCFSLineHdlr
CHKiRet(objUse(errmsg, CORE_COMPONENT));
CHKiRet(objUse(glbl, CORE_COMPONENT));
+ CHKiRet(objUse(statsobj, CORE_COMPONENT));
CHKiRet(objUse(datetime, CORE_COMPONENT));
CHKiRet(objUse(prop, CORE_COMPONENT));
CHKiRet(objUse(ruleset, CORE_COMPONENT));
diff --git a/plugins/imuxsock/imuxsock.c b/plugins/imuxsock/imuxsock.c
index 5867f1c9..0e91a977 100644
--- a/plugins/imuxsock/imuxsock.c
+++ b/plugins/imuxsock/imuxsock.c
@@ -89,6 +89,7 @@ DEF_IMOD_STATIC_DATA
DEFobjCurrIf(errmsg)
DEFobjCurrIf(glbl)
DEFobjCurrIf(prop)
+DEFobjCurrIf(net)
DEFobjCurrIf(parser)
DEFobjCurrIf(datetime)
DEFobjCurrIf(statsobj)
@@ -157,6 +158,7 @@ static int sd_fds = 0; /* number of systemd activated sockets */
/* config settings */
static int bOmitLocalLogging = 0;
static uchar *pLogSockName = NULL;
+static uchar *pLocalIPIF = NULL;
static uchar *pLogHostName = NULL; /* host name to use with this socket */
static int bUseFlowCtl = 0; /* use flow control or not (if yes, only LIGHT is used! */
static int bIgnoreTimestamp = 1; /* ignore timestamps present in the incoming message? */
@@ -764,6 +766,8 @@ ENDrunInput
BEGINwillRun
+ uchar myIP[128];
+ rsRetVal localRet;
CODESTARTwillRun
register int i;
int actSocks;
@@ -784,7 +788,7 @@ CODESTARTwillRun
else if(sd_booted()) {
struct stat st;
if(stat(SYSTEMD_JOURNAL, &st) != -1 && S_ISDIR(st.st_mode)) {
- listeners[0].sockName = (uchar*) SYSTEMD_PATH_LOG;
+ listeners[0].sockName = (uchar*)SYSTEMD_PATH_LOG;
}
}
if(ratelimitIntervalSysSock > 0) {
@@ -826,6 +830,23 @@ CODESTARTwillRun
CHKiRet(prop.SetString(pInputName, UCHAR_CONSTANT("imuxsock"), sizeof("imuxsock") - 1));
CHKiRet(prop.ConstructFinalize(pInputName));
+ if(pLocalIPIF == NULL) {
+ strcpy((char*)myIP, "127.0.0.1");
+ } else {
+ localRet = net.GetIFIPAddr((uchar*)pLocalIPIF, AF_UNSPEC, myIP, (int) sizeof(myIP));
+ if(localRet != RS_RET_OK) {
+ DBGPRINTF("imuxsock: could not obtain my IP, using 127.0.0.1 instead\n");
+ strcpy((char*)myIP, "127.0.0.1");
+ }
+ }
+
+ DBGPRINTF("imuxsock: using '%s' as localhost IP\n", myIP);
+
+ CHKiRet(prop.Construct(&pLocalHostIP));
+ CHKiRet(prop.SetString(pLocalHostIP, myIP, ustrlen(myIP)));
+ CHKiRet(prop.ConstructFinalize(pLocalHostIP));
+
+
finalize_it:
ENDwillRun
@@ -857,6 +878,7 @@ CODESTARTafterRun
}
/* free no longer needed string */
free(pLogSockName);
+ free(pLocalIPIF);
free(pLogHostName);
discardLogSockets();
@@ -901,6 +923,10 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
free(pLogSockName);
pLogSockName = NULL;
}
+ if(pLocalIPIF != NULL) {
+ free(pLocalIPIF);
+ pLocalIPIF = NULL;
+ }
if(pLogHostName != NULL) {
free(pLogHostName);
pLogHostName = NULL;
@@ -931,6 +957,7 @@ CODESTARTmodInit
CODEmodInit_QueryRegCFSLineHdlr
CHKiRet(objUse(errmsg, CORE_COMPONENT));
CHKiRet(objUse(glbl, CORE_COMPONENT));
+ CHKiRet(objUse(net, CORE_COMPONENT));
CHKiRet(objUse(prop, CORE_COMPONENT));
CHKiRet(objUse(statsobj, CORE_COMPONENT));
CHKiRet(objUse(datetime, CORE_COMPONENT));
@@ -954,10 +981,6 @@ CODEmodInit_QueryRegCFSLineHdlr
listeners[i].fd = -1;
}
- CHKiRet(prop.Construct(&pLocalHostIP));
- CHKiRet(prop.SetString(pLocalHostIP, UCHAR_CONSTANT("127.0.0.1"), sizeof("127.0.0.1") - 1));
- CHKiRet(prop.ConstructFinalize(pLocalHostIP));
-
/* now init listen socket zero, the local log socket */
CHKiRet(prop.Construct(&(listeners[0].hostName)));
CHKiRet(prop.SetString(listeners[0].hostName, glbl.GetLocalHostName(), ustrlen(glbl.GetLocalHostName())));
@@ -988,6 +1011,8 @@ CODEmodInit_QueryRegCFSLineHdlr
NULL, &ratelimitSeverity, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler,
resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr((uchar *)"imuxsocklocalipif", 0, eCmdHdlrGetWord,
+ NULL, &pLocalIPIF, STD_LOADABLE_MODULE_ID));
/* the following one is a (dirty) trick: the system log socket is not added via
* an "addUnixListenSocket" config format. As such, it's properties can not be modified
* via $InputUnixListenSocket*". So we need to add a special directive
diff --git a/runtime/msg.c b/runtime/msg.c
index 31863b2d..810a396e 100644
--- a/runtime/msg.c
+++ b/runtime/msg.c
@@ -1671,8 +1671,6 @@ void MsgSetTAG(msg_t *pMsg, uchar* pszBuf, size_t lenBuf)
uchar *pBuf;
assert(pMsg != NULL);
-dbgprintf("MsgSetTAG in: len %d, pszBuf: %s\n", lenBuf, pszBuf);
-
freeTAG(pMsg);
pMsg->iLenTAG = lenBuf;
@@ -1691,8 +1689,6 @@ dbgprintf("MsgSetTAG in: len %d, pszBuf: %s\n", lenBuf, pszBuf);
memcpy(pBuf, pszBuf, pMsg->iLenTAG);
pBuf[pMsg->iLenTAG] = '\0'; /* this also works with truncation! */
-
-dbgprintf("MsgSetTAG exit: pMsg->iLenTAG %d, pMsg->TAG.szBuf: %s\n", pMsg->iLenTAG, pMsg->TAG.szBuf);
}
diff --git a/runtime/net.c b/runtime/net.c
index 4781739f..1264b2cb 100644
--- a/runtime/net.c
+++ b/runtime/net.c
@@ -12,7 +12,7 @@
* long term, but it is good to have it out of syslogd.c. Maybe this here is
* an interim location ;)
*
- * Copyright 2007, 2008 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH.
*
* rgerhards, 2008-04-16: I changed this code to LGPL today. I carefully analyzed
* that it does not borrow code from the original sysklogd and that I have
@@ -54,6 +54,9 @@
#include <fnmatch.h>
#include <fcntl.h>
#include <unistd.h>
+#include <ifaddrs.h>
+#include <sys/types.h>
+#include <arpa/inet.h>
#include "syslogd-types.h"
#include "module-template.h"
@@ -1581,6 +1584,54 @@ finalize_it:
}
+/* return the IP address (IPv4/6) for the provided interface. Returns
+ * RS_RET_NOT_FOUND if interface can not be found in interface list.
+ * The family must be correct (AF_INET vs. AF_INET6, AF_UNSPEC means
+ * either of *these two*).
+ * The function re-queries the interface list (at least in theory).
+ * However, it caches entries in order to avoid too-frequent requery.
+ * rgerhards, 2012-03-06
+ */
+static rsRetVal
+getIFIPAddr(uchar *szif, int family, uchar *pszbuf, int lenBuf)
+{
+ struct ifaddrs * ifaddrs = NULL;
+ struct ifaddrs * ifa;
+ void * pAddr;
+ DEFiRet;
+
+ if(getifaddrs(&ifaddrs) != 0) {
+ ABORT_FINALIZE(RS_RET_ERR);
+ }
+
+ for (ifa = ifaddrs; ifa != NULL; ifa = ifa->ifa_next) {
+ if(strcmp(ifa->ifa_name, (char*)szif))
+ continue;
+ if( (family == AF_INET6 || family == AF_UNSPEC)
+ && ifa->ifa_addr->sa_family == AF_INET6) {
+ pAddr = &((struct sockaddr_in6 *)ifa->ifa_addr)->sin6_addr;
+ inet_ntop(AF_INET6, pAddr, (char*)pszbuf, lenBuf);
+ break;
+ } else if(/* (family == AF_INET || family == AF_UNSPEC)
+ &&*/ ifa->ifa_addr->sa_family == AF_INET) {
+ pAddr = &((struct sockaddr_in *)ifa->ifa_addr)->sin_addr;
+ inet_ntop(AF_INET, pAddr, (char*)pszbuf, lenBuf);
+ break;
+ }
+ }
+
+ if(ifaddrs != NULL)
+ freeifaddrs(ifaddrs);
+
+ if(ifa == NULL)
+ iRet = RS_RET_NOT_FOUND;
+
+finalize_it:
+ RETiRet;
+
+}
+
+
/* queryInterface function
* rgerhards, 2008-03-05
*/
@@ -1612,6 +1663,7 @@ CODESTARTobjQueryInterface(net)
pIf->PermittedPeerWildcardMatch = PermittedPeerWildcardMatch;
pIf->CmpHost = CmpHost;
pIf->HasRestrictions = HasRestrictions;
+ pIf->GetIFIPAddr = getIFIPAddr;
/* data members */
pIf->pACLAddHostnameOnFail = &ACLAddHostnameOnFail;
pIf->pACLDontResolve = &ACLDontResolve;
diff --git a/runtime/net.h b/runtime/net.h
index 101ce79d..1b41c81c 100644
--- a/runtime/net.h
+++ b/runtime/net.h
@@ -1,6 +1,6 @@
/* Definitions for network-related stuff.
*
- * Copyright 2007, 2008 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of the rsyslog runtime library.
*
@@ -151,11 +151,13 @@ BEGINinterface(net) /* name must also be changed in ENDinterface macro! */
/* v6 interface additions - 2009-11-16 */
rsRetVal (*HasRestrictions)(uchar *, int *bHasRestrictions);
int (*isAllowedSender2)(uchar *pszType, struct sockaddr *pFrom, const char *pszFromHost, int bChkDNS);
+ /* v7 interface additions - 2012-03-06 */
+ rsRetVal (*GetIFIPAddr)(uchar *szif, int family, uchar *pszbuf, int lenBuf);
/* data members - these should go away over time... TODO */
int *pACLAddHostnameOnFail; /* add hostname to acl when DNS resolving has failed */
int *pACLDontResolve; /* add hostname to acl instead of resolving it to IP(s) */
ENDinterface(net)
-#define netCURR_IF_VERSION 6 /* increment whenever you change the interface structure! */
+#define netCURR_IF_VERSION 7 /* increment whenever you change the interface structure! */
/* prototypes */
PROTOTYPEObj(net);
diff --git a/runtime/queue.c b/runtime/queue.c
index 621a4eed..0b717289 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, 2009 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2008-2011 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of the rsyslog runtime library.
*
@@ -1312,6 +1312,7 @@ static int qqueueChkDiscardMsg(qqueue_t *pThis, int iQueueSize, void *pUsr)
if(iRetLocal == RS_RET_OK && iSeverity >= pThis->iDiscardSeverity) {
DBGOPRINT((obj_t*) pThis, "queue nearly full (%d entries), discarded severity %d message\n",
iQueueSize, iSeverity);
+ STATSCOUNTER_INC(pThis->ctrNFDscrd, pThis->mutCtrNFDscrd);
objDestruct(pUsr);
ABORT_FINALIZE(RS_RET_QUEUE_FULL);
} else {
@@ -1942,6 +1943,13 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */
CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("full"),
ctrType_IntCtr, &pThis->ctrFull));
+ STATSCOUNTER_INIT(pThis->ctrFDscrd, pThis->mutCtrFDscrd);
+ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("discarded.full"),
+ ctrType_IntCtr, &pThis->ctrFDscrd));
+ STATSCOUNTER_INIT(pThis->ctrNFDscrd, pThis->mutCtrNFDscrd);
+ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("discarded.nf"),
+ ctrType_IntCtr, &pThis->ctrNFDscrd));
+
pThis->ctrMaxqsize = 0; /* no mutex needed, thus no init call */
CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("maxqsize"),
ctrType_Int, &pThis->ctrMaxqsize));
@@ -2295,6 +2303,7 @@ doEnqSingleObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr)
// TODO : handle enqOnly => discard!
if(pthread_cond_timedwait(&pThis->notFull, pThis->mut, &t) != 0) {
DBGOPRINT((obj_t*) pThis, "enqueueMsg: cond timeout, dropping message!\n");
+ STATSCOUNTER_INC(pThis->ctrFDscrd, pThis->mutCtrFDscrd);
objDestruct(pUsr);
ABORT_FINALIZE(RS_RET_QUEUE_FULL);
}
@@ -2435,6 +2444,7 @@ DEFpropSetMeth(qqueue, iLowWtrMrk, int)
DEFpropSetMeth(qqueue, iDiscardMrk, int)
DEFpropSetMeth(qqueue, iFullDlyMrk, int)
DEFpropSetMeth(qqueue, iDiscardSeverity, int)
+DEFpropSetMeth(qqueue, iLightDlyMrk, int)
DEFpropSetMeth(qqueue, bIsDA, int)
DEFpropSetMeth(qqueue, iMinMsgsPerWrkr, int)
DEFpropSetMeth(qqueue, bSaveOnShutdown, int)
diff --git a/runtime/queue.h b/runtime/queue.h
index 7ef5673c..a575698f 100644
--- a/runtime/queue.h
+++ b/runtime/queue.h
@@ -169,6 +169,8 @@ struct queue_s {
statsobj_t *statsobj;
STATSCOUNTER_DEF(ctrEnqueued, mutCtrEnqueued);
STATSCOUNTER_DEF(ctrFull, mutCtrFull);
+ STATSCOUNTER_DEF(ctrFDscrd, mutCtrFDscrd);
+ STATSCOUNTER_DEF(ctrNFDscrd, mutCtrNFDscrd);
int ctrMaxqsize; /* NOT guarded by a mutex */
};
@@ -199,6 +201,7 @@ PROTOTYPEpropSetMeth(qqueue, toQShutdown, long);
PROTOTYPEpropSetMeth(qqueue, toActShutdown, long);
PROTOTYPEpropSetMeth(qqueue, toWrkShutdown, long);
PROTOTYPEpropSetMeth(qqueue, toEnq, long);
+PROTOTYPEpropSetMeth(qqueue, iLightDlyMrk, int);
PROTOTYPEpropSetMeth(qqueue, iHighWtrMrk, int);
PROTOTYPEpropSetMeth(qqueue, iLowWtrMrk, int);
PROTOTYPEpropSetMeth(qqueue, iDiscardMrk, int);
diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index 5ee2a55a..69968a52 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -500,6 +500,7 @@ debugPrintAll(void)
static rsRetVal
rulesetCreateQueue(void __attribute__((unused)) *pVal, int *pNewVal)
{
+ uchar *rulesetMainQName;
DEFiRet;
if(pCurrRuleset == NULL) {
@@ -518,7 +519,9 @@ rulesetCreateQueue(void __attribute__((unused)) *pVal, int *pNewVal)
FINALIZE; /* if it is turned off, we do not need to change anything ;) */
dbgprintf("adding a ruleset-specific \"main\" queue");
- CHKiRet(createMainQueue(&pCurrRuleset->pQueue, UCHAR_CONSTANT("ruleset")));
+ rulesetMainQName = (pCurrRuleset->pszName == NULL)? UCHAR_CONSTANT("ruleset") :
+ pCurrRuleset->pszName;
+ CHKiRet(createMainQueue(&pCurrRuleset->pQueue, rulesetMainQName));
finalize_it:
RETiRet;
diff --git a/tcps_sess.c b/tcps_sess.c
index 15423cc8..854d3742 100644
--- a/tcps_sess.c
+++ b/tcps_sess.c
@@ -259,6 +259,7 @@ defaultDoSubmitMessage(tcps_sess_t *pThis, struct syslogTime *stTime, time_t ttG
CHKiRet(MsgSetRcvFromIP(pMsg, pThis->fromHostIP));
MsgSetRuleset(pMsg, pThis->pLstnInfo->pRuleset);
+ STATSCOUNTER_INC(pThis->pLstnInfo->ctrSubmit, pThis->pLstnInfo->mutCtrSubmit);
if(pMultiSub == NULL) {
CHKiRet(submitMsg(pMsg));
} else {
diff --git a/tcpsrv.c b/tcpsrv.c
index 95c45780..86175ff4 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -39,8 +39,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
#include "config.h"
+#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
@@ -93,6 +93,7 @@ DEFobjCurrIf(netstrm)
DEFobjCurrIf(nssel)
DEFobjCurrIf(nspoll)
DEFobjCurrIf(prop)
+DEFobjCurrIf(statsobj)
/* add new listener port to listener port list
@@ -102,6 +103,7 @@ static inline rsRetVal
addNewLstnPort(tcpsrv_t *pThis, uchar *pszPort)
{
tcpLstnPortList_t *pEntry;
+ uchar statname[64];
DEFiRet;
ISOBJ_TYPE_assert(pThis, tcpsrv);
@@ -121,6 +123,15 @@ addNewLstnPort(tcpsrv_t *pThis, uchar *pszPort)
pEntry->pNext = pThis->pLstnPorts;
pThis->pLstnPorts = pEntry;
+ /* support statistics gathering */
+ CHKiRet(statsobj.Construct(&(pEntry->stats)));
+ 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(statsobj.AddCounter(pEntry->stats, UCHAR_CONSTANT("submitted"),
+ ctrType_IntCtr, &(pEntry->ctrSubmit)));
+ CHKiRet(statsobj.ConstructFinalize(pEntry->stats));
+
finalize_it:
RETiRet;
}
@@ -1071,6 +1082,7 @@ CODESTARTObjClassExit(tcpsrv)
objRelease(tcps_sess, DONT_LOAD_LIB);
objRelease(conf, CORE_COMPONENT);
objRelease(prop, CORE_COMPONENT);
+ objRelease(statsobj, CORE_COMPONENT);
objRelease(ruleset, CORE_COMPONENT);
objRelease(glbl, CORE_COMPONENT);
objRelease(errmsg, CORE_COMPONENT);
@@ -1097,6 +1109,7 @@ BEGINObjClassInit(tcpsrv, 1, OBJ_IS_LOADABLE_MODULE) /* class, version - CHANGE
CHKiRet(objUse(conf, CORE_COMPONENT));
CHKiRet(objUse(glbl, CORE_COMPONENT));
CHKiRet(objUse(ruleset, CORE_COMPONENT));
+ CHKiRet(objUse(statsobj, CORE_COMPONENT));
CHKiRet(objUse(prop, CORE_COMPONENT));
/* set our own handlers */
diff --git a/tcpsrv.h b/tcpsrv.h
index 269863c2..f958f917 100644
--- a/tcpsrv.h
+++ b/tcpsrv.h
@@ -24,6 +24,7 @@
#include "obj.h"
#include "prop.h"
#include "tcps_sess.h"
+#include "statsobj.h"
/* support for framing anomalies */
typedef enum ETCPsyslogFramingAnomaly {
@@ -39,6 +40,8 @@ struct tcpLstnPortList_s {
prop_t *pInputName;
tcpsrv_t *pSrv; /**< pointer to higher-level server instance */
ruleset_t *pRuleset; /**< associated ruleset */
+ statsobj_t *stats; /**< associated stats object */
+ STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit)
tcpLstnPortList_t *pNext; /**< next port or NULL */
};
diff --git a/tools/syslogd.c b/tools/syslogd.c
index 0b7bbc96..02a0dad6 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -262,6 +262,7 @@ static uchar *pszConfDAGFile = NULL; /* name of config DAG file, non-NULL mea
qqueue_t *pMsgQueue = NULL; /* the main message queue */
static int iMainMsgQueueSize = 10000; /* size of the main message queue above */
static int iMainMsgQHighWtrMark = 8000; /* high water mark for disk-assisted queues */
+static int iMainMsgQLightDlyMark = -1; /* light delay mark for disk-assisted queues */
static int iMainMsgQLowWtrMark = 2000; /* low water mark for disk-assisted queues */
static int iMainMsgQDiscardMark = 9800; /* begin to discard messages */
static int iMainMsgQDiscardSeverity = 8; /* by default, discard nothing to prevent unintentional loss */
@@ -300,6 +301,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
pszMainMsgQFName = NULL;
iMainMsgQueueSize = 10000;
iMainMsgQHighWtrMark = 8000;
+ iMainMsgQLightDlyMark = -1;
iMainMsgQLowWtrMark = 2000;
iMainMsgQDiscardMark = 9800;
iMainMsgQDiscardSeverity = 8;
@@ -1457,8 +1459,8 @@ static void dbgPrintInitInfo(void)
iMainMsgQueueNumWorkers, iMainMsgQtoWrkShutdown, iMainMsgQPersistUpdCnt);
DBGPRINTF("Main queue timeouts: shutdown: %d, action completion shutdown: %d, enq: %d\n",
iMainMsgQtoQShutdown, iMainMsgQtoActShutdown, iMainMsgQtoEnq);
- DBGPRINTF("Main queue watermarks: high: %d, low: %d, discard: %d, discard-severity: %d\n",
- iMainMsgQHighWtrMark, iMainMsgQLowWtrMark, iMainMsgQDiscardMark, iMainMsgQDiscardSeverity);
+ DBGPRINTF("Main queue watermarks: high: %d, low: %d, discard: %d, discard-severity: %d, light-delay %d\n",
+ iMainMsgQHighWtrMark, iMainMsgQLowWtrMark, iMainMsgQDiscardMark, iMainMsgQDiscardSeverity, iMainMsgQLightDlyMark);
DBGPRINTF("Main queue save on shutdown %d, max disk space allowed %lld\n",
bMainMsgQSaveOnShutdown, iMainMsgQueMaxDiskSpace);
/* TODO: add
@@ -1574,6 +1576,9 @@ rsRetVal createMainQueue(qqueue_t **ppQueue, uchar *pszQueueName)
setQPROP(qqueueSetiHighWtrMrk, "$MainMsgQueueHighWaterMark", iMainMsgQHighWtrMark);
setQPROP(qqueueSetiLowWtrMrk, "$MainMsgQueueLowWaterMark", iMainMsgQLowWtrMark);
setQPROP(qqueueSetiDiscardMrk, "$MainMsgQueueDiscardMark", iMainMsgQDiscardMark);
+ if(iMainMsgQLightDlyMark > 0) {
+ setQPROP(qqueueSetiLightDlyMrk, "$MainMsgQueueLightDelayMark", iMainMsgQLightDlyMark);
+ }
setQPROP(qqueueSetiDiscardSeverity, "$MainMsgQueueDiscardSeverity", iMainMsgQDiscardSeverity);
setQPROP(qqueueSetiMinMsgsPerWrkr, "$MainMsgQueueWorkerThreadMinimumMessages", iMainMsgQWrkMinMsgs);
setQPROP(qqueueSetbSaveOnShutdown, "$MainMsgQueueSaveOnShutdown", bMainMsgQSaveOnShutdown);
@@ -2051,6 +2056,7 @@ static rsRetVal loadBuildInModules(void)
CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuehighwatermark", 0, eCmdHdlrInt, NULL, &iMainMsgQHighWtrMark, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuelowwatermark", 0, eCmdHdlrInt, NULL, &iMainMsgQLowWtrMark, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuediscardmark", 0, eCmdHdlrInt, NULL, &iMainMsgQDiscardMark, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuelightdelaymark", 0, eCmdHdlrInt, NULL, &iMainMsgQLightDlyMark, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuediscardseverity", 0, eCmdHdlrSeverity, NULL, &iMainMsgQDiscardSeverity, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuecheckpointinterval", 0, eCmdHdlrInt, NULL, &iMainMsgQPersistUpdCnt, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"mainmsgqueuesyncqueuefiles", 0, eCmdHdlrBinary, NULL, &bMainMsgQSyncQeueFiles, NULL));
@@ -2706,6 +2712,7 @@ int realMain(int argc, char **argv)
CHKmalloc(LocalHostName = (uchar*)strdup(hent->h_aliases[i]));
} else {
CHKmalloc(LocalHostName = (uchar*)strdup(hent->h_name));
+ int i;
}
if((p = (uchar*)strchr((char*)LocalHostName, '.')))
generated by cgit (git 2.34.1) at 2024-05-20 01:43:58 +0000