diff options
| -rw-r--r-- | ChangeLog | 25 | ||||
| -rw-r--r-- | action.c | 10 | ||||
| -rw-r--r-- | configure.ac | 6 | ||||
| -rw-r--r-- | dirty.h | 16 | ||||
| -rw-r--r-- | doc/queues.html | 7 | ||||
| -rw-r--r-- | doc/rsyslog_conf.html | 8 | ||||
| -rw-r--r-- | plugins/immark/immark.c | 1 | ||||
| -rw-r--r-- | plugins/imrelp/imrelp.c | 5 | ||||
| -rw-r--r-- | plugins/imudp/imudp.c | 193 | ||||
| -rw-r--r-- | plugins/imuxsock/imuxsock.c | 4 | ||||
| -rw-r--r-- | runtime/Makefile.am | 2 | ||||
| -rw-r--r-- | runtime/atomic.h | 2 | ||||
| -rw-r--r-- | runtime/datetime.c | 12 | ||||
| -rw-r--r-- | runtime/datetime.h | 2 | ||||
| -rw-r--r-- | runtime/debug.c | 5 | ||||
| -rw-r--r-- | runtime/debug.h | 2 | ||||
| -rw-r--r-- | runtime/modules.h | 5 | ||||
| -rw-r--r-- | runtime/msg.c | 91 | ||||
| -rw-r--r-- | runtime/msg.h | 29 | ||||
| -rw-r--r-- | runtime/parser.c | 313 | ||||
| -rw-r--r-- | runtime/parser.h | 30 | ||||
| -rw-r--r-- | runtime/queue.c | 26 | ||||
| -rw-r--r-- | runtime/rsyslog.h | 1 | ||||
| -rw-r--r-- | runtime/srutils.c | 4 | ||||
| -rw-r--r-- | runtime/sysvar.c | 2 | ||||
| -rw-r--r-- | runtime/wti.c | 6 | ||||
| -rw-r--r-- | runtime/wtp.c | 2 | ||||
| -rw-r--r-- | tcps_sess.c | 11 | ||||
| -rw-r--r-- | threads.c | 2 | ||||
| -rw-r--r-- | tools/Makefile.am | 3 | ||||
| -rw-r--r-- | tools/msggen.c | 38 | ||||
| -rw-r--r-- | tools/syslogd.c | 68 |
32 files changed, 776 insertions, 155 deletions
@@ -1,10 +1,35 @@ --------------------------------------------------------------------------- +Version 3.23.1 [DEVEL] (rgerhards), 2008-10-?? + +********************************* WARNING ********************************* +This version has a slightly different on-disk format for message entries. +As a consequence, old queue files being read by this version may have +an invalid output timestamp, which could result to some malfunction inside +the output driver. It is recommended to drain queues with the previous +version before switching to this one. +********************************* WARNING ********************************* + +- reordered imudp processing. Message parsing is now done as part of main + message queue worker processing (was part of the input thread) + This should also improve performance, as potentially more work is + done in parallel. +- bugfix: compressed syslog messages could be slightly mis-uncompressed + if the last byte of the compressed record was a NUL +- added $UDPServerTimeRequery option which enables to work with + less acurate timestamps in favor of performance. This enables querying + of the time only every n-th time if imudp is running in the tight + receive loop (aka receiving messsages at a high rate) +- doc bugfix: queue doc had wrong parameter name for setting controlling + worker thread shutdown period +--------------------------------------------------------------------------- Version 3.21.6 [DEVEL] (rgerhards), 2008-10-22 - consolidated time calls during msg object creation, improves performance and consistency - bugfix: solved a segfault condition - bugfix: subsecond time properties generated by imfile, imklog and internal messages could be slightly inconsistent +- bugfix: (potentially big) memory leak on HUP if queues could not be + drained before timeout - thanks to David Lang for pointing this out - added capability to support multiple module search pathes. Thank to Marius Tomaschewski for providing the patch. - bugfix: im3195 did no longer compile @@ -602,7 +602,7 @@ actionWriteToAction(action_t *pAction) * ... RAWMSG is a problem ... Please note that digital * signatures inside the message are also invalidated. */ - datetime.getCurrTime(&(pMsg->tRcvdAt)); + datetime.getCurrTime(&(pMsg->tRcvdAt), &(pMsg->ttGenTime)); memcpy(&pMsg->tTIMESTAMP, &pMsg->tRcvdAt, sizeof(struct syslogTime)); MsgSetMSG(pMsg, (char*)szRepMsg); MsgSetRawMsg(pMsg, (char*)szRepMsg); @@ -625,13 +625,13 @@ actionWriteToAction(action_t *pAction) dbgprintf("action not yet ready again to be executed, onceInterval %d, tCurr %d, tNext %d\n", (int) pAction->iSecsExecOnceInterval, (int) getActNow(pAction), (int) (pAction->iSecsExecOnceInterval + pAction->tLastExec)); + /* TODO: the time call below may use reception time, not dequeue time - under consideration. -- rgerhards, 2008-09-17 */ + pAction->tLastExec = getActNow(pAction); /* re-init time flags */ FINALIZE; } - pAction->f_time = pAction->tLastExec = getActNow(pAction); /* re-init time flags */ - /* Note: tLastExec could be set in the if block above, but f_time causes us a hard time - * so far, I do not see a solution to getting rid of it. -- rgerhards, 2008-09-16 - */ + /* TODO: the time call below may use reception time, not dequeue time - under consideration. -- rgerhards, 2008-09-17 */ + pAction->f_time = pAction->f_pMsg->ttGenTime; /* When we reach this point, we have a valid, non-disabled action. * So let's enqueue our message for execution. -- rgerhards, 2007-07-24 diff --git a/configure.ac b/configure.ac index 69744c8f..2f17e575 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],[3.21.6],[rsyslog@lists.adiscon.com]) +AC_INIT([rsyslog],[3.23.1],[rsyslog@lists.adiscon.com]) AM_INIT_AUTOMAKE AC_CONFIG_SRCDIR([ChangeLog]) AC_CONFIG_HEADERS([config.h]) @@ -53,7 +53,7 @@ AC_SUBST(dl_libs) AC_HEADER_RESOLV AC_HEADER_STDC AC_HEADER_SYS_WAIT -AC_CHECK_HEADERS([arpa/inet.h libgen.h fcntl.h locale.h netdb.h netinet/in.h paths.h stddef.h stdlib.h string.h sys/file.h sys/ioctl.h sys/param.h sys/socket.h sys/time.h sys/stat.h syslog.h unistd.h utmp.h]) +AC_CHECK_HEADERS([arpa/inet.h libgen.h fcntl.h locale.h netdb.h netinet/in.h paths.h stddef.h stdlib.h string.h sys/file.h sys/ioctl.h sys/param.h sys/socket.h sys/time.h sys/stat.h syslog.h unistd.h utmp.h sys/epoll.h]) # Checks for typedefs, structures, and compiler characteristics. AC_C_CONST @@ -88,7 +88,7 @@ AC_TYPE_SIGNAL AC_FUNC_STAT AC_FUNC_STRERROR_R AC_FUNC_VPRINTF -AC_CHECK_FUNCS([flock basename alarm clock_gettime gethostbyname gethostname gettimeofday localtime_r memset mkdir regcomp select setid socket strcasecmp strchr strdup strerror strndup strnlen strrchr strstr strtol strtoul uname ttyname_r]) +AC_CHECK_FUNCS([flock basename alarm clock_gettime gethostbyname gethostname gettimeofday localtime_r memset mkdir regcomp select setid socket strcasecmp strchr strdup strerror strndup strnlen strrchr strstr strtol strtoul uname ttyname_r epoll_wait]) # Check for MAXHOSTNAMELEN AC_MSG_CHECKING(for MAXHOSTNAMELEN) @@ -27,20 +27,9 @@ #ifndef DIRTY_H_INCLUDED #define DIRTY_H_INCLUDED 1 -/* Flags to logmsg(). - */ -#define NOFLAG 0x000 /* no flag is set (to be used when a flag must be specified and none is required) */ -#define INTERNAL_MSG 0x001 /* msg generated by logmsgInternal() --> special handling */ -/* NO LONGER USED: #define SYNC_FILE 0x002 / * do fsync on file after printing */ -#define IGNDATE 0x004 /* ignore, if given, date in message and use date of reception as msg date */ -#define MARK 0x008 /* this message is a mark */ - -#define MSG_PARSE_HOSTNAME 1 -#define MSG_DONT_PARSE_HOSTNAME 0 - rsRetVal submitMsg(msg_t *pMsg); rsRetVal logmsgInternal(int iErr, int pri, uchar *msg, int flags); -rsRetVal parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int bParseHost, int flags, flowControl_t flowCtlTypeu, uchar *pszInputName); +rsRetVal parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int flags, flowControl_t flowCtlTypeu, uchar *pszInputName, struct syslogTime *stTime, time_t ttGenTime); /* TODO: the following 2 need to go in conf obj interface... */ rsRetVal cflineParseTemplateName(uchar** pp, omodStringRequest_t *pOMSR, int iEntry, int iTplOpts, uchar *dfltTplName); @@ -61,6 +50,9 @@ extern int bReduceRepeatMsgs; #define BACKOFF(f) { if (++(f)->f_repeatcount > MAXREPEAT) \ (f)->f_repeatcount = MAXREPEAT; \ } +extern int bDropTrailingLF; +extern uchar cCCEscapeChar; +extern int bEscapeCCOnRcv; #ifdef USE_NETZIP /* config param: minimum message size to try compression. The smaller * the message, the less likely is any compression gain. We check for diff --git a/doc/queues.html b/doc/queues.html index a2074d36..7461121b 100644 --- a/doc/queues.html +++ b/doc/queues.html @@ -219,11 +219,12 @@ parall. Thus, the upper limit ca be set via "<i>$<object>QueueWorkerThread If it, for example, is set to four, no more than four workers will ever be started, no matter how many elements are enqueued. </p> <p>Worker threads that have been started are kept running until an inactivity -timeout happens. The timeout can be set via "<i>$<object>QueueWorkerTimeoutShutdown</i>" +timeout happens. The timeout can be set via "<i>$<object>QueueWorkerTimeoutThreadShutdown</i>" and is specified in milliseconds. If you do not like to keep the workers running, simply set it to 0, which means immediate timeout and thus immediate shutdown. But consider that creating threads involves some overhead, and this is -why we keep them running.</p> +why we keep them running. If you would like to never shutdown any worker +threads, specify -1 for this parameter.</p> <h2>Discarding Messages</h2> <p>If the queue reaches the so called "discard watermark" (a number of queued elements), less important messages can automatically be discarded. This is in an @@ -357,4 +358,4 @@ parameters, because not all are applicable. For example, in current output module design, actions do not support multi-threading. Consequently, the number of worker threads is fixed to one for action queues and can not be changed.</p> -</body></html>
\ No newline at end of file +</body></html> diff --git a/doc/rsyslog_conf.html b/doc/rsyslog_conf.html index 7cd40cb7..2b773476 100644 --- a/doc/rsyslog_conf.html +++ b/doc/rsyslog_conf.html @@ -243,6 +243,14 @@ address (or name) the UDP listens should bind to</li> <li>$UDPServerRun <port> (imudp) -- former -r<port> option, default 514, start UDP server on this port, "*" means all addresses</li> +<li>$UDPServerTimeRequery <nbr-of-times> (imudp) -- this is a performance +optimization. Getting the system time is very costly. With this setting, imudp can +be instructed to obtain the precise time only once every n-times. This logic is +only activated if messages come in at a very fast rate, so doing less frequent +time calls should usually be acceptable. The default value is two, because we have +seen that even without optimization the kernel often returns twice the identical time. +You can set this value as high as you like, but do so at your own risk. The higher +the value, the less precise the timestamp. <li><a href="rsconf1_umask.html">$UMASK</a></li> </ul> <p><b>Where <size_nbr> is specified above,</b> diff --git a/plugins/immark/immark.c b/plugins/immark/immark.c index 323da3fe..8504f872 100644 --- a/plugins/immark/immark.c +++ b/plugins/immark/immark.c @@ -41,6 +41,7 @@ #include "cfsysline.h" #include "module-template.h" #include "errmsg.h" +#include "msg.h" MODULE_TYPE_INPUT diff --git a/plugins/imrelp/imrelp.c b/plugins/imrelp/imrelp.c index b01dd98b..2255e643 100644 --- a/plugins/imrelp/imrelp.c +++ b/plugins/imrelp/imrelp.c @@ -42,6 +42,7 @@ #include "cfsysline.h" #include "module-template.h" #include "net.h" +#include "msg.h" MODULE_TYPE_INPUT @@ -83,8 +84,8 @@ static relpRetVal onSyslogRcv(uchar *pHostname, uchar __attribute__((unused)) *pIP, uchar *pMsg, size_t lenMsg) { DEFiRet; - parseAndSubmitMessage(pHostname, (uchar*) "[unset]", pMsg, lenMsg, MSG_PARSE_HOSTNAME, - NOFLAG, eFLOWCTL_LIGHT_DELAY, (uchar*)"imrelp"); + parseAndSubmitMessage(pHostname, (uchar*) "[unset]", pMsg, lenMsg, PARSE_HOSTNAME, + eFLOWCTL_LIGHT_DELAY, (uchar*)"imrelp", NULL, 0); RETiRet; } diff --git a/plugins/imudp/imudp.c b/plugins/imudp/imudp.c index 5792a999..a49378cf 100644 --- a/plugins/imudp/imudp.c +++ b/plugins/imudp/imudp.c @@ -40,6 +40,9 @@ #include "srUtils.h" #include "errmsg.h" #include "glbl.h" +#include "msg.h" +#include "parser.h" +#include "datetime.h" MODULE_TYPE_INPUT @@ -50,6 +53,7 @@ DEF_IMOD_STATIC_DATA DEFobjCurrIf(errmsg) DEFobjCurrIf(glbl) DEFobjCurrIf(net) +DEFobjCurrIf(datetime) static int iMaxLine; /* maximum UDP message size supported */ static int *udpLstnSocks = NULL; /* Internet datagram sockets, first element is nbr of elements @@ -59,6 +63,8 @@ static uchar *pRcvBuf = NULL; /* receive buffer (for a single packet). We use a * it so that we can check available memory in willRun() and request * termination if we can not get it. -- rgerhards, 2007-12-27 */ +#define TIME_REQUERY_DFLT 2 +static int iTimeRequery = TIME_REQUERY_DFLT;/* how often is time to be queried inside tight recv loop? 0=always */ /* config settings */ @@ -126,20 +132,126 @@ finalize_it: } +/* This function is a helper to runInput. I have extracted it + * from the main loop just so that we do not have that large amount of code + * in a single place. This function takes a socket and pulls messages from + * it until the socket does not have any more waiting. + * rgerhards, 2008-01-08 + * We try to read from the file descriptor until there + * is no more data. This is done in the hope to get better performance + * out of the system. However, this also means that a descriptor + * monopolizes processing while it contains data. This can lead to + * data loss in other descriptors. However, if the system is incapable of + * handling the workload, we will loss data in any case. So it doesn't really + * matter where the actual loss occurs - it is always random, because we depend + * on scheduling order. -- rgerhards, 2008-10-02 + */ +static inline rsRetVal +processSocket(int fd, struct sockaddr_storage *frominetPrev, int *pbIsPermitted, + uchar *fromHost, uchar *fromHostFQDN, uchar *fromHostIP) +{ + DEFiRet; + int iNbrTimeUsed; + time_t ttGenTime; + struct syslogTime stTime; + socklen_t socklen; + ssize_t lenRcvBuf; + struct sockaddr_storage frominet; + msg_t *pMsg; + char errStr[1024]; + + iNbrTimeUsed = 0; + while(1) { /* loop is terminated if we have a bad receive, done below in the body */ + socklen = sizeof(struct sockaddr_storage); + lenRcvBuf = recvfrom(fd, (char*) pRcvBuf, iMaxLine, 0, (struct sockaddr *)&frominet, &socklen); + if(lenRcvBuf < 0) { + if(errno != EINTR && errno != EAGAIN) { + rs_strerror_r(errno, errStr, sizeof(errStr)); + DBGPRINTF("INET socket error: %d = %s.\n", errno, errStr); + errmsg.LogError(errno, NO_ERRCODE, "recvfrom inet"); + } + ABORT_FINALIZE(RS_RET_ERR); + } + + /* if we reach this point, we had a good receive and can process the packet received */ + /* check if we have a different sender than before, if so, we need to query some new values */ + if(memcmp(&frominet, frominetPrev, socklen) != 0) { + CHKiRet(net.cvthname(&frominet, fromHost, fromHostFQDN, fromHostIP)); + memcpy(frominetPrev, &frominet, socklen); /* update cache indicator */ + /* Here we check if a host is permitted to send us + * syslog messages. If it isn't, we do not further + * process the message but log a warning (if we are + * configured to do this). + * rgerhards, 2005-09-26 + */ + *pbIsPermitted = net.isAllowedSender(net.pAllowedSenders_UDP, + (struct sockaddr *)&frominet, (char*)fromHostFQDN); + + if(!*pbIsPermitted) { + DBGPRINTF("%s is not an allowed sender\n", (char*)fromHostFQDN); + if(glbl.GetOption_DisallowWarning) { + errmsg.LogError(0, NO_ERRCODE, "UDP message from disallowed sender %s discarded", + (char*)fromHost); + } + } + } + + DBGPRINTF("recv(%d,%d)/%s,acl:%d,msg:%.80s\n", fd, (int) lenRcvBuf, fromHost, *pbIsPermitted, pRcvBuf); + + if(*pbIsPermitted) { + if((iTimeRequery == 0) || (iNbrTimeUsed++ % iTimeRequery) == 0) { + datetime.getCurrTime(&stTime, &ttGenTime); + } + /* we now create our own message object and submit it to the queue */ + CHKiRet(msgConstructWithTime(&pMsg, &stTime, ttGenTime)); + /* first trim the buffer to what we have actually received */ + CHKmalloc(pMsg->pszRawMsg = malloc(sizeof(uchar)* lenRcvBuf)); + memcpy(pMsg->pszRawMsg, pRcvBuf, lenRcvBuf); + pMsg->iLenRawMsg = lenRcvBuf; + MsgSetInputName(pMsg, "imudp"); + MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY); + pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME; + MsgSetRcvFrom(pMsg, (char*)fromHost); + CHKiRet(MsgSetRcvFromIP(pMsg, fromHostIP)); + CHKiRet(submitMsg(pMsg)); + } + } + + +finalize_it: + RETiRet; +} + + /* This function is called to gather input. + * Note that udpLstnSocks 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 + * rgerhards, 2008-10-07: I have implemented a very simple, yet in most cases probably + * highly efficient "name caching". Before querying a name, I now check if the name to be + * queried is the same as the one queried in the last message processed. If that is the + * case, we can simple re-use the previous value. This algorithm works quite well with + * few sender, especially if they emit messages in bursts. The more sender and the + * more intermixed messages arrive, the less this algorithm works, but the overhead + * is so minimal (a simple memory compare and move) that this does not hurt. Even + * with a real name lookup cache, this optimization here is useful as it is quicker + * than even a cache lookup). */ BEGINrunInput int maxfds; int nfds; int i; fd_set readfds; - struct sockaddr_storage frominet; - socklen_t socklen; + struct sockaddr_storage frominetPrev; + int bIsPermitted; uchar fromHost[NI_MAXHOST]; uchar fromHostIP[NI_MAXHOST]; uchar fromHostFQDN[NI_MAXHOST]; - ssize_t l; CODESTARTrunInput + /* start "name caching" algo by making sure the previous system indicator + * is invalidated. + */ + bIsPermitted = 0; + memset(&frominetPrev, 0, sizeof(frominetPrev)); /* this is an endless loop - it is terminated when the thread is * signalled to do so. This, however, is handled by the framework, * right into the sleep below. @@ -154,17 +266,14 @@ CODESTARTrunInput maxfds = 0; FD_ZERO (&readfds); - /* Add the UDP listen sockets to the list of read descriptors. - */ - if(udpLstnSocks != NULL) { - for (i = 0; i < *udpLstnSocks; i++) { - if (udpLstnSocks[i+1] != -1) { - if(Debug) - net.debugListenInfo(udpLstnSocks[i+1], "UDP"); - FD_SET(udpLstnSocks[i+1], &readfds); - if(udpLstnSocks[i+1]>maxfds) maxfds=udpLstnSocks[i+1]; - } - } + /* Add the UDP listen sockets to the list of read descriptors. */ + for (i = 0; i < *udpLstnSocks; i++) { + if (udpLstnSocks[i+1] != -1) { + if(Debug) + net.debugListenInfo(udpLstnSocks[i+1], "UDP"); + FD_SET(udpLstnSocks[i+1], &readfds); + if(udpLstnSocks[i+1]>maxfds) maxfds=udpLstnSocks[i+1]; + } } if(Debug) { dbgprintf("--------imUDP calling select, active file descriptors (max %d): ", maxfds); @@ -177,46 +286,14 @@ CODESTARTrunInput /* wait for io to become ready */ nfds = select(maxfds+1, (fd_set *) &readfds, NULL, NULL, NULL); - if(udpLstnSocks != NULL) { - for (i = 0; nfds && i < *udpLstnSocks; i++) { - if (FD_ISSET(udpLstnSocks[i+1], &readfds)) { - socklen = sizeof(frominet); - l = recvfrom(udpLstnSocks[i+1], (char*) pRcvBuf, iMaxLine, 0, - (struct sockaddr *)&frominet, &socklen); - if (l > 0) { - if(net.cvthname(&frominet, fromHost, fromHostFQDN, fromHostIP) == RS_RET_OK) { - dbgprintf("Message from inetd socket: #%d, host: %s\n", - udpLstnSocks[i+1], fromHost); - /* Here we check if a host is permitted to send us - * syslog messages. If it isn't, we do not further - * process the message but log a warning (if we are - * configured to do this). - * rgerhards, 2005-09-26 - */ - if(net.isAllowedSender(net.pAllowedSenders_UDP, - (struct sockaddr *)&frominet, (char*)fromHostFQDN)) { - parseAndSubmitMessage(fromHost, fromHostIP, pRcvBuf, l, - MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_NO_DELAY, (uchar*)"imudp"); - } else { - dbgprintf("%s is not an allowed sender\n", (char*)fromHostFQDN); - if(glbl.GetOption_DisallowWarning) { - errmsg.LogError(0, NO_ERRCODE, "UDP message from disallowed sender %s discarded", - (char*)fromHost); - } - } - } - } else if (l < 0 && errno != EINTR && errno != EAGAIN) { - char errStr[1024]; - rs_strerror_r(errno, errStr, sizeof(errStr)); - dbgprintf("INET socket error: %d = %s.\n", errno, errStr); - errmsg.LogError(errno, NO_ERRCODE, "recvfrom inet"); - /* should be harmless */ - sleep(1); - } - --nfds; /* indicate we have processed one */ - } - } - } + for (i = 0; nfds && i < *udpLstnSocks; i++) { + if (FD_ISSET(udpLstnSocks[i+1], &readfds)) { + processSocket(udpLstnSocks[i+1], &frominetPrev, &bIsPermitted, + fromHost, fromHostFQDN, fromHostIP); + } + --nfds; /* indicate we have processed one descriptor */ + } + /* end of a run, back to loop for next recv() */ } return iRet; @@ -264,6 +341,7 @@ CODESTARTmodExit /* release what we no longer need */ objRelease(errmsg, CORE_COMPONENT); objRelease(glbl, CORE_COMPONENT); + objRelease(datetime, CORE_COMPONENT); objRelease(net, LM_NET_FILENAME); ENDmodExit @@ -283,6 +361,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a net.closeUDPListenSockets(udpLstnSocks); udpLstnSocks = NULL; } + iTimeRequery = TIME_REQUERY_DFLT;/* the default is to query only every second time */ return RS_RET_OK; } @@ -293,6 +372,7 @@ CODESTARTmodInit CODEmodInit_QueryRegCFSLineHdlr CHKiRet(objUse(errmsg, CORE_COMPONENT)); CHKiRet(objUse(glbl, CORE_COMPONENT)); + CHKiRet(objUse(datetime, CORE_COMPONENT)); CHKiRet(objUse(net, LM_NET_FILENAME)); /* register config file handlers */ @@ -300,9 +380,10 @@ CODEmodInit_QueryRegCFSLineHdlr addListner, NULL, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"udpserveraddress", 0, eCmdHdlrGetWord, NULL, &pszBindAddr, STD_LOADABLE_MODULE_ID)); + CHKiRet(omsdRegCFSLineHdlr((uchar *)"udpservertimerequery", 0, eCmdHdlrInt, + NULL, &iTimeRequery, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID)); ENDmodInit -/* - * vi:set ai: +/* vim:set ai: */ diff --git a/plugins/imuxsock/imuxsock.c b/plugins/imuxsock/imuxsock.c index 55b8b2df..1d88a2b5 100644 --- a/plugins/imuxsock/imuxsock.c +++ b/plugins/imuxsock/imuxsock.c @@ -42,6 +42,7 @@ #include "errmsg.h" #include "net.h" #include "glbl.h" +#include "msg.h" MODULE_TYPE_INPUT @@ -221,7 +222,8 @@ static rsRetVal readSocket(int fd, int iSock) if (iRcvd > 0) { parseAndSubmitMessage(funixHName[iSock] == NULL ? glbl.GetLocalHostName() : funixHName[iSock], (uchar*)"127.0.0.1", pRcv, - iRcvd, funixParseHost[iSock], funixFlags[iSock], funixFlowCtl[iSock], (uchar*)"imuxsock"); + iRcvd, funixParseHost[iSock] ? (funixFlags[iSock] | PARSE_HOSTNAME) : funixFlags[iSock], + funixFlowCtl[iSock], (uchar*)"imuxsock", NULL, 0); } else if (iRcvd < 0 && errno != EINTR) { char errStr[1024]; rs_strerror_r(errno, errStr, sizeof(errStr)); diff --git a/runtime/Makefile.am b/runtime/Makefile.am index 8dd8ad12..5e5d08ea 100644 --- a/runtime/Makefile.am +++ b/runtime/Makefile.am @@ -16,6 +16,8 @@ librsyslog_la_SOURCES = \ glbl.c \ conf.c \ conf.h \ + parser.h \ + parser.c \ msg.c \ msg.h \ linkedlist.c \ diff --git a/runtime/atomic.h b/runtime/atomic.h index 2dbe7f52..7ad8e2e4 100644 --- a/runtime/atomic.h +++ b/runtime/atomic.h @@ -42,12 +42,14 @@ */ #ifdef HAVE_ATOMIC_BUILTINS # define ATOMIC_INC(data) ((void) __sync_fetch_and_add(&(data), 1)) +# define ATOMIC_DEC(data) ((void) __sync_sub_and_fetch(&(data), 1)) # define ATOMIC_DEC_AND_FETCH(data) __sync_sub_and_fetch(&(data), 1) # define ATOMIC_FETCH_32BIT(data) ((unsigned) __sync_fetch_and_and(&(data), 0xffffffff)) # define ATOMIC_STORE_1_TO_32BIT(data) __sync_lock_test_and_set(&(data), 1) #else # warning "atomic builtins not available, using nul operations" # define ATOMIC_INC(data) (++(data)) +# define ATOMIC_DEC(data) (--(data)) # define ATOMIC_DEC_AND_FETCH(data) (--(data)) # define ATOMIC_FETCH_32BIT(data) (data) # define ATOMIC_STORE_1_TO_32BIT(data) (data) = 1 diff --git a/runtime/datetime.c b/runtime/datetime.c index 20ca6191..aa1956d7 100644 --- a/runtime/datetime.c +++ b/runtime/datetime.c @@ -62,9 +62,14 @@ DEFobjCurrIf(errmsg) * most portable and removes the need for additional structures * (but I have to admit it is somewhat "bulky";)). * - * Obviously, all caller-provided pointers must not be NULL... + * Obviously, *t must not be NULL... + * + * rgerhards, 2008-10-07: added ttSeconds to provide a way to + * obtain the second-resolution UNIX timestamp. This is needed + * in some situations to minimize time() calls (namely when doing + * output processing). This can be left NULL if not needed. */ -static void getCurrTime(struct syslogTime *t) +static void getCurrTime(struct syslogTime *t, time_t *ttSeconds) { struct timeval tp; struct tm *tm; @@ -83,6 +88,9 @@ static void getCurrTime(struct syslogTime *t) # else gettimeofday(&tp, NULL); # endif + if(ttSeconds != NULL) + *ttSeconds = tp.tv_sec; + tm = localtime_r((time_t*) &(tp.tv_sec), &tmBuf); t->year = tm->tm_year + 1900; diff --git a/runtime/datetime.h b/runtime/datetime.h index 2210af02..0739588d 100644 --- a/runtime/datetime.h +++ b/runtime/datetime.h @@ -35,7 +35,7 @@ typedef struct datetime_s { /* interfaces */ BEGINinterface(datetime) /* name must also be changed in ENDinterface macro! */ - void (*getCurrTime)(struct syslogTime *t); + void (*getCurrTime)(struct syslogTime *t, time_t *ttSeconds); rsRetVal (*ParseTIMESTAMP3339)(struct syslogTime *pTime, char** ppszTS); rsRetVal (*ParseTIMESTAMP3164)(struct syslogTime *pTime, char** pszTS); int (*formatTimestampToMySQL)(struct syslogTime *ts, char* pDst, size_t iLenDst); diff --git a/runtime/debug.c b/runtime/debug.c index b0bf76ea..102d5d0f 100644 --- a/runtime/debug.c +++ b/runtime/debug.c @@ -1,3 +1,4 @@ +#include <sys/syscall.h> /* debug.c * * This file proides debug and run time error analysis support. Some of the @@ -817,7 +818,9 @@ dbgprint(obj_t *pObj, char *pszMsg, size_t lenMsg) if(stddbg != -1) write(stddbg, pszWriteBuf, lenWriteBuf); if(altdbg != -1) write(altdbg, pszWriteBuf, lenWriteBuf); } - lenWriteBuf = snprintf(pszWriteBuf, sizeof(pszWriteBuf), "%s: ", pszThrdName); + + // old, reenable TODO lenWriteBuf = snprintf(pszWriteBuf, sizeof(pszWriteBuf), "%s: ", pszThrdName); + lenWriteBuf = snprintf(pszWriteBuf, sizeof(pszWriteBuf), "{%ld}%s: ", (long) syscall(SYS_gettid), pszThrdName); if(stddbg != -1) write(stddbg, pszWriteBuf, lenWriteBuf); if(altdbg != -1) write(altdbg, pszWriteBuf, lenWriteBuf); /* print object name header if we have an object */ diff --git a/runtime/debug.h b/runtime/debug.h index d9d576b5..59b3e776 100644 --- a/runtime/debug.h +++ b/runtime/debug.h @@ -100,6 +100,8 @@ void dbgSetThrdName(uchar *pszName); void dbgPrintAllDebugInfo(void); /* macros */ +#define DBGPRINTF(...) if(Debug) { dbgprintf(__VA_ARGS__); } +#define DBGOPRINT(...) if(Debug) { dbgoprint(__VA_ARGS__); } #ifdef RTINST # define BEGINfunc static dbgFuncDB_t *pdbgFuncDB; int dbgCALLStaCK_POP_POINT = dbgEntrFunc(&pdbgFuncDB, __FILE__, __func__, __LINE__); # define ENDfunc dbgExitFunc(pdbgFuncDB, dbgCALLStaCK_POP_POINT, RS_RET_NO_IRET); diff --git a/runtime/modules.h b/runtime/modules.h index 7d34bcf7..96016082 100644 --- a/runtime/modules.h +++ b/runtime/modules.h @@ -44,8 +44,11 @@ * rgerhards, 2008-03-04 * version 3 adds modInfo_t ptr to call of modInit -- rgerhards, 2008-03-10 * version 4 removes needUDPSocket OM callback -- rgerhards, 2008-03-22 + * version 5 changes the way parsing works for input modules. This is + * an important change, parseAndSubmitMessage() goes away. Other + * module types are not affected. -- rgerhards, 2008-10-09 */ -#define CURR_MOD_IF_VERSION 4 +#define CURR_MOD_IF_VERSION 5 typedef enum eModType_ { eMOD_IN, /* input module */ diff --git a/runtime/msg.c b/runtime/msg.c index 9c2e3f17..c030fa45 100644 --- a/runtime/msg.c +++ b/runtime/msg.c @@ -140,8 +140,8 @@ void (*funcMsgPrepareEnqueue)(msg_t *pMsg); #define MsgLock(pMsg) funcLock(pMsg) #define MsgUnlock(pMsg) funcUnlock(pMsg) #else -#define MsgLock(pMsg) {dbgprintf("line %d\n - ", __LINE__); funcLock(pMsg);; } -#define MsgUnlock(pMsg) {dbgprintf("line %d - ", __LINE__); funcUnlock(pMsg); } +#define MsgLock(pMsg) {dbgprintf("MsgLock line %d\n - ", __LINE__); funcLock(pMsg);; } +#define MsgUnlock(pMsg) {dbgprintf("MsgUnlock line %d - ", __LINE__); funcUnlock(pMsg); } #endif /* the next function is a dummy to be used by the looking functions @@ -239,12 +239,21 @@ rsRetVal MsgEnableThreadSafety(void) /* end locking functions */ -/* "Constructor" for a msg "object". Returns a pointer to +/* This is common code for all Constructors. It is defined in an + * inline'able function so that we can save a function call in the + * actual constructors (otherwise, the msgConstruct would need + * to call msgConstructWithTime(), which would require a + * function call). Now, both can use this inline function. This + * enables us to be optimal, but still have the code just once. * the new object or NULL if no such object could be allocated. * An object constructed via this function should only be destroyed - * via "msgDestruct()". + * via "msgDestruct()". This constructor does not query system time + * itself but rather uses a user-supplied value. This enables the caller + * to do some tricks to save processing time (done, for example, in the + * udp input). + * rgerhards, 2008-10-06 */ -rsRetVal msgConstruct(msg_t **ppThis) +static inline rsRetVal msgBaseConstruct(msg_t **ppThis) { DEFiRet; msg_t *pM; @@ -257,21 +266,59 @@ rsRetVal msgConstruct(msg_t **ppThis) pM->iRefCount = 1; pM->iSeverity = -1; pM->iFacility = -1; + objConstructSetObjInfo(pM); + + /* DEV debugging only! dbgprintf("msgConstruct\t0x%x, ref 1\n", (int)pM);*/ + + *ppThis = pM; + +finalize_it: + RETiRet; +} + + +/* "Constructor" for a msg "object". Returns a pointer to + * the new object or NULL if no such object could be allocated. + * An object constructed via this function should only be destroyed + * via "msgDestruct()". This constructor does not query system time + * itself but rather uses a user-supplied value. This enables the caller + * to do some tricks to save processing time (done, for example, in the + * udp input). + * rgerhards, 2008-10-06 + */ +rsRetVal msgConstructWithTime(msg_t **ppThis, struct syslogTime *stTime, time_t ttGenTime) +{ + DEFiRet; + + CHKiRet(msgBaseConstruct(ppThis)); + (*ppThis)->ttGenTime = ttGenTime; + memcpy(&(*ppThis)->tRcvdAt, stTime, sizeof(struct syslogTime)); + memcpy(&(*ppThis)->tTIMESTAMP, stTime, sizeof(struct syslogTime)); + +finalize_it: + RETiRet; +} + + +/* "Constructor" for a msg "object". Returns a pointer to + * the new object or NULL if no such object could be allocated. + * An object constructed via this function should only be destroyed + * via "msgDestruct()". This constructor, for historical reasons, + * also sets the two timestamps to the current time. + */ +rsRetVal msgConstruct(msg_t **ppThis) +{ + DEFiRet; + CHKiRet(msgBaseConstruct(ppThis)); /* we initialize both timestamps to contain the current time, so that they * are consistent. Also, this saves us from doing any further time calls just * to obtain a timestamp. The memcpy() should not really make a difference, * especially as I think there is no codepath currently where it would not be * required (after I have cleaned up the pathes ;)). -- rgerhards, 2008-10-02 */ - datetime.getCurrTime(&(pM->tRcvdAt)); - memcpy(&pM->tTIMESTAMP, &pM->tRcvdAt, sizeof(struct syslogTime)); - - objConstructSetObjInfo(pM); - - /* DEV debugging only! dbgprintf("msgConstruct\t0x%x, ref 1\n", (int)pM);*/ - - *ppThis = pM; + datetime.getCurrTime(&((*ppThis)->tRcvdAt), &((*ppThis)->ttGenTime)); + memcpy(&(*ppThis)->tTIMESTAMP, &(*ppThis)->tRcvdAt, sizeof(struct syslogTime)); finalize_it: RETiRet; @@ -396,7 +443,7 @@ msg_t* MsgDup(msg_t* pOld) assert(pOld != NULL); BEGINfunc - if(msgConstruct(&pNew) != RS_RET_OK) { + if(msgConstructWithTime(&pNew, &pOld->tTIMESTAMP, pOld->ttGenTime) != RS_RET_OK) { return NULL; } @@ -407,8 +454,7 @@ msg_t* MsgDup(msg_t* pOld) pNew->bParseHOSTNAME = pOld->bParseHOSTNAME; pNew->msgFlags = pOld->msgFlags; pNew->iProtocolVersion = pOld->iProtocolVersion; - memcpy(&pNew->tRcvdAt, &pOld->tRcvdAt, sizeof(struct syslogTime)); - memcpy(&pNew->tTIMESTAMP, &pOld->tTIMESTAMP, sizeof(struct syslogTime)); + pNew->ttGenTime = pOld->ttGenTime; tmpCOPYSZ(Severity); tmpCOPYSZ(SeverityStr); tmpCOPYSZ(Facility); @@ -462,6 +508,7 @@ static rsRetVal MsgSerialize(msg_t *pThis, strm_t *pStrm) objSerializeSCALAR(pStrm, iSeverity, SHORT); objSerializeSCALAR(pStrm, iFacility, SHORT); objSerializeSCALAR(pStrm, msgFlags, INT); + objSerializeSCALAR(pStrm, ttGenTime, INT); objSerializeSCALAR(pStrm, tRcvdAt, SYSLOGTIME); objSerializeSCALAR(pStrm, tTIMESTAMP, SYSLOGTIME); @@ -731,6 +778,7 @@ int getPRIi(msg_t *pM) char *getTimeReported(msg_t *pM, enum tplFormatTypes eFmt) { + BEGINfunc if(pM == NULL) return ""; @@ -802,11 +850,13 @@ char *getTimeReported(msg_t *pM, enum tplFormatTypes eFmt) MsgUnlock(pM); return(pM->pszTIMESTAMP_SecFrac); } + ENDfunc return "INVALID eFmt OPTION!"; } char *getTimeGenerated(msg_t *pM, enum tplFormatTypes eFmt) { + BEGINfunc if(pM == NULL) return ""; @@ -878,6 +928,7 @@ char *getTimeGenerated(msg_t *pM, enum tplFormatTypes eFmt) MsgUnlock(pM); return(pM->pszRcvdAt_SecFrac); } + ENDfunc return "INVALID eFmt OPTION!"; } @@ -1623,7 +1674,7 @@ static uchar *getNOW(eNOWType eNow) return NULL; } - datetime.getCurrTime(&t); + datetime.getCurrTime(&t, NULL); switch(eNow) { case NOW_NOW: snprintf((char*) pBuf, tmpBUFSIZE, "%4.4d-%2.2d-%2.2d", t.year, t.month, t.day); @@ -2431,6 +2482,8 @@ rsRetVal MsgSetProperty(msg_t *pThis, var_t *pProp) MsgSetPROCID(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); } else if(isProp("pCSMSGID")) { MsgSetMSGID(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); + } else if(isProp("ttGenTime")) { + pThis->ttGenTime = pProp->val.num; } else if(isProp("tRcvdAt")) { memcpy(&pThis->tRcvdAt, &pProp->val.vSyslogTime, sizeof(struct syslogTime)); } else if(isProp("tTIMESTAMP")) { @@ -2492,7 +2545,5 @@ BEGINObjClassInit(msg, 1, OBJ_IS_CORE_MODULE) funcDeleteMutex = MsgLockingDummy; funcMsgPrepareEnqueue = MsgLockingDummy; ENDObjClassInit(msg) - -/* - * vi:set ai: +/* vim:set ai: */ diff --git a/runtime/msg.h b/runtime/msg.h index 21cb2c64..d98111a8 100644 --- a/runtime/msg.h +++ b/runtime/msg.h @@ -52,7 +52,9 @@ struct msg { BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ pthread_mutexattr_t mutAttr; pthread_mutex_t mut; - int iRefCount; /* reference counter (0 = unused) */ + flowControl_t flowCtlType; /**< type of flow control we can apply, for enqueueing, needs not to be persisted because + once data has entered the queue, this property is no longer needed. */ + short iRefCount; /* reference counter (0 = unused) */ short bParseHOSTNAME; /* should the hostname be parsed from the message? */ /* background: the hostname is not present on "regular" messages * received via UNIX domain sockets from the same machine. However, @@ -60,8 +62,6 @@ struct msg { * sockets. All in all, the parser would need parse templates, that would * resolve all these issues... rgerhards, 2005-10-06 */ - flowControl_t flowCtlType; /**< type of flow control we can apply, for enqueueing, needs not to be persisted because - once data has entered the queue, this property is no longer needed. */ short iSeverity; /* the severity 0..7 */ uchar *pszSeverity; /* severity as string... */ int iLenSeverity; /* ... and its length. */ @@ -99,6 +99,13 @@ struct msg { cstr_t *pCSAPPNAME; /* APP-NAME */ cstr_t *pCSPROCID; /* PROCID */ cstr_t *pCSMSGID; /* MSGID */ + time_t ttGenTime; /* time msg object was generated, same as tRcvdAt, but a Unix timestamp. + While this field looks redundant, it is required because a Unix timestamp + is used at later processing stages (namely in the output arena). Thanks to + the subleties of how time is defined, there is no reliable way to reconstruct + the Unix timestamp from the syslogTime fields (in practice, we may be close + enough to reliable, but I prefer to leave the subtle things to the OS, where + it obviously is solved in way or another...). */ struct syslogTime tRcvdAt;/* time the message entered this program */ char *pszRcvdAt3164; /* time as RFC3164 formatted string (always 15 charcters) */ char *pszRcvdAt3339; /* time as RFC3164 formatted string (32 charcters at most) */ @@ -114,11 +121,24 @@ struct msg { int msgFlags; /* flags associated with this message */ }; + +/* message flags (msgFlags), not an enum for historical reasons + */ +#define NOFLAG 0x000 /* no flag is set (to be used when a flag must be specified and none is required) */ +#define INTERNAL_MSG 0x001 /* msg generated by logmsgInternal() --> special handling */ +/* 0x002 not used because it was previously a known value - rgerhards, 2008-10-09 */ +#define IGNDATE 0x004 /* ignore, if given, date in message and use date of reception as msg date */ +#define MARK 0x008 /* this message is a mark */ +#define NEEDS_PARSING 0x010 /* raw message, must be parsed before processing can be done */ +#define PARSE_HOSTNAME 0x020 /* parse the hostname during message parsing */ + + /* function prototypes */ PROTOTYPEObjClassInit(msg); char* getProgramName(msg_t*); rsRetVal msgConstruct(msg_t **ppThis); +rsRetVal msgConstructWithTime(msg_t **ppThis, struct syslogTime *stTime, time_t ttGenTime); rsRetVal msgDestruct(msg_t **ppM); msg_t* MsgDup(msg_t* pOld); msg_t *MsgAddRef(msg_t *pM); @@ -180,6 +200,5 @@ extern void (*funcMsgPrepareEnqueue)(msg_t *pMsg); #define MsgPrepareEnqueue(pMsg) funcMsgPrepareEnqueue(pMsg) #endif /* #ifndef MSG_H_INCLUDED */ -/* - * vi:set ai: +/* vim:set ai: */ diff --git a/runtime/parser.c b/runtime/parser.c new file mode 100644 index 00000000..fbdeebeb --- /dev/null +++ b/runtime/parser.c @@ -0,0 +1,313 @@ +/* parser.c + * This module contains functions for message parsers. It still needs to be + * converted into an object (and much extended). + * + * Module begun 2008-10-09 by Rainer Gerhards (based on previous code from syslogd.c) + * + * Copyright 2008 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#include "config.h" +#include <stdlib.h> +#include <ctype.h> +#include <assert.h> +#ifdef USE_NETZIP +#include <zlib.h> +#endif + +#include "rsyslog.h" +#include "dirty.h" +#include "msg.h" +#include "obj.h" +#include "errmsg.h" + +/* some defines */ +#define DEFUPRI (LOG_USER|LOG_NOTICE) + +/* definitions for objects we access */ +DEFobjStaticHelpers +DEFobjCurrIf(glbl) +DEFobjCurrIf(errmsg) + +/* static data */ + + +/* this is a dummy class init + */ +rsRetVal parserClassInit(void) +{ + DEFiRet; + + /* request objects we use */ + CHKiRet(objGetObjInterface(&obj)); /* this provides the root pointer for all other queries */ + CHKiRet(objUse(glbl, CORE_COMPONENT)); + CHKiRet(objUse(errmsg, CORE_COMPONENT)); +// TODO: free components! see action.c +finalize_it: + RETiRet; +} + + +/* uncompress a received message if it is compressed. + * pMsg->pszRawMsg buffer is updated. + * rgerhards, 2008-10-09 + */ +static inline rsRetVal uncompressMessage(msg_t *pMsg) +{ + DEFiRet; +# ifdef USE_NETZIP + uchar *deflateBuf = NULL; + uLongf iLenDefBuf; + uchar *pszMsg; + size_t lenMsg; + + assert(pMsg != NULL); + pszMsg = pMsg->pszRawMsg; + lenMsg = pMsg->iLenRawMsg; + + /* we first need to check if we have a compressed record. If so, + * we must decompress it. + */ + if(lenMsg > 0 && *pszMsg == 'z') { /* compressed data present? (do NOT change order if conditions!) */ + /* we have compressed data, so let's deflate it. We support a maximum + * message size of iMaxLine. If it is larger, an error message is logged + * and the message is dropped. We do NOT try to decompress larger messages + * as such might be used for denial of service. It might happen to later + * builds that such functionality be added as an optional, operator-configurable + * feature. + */ + int ret; + iLenDefBuf = glbl.GetMaxLine(); + CHKmalloc(deflateBuf = malloc(sizeof(uchar) * (iLenDefBuf + 1))); + ret = uncompress((uchar *) deflateBuf, &iLenDefBuf, (uchar *) pszMsg+1, lenMsg-1); + DBGPRINTF("Compressed message uncompressed with status %d, length: new %ld, old %d.\n", + ret, (long) iLenDefBuf, (int) (lenMsg-1)); + /* Now check if the uncompression worked. If not, there is not much we can do. In + * that case, we log an error message but ignore the message itself. Storing the + * compressed text is dangerous, as it contains control characters. So we do + * not do this. If someone would like to have a copy, this code here could be + * modified to do a hex-dump of the buffer in question. We do not include + * this functionality right now. + * rgerhards, 2006-12-07 + */ + if(ret != Z_OK) { + errmsg.LogError(0, NO_ERRCODE, "Uncompression of a message failed with return code %d " + "- enable debug logging if you need further information. " + "Message ignored.", ret); + FINALIZE; /* unconditional exit, nothing left to do... */ + } + free(pMsg->pszRawMsg); + pMsg->pszRawMsg = deflateBuf; + pMsg->iLenRawMsg = iLenDefBuf; + deflateBuf = NULL; /* logically "freed" - caller is now responsible */ + } +finalize_it: + if(deflateBuf != NULL) + free(deflateBuf); + +# else /* ifdef USE_NETZIP */ + + /* in this case, we still need to check if the message is compressed. If so, we must + * tell the user we can not accept it. + */ + if(len > 0 && *msg == 'z') { + errmsg.LogError(0, NO_ERRCODE, "Received a compressed message, but rsyslogd does not have compression " + "support enabled. The message will be ignored."); + ABORT_FINALIZE(RS_RET_NO_ZIP); + } + +# endif /* ifdef USE_NETZIP */ + + RETiRet; +} + + +/* sanitize a received message + * if a message gets to large during sanitization, it is truncated. This is + * as specified in the upcoming syslog RFC series. + * rgerhards, 2008-10-09 + * We check if we have a NUL character at the very end of the + * message. This seems to be a frequent problem with a number of senders. + * So I have now decided to drop these NULs. However, if they are intentional, + * that may cause us some problems, e.g. with syslog-sign. On the other hand, + * current code always has problems with intentional NULs (as it needs to escape + * them to prevent problems with the C string libraries), so that does not + * really matter. Just to be on the save side, we'll log destruction of such + * NULs in the debug log. + * rgerhards, 2007-09-14 + */ +static inline rsRetVal +sanitizeMessage(msg_t *pMsg) +{ + DEFiRet; + uchar *pszMsg; + uchar *pDst; /* destination for copy job */ + size_t lenMsg; + size_t iSrc; + size_t iDst; + size_t iMaxLine; + + assert(pMsg != NULL); + +# ifdef USE_NETZIP + CHKiRet(uncompressMessage(pMsg)); +# endif + + pszMsg = pMsg->pszRawMsg; + lenMsg = pMsg->iLenRawMsg; + + /* remove NUL character at end of message (see comment in function header) */ + if(pszMsg[lenMsg-1] == '\0') { + DBGPRINTF("dropped NUL at very end of message\n"); + lenMsg--; + } + + /* then we check if we need to drop trailing LFs, which often make + * their way into syslog messages unintentionally. In order to remain + * compatible to recent IETF developments, we allow the user to + * turn on/off this handling. rgerhards, 2007-07-23 + */ + if(bDropTrailingLF && pszMsg[lenMsg-1] == '\n') { + DBGPRINTF("dropped LF at very end of message (DropTrailingLF is set)\n"); + lenMsg--; + } + + /* now copy over the message and sanitize it */ + /* TODO: can we get cheaper memory alloc? {alloca()?}*/ + iMaxLine = glbl.GetMaxLine(); + CHKmalloc(pDst = malloc(sizeof(uchar) * (iMaxLine + 1))); + iSrc = iDst = 0; + while(iSrc < lenMsg && iDst < iMaxLine) { + if(pszMsg[iSrc] == '\0') { /* guard against \0 characters... */ + /* changed to the sequence (somewhat) proposed in + * draft-ietf-syslog-protocol-19. rgerhards, 2006-11-30 + */ + if(iDst + 3 < iMaxLine) { /* do we have space? */ + pDst[iDst++] = cCCEscapeChar; + pDst[iDst++] = '0'; + pDst[iDst++] = '0'; + pDst[iDst++] = '0'; + } /* if we do not have space, we simply ignore the '\0'... */ + /* log an error? Very questionable... rgerhards, 2006-11-30 */ + /* decided: we do not log an error, it won't help... rger, 2007-06-21 */ + } else if(bEscapeCCOnRcv && iscntrl((int) pszMsg[iSrc])) { + /* we are configured to escape control characters. Please note + * that this most probably break non-western character sets like + * Japanese, Korean or Chinese. rgerhards, 2007-07-17 + * Note: sysklogd logs octal values only for DEL and CCs above 127. + * For others, it logs ^n where n is the control char converted to an + * alphabet character. We like consistency and thus escape it to octal + * in all cases. If someone complains, we may change the mode. At least + * we known now what's going on. + * rgerhards, 2007-07-17 + */ + if(iDst + 3 < iMaxLine) { /* do we have space? */ + pDst[iDst++] = cCCEscapeChar; + pDst[iDst++] = '0' + ((pszMsg[iSrc] & 0300) >> 6); + pDst[iDst++] = '0' + ((pszMsg[iSrc] & 0070) >> 3); + pDst[iDst++] = '0' + ((pszMsg[iSrc] & 0007)); + } /* again, if we do not have space, we ignore the char - see comment at '\0' */ + } else { + pDst[iDst++] = pszMsg[iSrc]; + } + ++iSrc; + } + pDst[iDst] = '\0'; /* space *is* reserved for this! */ + + /* we have a sanitized string. Let's save it now */ + free(pMsg->pszRawMsg); + if((pMsg->pszRawMsg = malloc((iDst+1) * sizeof(uchar))) == NULL) { + /* when we get no new buffer, we use what we already have ;) */ + pMsg->pszRawMsg = pDst; + } else { + /* trim buffer */ + memcpy(pMsg->pszRawMsg, pDst, iDst+1); + free(pDst); /* too big! */ + pMsg->iLenRawMsg = iDst; + } + +finalize_it: + RETiRet; +} + +/* Parse a received message. The object's rawmsg property is taken and + * parsed according to the relevant standards. This can later be + * extended to support configured parsers. + * rgerhards, 2008-10-09 + */ +rsRetVal parseMsg(msg_t *pMsg) +{ + DEFiRet; + uchar *msg; + int pri; + + CHKiRet(sanitizeMessage(pMsg)); + + /* we needed to sanitize first, because we otherwise do not have a C-string we can print... */ + DBGPRINTF("msg parser: flags %x, from '%s', msg %s\n", pMsg->msgFlags, pMsg->pszRcvFrom, pMsg->pszRawMsg); + + /* pull PRI */ + pri = DEFUPRI; + msg = pMsg->pszRawMsg; + if(*msg == '<') { + pri = 0; + while(isdigit((int) *++msg)) { + pri = 10 * pri + (*msg - '0'); + } + if(*msg == '>') + ++msg; + if(pri & ~(LOG_FACMASK|LOG_PRIMASK)) + pri = DEFUPRI; + } + pMsg->iFacility = LOG_FAC(pri); + pMsg->iSeverity = LOG_PRI(pri); + MsgSetUxTradMsg(pMsg, (char*) msg); + + if(pMsg->bParseHOSTNAME == 0) + MsgSetHOSTNAME(pMsg, (char*) pMsg->pszRcvFrom); + + /* rger 2005-11-24 (happy thanksgiving!): we now need to check if we have + * a traditional syslog message or one formatted according to syslog-protocol. + * We need to apply different parsers depending on that. We use the + * -protocol VERSION field for the detection. + */ + if(msg[0] == '1' && msg[1] == ' ') { + dbgprintf("Message has syslog-protocol format.\n"); + setProtocolVersion(pMsg, 1); + if(parseRFCSyslogMsg(pMsg, pMsg->msgFlags) == 1) { // TODO: parseRFC... should pull flags from pMsg + msgDestruct(&pMsg); + ABORT_FINALIZE(RS_RET_ERR); // TODO: we need to handle these cases! + } + } else { /* we have legacy syslog */ + dbgprintf("Message has legacy syslog format.\n"); + setProtocolVersion(pMsg, 0); + if(parseLegacySyslogMsg(pMsg, pMsg->msgFlags) == 1) { + msgDestruct(&pMsg); + ABORT_FINALIZE(RS_RET_ERR); // TODO: we need to handle these cases! + } + } + + /* finalize message object */ + pMsg->msgFlags &= ~NEEDS_PARSING; /* this message is now parsed */ + MsgPrepareEnqueue(pMsg); /* "historical" name - preparese for multi-threading */ + +finalize_it: + RETiRet; +} diff --git a/runtime/parser.h b/runtime/parser.h new file mode 100644 index 00000000..cec9c083 --- /dev/null +++ b/runtime/parser.h @@ -0,0 +1,30 @@ +/* header for parser.c + * This is not yet an object, but contains all those code necessary to + * parse syslog messages. + * + * Copyright 2008 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#ifndef INCLUDED_PARSE_H +#define INCLUDED_PARSE_H + +extern rsRetVal parserClassInit(void); +extern rsRetVal parseMsg(msg_t*); + +#endif /* #ifndef INCLUDED_PARSE_H */ diff --git a/runtime/queue.c b/runtime/queue.c index 93b33967..7fa2df91 100644 --- a/runtime/queue.c +++ b/runtime/queue.c @@ -49,6 +49,7 @@ #include "obj.h" #include "wtp.h" #include "wti.h" +#include "atomic.h" /* static data */ DEFobjStaticHelpers @@ -1015,7 +1016,7 @@ queueAdd(queue_t *pThis, void *pUsr) CHKiRet(pThis->qAdd(pThis, pUsr)); if(pThis->qType != QUEUETYPE_DIRECT) { - ++pThis->iQueueSize; + ATOMIC_INC(pThis->iQueueSize); dbgoprint((obj_t*) pThis, "entry added, size now %d entries\n", pThis->iQueueSize); } @@ -1044,7 +1045,7 @@ queueDel(queue_t *pThis, void *pUsr) iRet = queueGetUngottenObj(pThis, (obj_t**) pUsr); } else { iRet = pThis->qDel(pThis, pUsr); - --pThis->iQueueSize; + ATOMIC_DEC(pThis->iQueueSize); } dbgoprint((obj_t*) pThis, "entry deleted, state %d, size now %d entries\n", @@ -1532,8 +1533,6 @@ queueRateLimiter(queue_t *pThis) ISOBJ_TYPE_assert(pThis, queue); - dbgoprint((obj_t*) pThis, "entering rate limiter\n"); - iDelay = 0; if(pThis->iDeqtWinToHr != 25) { /* 25 means disabled */ /* time calls are expensive, so only do them when needed */ @@ -2120,6 +2119,15 @@ queueEnqObj(queue_t *pThis, flowControl_t flowCtlType, void *pUsr) ISOBJ_TYPE_assert(pThis, queue); + /* first check if we need to discard this message (which will cause CHKiRet() to exit) + * rgerhards, 2008-10-07: It is OK to do this outside of mutex protection. The iQueueSize + * and bRunsDA parameters may not reflect the correct settings here, but they are + * "good enough" in the sense that they can be used to drive the decision. Valgrind's + * threading tools may point this access to be an error, but this is done + * intentional. I do not see this causes problems to us. + */ + CHKiRet(queueChkDiscardMsg(pThis, pThis->iQueueSize, pThis->bRunsDA, pUsr)); + /* Please note that this function is not cancel-safe and consequently * sets the calling thread's cancelibility state to PTHREAD_CANCEL_DISABLE * during its execution. If that is not done, race conditions occur if the @@ -2131,9 +2139,6 @@ queueEnqObj(queue_t *pThis, flowControl_t flowCtlType, void *pUsr) d_pthread_mutex_lock(pThis->mut); } - /* first check if we need to discard this message (which will cause CHKiRet() to exit) */ - CHKiRet(queueChkDiscardMsg(pThis, pThis->iQueueSize, pThis->bRunsDA, pUsr)); - /* then check if we need to add an assistance disk queue */ if(pThis->bIsDA) CHKiRet(queueChkStrtDA(pThis)); @@ -2196,10 +2201,15 @@ finalize_it: if(pThis->qType != QUEUETYPE_DIRECT) { /* make sure at least one worker is running. */ queueAdviseMaxWorkers(pThis); - dbgoprint((obj_t*) pThis, "EnqueueMsg advised worker start\n"); /* and release the mutex */ d_pthread_mutex_unlock(pThis->mut); pthread_setcancelstate(iCancelStateSave, NULL); + dbgoprint((obj_t*) pThis, "EnqueueMsg advised worker start\n"); + /* the following pthread_yield is experimental, but brought us performance + * benefit. For details, please see http://kb.monitorware.com/post14216.html#p14216 + * rgerhards, 2008-10-09 + */ + pthread_yield(); } RETiRet; diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h index 361bfb47..619343bd 100644 --- a/runtime/rsyslog.h +++ b/runtime/rsyslog.h @@ -252,6 +252,7 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth RS_RET_QUEUE_FULL = -2105, /**< queue is full, operation could not be completed */ RS_RET_ACCEPT_ERR = -2106, /**< error during accept() system call */ RS_RET_INVLD_TIME = -2107, /**< invalid timestamp (e.g. could not be parsed) */ + RS_RET_NO_ZIP = -2108, /**< ZIP functionality is not present */ /* RainerScript error messages (range 1000.. 1999) */ RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */ diff --git a/runtime/srutils.c b/runtime/srutils.c index 97cc3252..1280e40d 100644 --- a/runtime/srutils.c +++ b/runtime/srutils.c @@ -371,6 +371,7 @@ int getNumberDigits(long lNum) rsRetVal timeoutComp(struct timespec *pt, long iTimeout) { + BEGINfunc assert(pt != NULL); /* compute timeout */ clock_gettime(CLOCK_REALTIME, pt); @@ -379,6 +380,7 @@ timeoutComp(struct timespec *pt, long iTimeout) pt->tv_nsec -= 1000000000; } pt->tv_sec += iTimeout / 1000; + ENDfunc return RS_RET_OK; /* so far, this is static... */ } @@ -393,6 +395,7 @@ timeoutVal(struct timespec *pt) { struct timespec t; long iTimeout; + BEGINfunc assert(pt != NULL); /* compute timeout */ @@ -403,6 +406,7 @@ timeoutVal(struct timespec *pt) if(iTimeout < 0) iTimeout = 0; + ENDfunc return iTimeout; } diff --git a/runtime/sysvar.c b/runtime/sysvar.c index 5eec8f67..c102d1f5 100644 --- a/runtime/sysvar.c +++ b/runtime/sysvar.c @@ -84,7 +84,7 @@ getNOW(eNOWType eNow, cstr_t **ppStr) uchar szBuf[16]; struct syslogTime t; - datetime.getCurrTime(&t); + datetime.getCurrTime(&t, NULL); switch(eNow) { case NOW_NOW: snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%4.4d-%2.2d-%2.2d", t.year, t.month, t.day); diff --git a/runtime/wti.c b/runtime/wti.c index 365b25d5..4abca090 100644 --- a/runtime/wti.c +++ b/runtime/wti.c @@ -299,7 +299,7 @@ wtiWorkerCancelCleanup(void *arg) pWtp = pThis->pWtp; ISOBJ_TYPE_assert(pWtp, wtp); - dbgprintf("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis)); /* call user supplied handler (that one e.g. requeues the element) */ pWtp->pfOnWorkerCancel(pThis->pWtp->pUsr, pThis->pUsrp); @@ -391,7 +391,7 @@ wtiWorker(wti_t *pThis) /* if we reach this point, we are still protected by the mutex */ if(pWtp->pfIsIdle(pWtp->pUsr, MUTEX_ALREADY_LOCKED)) { - dbgprintf("%s: worker IDLE, waiting for work.\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: worker IDLE, waiting for work.\n", wtiGetDbgHdr(pThis)); pWtp->pfOnIdle(pWtp->pUsr, MUTEX_ALREADY_LOCKED); if(pWtp->toWrkShutdown == -1) { @@ -400,7 +400,7 @@ wtiWorker(wti_t *pThis) } else { timeoutComp(&t, pWtp->toWrkShutdown);/* get absolute timeout */ if(d_pthread_cond_timedwait(pWtp->pcondBusy, pWtp->pmutUsr, &t) != 0) { - dbgprintf("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis)); bInactivityTOOccured = 1; /* indicate we had a timeout */ } } diff --git a/runtime/wtp.c b/runtime/wtp.c index 734c8d57..45034fa7 100644 --- a/runtime/wtp.c +++ b/runtime/wtp.c @@ -456,7 +456,7 @@ wtpStartWrkr(wtp_t *pThis, int bLockMutex) ISOBJ_TYPE_assert(pThis, wtp); - wtpProcessThrdChanges(pThis); + wtpProcessThrdChanges(pThis); // TODO: Performance: this causes a lot of FUTEX calls BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex); diff --git a/tcps_sess.c b/tcps_sess.c index f5420fc0..e8bef5b1 100644 --- a/tcps_sess.c +++ b/tcps_sess.c @@ -42,6 +42,7 @@ #include "obj.h" #include "errmsg.h" #include "netstrm.h" +#include "msg.h" /* static data */ @@ -229,8 +230,8 @@ PrepareClose(tcps_sess_t *pThis) * this case. */ dbgprintf("Extra data at end of stream in legacy syslog/tcp message - processing\n"); - parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg, MSG_PARSE_HOSTNAME, - NOFLAG, eFLOWCTL_LIGHT_DELAY, NULL); /* TODO: add real InputName */ + parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg, + PARSE_HOSTNAME, eFLOWCTL_LIGHT_DELAY, NULL, NULL, 0); /* TODO: add real InputName */ pThis->bAtStrtOfFram = 1; } @@ -313,7 +314,7 @@ processDataRcvd(tcps_sess_t *pThis, char c) /* emergency, we now need to flush, no matter if we are at end of message or not... */ dbgprintf("error: message received is larger than max msg size, we split it\n"); parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg, - MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_LIGHT_DELAY, NULL); /* TODO: add real InputName */ + PARSE_HOSTNAME, eFLOWCTL_LIGHT_DELAY, NULL, NULL, 0); /* TODO: add real InputName */ pThis->iMsg = 0; /* we might think if it is better to ignore the rest of the * message than to treat it as a new one. Maybe this is a good @@ -324,7 +325,7 @@ processDataRcvd(tcps_sess_t *pThis, char c) if(c == '\n' && pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delemiter? */ parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg, - MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_LIGHT_DELAY, NULL); /* TODO: add real InputName */ + PARSE_HOSTNAME, eFLOWCTL_LIGHT_DELAY, NULL, NULL, 0); /* TODO: add real InputName */ pThis->iMsg = 0; pThis->inputState = eAtStrtFram; } else { @@ -343,7 +344,7 @@ processDataRcvd(tcps_sess_t *pThis, char c) if(pThis->iOctetsRemain < 1) { /* we have end of frame! */ parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg, - MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_LIGHT_DELAY, NULL); /* TODO: add real InputName */ + PARSE_HOSTNAME, eFLOWCTL_LIGHT_DELAY, NULL, NULL, 0); /* TODO: add real InputName */ pThis->iMsg = 0; pThis->inputState = eAtStrtFram; } @@ -127,7 +127,7 @@ static void* thrdStarter(void *arg) assert(pThis != NULL); assert(pThis->pUsrThrdMain != NULL); - /* block all signalsi */ + /* block all signals */ sigset_t sigSet; sigfillset(&sigSet); pthread_sigmask(SIG_BLOCK, &sigSet, NULL); diff --git a/tools/Makefile.am b/tools/Makefile.am index a265af9c..b66b8e0c 100644 --- a/tools/Makefile.am +++ b/tools/Makefile.am @@ -26,8 +26,9 @@ rsyslogd_LDADD = $(zlib_libs) $(pthreads_libs) $(rsrt_libs) rsyslogd_LDFLAGS = -export-dynamic if ENABLE_DIAGTOOLS -sbin_PROGRAMS += rsyslog_diag_hostname +sbin_PROGRAMS += rsyslog_diag_hostname msggen rsyslog_diag_hostname_SOURCES = gethostn.c +msggen_SOURCES = msggen.c endif EXTRA_DIST = $(man_MANS) diff --git a/tools/msggen.c b/tools/msggen.c new file mode 100644 index 00000000..7990a3c8 --- /dev/null +++ b/tools/msggen.c @@ -0,0 +1,38 @@ +/* msggen - a small diagnostic utility that does very quick + * syslog() calls. + * + * Copyright 2008 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of rsyslog. + * + * Rsyslog is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * Rsyslog is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with Rsyslog. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + */ + +#include <stdio.h> +#include <syslog.h> + +int main(int argc, char *argv[]) +{ + int i; + + openlog("msggen", 0 , LOG_LOCAL0); + + for(i = 0 ; i < 10 ; ++i) + syslog(LOG_NOTICE, "This is message number %d", i); + + closelog(); + return 0; +} diff --git a/tools/syslogd.c b/tools/syslogd.c index 7a99fc1d..13696955 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -128,6 +128,7 @@ #include "vm.h" #include "errmsg.h" #include "datetime.h" +#include "parser.h" #include "sysvar.h" /* definitions for objects we access */ @@ -249,15 +250,15 @@ typedef struct legacyOptsLL_s { legacyOptsLL_t *pLegacyOptsLL = NULL; /* global variables for config file state */ -static int bDropTrailingLF = 1; /* drop trailing LF's on reception? */ +int bDropTrailingLF = 1; /* drop trailing LF's on reception? */ int iCompatibilityMode = 0; /* version we should be compatible with; 0 means sysklogd. It is the default, so if no -c<n> option is given, we make ourselvs as compatible to sysklogd as possible. */ static int bDebugPrintTemplateList = 1;/* output template list in debug mode? */ static int bDebugPrintCfSysLineHandlerList = 1;/* output cfsyslinehandler list in debug mode? */ static int bDebugPrintModuleList = 1;/* output module list in debug mode? */ -static uchar cCCEscapeChar = '\\';/* character to be used to start an escape sequence for control chars */ -static int bEscapeCCOnRcv = 1; /* escape control characters on reception: 0 - no, 1 - yes */ +uchar cCCEscapeChar = '\\';/* character to be used to start an escape sequence for control chars */ +int bEscapeCCOnRcv = 1; /* escape control characters on reception: 0 - no, 1 - yes */ static int bErrMsgToStderr = 1; /* print error messages to stderr (in addition to everything else)? */ int bReduceRepeatMsgs; /* reduce repeated message - 0 - no, 1 - yes */ int bActExecWhenPrevSusp; /* execute action only when previous one was suspended? */ @@ -581,9 +582,18 @@ void untty(void) * Interface change: added new parameter "InputName", permits the input to provide * a string that identifies it. May be NULL, but must be a valid char* pointer if * non-NULL. + * + * rgerhards, 2008-10-06: + * Interface change: added new parameter "stTime", which enables the caller to provide + * a timestamp that is to be used as timegenerated instead of the current system time. + * This is meant to facilitate performance optimization. Some inputs support such modes. + * If stTime is NULL, the current system time is used. + * + * rgerhards, 2008-10-09: + * interface change: bParseHostname removed, now in flags */ -rsRetVal printline(uchar *hname, uchar *hnameIP, uchar *msg, int bParseHost, int flags, flowControl_t flowCtlType, - uchar *pszInputName) +static inline rsRetVal printline(uchar *hname, uchar *hnameIP, uchar *msg, int flags, flowControl_t flowCtlType, + uchar *pszInputName, struct syslogTime *stTime, time_t ttGenTime) { DEFiRet; register uchar *p; @@ -591,13 +601,16 @@ rsRetVal printline(uchar *hname, uchar *hnameIP, uchar *msg, int bParseHost, int msg_t *pMsg; /* Now it is time to create the message object (rgerhards) */ - CHKiRet(msgConstruct(&pMsg)); + if(stTime == NULL) { + CHKiRet(msgConstruct(&pMsg)); + } else { + CHKiRet(msgConstructWithTime(&pMsg, stTime, ttGenTime)); + } if(pszInputName != NULL) MsgSetInputName(pMsg, (char*) pszInputName); MsgSetFlowControlType(pMsg, flowCtlType); MsgSetRawMsg(pMsg, (char*)msg); - pMsg->bParseHOSTNAME = bParseHost; /* test for special codes */ pri = DEFUPRI; p = msg; @@ -622,7 +635,7 @@ rsRetVal printline(uchar *hname, uchar *hnameIP, uchar *msg, int bParseHost, int * the message was received from (that, for obvious reasons, * being the local host). rgerhards 2004-11-16 */ - if(bParseHost == 0) + if((pMsg->msgFlags & PARSE_HOSTNAME) == 0) MsgSetHOSTNAME(pMsg, (char*)hname); MsgSetRcvFrom(pMsg, (char*)hname); CHKiRet(MsgSetRcvFromIP(pMsg, hnameIP)); @@ -684,10 +697,19 @@ finalize_it: * Interface change: added new parameter "InputName", permits the input to provide * a string that identifies it. May be NULL, but must be a valid char* pointer if * non-NULL. + * + * rgerhards, 2008-10-06: + * Interface change: added new parameter "stTime", which enables the caller to provide + * a timestamp that is to be used as timegenerated instead of the current system time. + * This is meant to facilitate performance optimization. Some inputs support such modes. + * If stTime is NULL, the current system time is used. + * + * rgerhards, 2008-10-09: + * interface change: bParseHostname removed, now in flags */ rsRetVal -parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int bParseHost, int flags, flowControl_t flowCtlType, - uchar *pszInputName) +parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int flags, flowControl_t flowCtlType, + uchar *pszInputName, struct syslogTime *stTime, time_t ttGenTime) { DEFiRet; register int iMsg; @@ -714,9 +736,6 @@ parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int bPa * TODO: optimize buffer handling */ iMaxLine = glbl.GetMaxLine(); CHKmalloc(tmpline = malloc(sizeof(uchar) * (iMaxLine + 1))); -# ifdef USE_NETZIP - CHKmalloc(deflateBuf = malloc(sizeof(uchar) * (iMaxLine + 1))); -# endif /* we first check if we have a NUL character at the very end of the * message. This seems to be a frequent problem with a number of senders. @@ -762,6 +781,7 @@ parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int bPa */ int ret; iLenDefBuf = iMaxLine; + CHKmalloc(deflateBuf = malloc(sizeof(uchar) * (iMaxLine + 1))); ret = uncompress((uchar *) deflateBuf, &iLenDefBuf, (uchar *) msg+1, len-1); dbgprintf("Compressed message uncompressed with status %d, length: new %ld, old %d.\n", ret, (long) iLenDefBuf, len-1); @@ -800,7 +820,7 @@ parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int bPa */ if(iMsg == iMaxLine) { *(pMsg + iMsg) = '\0'; /* space *is* reserved for this! */ - printline(hname, hnameIP, tmpline, bParseHost, flags, flowCtlType, pszInputName); + printline(hname, hnameIP, tmpline, flags, flowCtlType, pszInputName, stTime, ttGenTime); } else { /* This case in theory never can happen. If it happens, we have * a logic error. I am checking for it, because if I would not, @@ -852,7 +872,7 @@ parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int bPa *(pMsg + iMsg) = '\0'; /* space *is* reserved for this! */ /* typically, we should end up here! */ - printline(hname, hnameIP, tmpline, bParseHost, flags, flowCtlType, pszInputName); + printline(hname, hnameIP, tmpline, flags, flowCtlType, pszInputName, stTime, ttGenTime); finalize_it: if(tmpline != NULL) @@ -1176,6 +1196,9 @@ msgConsumer(void __attribute__((unused)) *notNeeded, void *pUsr) assert(pMsg != NULL); + if((pMsg->msgFlags & NEEDS_PARSING) != 0) { + parseMsg(pMsg); + } processMsg(pMsg); msgDestruct(&pMsg); @@ -1297,7 +1320,7 @@ static int parseRFCStructuredData(char **pp2parse, char *pResult) * * rger, 2005-11-24 */ -static int parseRFCSyslogMsg(msg_t *pMsg, int flags) +int parseRFCSyslogMsg(msg_t *pMsg, int flags) { char *p2parse; char *pBuf; @@ -1393,7 +1416,7 @@ static int parseRFCSyslogMsg(msg_t *pMsg, int flags) * but I thought I log it in this comment. * rgerhards, 2006-01-10 */ -static int parseLegacySyslogMsg(msg_t *pMsg, int flags) +int parseLegacySyslogMsg(msg_t *pMsg, int flags) { char *p2parse; char *pBuf; @@ -1931,7 +1954,7 @@ die(int sig) /* close the inputs */ dbgprintf("Terminating input threads...\n"); - thrdTerminateAll(); /* TODO: inputs only, please */ + thrdTerminateAll(); /* and THEN send the termination log message (see long comment above) */ if (sig) { @@ -2171,8 +2194,8 @@ static void dbgPrintInitInfo(void) cCCEscapeChar); dbgprintf("Main queue size %d messages.\n", iMainMsgQueueSize); - dbgprintf("Main queue worker threads: %d, Perists every %d updates.\n", - iMainMsgQueueNumWorkers, iMainMsgQPersistUpdCnt); + dbgprintf("Main queue worker threads: %d, wThread shutdown: %d, Perists every %d updates.\n", + 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", @@ -2182,11 +2205,9 @@ static void dbgPrintInitInfo(void) /* TODO: add iActionRetryCount = 0; iActionRetryInterval = 30000; - static int iMainMsgQtoWrkShutdown = 60000; static int iMainMsgQtoWrkMinMsgs = 100; static int iMainMsgQbSaveOnShutdown = 1; iMainMsgQueMaxDiskSpace = 0; - setQPROP(queueSettoWrkShutdown, "$MainMsgQueueTimeoutWorkerThreadShutdown", 5000); setQPROP(queueSetiMinMsgsPerWrkr, "$MainMsgQueueWorkerThreadMinimumMessages", 100); setQPROP(queueSetbSaveOnShutdown, "$MainMsgQueueSaveOnShutdown", 1); */ @@ -2896,6 +2917,8 @@ InitGlobalClasses(void) CHKiRet(actionClassInit()); pErrObj = "template"; CHKiRet(templateInit()); + pErrObj = "parser"; + CHKiRet(parserClassInit()); /* TODO: the dependency on net shall go away! -- rgerhards, 2008-03-07 */ pErrObj = "net"; @@ -3514,6 +3537,5 @@ int main(int argc, char **argv) dbgClassInit(); return realMain(argc, argv); } - /* vim:set ai: */ |