summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--action.c101
-rw-r--r--action.h8
-rw-r--r--grammar/rainerscript.c2
-rw-r--r--plugins/mmanon/mmanon.c15
-rw-r--r--plugins/omruleset/omruleset.c4
-rw-r--r--plugins/omtesting/omtesting.c5
-rw-r--r--runtime/module-template.h51
-rw-r--r--runtime/modules.c3
-rw-r--r--runtime/modules.h2
-rw-r--r--runtime/queue.c26
-rw-r--r--runtime/queue.h8
-rw-r--r--runtime/rsconf.c1
-rw-r--r--runtime/ruleset.c46
-rw-r--r--runtime/ruleset.h5
-rw-r--r--runtime/wti.c8
-rw-r--r--runtime/wti.h1
-rw-r--r--runtime/wtp.c6
-rw-r--r--tools/omdiscard.c17
-rw-r--r--tools/omfile.c107
-rw-r--r--tools/omfwd.c15
-rw-r--r--tools/ompipe.c17
-rw-r--r--tools/omshell.c15
-rw-r--r--tools/omusrmsg.c17
-rw-r--r--tools/syslogd.c4
24 files changed, 364 insertions, 120 deletions
diff --git a/action.c b/action.c
index eb6301d8..f9297a55 100644
--- a/action.c
+++ b/action.c
@@ -119,10 +119,10 @@
#define NO_TIME_PROVIDED 0 /* indicate we do not provide any cached time */
/* forward definitions */
-static rsRetVal processBatchMain(action_t *pAction, batch_t *pBatch, int*);
-static rsRetVal doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch);
-static rsRetVal doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch);
-static rsRetVal doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch);
+static rsRetVal processBatchMain(void *pVoid, batch_t *pBatch, wti_t *pWti, int*);
+static rsRetVal doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti);
+static rsRetVal doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti);
+static rsRetVal doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti);
/* object static data (once for all instances) */
/* TODO: make this an object! DEFobjStaticHelpers -- rgerhards, 2008-03-05 */
@@ -175,10 +175,9 @@ configSettings_t cs_save; /* our saved (scope!) config settings */
/* the counter below counts actions created. It is used to obtain unique IDs for the action. They
* should not be relied on for any long-term activity (e.g. disk queue names!), but they are nice
* to have during one instance of an rsyslogd run. For example, I use them to name actions when there
- * is no better name available. Note that I do NOT recover previous numbers on HUP - we simply keep
- * counting. -- rgerhards, 2008-01-29
+ * is no better name available.
*/
-static int iActionNbr = 0;
+int iActionNbr = 0;
/* tables for interfacing with the v6 config system */
static struct cnfparamdescr cnfparamdescr[] = {
@@ -341,6 +340,7 @@ rsRetVal actionConstruct(action_t **ppThis)
pThis->bExecWhenPrevSusp = 0;
pThis->bRepMsgHasMsg = 0;
pThis->tLastOccur = datetime.GetTime(NULL); /* done once per action on startup only */
+ pThis->iActionNbr = iActionNbr;
pthread_mutex_init(&pThis->mutActExec, NULL);
pthread_mutex_init(&pThis->mutAction, NULL);
INIT_ATOMIC_HELPER_MUT(pThis->mutCAS);
@@ -428,7 +428,7 @@ actionConstructFinalize(action_t *pThis, struct nvlst *lst)
* spec. -- rgerhards, 2008-01-30
*/
CHKiRet(qqueueConstruct(&pThis->pQueue, cs.ActionQueType, 1, cs.iActionQueueSize,
- (rsRetVal (*)(void*, batch_t*, int*))processBatchMain));
+ processBatchMain));
obj.SetName((obj_t*) pThis->pQueue, pszAName);
qqueueSetpAction(pThis->pQueue, pThis);
@@ -905,17 +905,25 @@ done: RETiRet;
* rgerhards, 2008-01-28
*/
rsRetVal
-actionCallDoAction(action_t *pThis, msg_t *pMsg, void *actParams)
+actionCallDoAction(action_t *pThis, msg_t *pMsg, void *actParams, wti_t *pWti)
{
DEFiRet;
ASSERT(pThis != NULL);
ISOBJ_TYPE_assert(pMsg, msg);
- DBGPRINTF("entering actionCalldoAction(), state: %s\n", getActStateName(pThis));
+ DBGPRINTF("entering actionCalldoAction(), state: %s, actionNbr %d\n",
+ getActStateName(pThis), pThis->iActionNbr);
+
+ if(pWti->actWrkrData[pThis->iActionNbr] == NULL) {
+ DBGPRINTF("we need to create a new action worker instance for "
+ "action %d\n", pThis->iActionNbr);
+ CHKiRet(pThis->pMod->mod.om.createWrkrInstance(&(pWti->actWrkrData[pThis->iActionNbr]), pThis->pModData));
+ }
pThis->bHadAutoCommit = 0;
- iRet = pThis->pMod->mod.om.doAction(actParams, pMsg->msgFlags, pThis->pModData);
+ iRet = pThis->pMod->mod.om.doAction(actParams, pMsg->msgFlags,
+ pWti->actWrkrData[pThis->iActionNbr]);
switch(iRet) {
case RS_RET_OK:
actionCommitted(pThis);
@@ -952,8 +960,8 @@ finalize_it:
* this readies the action and then calls doAction()
* rgerhards, 2008-01-28
*/
-static inline rsRetVal
-actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams, int *pbShutdownImmediate)
+static rsRetVal
+actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams, int *pbShutdownImmediate, wti_t *pWti)
{
DEFiRet;
@@ -964,7 +972,7 @@ actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams, int *pbShutd
if(pThis->pMod->mod.om.SetShutdownImmdtPtr != NULL)
pThis->pMod->mod.om.SetShutdownImmdtPtr(pThis->pModData, pbShutdownImmediate);
if(pThis->eState == ACT_STATE_ITX)
- CHKiRet(actionCallDoAction(pThis, pMsg, actParams));
+ CHKiRet(actionCallDoAction(pThis, pMsg, actParams, pWti));
iRet = getReturnCode(pThis);
finalize_it:
@@ -977,7 +985,7 @@ finalize_it:
* rgerhards, 2008-01-28
*/
static rsRetVal
-finishBatch(action_t *pThis, batch_t *pBatch)
+finishBatch(action_t *pThis, batch_t *pBatch, wti_t *pWti)
{
int i;
DEFiRet;
@@ -991,7 +999,7 @@ finishBatch(action_t *pThis, batch_t *pBatch)
CHKiRet(actionPrepare(pThis, pBatch->pbShutdownImmediate));
if(pThis->eState == ACT_STATE_ITX) {
- iRet = pThis->pMod->mod.om.endTransaction(pThis->pModData);
+ iRet = pThis->pMod->mod.om.endTransaction(pWti->actWrkrData[pThis->iActionNbr]);
switch(iRet) {
case RS_RET_OK:
actionCommitted(pThis);
@@ -1034,7 +1042,7 @@ finalize_it:
* rgerhards, 2009-05-12
*/
static inline rsRetVal
-tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem)
+tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem, wti_t *pWti)
{
int i;
int iElemProcessed;
@@ -1059,7 +1067,7 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem)
if(batchIsValidElem(pBatch, i)) {
pMsg = pBatch->pElem[i].pMsg;
localRet = actionProcessMessage(pAction, pMsg, pBatch->pElem[i].staticActParams,
- pBatch->pbShutdownImmediate);
+ pBatch->pbShutdownImmediate, pWti);
DBGPRINTF("action %p call returned %d\n", pAction, localRet);
/* Note: we directly modify the batch object state, because we know that
* wo do not overwrite BATCH_STATE_DISC indicators!
@@ -1111,7 +1119,7 @@ finalize_it:
* rgerhards, 2009-05-12
*/
static rsRetVal
-submitBatch(action_t *pAction, batch_t *pBatch, int nElem)
+submitBatch(action_t *pAction, batch_t *pBatch, int nElem, wti_t *pWti)
{
int i;
int bDone;
@@ -1125,7 +1133,7 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem)
wasDoneTo = pBatch->iDoneUpTo;
bDone = 0;
do {
- localRet = tryDoAction(pAction, pBatch, &nElem);
+ localRet = tryDoAction(pAction, pBatch, &nElem, pWti);
if(localRet == RS_RET_FORCE_TERM) {
ABORT_FINALIZE(RS_RET_FORCE_TERM);
}
@@ -1135,7 +1143,7 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem)
/* try commit transaction, once done, we can simply do so as if
* that return state was returned from tryDoAction().
*/
- localRet = finishBatch(pAction, pBatch);
+ localRet = finishBatch(pAction, pBatch, pWti);
}
if( localRet == RS_RET_OK
@@ -1164,8 +1172,8 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem)
/* retry with half as much. Depth is log_2 batchsize, so recursion is not too deep */
DBGPRINTF("submitBatch recursing trying to find and exclude the culprit "
"for iRet %d\n", localRet);
- submitBatch(pAction, pBatch, nElem / 2);
- submitBatch(pAction, pBatch, nElem - (nElem / 2));
+ submitBatch(pAction, pBatch, nElem / 2, pWti);
+ submitBatch(pAction, pBatch, nElem - (nElem / 2), pWti);
bDone = 1;
}
}
@@ -1245,13 +1253,13 @@ prepareBatch(action_t *pAction, batch_t *pBatch, sbool **activeSave, int *bMustR
* rgerhards, 2009-05-12
*/
static inline rsRetVal
-processAction(action_t *pAction, batch_t *pBatch)
+processAction(action_t *pAction, batch_t *pBatch, wti_t* pWti)
{
DEFiRet;
assert(pBatch != NULL);
- CHKiRet(submitBatch(pAction, pBatch, pBatch->nElem));
- iRet = finishBatch(pAction, pBatch);
+ CHKiRet(submitBatch(pAction, pBatch, pBatch->nElem, pWti));
+ iRet = finishBatch(pAction, pBatch, pWti);
finalize_it:
RETiRet;
@@ -1264,12 +1272,13 @@ finalize_it:
* rgerhards, 2009-04-22
*/
static rsRetVal
-processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate)
+processBatchMain(void *pVoid, batch_t *pBatch, wti_t *pWti, int *pbShutdownImmediate)
{
int *pbShutdownImmdtSave;
sbool *activeSave;
int bMustRestoreActivePtr = 0;
rsRetVal localRet;
+ action_t *pAction = (action_t*) pVoid;
DEFiRet;
assert(pBatch != NULL);
@@ -1288,7 +1297,7 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate)
d_pthread_mutex_lock(&pAction->mutActExec);
pthread_cleanup_push(mutexCancelCleanup, &pAction->mutActExec);
- iRet = processAction(pAction, pBatch);
+ iRet = processAction(pAction, pBatch, pWti);
pthread_cleanup_pop(1); /* unlock mutex */
@@ -1381,7 +1390,7 @@ static rsRetVal setActionQueType(void __attribute__((unused)) *pVal, uchar *pszT
* rgerhards, 2010-06-08
*/
static inline rsRetVal
-doSubmitToActionQ(action_t *pAction, msg_t *pMsg)
+doSubmitToActionQ(action_t *pAction, msg_t *pMsg, wti_t *pWti)
{
DEFiRet;
@@ -1392,7 +1401,7 @@ doSubmitToActionQ(action_t *pAction, msg_t *pMsg)
STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed);
if(pAction->pQueue->qType == QUEUETYPE_DIRECT)
- iRet = qqueueEnqMsgDirect(pAction->pQueue, MsgAddRef(pMsg));
+ iRet = qqueueEnqMsgDirect(pAction->pQueue, MsgAddRef(pMsg), pWti);
else
iRet = qqueueEnqMsg(pAction->pQueue, eFLOWCTL_NO_DELAY, MsgAddRef(pMsg));
@@ -1409,7 +1418,7 @@ finalize_it:
* be filtered out before calling us (what is done currently!).
*/
rsRetVal
-actionWriteToAction(action_t *pAction, msg_t *pMsg)
+actionWriteToAction(action_t *pAction, msg_t *pMsg, wti_t *pWti)
{
DEFiRet;
@@ -1464,7 +1473,7 @@ actionWriteToAction(action_t *pAction, msg_t *pMsg)
/* When we reach this point, we have a valid, non-disabled action.
* So let's enqueue our message for execution. -- rgerhards, 2007-07-24
*/
- iRet = doSubmitToActionQ(pAction, pMsg);
+ iRet = doSubmitToActionQ(pAction, pMsg, pWti);
finalize_it:
RETiRet;
@@ -1475,7 +1484,7 @@ finalize_it:
* pthread_cleanup_push() POSIX macro...
*/
static inline rsRetVal
-doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch)
+doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch, wti_t *pWti)
{
msg_t *pMsg;
DEFiRet;
@@ -1490,7 +1499,7 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch)
}
/* call the output driver */
- iRet = actionWriteToAction(pAction, pMsg);
+ iRet = actionWriteToAction(pAction, pMsg, pWti);
finalize_it:
/* we need to update the batch to handle failover processing correctly */
@@ -1550,7 +1559,7 @@ activateActions(void)
* rgerhards, 2010-06-08
*/
static rsRetVal
-doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch)
+doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti)
{
time_t now = 0;
time_t lastAct;
@@ -1588,7 +1597,7 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch)
}
}
- iRet = doSubmitToActionQBatch(pAction, pBatch);
+ iRet = doSubmitToActionQBatch(pAction, pBatch, pWti);
free(pBatch->active);
pBatch->active = activeSave;
@@ -1613,7 +1622,7 @@ countStatsBatchEnq(action_t *pAction, batch_t *pBatch)
* rgerhards, 2011-06-16
*/
static inline rsRetVal
-doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch)
+doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti)
{
sbool bNeedSubmit;
sbool *activeSave;
@@ -1645,14 +1654,14 @@ doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch)
}
if(bNeedSubmit) {
/* note: stats were already computed above */
- iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch);
+ iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch, pWti);
} else {
DBGPRINTF("no need to submit batch, all invalid\n");
}
} else {
if(GatherStats)
countStatsBatchEnq(pAction, pBatch);
- iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch);
+ iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch, pWti);
}
free(pBatch->active);
@@ -1666,7 +1675,7 @@ doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch)
* rgerhards, 2010-06-08
*/
static rsRetVal
-doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch)
+doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti)
{
int i;
DEFiRet;
@@ -1674,7 +1683,7 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch)
DBGPRINTF("Called action(Batch), logging to %s\n", module.GetStateName(pAction->pMod));
if(pAction->pQueue->qType == QUEUETYPE_DIRECT) {
- iRet = doQueueEnqObjDirectBatch(pAction, pBatch);
+ iRet = doQueueEnqObjDirectBatch(pAction, pBatch, pWti);
} else {/* in this case, we do single submits to the queue.
* TODO: optimize this, we may do at least a multi-submit!
*/
@@ -1684,7 +1693,7 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch)
pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
if( batchIsValidElem(pBatch, i)
&& (pAction->bExecWhenPrevSusp == 0 || pBatch->pElem[i].bPrevWasSuspended == 1)) {
- doSubmitToActionQ(pAction, pBatch->pElem[i].pMsg);
+ doSubmitToActionQ(pAction, pBatch->pElem[i].pMsg, pWti);
}
}
}
@@ -1699,7 +1708,7 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch)
* rgerhards, 2010-06-23
*/
static inline rsRetVal
-helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch)
+helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti)
{
int i;
DEFiRet;
@@ -1712,7 +1721,7 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch)
pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended);
if( batchIsValidElem(pBatch, i)
&& ((pAction->bExecWhenPrevSusp == 0) || pBatch->pElem[i].bPrevWasSuspended) ) {
- doActionCallAction(pAction, pBatch, i);
+ doActionCallAction(pAction, pBatch, i, pWti);
}
}
@@ -1724,13 +1733,13 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch)
*/
#pragma GCC diagnostic ignored "-Wempty-body"
static rsRetVal
-doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch)
+doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti)
{
DEFiRet;
d_pthread_mutex_lock(&pAction->mutAction);
pthread_cleanup_push(mutexCancelCleanup, &pAction->mutAction);
- iRet = helperSubmitToActionQComplexBatch(pAction, pBatch);
+ iRet = helperSubmitToActionQComplexBatch(pAction, pBatch, pWti);
d_pthread_mutex_unlock(&pAction->mutAction);
pthread_cleanup_pop(0); /* remove mutex cleanup handler */
diff --git a/action.h b/action.h
index 54cdb54c..0724dc78 100644
--- a/action.h
+++ b/action.h
@@ -51,6 +51,7 @@ struct action_s {
time_t tActNow; /* the current time for an action execution. Initially set to -1 and
populated on an as-needed basis. This is a performance optimization. */
time_t tLastExec; /* time this action was last executed */
+ int iActionNbr; /* this action's number (ID) */
sbool bExecWhenPrevSusp;/* execute only when previous action is suspended? */
sbool bWriteAllMarkMsgs;/* should all mark msgs be written (not matter how recent the action was executed)? */
int iSecsExecOnceInterval; /* if non-zero, minimum seconds to wait until action is executed again */
@@ -68,7 +69,7 @@ struct action_s {
struct modInfo_s *pMod;/* pointer to output module handling this selector */
void *pModData; /* pointer to module data - content is module-specific */
sbool bRepMsgHasMsg; /* "message repeated..." has msg fragment in it (0-no, 1-yes) */
- rsRetVal (*submitToActQ)(action_t *, batch_t *);/* function submit message to action queue */
+ rsRetVal (*submitToActQ)(action_t *, batch_t *, wti_t*);/* function submit message to action queue */
rsRetVal (*qConstruct)(struct queue_s *pThis);
enum { ACT_STRING_PASSING = 0, ACT_ARRAY_PASSING = 1, ACT_MSG_PASSING = 2,
ACT_JSON_PASSING = 3}
@@ -96,7 +97,7 @@ rsRetVal actionDestruct(action_t *pThis);
rsRetVal actionDbgPrint(action_t *pThis);
rsRetVal actionSetGlobalResumeInterval(int iNewVal);
rsRetVal actionDoAction(action_t *pAction);
-rsRetVal actionWriteToAction(action_t *pAction, msg_t *pMsg);
+rsRetVal actionWriteToAction(action_t *pAction, msg_t *pMsg, wti_t*);
rsRetVal actionCallHUPHdlr(action_t *pAction);
rsRetVal actionClassInit(void);
rsRetVal addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringRequest_t *pOMSR, struct cnfparamvals *actParams, struct nvlst *lst, int bSuspended);
@@ -104,4 +105,7 @@ rsRetVal activateActions(void);
rsRetVal actionNewInst(struct nvlst *lst, action_t **ppAction);
rsRetVal actionProcessCnf(struct cnfobj *o);
+/* external data */
+extern int iActionNbr;
+
#endif /* #ifndef ACTION_H_INCLUDED */
diff --git a/grammar/rainerscript.c b/grammar/rainerscript.c
index 53839474..76b91940 100644
--- a/grammar/rainerscript.c
+++ b/grammar/rainerscript.c
@@ -2427,7 +2427,7 @@ cnfstmtPrintOnly(struct cnfstmt *stmt, int indent, sbool subtree)
free(cstr);
break;
case S_ACT:
- doIndent(indent); dbgprintf("ACTION %p [%s:%s]\n", stmt->d.act,
+ doIndent(indent); dbgprintf("ACTION %d [%s:%s]\n", stmt->d.act->iActionNbr,
modGetName(stmt->d.act->pMod), stmt->printable);
break;
case S_IF:
diff --git a/plugins/mmanon/mmanon.c b/plugins/mmanon/mmanon.c
index 16a4f34b..c714706a 100644
--- a/plugins/mmanon/mmanon.c
+++ b/plugins/mmanon/mmanon.c
@@ -71,6 +71,10 @@ typedef struct _instanceData {
} ipv4;
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
struct modConfData_s {
rsconf_t *pConf; /* our overall config object */
};
@@ -119,6 +123,10 @@ BEGINcreateInstance
CODESTARTcreateInstance
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
BEGINisCompatibleWithFeature
CODESTARTisCompatibleWithFeature
@@ -130,6 +138,11 @@ CODESTARTfreeInstance
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
static inline void
setInstParamDefaults(instanceData *pData)
{
@@ -354,6 +367,7 @@ BEGINdoAction
int lenMsg;
int i;
CODESTARTdoAction
+ pData = pWrkrData->pData;
pMsg = (msg_t*) ppString[0];
lenMsg = getMSGLen(pMsg);
msg = getMSG(pMsg);
@@ -387,6 +401,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
CODEqueryEtryPt_STD_CONF2_QUERIES
ENDqueryEtryPt
diff --git a/plugins/omruleset/omruleset.c b/plugins/omruleset/omruleset.c
index 11765507..2908095e 100644
--- a/plugins/omruleset/omruleset.c
+++ b/plugins/omruleset/omruleset.c
@@ -70,6 +70,10 @@ typedef struct _instanceData {
uchar *pszRulesetName; /* primarily for debugging/display purposes */
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
typedef struct configSettings_s {
ruleset_t *pRuleset; /* ruleset to enqueue message to (NULL = Default, not recommended) */
uchar *pszRulesetName;
diff --git a/plugins/omtesting/omtesting.c b/plugins/omtesting/omtesting.c
index c9f1e06b..7c793414 100644
--- a/plugins/omtesting/omtesting.c
+++ b/plugins/omtesting/omtesting.c
@@ -63,7 +63,6 @@ MODULE_CNFNAME("omtesting")
*/
DEF_OMOD_STATIC_DATA
-
typedef struct _instanceData {
enum { MD_SLEEP, MD_FAIL, MD_RANDFAIL, MD_ALWAYS_SUSPEND }
mode;
@@ -76,6 +75,10 @@ typedef struct _instanceData {
int iCurrRetries;
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
typedef struct configSettings_s {
int bEchoStdout; /* echo non-failed messages to stdout */
} configSettings_t;
diff --git a/runtime/module-template.h b/runtime/module-template.h
index 8a958f90..7657a985 100644
--- a/runtime/module-template.h
+++ b/runtime/module-template.h
@@ -175,6 +175,44 @@ static rsRetVal freeInstance(void* pModData)\
RETiRet;\
}
+/* createWrkrInstance()
+ */
+#define BEGINcreateWrkrInstance \
+static rsRetVal createWrkrInstance(wrkrInstanceData_t **ppWrkrData, instanceData *pData)\
+ {\
+ DEFiRet; /* store error code here */\
+ wrkrInstanceData_t *pWrkrData; /* use this to point to data elements */
+
+#define CODESTARTcreateWrkrInstance \
+ if((pWrkrData = calloc(1, sizeof(wrkrInstanceData_t))) == NULL) {\
+ *ppWrkrData = NULL;\
+ ENDfunc \
+ return RS_RET_OUT_OF_MEMORY;\
+ } \
+ pWrkrData->pData = pData;
+
+#define ENDcreateWrkrInstance \
+ *ppWrkrData = pWrkrData;\
+ RETiRet;\
+}
+
+/* freeWrkrInstance */
+#define BEGINfreeWrkrInstance \
+static rsRetVal freeWrkrInstance(void* pd)\
+{\
+ DEFiRet;\
+ wrkrInstanceData_t *pWrkrData;
+
+#define CODESTARTfreeWrkrInstance \
+ pWrkrData = (wrkrInstanceData_t*) pd;
+
+#define ENDfreeWrkrInstance \
+ if(pWrkrData != NULL)\
+ free(pWrkrData); /* we need to free this in any case */\
+ RETiRet;\
+}
+
+
/* isCompatibleWithFeature()
*/
#define BEGINisCompatibleWithFeature \
@@ -209,8 +247,9 @@ static rsRetVal beginTransaction(instanceData __attribute__((unused)) *pData)\
* introduced in v4.3.3 -- rgerhards, 2009-04-27
*/
#define BEGINendTransaction \
-static rsRetVal endTransaction(instanceData __attribute__((unused)) *pData)\
+static rsRetVal endTransaction(wrkrInstanceData_t __attribute__((unused)) *pWrkrData)\
{\
+ instanceData *pData = NULL; /* deliberately make module abort if it does not support new IF */\
DEFiRet;
#define CODESTARTendTransaction /* currently empty, but may be extended */
@@ -223,8 +262,9 @@ static rsRetVal endTransaction(instanceData __attribute__((unused)) *pData)\
/* doAction()
*/
#define BEGINdoAction \
-static rsRetVal doAction(uchar __attribute__((unused)) **ppString, unsigned __attribute__((unused)) iMsgOpts, instanceData __attribute__((unused)) *pData)\
+static rsRetVal doAction(uchar __attribute__((unused)) **ppString, unsigned __attribute__((unused)) iMsgOpts, wrkrInstanceData_t __attribute__((unused)) *pWrkrData)\
{\
+ instanceData *pData = NULL; /* deliberately make module abort if it does not support new IF */\
DEFiRet;
#define CODESTARTdoAction \
@@ -467,6 +507,13 @@ static rsRetVal queryEtryPt(uchar *name, rsRetVal (**pEtryPoint)())\
*pEtryPoint = tryResume;\
}
+/* standard queries for output module interface in rsyslog v8+ */
+#define CODEqueryEtryPt_STD_OMOD8_QUERIES \
+ else if(!strcmp((char*) name, "createWrkrInstance")) {\
+ *pEtryPoint = createWrkrInstance;\
+ } else if(!strcmp((char*) name, "freeWrkrInstance")) {\
+ *pEtryPoint = freeWrkrInstance;\
+ }
/* the following definition is queryEtryPt block that must be added
* if an output module supports the transactional interface.
diff --git a/runtime/modules.c b/runtime/modules.c
index 56606306..52096082 100644
--- a/runtime/modules.c
+++ b/runtime/modules.c
@@ -652,6 +652,9 @@ doModInit(rsRetVal (*modInit)(int, int*, rsRetVal(**)(), rsRetVal(*)(), modInfo_
CHKiRet((*pNew->modQueryEtryPt)((uchar*)"doAction", &pNew->mod.om.doAction));
CHKiRet((*pNew->modQueryEtryPt)((uchar*)"parseSelectorAct", &pNew->mod.om.parseSelectorAct));
CHKiRet((*pNew->modQueryEtryPt)((uchar*)"tryResume", &pNew->tryResume));
+ CHKiRet((*pNew->modQueryEtryPt)((uchar*)"createWrkrInstance", &pNew->mod.om.createWrkrInstance));
+ CHKiRet((*pNew->modQueryEtryPt)((uchar*)"freeWrkrInstance", &pNew->mod.om.freeWrkrInstance));
+
/* try load optional interfaces */
localRet = (*pNew->modQueryEtryPt)((uchar*)"doHUP", &pNew->doHUP);
if(localRet != RS_RET_OK && localRet != RS_RET_MODULE_ENTRY_POINT_NOT_FOUND)
diff --git a/runtime/modules.h b/runtime/modules.h
index 23df22d6..a0637ac9 100644
--- a/runtime/modules.h
+++ b/runtime/modules.h
@@ -139,6 +139,8 @@ struct modInfo_s {
rsRetVal (*parseSelectorAct)(uchar**, void**,omodStringRequest_t**);
rsRetVal (*newActInst)(uchar *modName, struct nvlst *lst, void **, omodStringRequest_t **);
rsRetVal (*SetShutdownImmdtPtr)(void *pData, void *pPtr);
+ rsRetVal (*createWrkrInstance)(void*ppWrkrData, void*pData);
+ rsRetVal (*freeWrkrInstance)(void*pWrkrData);
} om;
struct { /* data for library modules */
char dummy;
diff --git a/runtime/queue.c b/runtime/queue.c
index 66cb7218..968c016e 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -81,8 +81,8 @@ static rsRetVal GetDeqBatchSize(qqueue_t *pThis, int *pVal);
static rsRetVal ConsumerDA(qqueue_t *pThis, wti_t *pWti);
static rsRetVal batchProcessed(qqueue_t *pThis, wti_t *pWti);
static rsRetVal qqueueMultiEnqObjNonDirect(qqueue_t *pThis, multi_submit_t *pMultiSub);
-static rsRetVal qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub);
-static rsRetVal qAddDirect(qqueue_t *pThis, msg_t *pMsg);
+static rsRetVal qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub, wti_t *pWti);
+static rsRetVal qAddDirect(qqueue_t *pThis, msg_t *pMsg, wti_t *pWti);
static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis);
static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis);
static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis);
@@ -959,7 +959,7 @@ static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis)
return RS_RET_OK;
}
-static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg)
+static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg, wti_t *pWti)
{
batch_t singleBatch;
batch_obj_t batchObj;
@@ -986,7 +986,7 @@ static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg)
singleBatch.pElem = &batchObj;
singleBatch.eltState = &batchState;
singleBatch.active = &active;
- iRet = pThis->pConsumer(pThis->pAction, &singleBatch, &pThis->bShutdownImmediate);
+ iRet = pThis->pConsumer(pThis->pAction, &singleBatch, pWti, &pThis->bShutdownImmediate);
/* delete the batch string params: TODO: create its own "class" for this */
for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) {
free(batchObj.staticActStrings[i]);
@@ -999,7 +999,7 @@ static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg)
/* "enqueue" a batch in direct mode. This is a shortcut which saves all the overhead
* otherwise incured. -- rgerhards, ~2010-06-23
*/
-rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch)
+rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch, wti_t *pWti)
{
DEFiRet;
@@ -1013,7 +1013,7 @@ rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch)
* We use our knowledge about the batch_t structure below, but without that, we
* pay a too-large performance toll... -- rgerhards, 2009-04-22
*/
- iRet = pThis->pConsumer(pThis->pAction, pBatch, NULL);
+ iRet = pThis->pConsumer(pThis->pAction, pBatch, pWti, NULL);
RETiRet;
}
@@ -1319,7 +1319,7 @@ finalize_it:
* to modify some parameters before the queue is actually started.
*/
rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads,
- int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*,int*))
+ int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*, wti_t*, int*))
{
DEFiRet;
qqueue_t *pThis;
@@ -1878,7 +1878,7 @@ ConsumerReg(qqueue_t *pThis, wti_t *pWti)
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave);
- CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, &pThis->bShutdownImmediate));
+ CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, pWti, &pThis->bShutdownImmediate));
/* we now need to check if we should deliberately delay processing a bit
* and, if so, do that. -- rgerhards, 2008-01-30
@@ -2675,7 +2675,7 @@ finalize_it:
/* now, the same function, but for direct mode */
static rsRetVal
-qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub)
+qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub, wti_t *pWti)
{
int i;
DEFiRet;
@@ -2684,7 +2684,7 @@ qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub)
assert(pMultiSub != NULL);
for(i = 0 ; i < pMultiSub->nElem ; ++i) {
- CHKiRet(qAddDirect(pThis, (void*)pMultiSub->ppMsgs[i]));
+ CHKiRet(qAddDirect(pThis, (void*)pMultiSub->ppMsgs[i], pWti));
}
finalize_it:
@@ -2699,16 +2699,16 @@ finalize_it:
* Enqueues the new element and awakes worker thread.
*/
rsRetVal
-qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg)
+qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg, wti_t *pWti)
{
DEFiRet;
ISOBJ_TYPE_assert(pThis, qqueue);
- iRet = qAddDirect(pThis, pMsg);
+ iRet = qAddDirect(pThis, pMsg, pWti);
RETiRet;
}
-/* enqueue a new user data element
+/* enqueue a new user data element
* Enqueues the new element and awakes worker thread.
*/
rsRetVal
diff --git a/runtime/queue.h b/runtime/queue.h
index 844523ad..91900b30 100644
--- a/runtime/queue.h
+++ b/runtime/queue.h
@@ -103,7 +103,7 @@ struct queue_s {
* the user really wanted...). -- rgerhards, 2008-04-02
*/
/* end dequeue time window */
- rsRetVal (*pConsumer)(void *,batch_t*,int*); /* user-supplied consumer function for dequeued messages */
+ rsRetVal (*pConsumer)(void *,batch_t*, wti_t*,int*); /* user-supplied consumer function for dequeued messages */
/* calling interface for pConsumer: arg1 is the global user pointer from this structure, arg2 is the
* user pointer array that was dequeued (actual sample: for actions, arg1 is the pAction and arg2
* is pointer to an array of message message pointers), arg3 is a pointer to an interger which is zero
@@ -195,14 +195,14 @@ struct queue_s {
/* prototypes */
rsRetVal qqueueDestruct(qqueue_t **ppThis);
-rsRetVal qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg);
+rsRetVal qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg, wti_t *pWti);
rsRetVal qqueueEnqMsg(qqueue_t *pThis, flowControl_t flwCtlType, msg_t *pMsg);
rsRetVal qqueueStart(qqueue_t *pThis);
rsRetVal qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize);
rsRetVal qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix);
rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads,
- int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, int*));
-rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch);
+ int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, wti_t *, int*));
+rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch, wti_t *pWti);
int queueCnfParamsSet(struct nvlst *lst);
rsRetVal qqueueApplyCnfParam(qqueue_t *pThis, struct nvlst *lst);
void qqueueSetDefaultsRulesetQueue(qqueue_t *pThis);
diff --git a/runtime/rsconf.c b/runtime/rsconf.c
index 8c808786..abce53b8 100644
--- a/runtime/rsconf.c
+++ b/runtime/rsconf.c
@@ -1205,6 +1205,7 @@ ourConf = loadConf; // TODO: remove, once ourConf is gone!
ABORT_FINALIZE(RS_RET_NO_ACTIONS);
}
tellLexEndParsing();
+ DBGPRINTF("Number of actions in this configuration: %d\n", iActionNbr);
rulesetOptimizeAll(loadConf);
tellCoreConfigLoadDone();
diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index 57cf6afe..db253d28 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -67,8 +67,8 @@ static struct cnfparamblk rspblk =
};
/* forward definitions */
-static rsRetVal processBatch(batch_t *pBatch);
-static rsRetVal scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active);
+static rsRetVal processBatch(batch_t *pBatch, wti_t *pWti);
+static rsRetVal scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active, wti_t *pWti);
/* ---------- linked-list key handling functions (ruleset) ---------- */
@@ -168,7 +168,7 @@ finalize_it:
* rgerhards, 2010-06-15
*/
static inline rsRetVal
-processBatchMultiRuleset(batch_t *pBatch)
+processBatchMultiRuleset(batch_t *pBatch, wti_t *pWti)
{
ruleset_t *currRuleset;
batch_t snglRuleBatch;
@@ -206,7 +206,7 @@ processBatchMultiRuleset(batch_t *pBatch)
snglRuleBatch.nElem = iNew; /* was left just right by the for loop */
batchSetSingleRuleset(&snglRuleBatch, 1);
/* process temp batch */
- processBatch(&snglRuleBatch);
+ processBatch(&snglRuleBatch, pWti);
batchFree(&snglRuleBatch);
} while(bHaveUnprocessed == 1);
@@ -226,12 +226,12 @@ static inline void freeActive(sbool *active) { free(active); }
/* for details, see scriptExec() header comment! */
/* call action for all messages with filter on */
static rsRetVal
-execAct(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
+execAct(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti)
{
DEFiRet;
dbgprintf("RRRR: execAct [%s]: batch of %d elements, active %p\n", modGetName(stmt->d.act->pMod), batchNumMsgs(pBatch), active);
pBatch->active = active;
- stmt->d.act->submitToActQ(stmt->d.act, pBatch);
+ stmt->d.act->submitToActQ(stmt->d.act, pBatch, pWti);
RETiRet;
}
@@ -292,7 +292,7 @@ execStop(batch_t *pBatch, sbool *active)
// set new filter, inverted
// perform else (if any messages)
static rsRetVal
-execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
+execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti)
{
sbool *newAct;
int i;
@@ -321,7 +321,7 @@ execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
}
if(stmt->d.s_if.t_then != NULL) {
- scriptExec(stmt->d.s_if.t_then, pBatch, newAct);
+ scriptExec(stmt->d.s_if.t_then, pBatch, newAct, pWti);
}
if(stmt->d.s_if.t_else != NULL) {
for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
@@ -331,7 +331,7 @@ execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
&& (active == NULL || active[i]))
newAct[i] = !newAct[i];
}
- scriptExec(stmt->d.s_if.t_else, pBatch, newAct);
+ scriptExec(stmt->d.s_if.t_else, pBatch, newAct, pWti);
}
freeActive(newAct);
finalize_it:
@@ -340,7 +340,7 @@ finalize_it:
/* for details, see scriptExec() header comment! */
static void
-execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
+execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti)
{
sbool *newAct;
msg_t *pMsg;
@@ -367,7 +367,7 @@ execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
}
if(stmt->d.s_prifilt.t_then != NULL) {
- scriptExec(stmt->d.s_prifilt.t_then, pBatch, newAct);
+ scriptExec(stmt->d.s_prifilt.t_then, pBatch, newAct, pWti);
}
if(stmt->d.s_prifilt.t_else != NULL) {
for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
@@ -377,7 +377,7 @@ execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
&& (active == NULL || active[i]))
newAct[i] = !newAct[i];
}
- scriptExec(stmt->d.s_prifilt.t_else, pBatch, newAct);
+ scriptExec(stmt->d.s_prifilt.t_else, pBatch, newAct, pWti);
}
freeActive(newAct);
}
@@ -476,7 +476,7 @@ done:
/* for details, see scriptExec() header comment! */
static void
-execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
+execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti)
{
sbool *thenAct;
sbool bRet;
@@ -495,7 +495,7 @@ execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
DBGPRINTF("batch: item %d PROPFILT %d\n", i, thenAct[i]);
}
- scriptExec(stmt->d.s_propfilt.t_then, pBatch, thenAct);
+ scriptExec(stmt->d.s_propfilt.t_then, pBatch, thenAct, pWti);
freeActive(thenAct);
}
@@ -510,7 +510,7 @@ execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
* rgerhards, 2012-09-04
*/
static rsRetVal
-scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active)
+scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active, wti_t *pWti)
{
DEFiRet;
struct cnfstmt *stmt;
@@ -528,7 +528,7 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active)
execStop(pBatch, active);
break;
case S_ACT:
- execAct(stmt, pBatch, active);
+ execAct(stmt, pBatch, active, pWti);
break;
case S_SET:
execSet(stmt, pBatch, active);
@@ -537,16 +537,16 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active)
execUnset(stmt, pBatch, active);
break;
case S_CALL:
- scriptExec(stmt->d.s_call.stmt, pBatch, active);
+ scriptExec(stmt->d.s_call.stmt, pBatch, active, pWti);
break;
case S_IF:
- execIf(stmt, pBatch, active);
+ execIf(stmt, pBatch, active, pWti);
break;
case S_PRIFILT:
- execPRIFILT(stmt, pBatch, active);
+ execPRIFILT(stmt, pBatch, active, pWti);
break;
case S_PROPFILT:
- execPROPFILT(stmt, pBatch, active);
+ execPROPFILT(stmt, pBatch, active, pWti);
break;
default:
dbgprintf("error: unknown stmt type %u during exec\n",
@@ -565,7 +565,7 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active)
* rgerhards, 2005-10-13
*/
static rsRetVal
-processBatch(batch_t *pBatch)
+processBatch(batch_t *pBatch, wti_t *pWti)
{
ruleset_t *pThis;
DEFiRet;
@@ -577,9 +577,9 @@ processBatch(batch_t *pBatch)
if(pThis == NULL)
pThis = ourConf->rulesets.pDflt;
ISOBJ_TYPE_assert(pThis, ruleset);
- CHKiRet(scriptExec(pThis->root, pBatch, NULL));
+ CHKiRet(scriptExec(pThis->root, pBatch, NULL, pWti));
} else {
- CHKiRet(processBatchMultiRuleset(pBatch));
+ CHKiRet(processBatchMultiRuleset(pBatch, pWti));
}
finalize_it:
diff --git a/runtime/ruleset.h b/runtime/ruleset.h
index 8bfd4920..05da144d 100644
--- a/runtime/ruleset.h
+++ b/runtime/ruleset.h
@@ -46,7 +46,7 @@ BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */
rsRetVal (*Destruct)(ruleset_t **ppThis);
rsRetVal (*DestructAllActions)(rsconf_t *conf);
rsRetVal (*SetName)(ruleset_t *pThis, uchar *pszName);
- rsRetVal (*ProcessBatch)(batch_t*);
+ rsRetVal (*ProcessBatch)(batch_t*, wti_t *);
rsRetVal (*GetRuleset)(rsconf_t *conf, ruleset_t **ppThis, uchar*);
rsRetVal (*SetDefaultRuleset)(rsconf_t *conf, uchar*);
rsRetVal (*SetCurrRuleset)(rsconf_t *conf, uchar*);
@@ -64,8 +64,9 @@ BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */
/* AddRule() removed */
/*TODO:REMOVE*/rsRetVal (*IterateAllActions)(rsconf_t *conf, rsRetVal (*pFunc)(void*, void*), void* pParam);
void (*AddScript)(ruleset_t *pThis, struct cnfstmt *script);
+ /* v8: changed processBatch interface */
ENDinterface(ruleset)
-#define rulesetCURR_IF_VERSION 7 /* increment whenever you change the interface structure! */
+#define rulesetCURR_IF_VERSION 8 /* increment whenever you change the interface structure! */
/* prototypes */
diff --git a/runtime/wti.c b/runtime/wti.c
index f91fb5a9..e9d3599d 100644
--- a/runtime/wti.c
+++ b/runtime/wti.c
@@ -44,6 +44,7 @@
#include "wti.h"
#include "obj.h"
#include "glbl.h"
+#include "action.h"
#include "atomic.h"
/* static data */
@@ -171,6 +172,7 @@ BEGINobjDestruct(wti) /* be sure to specify the object type also in END and CODE
CODESTARTobjDestruct(wti)
/* actual destruction */
batchFree(&pThis->batch);
+ free(pThis->actWrkrData);
DESTROY_ATOMIC_HELPER_MUT(pThis->mutIsRunning);
free(pThis->pszDbgHdr);
@@ -195,11 +197,15 @@ wtiConstructFinalize(wti_t *pThis)
ISOBJ_TYPE_assert(pThis, wti);
- DBGPRINTF("%s: finalizing construction of worker instance data\n", wtiGetDbgHdr(pThis));
+ DBGPRINTF("%s: finalizing construction of worker instance data (for %d actions)\n",
+ wtiGetDbgHdr(pThis), iActionNbr);
/* initialize our thread instance descriptor (no concurrency here) */
pThis->bIsRunning = RSFALSE;
+ /* must use calloc as we need zeto-init */
+ CHKmalloc(pThis->actWrkrData = calloc(iActionNbr, sizeof(void*)));
+
/* we now alloc the array for user pointers. We obtain the max from the queue itself. */
CHKiRet(pThis->pWtp->pfGetDeqBatchSize(pThis->pWtp->pUsr, &iDeqBatchSize));
CHKiRet(batchInit(&pThis->batch, iDeqBatchSize));
diff --git a/runtime/wti.h b/runtime/wti.h
index 014251f0..297fb999 100644
--- a/runtime/wti.h
+++ b/runtime/wti.h
@@ -37,6 +37,7 @@ struct wti_s {
wtp_t *pWtp; /* my worker thread pool (important if only the work thread instance is passed! */
batch_t batch; /* pointer to an object array meaningful for current user pointer (e.g. queue pUsr data elemt) */
uchar *pszDbgHdr; /* header string for debug messages */
+ void **actWrkrData; /* *array* of action wrkr data pointers (sized for max nbr of actions in config!) */
DEF_ATOMIC_HELPER_MUT(mutIsRunning);
};
diff --git a/runtime/wtp.c b/runtime/wtp.c
index 19151e7c..1b960eb9 100644
--- a/runtime/wtp.c
+++ b/runtime/wtp.c
@@ -8,7 +8,7 @@
* (and in the web doc set on http://www.rsyslog.com/doc). Be sure to read it
* if you are getting aquainted to the object.
*
- * Copyright 2008,2009 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2008-2013 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of the rsyslog runtime library.
*
@@ -91,6 +91,7 @@ BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro!
pthread_cond_init(&pThis->condThrdTrm, NULL);
pthread_attr_init(&pThis->attrThrd);
/* Set thread scheduling policy to default */
+#warning do we need this any longer? I think it was a cure for an already fixed bug..
#ifdef HAVE_PTHREAD_SETSCHEDPARAM
pthread_attr_setschedpolicy(&pThis->attrThrd, default_thr_sched_policy);
pthread_attr_setschedparam(&pThis->attrThrd, &default_sched_param);
@@ -121,7 +122,8 @@ wtpConstructFinalize(wtp_t *pThis)
ISOBJ_TYPE_assert(pThis, wtp);
- DBGPRINTF("%s: finalizing construction of worker thread pool\n", wtpGetDbgHdr(pThis));
+ DBGPRINTF("%s: finalizing construction of worker thread pool (numworkerThreads %d)\n",
+ wtpGetDbgHdr(pThis), pThis->iNumWorkerThreads);
/* alloc and construct workers - this can only be done in finalizer as we previously do
* not know the max number of workers
*/
diff --git a/tools/omdiscard.c b/tools/omdiscard.c
index 15c6ea82..a76bcc33 100644
--- a/tools/omdiscard.c
+++ b/tools/omdiscard.c
@@ -6,7 +6,7 @@
*
* File begun on 2007-07-24 by RGerhards
*
- * Copyright 2007-2012 Adiscon GmbH.
+ * Copyright 2007-2013 Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -49,6 +49,10 @@ typedef struct _instanceData {
EMPTY_STRUCT
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
/* we do not need a createInstance()!
BEGINcreateInstance
CODESTARTcreateInstance
@@ -56,6 +60,11 @@ ENDcreateInstance
*/
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINdbgPrintInstInfo
CODESTARTdbgPrintInstInfo
/* do nothing */
@@ -87,6 +96,11 @@ CODESTARTfreeInstance
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINparseSelectorAct
CODESTARTparseSelectorAct
CODE_STD_STRING_REQUESTparseSelectorAct(0)
@@ -114,6 +128,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
ENDqueryEtryPt
diff --git a/tools/omfile.c b/tools/omfile.c
index fdcf355a..4300bb4d 100644
--- a/tools/omfile.c
+++ b/tools/omfile.c
@@ -133,6 +133,7 @@ typedef struct s_dynaFileCacheEntry dynaFileCacheEntry;
typedef struct _instanceData {
+ pthread_mutex_t mutWrite; /* guard against multiple instances writing to single file */
uchar *f_fname; /* file or template name (display only) */
uchar *tplName; /* name of assigned template */
strm_t *pStrm; /* our output stream */
@@ -181,6 +182,20 @@ typedef struct _instanceData {
STATSCOUNTER_DEF(ctrMax, mutCtrMax);
} instanceData;
+/* to build a linked list for temporary storage of lines while we cannot commit */
+typedef struct linebuf {
+ uchar *filename; /* for dynafiles, make go away */
+ uchar *ln;
+ unsigned iMsgOpts;
+ struct linebuf *pNext;
+} linebuf_t;
+
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+ linebuf_t *pRoot;
+ linebuf_t *pLast;
+} wrkrInstanceData_t;
+
typedef struct configSettings_s {
int iDynaFileCacheSize; /* max cache for dynamic files */
@@ -786,17 +801,19 @@ finalize_it:
/* rgerhards 2004-11-11: write to a file output. */
static rsRetVal
-writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pData)
+writeFile(instanceData *pData, linebuf_t *linebuf)
{
DEFiRet;
ASSERT(pData != NULL);
+ pthread_mutex_lock(&pData->mutWrite);
+
/* first check if we have a dynamic file name and, if so,
* check if it still is ok or a new file needs to be created
*/
if(pData->bDynamicName) {
- CHKiRet(prepareDynFile(pData, ppString[1], iMsgOpts));
+ CHKiRet(prepareDynFile(pData, linebuf->filename, linebuf->iMsgOpts));
} else { /* "regular", non-dynafile */
if(pData->pStrm == NULL) {
CHKiRet(prepareFile(pData, pData->f_fname));
@@ -806,9 +823,10 @@ writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pData)
}
}
- CHKiRet(doWrite(pData, ppString[0], strlen(CHAR_CONVERT(ppString[0]))));
+ CHKiRet(doWrite(pData, linebuf->ln, ustrlen(linebuf->ln)));
finalize_it:
+ pthread_mutex_unlock(&pData->mutWrite);
RETiRet;
}
@@ -888,9 +906,15 @@ ENDfreeCnf
BEGINcreateInstance
CODESTARTcreateInstance
pData->pStrm = NULL;
+ pthread_mutex_init(&pData->mutWrite, NULL);
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINfreeInstance
CODESTARTfreeInstance
free(pData->tplName);
@@ -913,9 +937,15 @@ CODESTARTfreeInstance
free(pData->cryprovName);
free(pData->cryprovNameFull);
}
+ pthread_mutex_destroy(&pData->mutWrite);
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINtryResume
CODESTARTtryResume
ENDtryResume
@@ -926,8 +956,61 @@ CODESTARTbeginTransaction
ENDbeginTransaction
+static rsRetVal
+bufferLine(wrkrInstanceData_t *pWrkrData, uchar *filename, uchar *line)
+{
+ linebuf_t *lb;
+ DEFiRet;
+
+ dbgprintf("DDDD: buffering root %p, line %s\n", pWrkrData->pRoot, line);
+ CHKmalloc(lb = (linebuf_t*) malloc(sizeof(linebuf_t)));
+ CHKmalloc(lb->filename = ustrdup(filename));
+ CHKmalloc(lb->ln = ustrdup(line));
+ lb->pNext = NULL;
+ if(pWrkrData->pRoot == NULL) {
+ pWrkrData->pRoot = pWrkrData->pLast = lb;
+ } else {
+ pWrkrData->pLast->pNext = lb;
+ pWrkrData->pLast = lb;
+ }
+finalize_it:
+ RETiRet;
+}
+
+static void
+submitCachedLines(wrkrInstanceData_t *pWrkrData, instanceData *pData)
+{
+ linebuf_t *curr, *todel;
+
+ for(curr = pWrkrData->pRoot ; curr != NULL ; ) {
+ DBGPRINTF("omfile: file to log to: %s\n", curr->filename);
+ DBGPRINTF("omfile: start of data: '%.128s'\n", curr->ln);
+ STATSCOUNTER_INC(pData->ctrRequests, pData->mutCtrRequests);
+ writeFile(pData, curr);
+
+ todel = curr;
+ curr = curr->pNext;
+ free(todel->filename);
+ free(todel->ln);
+ free(todel);
+ }
+ pWrkrData->pRoot = NULL;
+}
+
+
+BEGINdoAction
+CODESTARTdoAction
+ pData = pWrkrData->pData;
+ iRet = bufferLine(pWrkrData, (pData->bDynamicName) ? ppString[1] : pData->f_fname,
+ ppString[0]);
+ if(iRet == RS_RET_OK)
+ iRet = RS_RET_DEFER_COMMIT;
+ENDdoAction
+
BEGINendTransaction
CODESTARTendTransaction
+ pData = pWrkrData->pData;
+ submitCachedLines(pWrkrData, pData);
/* Note: pStrm may be NULL if there was an error opening the stream */
if(pData->bFlushOnTXEnd && pData->pStrm != NULL) {
/* if we have an async writer, it controls the flush via
@@ -941,21 +1024,6 @@ finalize_it:
ENDendTransaction
-BEGINdoAction
-CODESTARTdoAction
- DBGPRINTF("file to log to: %s\n",
- (pData->bDynamicName) ? ppString[1] : pData->f_fname);
- DBGPRINTF("omfile: start of data: '%.128s'\n", ppString[0]);
- STATSCOUNTER_INC(pData->ctrRequests, pData->mutCtrRequests);
- CHKiRet(writeFile(ppString, iMsgOpts, pData));
- if(!bCoreSupportsBatching && pData->bFlushOnTXEnd) {
- CHKiRet(strm.Flush(pData->pStrm));
- }
-finalize_it:
- if(iRet == RS_RET_OK)
- iRet = RS_RET_DEFER_COMMIT;
-ENDdoAction
-
static inline void
setInstParamDefaults(instanceData *pData)
@@ -1336,6 +1404,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
BEGINdoHUP
CODESTARTdoHUP
+ pthread_mutex_lock(&pData->mutWrite);
if(pData->bDynamicName) {
dynaFileFreeCacheEntries(pData);
} else {
@@ -1343,6 +1412,7 @@ CODESTARTdoHUP
closeFile(pData);
}
}
+ pthread_mutex_unlock(&pData->mutWrite);
ENDdoHUP
@@ -1358,6 +1428,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_STD_CONF2_QUERIES
CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES
CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
diff --git a/tools/omfwd.c b/tools/omfwd.c
index 6e5cf809..ca33cac2 100644
--- a/tools/omfwd.c
+++ b/tools/omfwd.c
@@ -112,6 +112,10 @@ typedef struct _instanceData {
int errsToReport; /* (remaining) number of errors to report */
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
/* config data */
typedef struct configSettings_s {
uchar *pszTplName; /* name of the default template to use */
@@ -335,6 +339,11 @@ CODESTARTcreateInstance
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINisCompatibleWithFeature
CODESTARTisCompatibleWithFeature
if(eFeat == sFEATURERepeatedMsgReduction)
@@ -360,6 +369,11 @@ CODESTARTfreeInstance
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINdbgPrintInstInfo
CODESTARTdbgPrintInstInfo
dbgprintf("%s", pData->target);
@@ -1251,6 +1265,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_STD_CONF2_QUERIES
CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES
CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
diff --git a/tools/ompipe.c b/tools/ompipe.c
index df8066b1..eab7f694 100644
--- a/tools/ompipe.c
+++ b/tools/ompipe.c
@@ -12,7 +12,7 @@
* NOTE: read comments in module-template.h to understand how this pipe
* works!
*
- * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2013 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -72,6 +72,10 @@ typedef struct _instanceData {
sbool bHadError; /* did we already have/report an error on this pipe? */
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
typedef struct configSettings_s {
EMPTY_STRUCT
} configSettings_t;
@@ -279,6 +283,11 @@ CODESTARTcreateInstance
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINfreeInstance
CODESTARTfreeInstance
free(pData->pipe);
@@ -287,6 +296,11 @@ CODESTARTfreeInstance
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINtryResume
CODESTARTtryResume
ENDtryResume
@@ -390,6 +404,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_doHUP
CODEqueryEtryPt_STD_CONF2_QUERIES
CODEqueryEtryPt_STD_CONF2_CNFNAME_QUERIES
diff --git a/tools/omshell.c b/tools/omshell.c
index ac62fa62..6680faab 100644
--- a/tools/omshell.c
+++ b/tools/omshell.c
@@ -63,11 +63,20 @@ typedef struct _instanceData {
uchar progName[MAXFNAME]; /* program to execute */
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
BEGINcreateInstance
CODESTARTcreateInstance
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINisCompatibleWithFeature
CODESTARTisCompatibleWithFeature
if(eFeat == sFEATURERepeatedMsgReduction)
@@ -80,6 +89,11 @@ CODESTARTfreeInstance
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINdbgPrintInstInfo
CODESTARTdbgPrintInstInfo
printf("%s", pData->progName);
@@ -139,6 +153,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
ENDqueryEtryPt
diff --git a/tools/omusrmsg.c b/tools/omusrmsg.c
index f4cc4094..cab2a4fe 100644
--- a/tools/omusrmsg.c
+++ b/tools/omusrmsg.c
@@ -8,7 +8,7 @@
* File begun on 2007-07-20 by RGerhards (extracted from syslogd.c, which at the
* time of the fork from sysklogd was under BSD license)
*
- * Copyright 2007-2012 Adiscon GmbH.
+ * Copyright 2007-2013 Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -87,6 +87,10 @@ typedef struct _instanceData {
uchar *tplName;
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
typedef struct configSettings_s {
EMPTY_STRUCT
} configSettings_t;
@@ -115,6 +119,11 @@ CODESTARTcreateInstance
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINisCompatibleWithFeature
CODESTARTisCompatibleWithFeature
if(eFeat == sFEATURERepeatedMsgReduction)
@@ -128,6 +137,11 @@ CODESTARTfreeInstance
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINdbgPrintInstInfo
register int i;
CODESTARTdbgPrintInstInfo
@@ -435,6 +449,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
ENDqueryEtryPt
diff --git a/tools/syslogd.c b/tools/syslogd.c
index aaeb9866..7597b05d 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -560,13 +560,13 @@ finalize_it:
* for the main queue.
*/
static rsRetVal
-msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch, int *pbShutdownImmediate)
+msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch, wti_t *pWti, int *pbShutdownImmediate)
{
DEFiRet;
assert(pBatch != NULL);
pBatch->pbShutdownImmediate = pbShutdownImmediate; /* TODO: move this to batch creation! */
preprocessBatch(pBatch);
- ruleset.ProcessBatch(pBatch);
+ ruleset.ProcessBatch(pBatch, pWti);
//TODO: the BATCH_STATE_COMM must be set somewhere down the road, but we
//do not have this yet and so we emulate -- 2010-06-10
int i;
generated by cgit v1.2.3 (git 2.25.1) at 2025-07-14 22:17:47 +0000