summaryrefslogtreecommitdiffstats
path: root/tools
diff options
context:
space:
mode:
Diffstat (limited to 'tools')
-rw-r--r--tools/omfile.c15
-rw-r--r--tools/syslogd.c84
2 files changed, 52 insertions, 47 deletions
diff --git a/tools/omfile.c b/tools/omfile.c
index 5b0bfb46..c7e0dc25 100644
--- a/tools/omfile.c
+++ b/tools/omfile.c
@@ -156,6 +156,7 @@ typedef struct _instanceData {
int iFlushInterval; /* how fast flush buffer on inactivity? */
sbool bFlushOnTXEnd; /* flush write buffers when transaction has ended? */
sbool bUseAsyncWriter; /* use async stream writer? */
+ sbool bVeryRobustZip;
} instanceData;
@@ -205,6 +206,7 @@ static struct cnfparamdescr actpdescr[] = {
{ "ziplevel", eCmdHdlrInt, 0 }, /* legacy: omfileziplevel */
{ "flushinterval", eCmdHdlrInt, 0 }, /* legacy: omfileflushinterval */
{ "asyncwriting", eCmdHdlrBinary, 0 }, /* legacy: omfileasyncwriting */
+ { "veryrobustzip", eCmdHdlrBinary, 0 },
{ "flushontxend", eCmdHdlrBinary, 0 }, /* legacy: omfileflushontxend */
{ "iobuffersize", eCmdHdlrSize, 0 }, /* legacy: omfileiobuffersize */
{ "dirowner", eCmdHdlrUID, 0 }, /* legacy: dirowner */
@@ -269,7 +271,8 @@ CODESTARTdbgPrintInstInfo
dbgprintf("\tflush on TX end=%d\n", pData->bFlushOnTXEnd);
dbgprintf("\tflush interval=%d\n", pData->iFlushInterval);
dbgprintf("\tfile cache size=%d\n", pData->iDynaFileCacheSize);
- dbgprintf("\tcreate directories: %s\n", pData->bCreateDirs ? "yes" : "no");
+ dbgprintf("\tcreate directories: %s\n", pData->bCreateDirs ? "on" : "off");
+ dbgprintf("\tvery robust zip: %s\n", pData->bCreateDirs ? "on" : "off");
dbgprintf("\tfile owner %d, group %d\n", (int) pData->fileUID, (int) pData->fileGID);
dbgprintf("\tdirectory owner %d, group %d\n", (int) pData->dirUID, (int) pData->dirGID);
dbgprintf("\tdir create mode 0%3.3o, file create mode 0%3.3o\n",
@@ -292,7 +295,7 @@ setLegacyDfltTpl(void __attribute__((unused)) *pVal, uchar* newVal)
if(loadModConf != NULL && loadModConf->tplName != NULL) {
free(newVal);
- errmsg.LogError(0, RS_RET_ERR, "omfile default template already set via module "
+ errmsg.LogError(0, RS_RET_ERR, "omfile: default template already set via module "
"global parameter - can no longer be changed");
ABORT_FINALIZE(RS_RET_ERR);
}
@@ -536,6 +539,7 @@ prepareFile(instanceData *pData, uchar *newFileName)
CHKiRet(strm.SetFName(pData->pStrm, szBaseName, ustrlen(szBaseName)));
CHKiRet(strm.SetDir(pData->pStrm, szDirName, ustrlen(szDirName)));
CHKiRet(strm.SetiZipLevel(pData->pStrm, pData->iZipLevel));
+ CHKiRet(strm.SetbVeryReliableZip(pData->pStrm, pData->bVeryRobustZip));
CHKiRet(strm.SetsIOBufSize(pData->pStrm, (size_t) pData->iIOBufSize));
CHKiRet(strm.SettOperationsMode(pData->pStrm, STREAMMODE_WRITE_APPEND));
CHKiRet(strm.SettOpenMode(pData->pStrm, cs.fCreateMode));
@@ -843,6 +847,7 @@ BEGINendTransaction
CODESTARTendTransaction
/* Note: pStrm may be NULL if there was an error opening the stream */
if(pData->bFlushOnTXEnd && pData->pStrm != NULL) {
+dbgprintf("AAAA: flusing stream, endTx\n");
CHKiRet(strm.Flush(pData->pStrm));
}
finalize_it:
@@ -854,6 +859,7 @@ CODESTARTdoAction
DBGPRINTF("file to log to: %s\n", pData->f_fname);
CHKiRet(writeFile(ppString, iMsgOpts, pData));
if(!bCoreSupportsBatching && pData->bFlushOnTXEnd) {
+dbgprintf("AAAA: flusing stream, in Tx\n");
CHKiRet(strm.Flush(pData->pStrm));
}
finalize_it:
@@ -878,6 +884,7 @@ setInstParamDefaults(instanceData *pData)
pData->bCreateDirs = 1;
pData->bSyncFile = 0;
pData->iZipLevel = 0;
+ pData->bVeryRobustZip = 0;
pData->bFlushOnTXEnd = FLUSHONTX_DFLT;
pData->iIOBufSize = IOBUF_DFLT_SIZE;
pData->iFlushInterval = FLUSH_INTRVL_DFLT;
@@ -915,6 +922,8 @@ CODESTARTnewActInst
pData->iZipLevel = (int) pvals[i].val.d.n;
} else if(!strcmp(actpblk.descr[i].name, "flushinterval")) {
pData->iFlushInterval = pvals[i].val.d.n;
+ } else if(!strcmp(actpblk.descr[i].name, "veryrobustzip")) {
+ pData->bVeryRobustZip = pvals[i].val.d.n;
} else if(!strcmp(actpblk.descr[i].name, "asyncwriting")) {
pData->bUseAsyncWriter = pvals[i].val.d.n;
} else if(!strcmp(actpblk.descr[i].name, "flushontxend")) {
@@ -1064,6 +1073,7 @@ CODESTARTparseSelectorAct
pData->iIOBufSize = (int) cs.iIOBufSize;
pData->iFlushInterval = cs.iFlushInterval;
pData->bUseAsyncWriter = cs.bUseAsyncWriter;
+ pData->bVeryRobustZip = 0; /* cannot be specified via legacy conf */
CODE_STD_FINALIZERparseSelectorAct
ENDparseSelectorAct
@@ -1090,7 +1100,6 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
cs.bUseAsyncWriter = USE_ASYNCWRITER_DFLT;
free(pszFileDfltTplName);
pszFileDfltTplName = NULL;
-
return RS_RET_OK;
}
diff --git a/tools/syslogd.c b/tools/syslogd.c
index edb546a1..cb6a47cd 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -125,6 +125,7 @@
#include "dnscache.h"
#include "sd-daemon.h"
#include "rainerscript.h"
+#include "ratelimit.h"
/* definitions for objects we access */
DEFobjCurrIf(obj)
@@ -220,6 +221,7 @@ struct queuefilenames_s {
} *queuefilenames = NULL;
+static ratelimit_t *dflt_ratelimiter = NULL; /* ratelimiter for submits without explicit one */
int MarkInterval = 20 * 60; /* interval between marks in seconds - read-only after startup */
int send_to_all = 0; /* send message to all IPv4/IPv6 addresses */
static int NoFork = 0; /* don't fork - don't run in daemon mode - read-only after startup */
@@ -411,7 +413,7 @@ parseAndSubmitMessage(uchar *hname, uchar *hnameIP, uchar *msg, int len, int fla
CHKiRet(prop.Destruct(&pProp));
CHKiRet(MsgSetRcvFromIPStr(pMsg, hnameIP, ustrlen(hnameIP), &pProp));
CHKiRet(prop.Destruct(&pProp));
- CHKiRet(submitMsg(pMsg));
+ CHKiRet(submitMsg2(pMsg));
finalize_it:
RETiRet;
@@ -431,6 +433,12 @@ submitErrMsg(int iErr, uchar *msg)
}
+static inline rsRetVal
+submitMsgWithDfltRatelimiter(msg_t *pMsg)
+{
+ return ratelimitAddMsg(dflt_ratelimiter, NULL, pMsg);
+}
+
/* rgerhards 2004-11-09: the following is a function that can be used
* to log a message orginating from the syslogd itself.
*/
@@ -482,50 +490,12 @@ logmsgInternal(int iErr, int pri, uchar *msg, int flags)
/* we have the queue, so we can simply provide the
* message to the queue engine.
*/
- submitMsg(pMsg);
+ submitMsgWithDfltRatelimiter(pMsg);
}
finalize_it:
RETiRet;
}
-/* check message against ACL set
- * rgerhards, 2009-11-16
- */
-#if 0
-static inline rsRetVal
-chkMsgAgainstACL() {
- /* 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(net.CmpHost(&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((uchar*)"UDP",
- (struct sockaddr *)&frominet, (char*)fromHostFQDN);
-
- if(!*pbIsPermitted) {
- DBGPRINTF("%s is not an allowed sender\n", (char*)fromHostFQDN);
- if(glbl.GetOption_DisallowWarning) {
- time_t tt;
-
- datetime.GetTime(&tt);
- if(tt > ttLastDiscard + 60) {
- ttLastDiscard = tt;
- errmsg.LogError(0, NO_ERRCODE,
- "UDP message from disallowed sender %s discarded",
- (char*)fromHost);
- }
- }
- }
- }
-}
-#endif
-
/* preprocess a batch of messages, that is ready them for actual processing. This is done
* as a first stage and totally in parallel to any other worker active in the system. So
@@ -618,7 +588,7 @@ int i;
* rgerhards, 2008-02-13
*/
rsRetVal
-submitMsg(msg_t *pMsg)
+submitMsg2(msg_t *pMsg)
{
qqueue_t *pQueue;
ruleset_t *pRuleset;
@@ -631,7 +601,7 @@ submitMsg(msg_t *pMsg)
/* if a plugin logs a message during shutdown, the queue may no longer exist */
if(pQueue == NULL) {
- DBGPRINTF("submitMsg() could not submit message - "
+ DBGPRINTF("submitMsg2() could not submit message - "
"queue does (no longer?) exist - ignored\n");
FINALIZE;
}
@@ -643,13 +613,19 @@ finalize_it:
RETiRet;
}
+rsRetVal
+submitMsg(msg_t *pMsg)
+{
+ return submitMsgWithDfltRatelimiter(pMsg);
+}
+
/* submit multiple messages at once, very similar to submitMsg, just
* for multi_submit_t. All messages need to go into the SAME queue!
* rgerhards, 2009-06-16
*/
rsRetVal
-multiSubmitMsg(multi_submit_t *pMultiSub)
+multiSubmitMsg2(multi_submit_t *pMultiSub)
{
int i;
qqueue_t *pQueue;
@@ -680,8 +656,23 @@ multiSubmitMsg(multi_submit_t *pMultiSub)
finalize_it:
RETiRet;
}
+rsRetVal
+multiSubmitMsg(multi_submit_t *pMultiSub) /* backward compat. level */
+{
+ return multiSubmitMsg2(pMultiSub);
+}
+/* flush multiSubmit, e.g. at end of read records */
+rsRetVal
+multiSubmitFlush(multi_submit_t *pMultiSub)
+{
+ DEFiRet;
+ if(pMultiSub->nElem > 0) {
+ iRet = multiSubmitMsg2(pMultiSub);
+ }
+ RETiRet;
+}
static void
@@ -1262,7 +1253,7 @@ static inline void processImInternal(void)
msg_t *pMsg;
while(iminternalRemoveMsg(&pMsg) == RS_RET_OK) {
- submitMsg(pMsg);
+ submitMsgWithDfltRatelimiter(pMsg);
}
}
@@ -1467,6 +1458,7 @@ InitGlobalClasses(void)
CHKiRet(objUse(net, LM_NET_FILENAME));
dnscacheInit();
initRainerscript();
+ ratelimitModInit();
finalize_it:
if(iRet != RS_RET_OK) {
@@ -1505,6 +1497,7 @@ GlobalClassExit(void)
/* TODO: implement the rest of the deinit */
/* dummy "classes */
strExit();
+ ratelimitModExit();
#if 0
CHKiRet(objGetObjInterface(&obj)); /* this provides the root pointer for all other queries */
@@ -2037,6 +2030,9 @@ int realMain(int argc, char **argv)
}
CHKiRet(localRet);
+ CHKiRet(ratelimitNew(&dflt_ratelimiter, "rsyslogd", NULL));
+ /* TODO: add linux-type limiting capability */
+
if(bChDirRoot) {
if(chdir("/") != 0)
fprintf(stderr, "Can not do 'cd /' - still trying to run\n");
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-17 15:14:37 +0000