From 9813a12789d08a00fa96c11f113cd531c7f2ce7b Mon Sep 17 00:00:00 2001 From: David Lang Date: Tue, 25 Jan 2011 14:24:50 +0100 Subject: enhanced imfile to support multi-line messages Signed-off-by: Rainer Gerhards --- ChangeLog | 1 + doc/imfile.html | 5 +++ plugins/imfile/imfile.c | 8 +++- runtime/stream.c | 111 ++++++++++++++++++++++++++++++++++++------------ runtime/stream.h | 5 ++- 5 files changed, 101 insertions(+), 29 deletions(-) diff --git a/ChangeLog b/ChangeLog index 18eb97b4..f4084f35 100644 --- a/ChangeLog +++ b/ChangeLog @@ -1,5 +1,6 @@ --------------------------------------------------------------------------- Version 5.7.3 [V5-DEVEL] (rgerhards), 2010-12-?? +- added support for processing multi-line messages in imfile - added $IMUDPSchedulingPolicy and $IMUDPSchedulingPriority config settings - added $LocalHostName config directive - bugfix: fixed build problems on some platforms diff --git a/doc/imfile.html b/doc/imfile.html index f6b140a7..66c13e06 100644 --- a/doc/imfile.html +++ b/doc/imfile.html @@ -96,6 +96,11 @@ been processed. This setting can be used to guard against message duplication du to fatal errors (like power fail). Note that this setting affects imfile performance, especially when set to a low value. Frequently writing the state file is very time consuming. +
  • $InputFileReadMode [mode]
    +Available in 5.7.3+ +
    +Mode to be used when reading lines. 0 (the default) means that each line is forwarded +as its own log message. Caveats/Known Bugs:

    So far, only 100 files can be monitored. If more are needed, diff --git a/plugins/imfile/imfile.c b/plugins/imfile/imfile.c index 8681ac8c..c205f60e 100644 --- a/plugins/imfile/imfile.c +++ b/plugins/imfile/imfile.c @@ -71,6 +71,7 @@ typedef struct fileInfo_s { int nRecords; /**< How many records did we process before persisting the stream? */ int iPersistStateInterval; /**< how often should state be persisted? (0=on close only) */ strm_t *pStrm; /* its stream (NULL if not assigned) */ + int readMode; /* which mode to use in ReadMulteLine call? */ } fileInfo_t; @@ -85,6 +86,7 @@ static int iPollInterval = 10; /* number of seconds to sleep when there was no f static int iPersistStateInterval = 0; /* how often if state file to be persisted? (default 0->never) */ static int iFacility = 128; /* local0 */ static int iSeverity = 5; /* notice, as of rfc 3164 */ +static int readMode = 0; /* mode to use for ReadMultiLine call */ static int iFilPtr = 0; /* number of files to be monitored; pointer to next free spot during config */ #define MAX_INPUT_FILES 100 @@ -212,7 +214,7 @@ static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData) /* loop below will be exited when strmReadLine() returns EOF */ while(1) { - CHKiRet(strm.ReadLine(pThis->pStrm, &pCStr)); + CHKiRet(strm.ReadLine(pThis->pStrm, &pCStr, pThis->readMode)); *pbHadFileData = 1; /* this is just a flag, so set it and forget it */ CHKiRet(enqLine(pThis, pCStr)); /* process line */ rsCStrDestruct(&pCStr); /* discard string (must be done by us!) */ @@ -447,6 +449,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a iPollInterval = 10; iFacility = 128; /* local0 */ iSeverity = 5; /* notice, as of rfc 3164 */ + readMode = 0; RETiRet; } @@ -489,6 +492,7 @@ static rsRetVal addMonitor(void __attribute__((unused)) *pVal, uchar *pNewVal) pThis->iFacility = iFacility; pThis->iPersistStateInterval = iPersistStateInterval; pThis->nRecords = 0; + pThis->readMode = readMode; iPersistStateInterval = 0; } else { errmsg.LogError(0, RS_RET_OUT_OF_DESRIPTORS, "Too many file monitors configured - ignoring this one"); @@ -535,6 +539,8 @@ CODEmodInit_QueryRegCFSLineHdlr NULL, &iFacility, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilepollinterval", 0, eCmdHdlrInt, NULL, &iPollInterval, STD_LOADABLE_MODULE_ID)); + CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilereadmode", 0, eCmdHdlrInt, + NULL, &readMode, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilepersiststateinterval", 0, eCmdHdlrInt, NULL, &iPersistStateInterval, STD_LOADABLE_MODULE_ID)); /* that command ads a new file! */ diff --git a/runtime/stream.c b/runtime/stream.c index 658aba11..16d41a2e 100644 --- a/runtime/stream.c +++ b/runtime/stream.c @@ -561,39 +561,98 @@ static rsRetVal strmUnreadChar(strm_t *pThis, uchar c) return RS_RET_OK; } - -/* read a line from a strm file. A line is terminated by LF. The LF is read, but it - * is not returned in the buffer (it is discared). The caller is responsible for - * destruction of the returned CStr object! -- rgerhards, 2008-01-07 - * rgerhards, 2008-03-27: I now use the ppCStr directly, without any interim - * string pointer. The reason is that this function my be called by inputs, which - * are pthread_killed() upon termination. So if we use their native pointer, they - * can cleanup (but only then). +/* read a 'paragraph' from a strm file. + * A paragraph may be terminated by a LF, by a LFLF, or by LF depending on the option set. + * The termination LF characters are read, but are + * not returned in the buffer (it is discared). The caller is responsible for + * destruction of the returned CStr object! -- dlang 2010-12-13 */ static rsRetVal -strmReadLine(strm_t *pThis, cstr_t **ppCStr) +strmReadLine(strm_t *pThis, cstr_t **ppCStr, int mode) { - DEFiRet; - uchar c; - - ASSERT(pThis != NULL); - ASSERT(ppCStr != NULL); - - CHKiRet(cstrConstruct(ppCStr)); - - /* now read the line */ - CHKiRet(strmReadChar(pThis, &c)); - while(c != '\n') { - CHKiRet(cstrAppendChar(*ppCStr, c)); - CHKiRet(strmReadChar(pThis, &c)); + /* mode = 0 single line mode (equivalent to ReadLine) + * mode = 1 LFLF mode (paragraph, blank line between entries) + * mode = 2 LF mode, a log line starts at the beginning of a line, but following lines that are indented are part of the same log entry + * This modal interface is not nearly as flexible as being able to define a regex for when a new record starts, but it's also not nearly as hard (or as slow) to implement + */ + DEFiRet; + uchar c; + uchar finished; + + ASSERT(pThis != NULL); + ASSERT(ppCStr != NULL); + + CHKiRet(cstrConstruct(ppCStr)); + + /* now read the line */ + CHKiRet(strmReadChar(pThis, &c)); + if (mode == 0){ + while(c != '\n') { + CHKiRet(cstrAppendChar(*ppCStr, c)); + CHKiRet(strmReadChar(pThis, &c)); + } + CHKiRet(cstrFinalize(*ppCStr)); + } + if (mode == 1){ + finished=0; + while(finished == 0){ + if(c != '\n') { + CHKiRet(cstrAppendChar(*ppCStr, c)); + CHKiRet(strmReadChar(pThis, &c)); + } else { + if ((((*ppCStr)->iStrLen) > 0) ){ + if ((*ppCStr)->pBuf[(*ppCStr)->iStrLen -1 ] == '\n'){ + rsCStrTruncate(*ppCStr,1); /* remove the prior newline */ + finished=1; + } else { + CHKiRet(cstrAppendChar(*ppCStr, c)); + CHKiRet(strmReadChar(pThis, &c)); + } + } else { + finished=1; /* this is a blank line, a \n with nothing since the last complete record */ + } + } + } + CHKiRet(cstrFinalize(*ppCStr)); + } + if (mode == 2){ +/* indented follow-up lines */ + finished=0; + while(finished == 0){ + if ((*ppCStr)->iStrLen == 0){ + if(c != '\n') { +/* nothing in the buffer, and it's not a newline, add it to the buffer */ + CHKiRet(cstrAppendChar(*ppCStr, c)); + CHKiRet(strmReadChar(pThis, &c)); + } else { + finished=1; /* this is a blank line, a \n with nothing since the last complete record */ + } + } else { + if ((*ppCStr)->pBuf[(*ppCStr)->iStrLen -1 ] != '\n'){ +/* not the first character after a newline, add it to the buffer */ + CHKiRet(cstrAppendChar(*ppCStr, c)); + CHKiRet(strmReadChar(pThis, &c)); + } else { + if ((c == ' ') || (c == '\t')){ + CHKiRet(cstrAppendChar(*ppCStr, c)); + CHKiRet(strmReadChar(pThis, &c)); + } else { +/* clean things up by putting the character we just read back into the input buffer and removing the LF character that is currently at the end of the output string */ + CHKiRet(strmUnreadChar(pThis, c)); + rsCStrTruncate(*ppCStr,1); + finished=1; + } + } + } + } + CHKiRet(cstrFinalize(*ppCStr)); } - CHKiRet(cstrFinalize(*ppCStr)); finalize_it: - if(iRet != RS_RET_OK && *ppCStr != NULL) - cstrDestruct(ppCStr); + if(iRet != RS_RET_OK && *ppCStr != NULL) + cstrDestruct(ppCStr); - RETiRet; + RETiRet; } diff --git a/runtime/stream.h b/runtime/stream.h index 37e9d570..60c68cb2 100644 --- a/runtime/stream.h +++ b/runtime/stream.h @@ -156,7 +156,6 @@ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */ rsRetVal (*SetFileName)(strm_t *pThis, uchar *pszName, size_t iLenName); rsRetVal (*ReadChar)(strm_t *pThis, uchar *pC); rsRetVal (*UnreadChar)(strm_t *pThis, uchar c); - rsRetVal (*ReadLine)(strm_t *pThis, cstr_t **ppCStr); rsRetVal (*SeekCurrOffs)(strm_t *pThis); rsRetVal (*Write)(strm_t *pThis, uchar *pBuf, size_t lenBuf); rsRetVal (*WriteChar)(strm_t *pThis, uchar c); @@ -183,8 +182,10 @@ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */ INTERFACEpropSetMeth(strm, iSizeLimit, off_t); INTERFACEpropSetMeth(strm, iFlushInterval, int); INTERFACEpropSetMeth(strm, pszSizeLimitCmd, uchar*); + /* v6 added */ + rsRetVal (*ReadLine)(strm_t *pThis, cstr_t **ppCStr, int mode); ENDinterface(strm) -#define strmCURR_IF_VERSION 5 /* increment whenever you change the interface structure! */ +#define strmCURR_IF_VERSION 6 /* increment whenever you change the interface structure! */ /* prototypes */ -- cgit