enhanced imfile to support multi-line messages

Signed-off-by: Rainer Gerhards <rgerhards@adiscon.com>

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.
+<li><b>$InputFileReadMode</b> [mode]</b><br>
+Available in 5.7.3+
+<br>
+Mode to be used when reading lines. 0 (the default) means that each line is forwarded
+as its own log message.
 </ul>
 <b>Caveats/Known Bugs:</b>
 <p>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<not whitespace> 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 <not whitespace> 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 */