implemented $OMFileFlushOnTXEnd directive

plus some cleanup...

5 files changed, 25 insertions, 12 deletions

diff --git a/ChangeLog b/ChangeLog
index 39ca00ec..86d72c9e 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -20,6 +20,7 @@ Version 4.3.2  [DEVEL] (rgerhards), 2009-??-??
 - added configuration commands (see doc for explanations)
   * $OMFileZipLevel
   * $OMFileIOBufferSize
+  * $OMFileFlushOnTXEnd
   * $MainMsgQueueSyncQueueFiles
   * $ActionQueueSyncQueueFiles
 ---------------------------------------------------------------------------
diff --git a/doc/rsyslog_conf_global.html b/doc/rsyslog_conf_global.html
index e9b1c082..a909b00c 100644
--- a/doc/rsyslog_conf_global.html
+++ b/doc/rsyslog_conf_global.html
@@ -193,6 +193,13 @@ supported in order to be compliant to the upcoming new syslog RFC series.
 of the output file. The higher the number, the better the compression, but also the
 more CPU is required for zipping.</li>
 <li><b>$OMFileIOBufferSize</b> &lt;size_nbr&gt;, default 4k, size of the buffer used to writing output data. The larger the buffer, the potentially better performance is. The default of 4k is quite conservative, it is useful to go up to 64k, and 128K if you used gzip compression (then, even higher sizes may make sense)</li>
+<li><b>$OMFileFlushOnTXEnd</b> &lt;[on/<b>off</b>]&gt;, default off, by default, omfile
+writes output using a buffered writer. Disk writes are only done when the buffer is
+full. So if an error happens during that write, data is potentially lost. In cases where
+this is unacceptable, set $OMFileFlushOnTXEnd to on. Then, data is written at the end
+of each transaction (for pre-v5 this means after <b>each</b> log message) and the usual
+error recovery thus can handle write errors without data loss. Note that this option
+severely reduces the effect of zip compression.</li>
 <li><b>$RepeatedMsgContainsOriginalMsg</b> [on/<b>off</b>] - "last message repeated n times" messages, if generated,
 have a different format that contains the message that is being repeated.
 Note that only the first "n" characters are included, with n to be at least 80 characters, most
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index 28f7bce9..9421ca67 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -125,14 +125,6 @@ typedef enum {
 	FIOP_EREREGEX = 5	/* matches a ERE regular expression? */
 } fiop_t;
 
-/* file types (omfile & stream) */
-typedef enum {
-	eTypeFILE,
-	eTypeTTY,
-	eTypeCONSOLE,
-	eTypePIPE
-} filetype_t;
-
 
 #ifndef _PATH_CONSOLE
 #define _PATH_CONSOLE	"/dev/console"
diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index a1454275..f9edde8b 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -68,7 +68,7 @@ static rsRetVal keyDestruct(void __attribute__((unused)) *pData)
 /* ---------- END linked-list key handling functions ---------- */
 
 
-/* dirver to iterate over all of this ruleset actions */
+/* driver to iterate over all of this ruleset actions */
 typedef struct iterateAllActions_s {
 	rsRetVal (*pFunc)(void*, void*);
 	void *pParam;
@@ -81,6 +81,7 @@ DEFFUNC_llExecFunc(doIterateRulesetActions)
 	iRet = rule.IterateAllActions(pRule, pMyParam->pFunc, pMyParam->pParam);
 	RETiRet;
 }
+#if 0
 /* iterate over all actions of THIS rule set.
  */
 static rsRetVal
@@ -99,7 +100,7 @@ finalize_it:
 }
 
 
-/* dirver to iterate over all actions */
+/* driver to iterate over all actions */
 DEFFUNC_llExecFunc(doIterateAllActions)
 {
 	DEFiRet;
@@ -108,6 +109,7 @@ DEFFUNC_llExecFunc(doIterateAllActions)
 	iRet = iterateRulesetAllActions(pThis, pMyParam->pFunc, pMyParam->pParam);
 	RETiRet;
 }
+#endif
 /* iterate over ALL actions present in the WHOLE system.
  * this is often needed, for example when HUP processing 
  * must be done or a shutdown is pending.
@@ -121,7 +123,8 @@ iterateAllActions(rsRetVal (*pFunc)(void*, void*), void* pParam)
 
 	params.pFunc = pFunc;
 	params.pParam = pParam;
-	CHKiRet(llExecFunc(&llRulesets, doIterateAllActions, &params));
+	//CHKiRet(llExecFunc(&llRulesets, doIterateAllActions, &params));
+	CHKiRet(llExecFunc(&llRulesets, doIterateRulesetActions, &params));
 
 finalize_it:
 	RETiRet;
diff --git a/tools/omfile.c b/tools/omfile.c
index 6377268d..6d9eb096 100644
--- a/tools/omfile.c
+++ b/tools/omfile.c
@@ -101,6 +101,7 @@ static uid_t	dirGID;		/* GID to be used for newly created directories */
 static int	bCreateDirs;	/* auto-create directories for dynaFiles: 0 - no, 1 - yes */
 static int	bEnableSync = 0;/* enable syncing of files (no dash in front of pathname in conf): 0 - no, 1 - yes */
 static int	iZipLevel = 0;	/* zip compression mode (0..9 as usual) */
+static bool	bFlushOnTXEnd = 0;/* flush write buffers when transaction has ended? */
 static int	iIOBufSize = IOBUF_DFLT_SIZE;	/* size of an io buffer */
 static uchar	*pszTplName = NULL; /* name of the default template to use */
 /* end globals for default values */
@@ -131,6 +132,7 @@ typedef struct _instanceData {
 	uchar	*pszSizeLimitCmd;	/* command to carry out when size limit is reached */
 	int 	iZipLevel;		/* zip mode to use for this selector */
 	int	iIOBufSize;		/* size of associated io buffer */
+	bool	bFlushOnTXEnd;		/* flush write buffers when transaction has ended? */
 } instanceData;
 
 
@@ -595,7 +597,12 @@ ENDtryResume
 BEGINdoAction
 CODESTARTdoAction
 	DBGPRINTF("file to log to: %s\n", pData->f_fname);
-	iRet = writeFile(ppString, iMsgOpts, pData);
+	CHKiRet(writeFile(ppString, iMsgOpts, pData));
+	if(pData->bFlushOnTXEnd) {
+		/* TODO v5: do this in endTransaction only! */
+		CHKiRet(strm.Flush(pData->pStrm));
+	}
+finalize_it:
 ENDdoAction
 
 
@@ -669,6 +676,7 @@ CODESTARTparseSelectorAct
 	pData->dirUID = dirUID;
 	pData->dirGID = dirGID;
 	pData->iZipLevel = iZipLevel;
+	pData->bFlushOnTXEnd = bFlushOnTXEnd;
 	pData->iIOBufSize = iIOBufSize;
 
 	if(pData->bDynamicName == 0) {
@@ -702,6 +710,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
 	bCreateDirs = 1;
 	bEnableSync = 0;
 	iZipLevel = 0;
+	bFlushOnTXEnd = 0;
 	iIOBufSize = IOBUF_DFLT_SIZE;
 	if(pszTplName != NULL) {
 		free(pszTplName);
@@ -751,6 +760,7 @@ CODEmodInit_QueryRegCFSLineHdlr
 	CHKiRet(objUse(strm, CORE_COMPONENT));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"dynafilecachesize", 0, eCmdHdlrInt, (void*) setDynaFileCacheSize, NULL, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"omfileziplevel", 0, eCmdHdlrInt, NULL, &iZipLevel, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"omfileflushontxend", 0, eCmdHdlrBinary, NULL, &bFlushOnTXEnd, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"omfileiobuffersize", 0, eCmdHdlrSize, NULL, &iIOBufSize, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"dirowner", 0, eCmdHdlrUID, NULL, &dirUID, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"dirgroup", 0, eCmdHdlrGID, NULL, &dirGID, STD_LOADABLE_MODULE_ID));