diff options
Diffstat (limited to 'action.c')
| -rw-r--r-- | action.c | 790 |
1 files changed, 542 insertions, 248 deletions
@@ -8,7 +8,7 @@ * the right code in question): For performance reasons, this module * uses different methods of message submission based on the user-selected * configuration. This code is similar, but can not be abstracted because - * of the performanse-affecting differences in it. As such, it is often + * of the performance-affecting differences in it. As such, it is often * necessary to triple-check that everything works well in *all* modes. * The different modes (and calling sequence) are: * @@ -20,7 +20,7 @@ * - actionWriteToAction * - qqueueEnqObj * (now queue engine processing) - * if(pThis->bWriteAllMarkMsgs == FALSE) - this is the DEFAULT + * if(pThis->bWriteAllMarkMsgs == RSFALSE) - this is the DEFAULT * - doSubmitToActionQNotAllMarkBatch * - doSubmitToActionQBatch (and from here like in the else case below!) * else @@ -98,20 +98,23 @@ #include <strings.h> #include <time.h> #include <errno.h> +#include <json/json.h> #include "dirty.h" #include "template.h" #include "action.h" #include "modules.h" -#include "sync.h" #include "cfsysline.h" #include "srUtils.h" #include "errmsg.h" #include "batch.h" #include "wti.h" +#include "rsconf.h" #include "datetime.h" #include "unicode-helper.h" #include "atomic.h" +#include "ruleset.h" +#include "statsobj.h" #define NO_TIME_PROVIDED 0 /* indicate we do not provide any cached time */ @@ -127,39 +130,47 @@ DEFobjCurrIf(obj) DEFobjCurrIf(datetime) DEFobjCurrIf(module) DEFobjCurrIf(errmsg) - -static int iActExecOnceInterval = 0; /* execute action once every nn seconds */ -static int iActExecEveryNthOccur = 0; /* execute action every n-th occurence (0,1=always) */ -static time_t iActExecEveryNthOccurTO = 0; /* timeout for n-occurence setting (in seconds, 0=never) */ -static int glbliActionResumeInterval = 30; -int glbliActionResumeRetryCount = 0; /* how often should suspended actions be retried? */ -static int bActionRepMsgHasMsg = 0; /* last messsage repeated... has msg fragment in it */ - -static int bActionWriteAllMarkMsgs = FALSE; /* should all mark messages be unconditionally written? */ -static uchar *pszActionName; /* short name for the action */ -/* action queue and its configuration parameters */ -static queueType_t ActionQueType = QUEUETYPE_DIRECT; /* type of the main message queue above */ -static int iActionQueueSize = 1000; /* size of the main message queue above */ -static int iActionQueueDeqBatchSize = 16; /* batch size for action queues */ -static int iActionQHighWtrMark = 800; /* high water mark for disk-assisted queues */ -static int iActionQLowWtrMark = 200; /* low water mark for disk-assisted queues */ -static int iActionQDiscardMark = 9800; /* begin to discard messages */ -static int iActionQDiscardSeverity = 8; /* by default, discard nothing to prevent unintentional loss */ -static int iActionQueueNumWorkers = 1; /* number of worker threads for the mm queue above */ -static uchar *pszActionQFName = NULL; /* prefix for the main message queue file */ -static int64 iActionQueMaxFileSize = 1024*1024; -static int iActionQPersistUpdCnt = 0; /* persist queue info every n updates */ -static int bActionQSyncQeueFiles = 0; /* sync queue files */ -static int iActionQtoQShutdown = 0; /* queue shutdown */ -static int iActionQtoActShutdown = 1000; /* action shutdown (in phase 2) */ -static int iActionQtoEnq = 2000; /* timeout for queue enque */ -static int iActionQtoWrkShutdown = 60000; /* timeout for worker thread shutdown */ -static int iActionQWrkMinMsgs = 100; /* minimum messages per worker needed to start a new one */ -static int bActionQSaveOnShutdown = 1; /* save queue on shutdown (when DA enabled)? */ -static int64 iActionQueMaxDiskSpace = 0; /* max disk space allocated 0 ==> unlimited */ -static int iActionQueueDeqSlowdown = 0; /* dequeue slowdown (simple rate limiting) */ -static int iActionQueueDeqtWinFromHr = 0; /* hour begin of time frame when queue is to be dequeued */ -static int iActionQueueDeqtWinToHr = 25; /* hour begin of time frame when queue is to be dequeued */ +DEFobjCurrIf(statsobj) +DEFobjCurrIf(ruleset) + + +typedef struct configSettings_s { + int bActExecWhenPrevSusp; /* execute action only when previous one was suspended? */ + int bActionWriteAllMarkMsgs; /* should all mark messages be unconditionally written? */ + int iActExecOnceInterval; /* execute action once every nn seconds */ + int iActExecEveryNthOccur; /* execute action every n-th occurence (0,1=always) */ + time_t iActExecEveryNthOccurTO; /* timeout for n-occurence setting (in seconds, 0=never) */ + int glbliActionResumeInterval; + int glbliActionResumeRetryCount; /* how often should suspended actions be retried? */ + int bActionRepMsgHasMsg; /* last messsage repeated... has msg fragment in it */ + uchar *pszActionName; /* short name for the action */ + /* action queue and its configuration parameters */ + queueType_t ActionQueType; /* type of the main message queue above */ + int iActionQueueSize; /* size of the main message queue above */ + int iActionQueueDeqBatchSize; /* batch size for action queues */ + int iActionQHighWtrMark; /* high water mark for disk-assisted queues */ + int iActionQLowWtrMark; /* low water mark for disk-assisted queues */ + int iActionQDiscardMark; /* begin to discard messages */ + int iActionQDiscardSeverity; /* by default, discard nothing to prevent unintentional loss */ + int iActionQueueNumWorkers; /* number of worker threads for the mm queue above */ + uchar *pszActionQFName; /* prefix for the main message queue file */ + int64 iActionQueMaxFileSize; + int iActionQPersistUpdCnt; /* persist queue info every n updates */ + int bActionQSyncQeueFiles; /* sync queue files */ + int iActionQtoQShutdown; /* queue shutdown */ + int iActionQtoActShutdown; /* action shutdown (in phase 2) */ + int iActionQtoEnq; /* timeout for queue enque */ + int iActionQtoWrkShutdown; /* timeout for worker thread shutdown */ + int iActionQWrkMinMsgs; /* minimum messages per worker needed to start a new one */ + int bActionQSaveOnShutdown; /* save queue on shutdown (when DA enabled)? */ + int64 iActionQueMaxDiskSpace; /* max disk space allocated 0 ==> unlimited */ + int iActionQueueDeqSlowdown; /* dequeue slowdown (simple rate limiting) */ + int iActionQueueDeqtWinFromHr; /* hour begin of time frame when queue is to be dequeued */ + int iActionQueueDeqtWinToHr; /* hour begin of time frame when queue is to be dequeued */ +} configSettings_t; + +configSettings_t cs; /* our current config settings */ +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 @@ -169,6 +180,25 @@ static int iActionQueueDeqtWinToHr = 25; /* hour begin of time frame when queu */ static int iActionNbr = 0; +/* tables for interfacing with the v6 config system */ +static struct cnfparamdescr cnfparamdescr[] = { + { "name", eCmdHdlrGetWord, 0 }, /* legacy: actionname */ + { "type", eCmdHdlrString, CNFPARAM_REQUIRED }, /* legacy: actionname */ + { "action.writeallmarkmessages", eCmdHdlrBinary, 0 }, /* legacy: actionwriteallmarkmessages */ + { "action.execonlyeverynthtime", eCmdHdlrInt, 0 }, /* legacy: actionexeconlyeverynthtime */ + { "action.execonlyeverynthtimetimeout", eCmdHdlrInt, 0 }, /* legacy: actionexeconlyeverynthtimetimeout */ + { "action.execonlyonceeveryinterval", eCmdHdlrInt, 0 }, /* legacy: actionexeconlyonceeveryinterval */ + { "action.execonlywhenpreviousissuspended", eCmdHdlrInt, 0 }, /* legacy: actionexeconlywhenpreviousissuspended */ + { "action.repeatedmsgcontainsoriginalmsg", eCmdHdlrBinary, 0 }, /* legacy: repeatedmsgcontainsoriginalmsg */ + { "action.resumeretrycount", eCmdHdlrInt, 0 }, /* legacy: actionresumeretrycount */ + { "action.resumeinterval", eCmdHdlrInt, 0 } +}; +static struct cnfparamblk pblk = + { CNFPARAMBLK_VERSION, + sizeof(cnfparamdescr)/sizeof(struct cnfparamdescr), + cnfparamdescr + }; + /* ------------------------------ methods ------------------------------ */ /* This function returns the "current" time for this action. Current time @@ -220,32 +250,32 @@ actionResetQueueParams(void) { DEFiRet; - ActionQueType = QUEUETYPE_DIRECT; /* type of the main message queue above */ - iActionQueueSize = 1000; /* size of the main message queue above */ - iActionQueueDeqBatchSize = 16; /* default batch size */ - iActionQHighWtrMark = 800; /* high water mark for disk-assisted queues */ - iActionQLowWtrMark = 200; /* low water mark for disk-assisted queues */ - iActionQDiscardMark = 9800; /* begin to discard messages */ - iActionQDiscardSeverity = 8; /* discard warning and above */ - iActionQueueNumWorkers = 1; /* number of worker threads for the mm queue above */ - iActionQueMaxFileSize = 1024*1024; - iActionQPersistUpdCnt = 0; /* persist queue info every n updates */ - bActionQSyncQeueFiles = 0; - iActionQtoQShutdown = 0; /* queue shutdown */ - iActionQtoActShutdown = 1000; /* action shutdown (in phase 2) */ - iActionQtoEnq = 2000; /* timeout for queue enque */ - iActionQtoWrkShutdown = 60000; /* timeout for worker thread shutdown */ - iActionQWrkMinMsgs = 100; /* minimum messages per worker needed to start a new one */ - bActionQSaveOnShutdown = 1; /* save queue on shutdown (when DA enabled)? */ - iActionQueMaxDiskSpace = 0; - iActionQueueDeqSlowdown = 0; - iActionQueueDeqtWinFromHr = 0; - iActionQueueDeqtWinToHr = 25; /* 25 disables time windowed dequeuing */ - - glbliActionResumeRetryCount = 0; /* I guess it is smart to reset this one, too */ - - d_free(pszActionQFName); - pszActionQFName = NULL; /* prefix for the main message queue file */ + cs.ActionQueType = QUEUETYPE_DIRECT; /* type of the main message queue above */ + cs.iActionQueueSize = 1000; /* size of the main message queue above */ + cs.iActionQueueDeqBatchSize = 16; /* default batch size */ + cs.iActionQHighWtrMark = 800; /* high water mark for disk-assisted queues */ + cs.iActionQLowWtrMark = 200; /* low water mark for disk-assisted queues */ + cs.iActionQDiscardMark = 9800; /* begin to discard messages */ + cs.iActionQDiscardSeverity = 8; /* discard warning and above */ + cs.iActionQueueNumWorkers = 1; /* number of worker threads for the mm queue above */ + cs.iActionQueMaxFileSize = 1024*1024; + cs.iActionQPersistUpdCnt = 0; /* persist queue info every n updates */ + cs.bActionQSyncQeueFiles = 0; + cs.iActionQtoQShutdown = 0; /* queue shutdown */ + cs.iActionQtoActShutdown = 1000; /* action shutdown (in phase 2) */ + cs.iActionQtoEnq = 50; /* timeout for queue enque */ + cs.iActionQtoWrkShutdown = 60000; /* timeout for worker thread shutdown */ + cs.iActionQWrkMinMsgs = 100; /* minimum messages per worker needed to start a new one */ + cs.bActionQSaveOnShutdown = 1; /* save queue on shutdown (when DA enabled)? */ + cs.iActionQueMaxDiskSpace = 0; + cs.iActionQueueDeqSlowdown = 0; + cs.iActionQueueDeqtWinFromHr = 0; + cs.iActionQueueDeqtWinToHr = 25; /* 25 disables time windowed dequeuing */ + + cs.glbliActionResumeRetryCount = 0; /* I guess it is smart to reset this one, too */ + + d_free(cs.pszActionQFName); + cs.pszActionQFName = NULL; /* prefix for the main message queue file */ RETiRet; } @@ -259,29 +289,42 @@ rsRetVal actionDestruct(action_t *pThis) DEFiRet; ASSERT(pThis != NULL); + if(!strcmp((char*)modGetName(pThis->pMod), "builtin:omdiscard")) { + /* discard actions will be optimized out */ + FINALIZE; + } + if(pThis->pQueue != NULL) { qqueueDestruct(&pThis->pQueue); } + /* destroy stats object, if we have one (may not always be + * be the case, e.g. if turned off) + */ + if(pThis->statsobj != NULL) + statsobj.Destruct(&pThis->statsobj); + if(pThis->pMod != NULL) pThis->pMod->freeInstance(pThis->pModData); if(pThis->f_pMsg != NULL) msgDestruct(&pThis->f_pMsg); - SYNC_OBJ_TOOL_EXIT(pThis); + pthread_mutex_destroy(&pThis->mutAction); pthread_mutex_destroy(&pThis->mutActExec); d_free(pThis->pszName); d_free(pThis->ppTpl); +finalize_it: d_free(pThis); - RETiRet; } /* create a new action descriptor object * rgerhards, 2007-08-01 + * Note that it is vital to set proper initial values as the v6 config + * system depends on these! */ rsRetVal actionConstruct(action_t **ppThis) { @@ -291,12 +334,19 @@ rsRetVal actionConstruct(action_t **ppThis) ASSERT(ppThis != NULL); CHKmalloc(pThis = (action_t*) calloc(1, sizeof(action_t))); - pThis->iResumeInterval = glbliActionResumeInterval; - pThis->iResumeRetryCount = glbliActionResumeRetryCount; + pThis->iResumeInterval = 30; + pThis->iResumeRetryCount = 0; + pThis->pszName = NULL; + pThis->bWriteAllMarkMsgs = RSFALSE; + pThis->iExecEveryNthOccur = 0; + pThis->iExecEveryNthOccurTO = 0; + pThis->iSecsExecOnceInterval = 0; + pThis->bExecWhenPrevSusp = 0; + pThis->bRepMsgHasMsg = 0; pThis->tLastOccur = datetime.GetTime(NULL); /* done once per action on startup only */ pthread_mutex_init(&pThis->mutActExec, NULL); + pthread_mutex_init(&pThis->mutAction, NULL); INIT_ATOMIC_HELPER_MUT(pThis->mutCAS); - SYNC_OBJ_TOOL_INIT(pThis); /* indicate we have a new action */ ++iActionNbr; @@ -310,16 +360,49 @@ finalize_it: /* action construction finalizer */ rsRetVal -actionConstructFinalize(action_t *pThis) +actionConstructFinalize(action_t *pThis, struct cnfparamvals *queueParams) { DEFiRet; - uchar pszQName[64]; /* friendly name of our queue */ + uchar pszAName[64]; /* friendly name of our action */ ASSERT(pThis != NULL); - /* find a name for our queue */ - snprintf((char*) pszQName, sizeof(pszQName)/sizeof(uchar), "action %d queue", iActionNbr); + if(!strcmp((char*)modGetName(pThis->pMod), "builtin:omdiscard")) { + /* discard actions will be optimized out */ + FINALIZE; + } + /* generate a friendly name for us action stats */ + if(pThis->pszName == NULL) { + snprintf((char*) pszAName, sizeof(pszAName)/sizeof(uchar), "action %d", iActionNbr); + } else { + ustrncpy(pszAName, pThis->pszName, sizeof(pszAName)); + pszAName[sizeof(pszAName)-1] = '\0'; /* to be on the save side */ + } + + /* support statistics gathering */ + CHKiRet(statsobj.Construct(&pThis->statsobj)); + CHKiRet(statsobj.SetName(pThis->statsobj, pszAName)); + + STATSCOUNTER_INIT(pThis->ctrProcessed, pThis->mutCtrProcessed); + CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("processed"), + ctrType_IntCtr, &pThis->ctrProcessed)); + + STATSCOUNTER_INIT(pThis->ctrFail, pThis->mutCtrFail); + CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("failed"), + ctrType_IntCtr, &pThis->ctrFail)); + + CHKiRet(statsobj.ConstructFinalize(pThis->statsobj)); + + /* create our queue */ + /* generate a friendly name for the queue */ + if(pThis->pszName == NULL) { + snprintf((char*) pszAName, sizeof(pszAName)/sizeof(uchar), "action %d queue", + iActionNbr); + } else { + ustrncpy(pszAName, pThis->pszName, sizeof(pszAName)); + pszAName[63] = '\0'; /* to be on the save side */ + } /* now check if we can run the action in "firehose mode" during stage one of * its processing (that is before messages are enqueued into the action q). * This is only possible if some features, which require strict sequence, are @@ -338,7 +421,7 @@ actionConstructFinalize(action_t *pThis) pThis->iSecsExecOnceInterval ); pThis->submitToActQ = doSubmitToActionQComplexBatch; - } else if(pThis->bWriteAllMarkMsgs == FALSE) { + } else if(pThis->bWriteAllMarkMsgs == RSFALSE) { /* nearly full-speed submission mode, default case */ pThis->submitToActQ = doSubmitToActionQNotAllMarkBatch; } else { @@ -351,7 +434,7 @@ actionConstructFinalize(action_t *pThis) * msg object thread safety in this case (this costs a bit performance and thus * is not enabled by default. -- rgerhards, 2008-02-20 */ - if(ActionQueType != QUEUETYPE_DIRECT) + if(cs.ActionQueType != QUEUETYPE_DIRECT) MsgEnableThreadSafety(); /* create queue */ @@ -360,49 +443,53 @@ actionConstructFinalize(action_t *pThis) * to be run on multiple threads. So far, this is forbidden by the interface * spec. -- rgerhards, 2008-01-30 */ - CHKiRet(qqueueConstruct(&pThis->pQueue, ActionQueType, 1, iActionQueueSize, + CHKiRet(qqueueConstruct(&pThis->pQueue, cs.ActionQueType, 1, cs.iActionQueueSize, (rsRetVal (*)(void*, batch_t*, int*))processBatchMain)); - obj.SetName((obj_t*) pThis->pQueue, pszQName); + obj.SetName((obj_t*) pThis->pQueue, pszAName); + qqueueSetpUsr(pThis->pQueue, pThis); - /* ... set some properties ... */ -# define setQPROP(func, directive, data) \ - CHKiRet_Hdlr(func(pThis->pQueue, data)) { \ - errmsg.LogError(0, NO_ERRCODE, "Invalid " #directive ", error %d. Ignored, running with default setting", iRet); \ - } -# define setQPROPstr(func, directive, data) \ - CHKiRet_Hdlr(func(pThis->pQueue, data, (data == NULL)? 0 : strlen((char*) data))) { \ - errmsg.LogError(0, NO_ERRCODE, "Invalid " #directive ", error %d. Ignored, running with default setting", iRet); \ + if(queueParams == NULL) { /* use legacy params? */ + /* ... set some properties ... */ +# define setQPROP(func, directive, data) \ + CHKiRet_Hdlr(func(pThis->pQueue, data)) { \ + errmsg.LogError(0, NO_ERRCODE, "Invalid " #directive ", \ + error %d. Ignored, running with default setting", iRet); \ + } +# define setQPROPstr(func, directive, data) \ + CHKiRet_Hdlr(func(pThis->pQueue, data, (data == NULL)? 0 : strlen((char*) data))) { \ + errmsg.LogError(0, NO_ERRCODE, "Invalid " #directive ", \ + error %d. Ignored, running with default setting", iRet); \ + } + setQPROP(qqueueSetsizeOnDiskMax, "$ActionQueueMaxDiskSpace", cs.iActionQueMaxDiskSpace); + setQPROP(qqueueSetiDeqBatchSize, "$ActionQueueDequeueBatchSize", cs.iActionQueueDeqBatchSize); + setQPROP(qqueueSetMaxFileSize, "$ActionQueueFileSize", cs.iActionQueMaxFileSize); + setQPROPstr(qqueueSetFilePrefix, "$ActionQueueFileName", cs.pszActionQFName); + setQPROP(qqueueSetiPersistUpdCnt, "$ActionQueueCheckpointInterval", cs.iActionQPersistUpdCnt); + setQPROP(qqueueSetbSyncQueueFiles, "$ActionQueueSyncQueueFiles", cs.bActionQSyncQeueFiles); + setQPROP(qqueueSettoQShutdown, "$ActionQueueTimeoutShutdown", cs.iActionQtoQShutdown ); + setQPROP(qqueueSettoActShutdown, "$ActionQueueTimeoutActionCompletion", cs.iActionQtoActShutdown); + setQPROP(qqueueSettoWrkShutdown, "$ActionQueueWorkerTimeoutThreadShutdown", cs.iActionQtoWrkShutdown); + setQPROP(qqueueSettoEnq, "$ActionQueueTimeoutEnqueue", cs.iActionQtoEnq); + setQPROP(qqueueSetiHighWtrMrk, "$ActionQueueHighWaterMark", cs.iActionQHighWtrMark); + setQPROP(qqueueSetiLowWtrMrk, "$ActionQueueLowWaterMark", cs.iActionQLowWtrMark); + setQPROP(qqueueSetiDiscardMrk, "$ActionQueueDiscardMark", cs.iActionQDiscardMark); + setQPROP(qqueueSetiDiscardSeverity, "$ActionQueueDiscardSeverity", cs.iActionQDiscardSeverity); + setQPROP(qqueueSetiMinMsgsPerWrkr, "$ActionQueueWorkerThreadMinimumMessages", cs.iActionQWrkMinMsgs); + setQPROP(qqueueSetbSaveOnShutdown, "$ActionQueueSaveOnShutdown", cs.bActionQSaveOnShutdown); + setQPROP(qqueueSetiDeqSlowdown, "$ActionQueueDequeueSlowdown", cs.iActionQueueDeqSlowdown); + setQPROP(qqueueSetiDeqtWinFromHr, "$ActionQueueDequeueTimeBegin", cs.iActionQueueDeqtWinFromHr); + setQPROP(qqueueSetiDeqtWinToHr, "$ActionQueueDequeueTimeEnd", cs.iActionQueueDeqtWinToHr); + } else { + /* we have v6-style config params */ + qqueueSetDefaultsActionQueue(pThis->pQueue); + qqueueApplyCnfParam(pThis->pQueue, queueParams); } - qqueueSetpUsr(pThis->pQueue, pThis); - setQPROP(qqueueSetsizeOnDiskMax, "$ActionQueueMaxDiskSpace", iActionQueMaxDiskSpace); - setQPROP(qqueueSetiDeqBatchSize, "$ActionQueueDequeueBatchSize", iActionQueueDeqBatchSize); - setQPROP(qqueueSetMaxFileSize, "$ActionQueueFileSize", iActionQueMaxFileSize); - setQPROPstr(qqueueSetFilePrefix, "$ActionQueueFileName", pszActionQFName); - setQPROP(qqueueSetiPersistUpdCnt, "$ActionQueueCheckpointInterval", iActionQPersistUpdCnt); - setQPROP(qqueueSetbSyncQueueFiles, "$ActionQueueSyncQueueFiles", bActionQSyncQeueFiles); - setQPROP(qqueueSettoQShutdown, "$ActionQueueTimeoutShutdown", iActionQtoQShutdown ); - setQPROP(qqueueSettoActShutdown, "$ActionQueueTimeoutActionCompletion", iActionQtoActShutdown); - setQPROP(qqueueSettoWrkShutdown, "$ActionQueueWorkerTimeoutThreadShutdown", iActionQtoWrkShutdown); - setQPROP(qqueueSettoEnq, "$ActionQueueTimeoutEnqueue", iActionQtoEnq); - setQPROP(qqueueSetiHighWtrMrk, "$ActionQueueHighWaterMark", iActionQHighWtrMark); - setQPROP(qqueueSetiLowWtrMrk, "$ActionQueueLowWaterMark", iActionQLowWtrMark); - setQPROP(qqueueSetiDiscardMrk, "$ActionQueueDiscardMark", iActionQDiscardMark); - setQPROP(qqueueSetiDiscardSeverity, "$ActionQueueDiscardSeverity", iActionQDiscardSeverity); - setQPROP(qqueueSetiMinMsgsPerWrkr, "$ActionQueueWorkerThreadMinimumMessages", iActionQWrkMinMsgs); - setQPROP(qqueueSetbSaveOnShutdown, "$ActionQueueSaveOnShutdown", bActionQSaveOnShutdown); - setQPROP(qqueueSetiDeqSlowdown, "$ActionQueueDequeueSlowdown", iActionQueueDeqSlowdown); - setQPROP(qqueueSetiDeqtWinFromHr, "$ActionQueueDequeueTimeBegin", iActionQueueDeqtWinFromHr); - setQPROP(qqueueSetiDeqtWinToHr, "$ActionQueueDequeueTimeEnd", iActionQueueDeqtWinToHr); - # undef setQPROP # undef setQPROPstr - dbgoprint((obj_t*) pThis->pQueue, "save on shutdown %d, max disk space allowed %lld\n", - bActionQSaveOnShutdown, iActionQueMaxDiskSpace); - + qqueueDbgPrint(pThis->pQueue); - CHKiRet(qqueueStart(pThis->pQueue)); DBGPRINTF("Action %p: queue %p created\n", pThis, pThis->pQueue); /* and now reset the queue params (see comment in its function header!) */ @@ -418,7 +505,7 @@ finalize_it: */ rsRetVal actionSetGlobalResumeInterval(int iNewVal) { - glbliActionResumeInterval = iNewVal; + cs.glbliActionResumeInterval = iNewVal; return RS_RET_OK; } @@ -693,7 +780,8 @@ rsRetVal actionDbgPrint(action_t *pThis) dbgprintf("%s: ", module.GetStateName(pThis->pMod)); pThis->pMod->dbgPrintInstInfo(pThis->pModData); - dbgprintf("\n\tInstance data: 0x%lx\n", (unsigned long) pThis->pModData); + dbgprintf("\n"); + dbgprintf("\tInstance data: 0x%lx\n", (unsigned long) pThis->pModData); dbgprintf("\tRepeatedMsgReduction: %d\n", pThis->f_ReduceRepeated); dbgprintf("\tResume Interval: %d\n", pThis->iResumeInterval); if(pThis->eState == ACT_STATE_SUSP) { @@ -725,6 +813,7 @@ static rsRetVal prepareDoActionParams(action_t *pAction, batch_obj_t *pElem) { int i; msg_t *pMsg; + struct json_object *json; DEFiRet; ASSERT(pAction != NULL); @@ -745,6 +834,10 @@ static rsRetVal prepareDoActionParams(action_t *pAction, batch_obj_t *pElem) case ACT_MSG_PASSING: pElem->staticActParams[i] = (void*) pMsg; break; + case ACT_JSON_PASSING: + CHKiRet(tplToJSON(pAction->ppTpl[i], pMsg, &json)); + pElem->staticActParams[i] = (void*) json; + break; default:dbgprintf("software bug/error: unknown pAction->eParamPassing %d in prepareDoActionParams\n", (int) pAction->eParamPassing); assert(0); /* software bug if this happens! */ @@ -777,7 +870,7 @@ static rsRetVal releaseBatch(action_t *pAction, batch_t *pBatch) for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { pElem = &(pBatch->pElem[i]); - if(pElem->bFilterOK && pElem->state != BATCH_STATE_DISC) { + if(batchIsValidElem(pBatch, i)) { switch(pAction->eParamPassing) { case ACT_ARRAY_PASSING: ppMsgs = (uchar***) pElem->staticActParams; @@ -807,6 +900,13 @@ static rsRetVal releaseBatch(action_t *pAction, batch_t *pBatch) ((uchar**)pElem->staticActParams)[j] = NULL; } break; + case ACT_JSON_PASSING: + for(j = 0 ; j < pAction->iNumTpls ; ++j) { + json_object_put((struct json_object*) + pElem->staticActParams[j]); + pElem->staticActParams[j] = NULL; + } + break; } } } @@ -961,15 +1061,14 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem) i = pBatch->iDoneUpTo; /* all messages below that index are processed */ iElemProcessed = 0; iCommittedUpTo = i; -dbgprintf("XXXXX: tryDoAction %p, pnElem %d, nElem %d\n", pAction, *pnElem, pBatch->nElem); + DBGPRINTF("tryDoAction %p, pnElem %d, nElem %d\n", pAction, *pnElem, pBatch->nElem); while(iElemProcessed <= *pnElem && i < pBatch->nElem) { if(*(pBatch->pbShutdownImmediate)) ABORT_FINALIZE(RS_RET_FORCE_TERM); /* NOTE: do NOT extend the filter below! Anything else must be done on the * enq side of the queue (see file header comment)! -- rgerhards, 2011-06-15 */ - if( pBatch->pElem[i].bFilterOK - && pBatch->pElem[i].state != BATCH_STATE_DISC) { + if(batchIsValidElem(pBatch, i)) { pMsg = (msg_t*) pBatch->pElem[i].pUsrp; localRet = actionProcessMessage(pAction, pMsg, pBatch->pElem[i].staticActParams, pBatch->pbShutdownImmediate); @@ -1016,16 +1115,6 @@ finalize_it: RETiRet; } -/* debug aid */ -static void displayBatchState(batch_t *pBatch) -{ - int i; - for(i = 0 ; i < pBatch->nElem ; ++i) { - dbgprintf("XXXXX: displayBatchState2 %p[%d]: %d\n", pBatch, i, pBatch->pElem[i].state); - } -} - - /* submit a batch for actual action processing. * The first nElem elements are processed. This function calls itself * recursively if it needs to handle errors. @@ -1073,6 +1162,7 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem) && pBatch->pElem[i].state != BATCH_STATE_COMM ) { pBatch->pElem[i].state = BATCH_STATE_BAD; pBatch->pElem[i].bPrevWasSuspended = 1; + STATSCOUNTER_INC(pAction->ctrFail, pAction->mutCtrFail); } } bDone = 1; @@ -1099,6 +1189,29 @@ finalize_it: } +/* copy "active" array of batch, as we need to modify it. The caller + * must make sure the new array is freed and the orginal batch + * pointer is restored (thus the caller must save it). If active + * is currently NULL, this is properly handled. + * Note: the batches active pointer is modified, so it must be + * saved BEFORE calling this function! + * rgerhards, 2012-09-12 + */ +static rsRetVal +copyActive(batch_t *pBatch) +{ + sbool *active; + DEFiRet; + + CHKmalloc(active = malloc(batchNumMsgs(pBatch) * sizeof(sbool))); + if(pBatch->active == NULL) + memset(active, 1, batchNumMsgs(pBatch)); + else + memcpy(active, pBatch->active, batchNumMsgs(pBatch)); + pBatch->active = active; +finalize_it: + RETiRet; +} /* The following function prepares a batch for processing, that it is * reinitializes batch states, generates strings and does everything else @@ -1109,7 +1222,7 @@ finalize_it: * rgerhards, 2010-06-14 */ static inline rsRetVal -prepareBatch(action_t *pAction, batch_t *pBatch) +prepareBatch(action_t *pAction, batch_t *pBatch, sbool **activeSave, int *bMustRestoreActivePtr) { int i; batch_obj_t *pElem; @@ -1118,10 +1231,16 @@ prepareBatch(action_t *pAction, batch_t *pBatch) pBatch->iDoneUpTo = 0; for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { pElem = &(pBatch->pElem[i]); - if(pElem->bFilterOK && pElem->state != BATCH_STATE_DISC) { + if(batchIsValidElem(pBatch, i)) { pElem->state = BATCH_STATE_RDY; - if(prepareDoActionParams(pAction, pElem) != RS_RET_OK) - pElem->bFilterOK = FALSE; + if(prepareDoActionParams(pAction, pElem) != RS_RET_OK) { + /* make sure we have our copy of "active" array */ + if(!*bMustRestoreActivePtr) { + *activeSave = pBatch->active; + copyActive(pBatch); + } + pBatch->active[i] = RSFALSE; + } } } RETiRet; @@ -1154,6 +1273,8 @@ static rsRetVal processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) { int *pbShutdownImmdtSave; + sbool *activeSave; + int bMustRestoreActivePtr = 0; rsRetVal localRet; DEFiRet; @@ -1161,7 +1282,7 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) pbShutdownImmdtSave = pBatch->pbShutdownImmediate; pBatch->pbShutdownImmediate = pbShutdownImmediate; - CHKiRet(prepareBatch(pAction, pBatch)); + CHKiRet(prepareBatch(pAction, pBatch, &activeSave, &bMustRestoreActivePtr)); /* We now must guard the output module against execution by multiple threads. The * plugin interface specifies that output modules must not be thread-safe (except @@ -1184,6 +1305,11 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) if(iRet == RS_RET_OK) iRet = localRet; + + if(bMustRestoreActivePtr) { + free(pBatch->active); + pBatch->active = activeSave; + } finalize_it: pBatch->pbShutdownImmediate = pbShutdownImmdtSave; @@ -1230,16 +1356,16 @@ static rsRetVal setActionQueType(void __attribute__((unused)) *pVal, uchar *pszT DEFiRet; if (!strcasecmp((char *) pszType, "fixedarray")) { - ActionQueType = QUEUETYPE_FIXED_ARRAY; + cs.ActionQueType = QUEUETYPE_FIXED_ARRAY; DBGPRINTF("action queue type set to FIXED_ARRAY\n"); } else if (!strcasecmp((char *) pszType, "linkedlist")) { - ActionQueType = QUEUETYPE_LINKEDLIST; + cs.ActionQueType = QUEUETYPE_LINKEDLIST; DBGPRINTF("action queue type set to LINKEDLIST\n"); } else if (!strcasecmp((char *) pszType, "disk")) { - ActionQueType = QUEUETYPE_DISK; + cs.ActionQueType = QUEUETYPE_DISK; DBGPRINTF("action queue type set to DISK\n"); } else if (!strcasecmp((char *) pszType, "direct")) { - ActionQueType = QUEUETYPE_DIRECT; + cs.ActionQueType = QUEUETYPE_DIRECT; DBGPRINTF("action queue type set to DIRECT (no queueing at all)\n"); } else { errmsg.LogError(0, RS_RET_INVALID_PARAMS, "unknown actionqueue parameter: %s", (char *) pszType); @@ -1262,11 +1388,18 @@ doSubmitToActionQ(action_t *pAction, msg_t *pMsg) { DEFiRet; + if(pAction->eState == ACT_STATE_DIED) { + DBGPRINTF("action %p died, do NOT execute\n", pAction); + FINALIZE; + } + + STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed); if(pAction->pQueue->qType == QUEUETYPE_DIRECT) iRet = qqueueEnqObjDirect(pAction->pQueue, (void*) MsgAddRef(pMsg)); else - iRet = qqueueEnqObj(pAction->pQueue, pMsg->flowCtlType, (void*) MsgAddRef(pMsg)); + iRet = qqueueEnqObj(pAction->pQueue, eFLOWCTL_NO_DELAY, (void*) MsgAddRef(pMsg)); +finalize_it: RETiRet; } @@ -1412,7 +1545,7 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch) pAction->tActNow = -1; /* we do not yet know our current time (clear prev. value) */ /* don't output marks to recently written outputs */ - if(pAction->bWriteAllMarkMsgs == FALSE + if(pAction->bWriteAllMarkMsgs == RSFALSE && (pMsg->msgFlags & MARK) && (getActNow(pAction) - pAction->f_time) < MarkInterval / 2) { ABORT_FINALIZE(RS_RET_OK); } @@ -1467,8 +1600,45 @@ finalize_it: } +/* helper to activateActions, it activates a specific action. + */ +DEFFUNC_llExecFunc(doActivateActions) +{ + rsRetVal localRet; + action_t *pThis = (action_t*) pData; + BEGINfunc + localRet = qqueueStart(pThis->pQueue); + if(localRet != RS_RET_OK) { + errmsg.LogError(0, localRet, "error starting up action queue"); + if(localRet == RS_RET_FILE_PREFIX_MISSING) { + errmsg.LogError(0, localRet, "file prefix (work directory?) " + "is missing"); + } + actionDisable(pThis); + } + DBGPRINTF("Action %s[%p]: queue %p started\n", modGetName(pThis->pMod), + pThis, pThis->pQueue); + ENDfunc + return RS_RET_OK; /* we ignore errors, we can not do anything either way */ +} + + +/* This function "activates" the action after privileges have been dropped. Currently, + * this means that the queues are started. + * rgerhards, 2011-05-02 + */ +rsRetVal +activateActions(void) +{ + DEFiRet; + iRet = ruleset.IterateAllActions(ourConf, doActivateActions, NULL); + RETiRet; +} + + + /* This submits the message to the action queue in case where we need to handle - * bWriteAllMarkMessage == FALSE only. Note that we use a non-blocking CAS loop + * bWriteAllMarkMessage == RSFALSE only. Note that we use a non-blocking CAS loop * for the synchronization. Here, we just modify the filter condition to be false when * a mark message must not be written. However, in this case we must save the previous * filter as we may need it in the next action (potential future optimization: check if this is @@ -1481,22 +1651,15 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) time_t now = 0; time_t lastAct; int i; - int bModifiedFilter; - sbool FilterSave[1024]; - sbool *pFilterSave; + sbool *activeSave; DEFiRet; - if(batchNumMsgs(pBatch) <= (int) (sizeof(FilterSave)/sizeof(sbool))) { - pFilterSave = FilterSave; - } else { - CHKmalloc(pFilterSave = malloc(batchNumMsgs(pBatch) * sizeof(sbool))); - } + activeSave = pBatch->active; + copyActive(pBatch); - bModifiedFilter = 0; for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { - if(!pBatch->pElem[i].bFilterOK) + if((pBatch->pElem[i].state == BATCH_STATE_DISC) || !pBatch->active[i]) continue; - pFilterSave[i] = pBatch->pElem[i].bFilterOK; if(now == 0) { now = datetime.GetTime(NULL); /* good time call - the only one done */ } @@ -1507,15 +1670,15 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) lastAct = pAction->f_time; if(((msg_t*)(pBatch->pElem[i].pUsrp))->msgFlags & MARK) { if((now - lastAct) < MarkInterval / 2) { - pBatch->pElem[i].bFilterOK = 0; - bModifiedFilter = 1; - DBGPRINTF("action was recently called, ignoring mark message\n"); + pBatch->active[i] = 0; + DBGPRINTF("batch item %d: action was recently called, ignoring " + "mark message\n", i); break; /* do not update timestamp for non-written mark messages */ } } } while(ATOMIC_CAS_time_t(&pAction->f_time, lastAct, ((msg_t*)(pBatch->pElem[i].pUsrp))->ttGenTime, &pAction->mutCAS) == 0); - if(pBatch->pElem[i].bFilterOK) { + if(pBatch->active[i]) { DBGPRINTF("Called action(NotAllMark), processing batch[%d] via '%s'\n", i, module.GetStateName(pAction->pMod)); } @@ -1523,21 +1686,23 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) iRet = doSubmitToActionQBatch(pAction, pBatch); - if(bModifiedFilter) { - /* in this case, we need to restore previous state */ - for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { - /* note: clang static code analyzer reports a false positive below */ - pBatch->pElem[i].bFilterOK = pFilterSave[i]; - } - } - -finalize_it: - if(pFilterSave != FilterSave) - free(pFilterSave); + free(pBatch->active); + pBatch->active = activeSave; RETiRet; } +static inline void +countStatsBatchEnq(action_t *pAction, batch_t *pBatch) +{ + int i; + for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { + if( batchIsValidElem(pBatch, i)) { + STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed); + } + } +} + /* enqueue a batch in direct mode. We have put this into its own function just to avoid * cluttering the actual submit function. @@ -1546,18 +1711,13 @@ finalize_it: static inline rsRetVal doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch) { - sbool FilterSave[1024]; - sbool *pFilterSave; sbool bNeedSubmit; - sbool bModifiedFilter; + sbool *activeSave; int i; DEFiRet; - if(batchNumMsgs(pBatch) <= (int) (sizeof(FilterSave)/sizeof(sbool))) { - pFilterSave = FilterSave; - } else { - CHKmalloc(pFilterSave = malloc(batchNumMsgs(pBatch) * sizeof(sbool))); - } + activeSave = pBatch->active; + copyActive(pBatch); /* note: for direct mode, we need to adjust the filter property. For non-direct * this is not necessary, because in that case we enqueue only what actually needs @@ -1565,40 +1725,34 @@ doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch) */ if(pAction->bExecWhenPrevSusp) { bNeedSubmit = 0; - bModifiedFilter = 0; for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { - pFilterSave[i] = pBatch->pElem[i].bFilterOK; if(!pBatch->pElem[i].bPrevWasSuspended) { - DBGPRINTF("action enq stage: change bFilterOK to 0 due to " + DBGPRINTF("action enq stage: change active to 0 due to " "failover case in elem %d\n", i); - pBatch->pElem[i].bFilterOK = 0; - bModifiedFilter = 1; + pBatch->active[i] = 0; } - if(pBatch->pElem[i].bFilterOK) + if(batchIsValidElem(pBatch, i)) { + STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed); bNeedSubmit = 1; - DBGPRINTF("action %p[%d]: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n", - pAction, i, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state, + } + DBGPRINTF("action %p[%d]: valid:%d state:%d execWhenPrev:%d prevWasSusp:%d\n", + pAction, i, batchIsValidElem(pBatch, i), pBatch->pElem[i].state, pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended); } if(bNeedSubmit) { + /* note: stats were already computed above */ iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch); } else { - DBGPRINTF("no need to submit batch, all bFilterOK==0\n"); - } - if(bModifiedFilter) { - for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { - DBGPRINTF("action %p: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n", - pAction, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state, - pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended); - /* note: clang static code analyzer reports a false positive below */ - pBatch->pElem[i].bFilterOK = pFilterSave[i]; - } + DBGPRINTF("no need to submit batch, all invalid\n"); } } else { + if(GatherStats) + countStatsBatchEnq(pAction, pBatch); iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch); } -finalize_it: + free(pBatch->active); + pBatch->active = activeSave; RETiRet; } @@ -1621,11 +1775,10 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch) * TODO: optimize this, we may do at least a multi-submit! */ for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { - DBGPRINTF("action %p: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n", - pAction, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state, + DBGPRINTF("action %p: valid:%d state:%d execWhenPrev:%d prevWasSusp:%d\n", + pAction, batchIsValidElem(pBatch, i), pBatch->pElem[i].state, pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended); - if( pBatch->pElem[i].bFilterOK - && pBatch->pElem[i].state != BATCH_STATE_DISC + if( batchIsValidElem(pBatch, i) && (pAction->bExecWhenPrevSusp == 0 || pBatch->pElem[i].bPrevWasSuspended == 1)) { doSubmitToActionQ(pAction, (msg_t*)(pBatch->pElem[i].pUsrp)); } @@ -1650,11 +1803,10 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) DBGPRINTF("Called action %p (complex case), logging to %s\n", pAction, module.GetStateName(pAction->pMod)); for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { - DBGPRINTF("action %p: filterOK:%d state:%d execWhenPrev:%d prevWasSusp:%d\n", - pAction, pBatch->pElem[i].bFilterOK, pBatch->pElem[i].state, + DBGPRINTF("action %p: valid:%d state:%d execWhenPrev:%d prevWasSusp:%d\n", + pAction, batchIsValidElem(pBatch, i), pBatch->pElem[i].state, pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended); - if( pBatch->pElem[i].bFilterOK - && pBatch->pElem[i].state != BATCH_STATE_DISC + if( batchIsValidElem(pBatch, i) && ((pAction->bExecWhenPrevSusp == 0) || pBatch->pElem[i].bPrevWasSuspended) ) { doActionCallAction(pAction, pBatch, i); } @@ -1672,23 +1824,68 @@ doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) { DEFiRet; - LockObj(pAction); - pthread_cleanup_push(mutexCancelCleanup, pAction->Sync_mut); + d_pthread_mutex_lock(&pAction->mutAction); + pthread_cleanup_push(mutexCancelCleanup, &pAction->mutAction); iRet = helperSubmitToActionQComplexBatch(pAction, pBatch); - UnlockObj(pAction); + d_pthread_mutex_unlock(&pAction->mutAction); pthread_cleanup_pop(0); /* remove mutex cleanup handler */ RETiRet; } #pragma GCC diagnostic warning "-Wempty-body" + +/* apply all params from param block to action. This supports the v6 config system. + * Defaults must have been set appropriately during action construct! + * rgerhards, 2011-08-01 + */ +static rsRetVal +actionApplyCnfParam(action_t *pAction, struct cnfparamvals *pvals) +{ + int i; + + for(i = 0 ; i < pblk.nParams ; ++i) { + if(!pvals[i].bUsed) + continue; + if(!strcmp(pblk.descr[i].name, "name")) { + pAction->pszName = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL); + } else if(!strcmp(pblk.descr[i].name, "type")) { + continue; /* this is handled seperately during module select! */ + } else if(!strcmp(pblk.descr[i].name, "action.writeallmarkmessages")) { + pAction->bWriteAllMarkMsgs = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "action.execonlyeverynthtime")) { + pAction->iExecEveryNthOccur = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "action.execonlyeverynthtimetimeout")) { + pAction->iExecEveryNthOccurTO = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "action.execonlyonceeveryinterval")) { + pAction->iSecsExecOnceInterval = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "action.execonlywhenpreviousissuspended")) { + pAction->bExecWhenPrevSusp = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "action.repeatedmsgcontainsoriginalmsg")) { + pAction->bRepMsgHasMsg = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "action.resumeretrycount")) { + pAction->iResumeRetryCount = pvals[i].val.d.n; + } else if(!strcmp(pblk.descr[i].name, "action.resumeinterval")) { + pAction->iResumeInterval = pvals[i].val.d.n; + } else { + dbgprintf("action: program error, non-handled " + "param '%s'\n", pblk.descr[i].name); + } + } + return RS_RET_OK; +} + + + /* add an Action to the current selector * The pOMSR is freed, as it is not needed after this function. * Note: this function pulls global data that specifies action config state. * rgerhards, 2007-07-27 */ rsRetVal -addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringRequest_t *pOMSR, int bSuspended) +addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, + omodStringRequest_t *pOMSR, struct cnfparamvals *actParams, + struct cnfparamvals *queueParams, int bSuspended) { DEFiRet; int i; @@ -1700,22 +1897,28 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringReques assert(ppAction != NULL); assert(pMod != NULL); assert(pOMSR != NULL); - DBGPRINTF("Module %s processed this config line.\n", module.GetName(pMod)); + DBGPRINTF("Module %s processes this action.\n", module.GetName(pMod)); CHKiRet(actionConstruct(&pAction)); /* create action object first */ pAction->pMod = pMod; pAction->pModData = pModData; - pAction->pszName = pszActionName; - pszActionName = NULL; /* free again! */ - pAction->bWriteAllMarkMsgs = bActionWriteAllMarkMsgs; - bActionWriteAllMarkMsgs = FALSE; /* reset */ - pAction->bExecWhenPrevSusp = bActExecWhenPrevSusp; - pAction->iSecsExecOnceInterval = iActExecOnceInterval; - pAction->iExecEveryNthOccur = iActExecEveryNthOccur; - pAction->iExecEveryNthOccurTO = iActExecEveryNthOccurTO; - pAction->bRepMsgHasMsg = bActionRepMsgHasMsg; - iActExecEveryNthOccur = 0; /* auto-reset */ - iActExecEveryNthOccurTO = 0; /* auto-reset */ + if(actParams == NULL) { /* use legacy systemn */ + pAction->pszName = cs.pszActionName; + pAction->iResumeInterval = cs.glbliActionResumeInterval; + pAction->iResumeRetryCount = cs.glbliActionResumeRetryCount; + pAction->bWriteAllMarkMsgs = cs.bActionWriteAllMarkMsgs; + pAction->bExecWhenPrevSusp = cs.bActExecWhenPrevSusp; + pAction->iSecsExecOnceInterval = cs.iActExecOnceInterval; + pAction->iExecEveryNthOccur = cs.iActExecEveryNthOccur; + pAction->iExecEveryNthOccurTO = cs.iActExecEveryNthOccurTO; + pAction->bRepMsgHasMsg = cs.bActionRepMsgHasMsg; + cs.iActExecEveryNthOccur = 0; /* auto-reset */ + cs.iActExecEveryNthOccurTO = 0; /* auto-reset */ + cs.bActionWriteAllMarkMsgs = RSFALSE; /* auto-reset */ + cs.pszActionName = NULL; /* free again! */ + } else { + actionApplyCnfParam(pAction, actParams); + } /* check if we can obtain the template pointers - TODO: move to separate function? */ pAction->iNumTpls = OMSRgetEntryCount(pOMSR); @@ -1734,7 +1937,9 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringReques /* Ok, we got everything, so it now is time to look up the template * (Hint: templates MUST be defined before they are used!) */ - if((pAction->ppTpl[i] = tplFind((char*)pTplName, strlen((char*)pTplName))) == NULL) { + if( !(iTplOpts & OMSR_TPL_AS_MSG) + && (pAction->ppTpl[i] = + tplFind(ourConf, (char*)pTplName, strlen((char*)pTplName))) == NULL) { snprintf(errMsg, sizeof(errMsg) / sizeof(char), " Could not find template '%s' - action disabled\n", pTplName); @@ -1744,7 +1949,7 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringReques } /* check required template options */ if( (iTplOpts & OMSR_RQD_TPL_OPT_SQL) - && (pAction->ppTpl[i]->optFormatForSQL == 0)) { + && (pAction->ppTpl[i]->optFormatEscape == 0)) { errno = 0; errmsg.LogError(0, RS_RET_RQD_TPLOPT_MISSING, "Action disabled. To use this action, you have to specify " "the SQL or stdSQL option in your template!\n"); @@ -1756,6 +1961,8 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringReques pAction->eParamPassing = ACT_ARRAY_PASSING; } else if(iTplOpts & OMSR_TPL_AS_MSG) { pAction->eParamPassing = ACT_MSG_PASSING; + } else if(iTplOpts & OMSR_TPL_AS_JSON) { + pAction->eParamPassing = ACT_JSON_PASSING; } else { pAction->eParamPassing = ACT_STRING_PASSING; } @@ -1766,9 +1973,9 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringReques pAction->pMod = pMod; pAction->pModData = pModData; /* now check if the module is compatible with select features */ - if(pMod->isCompatibleWithFeature(sFEATURERepeatedMsgReduction) == RS_RET_OK) - pAction->f_ReduceRepeated = bReduceRepeatMsgs; - else { + if(pMod->isCompatibleWithFeature(sFEATURERepeatedMsgReduction) == RS_RET_OK) { + pAction->f_ReduceRepeated = loadConf->globals.bReduceRepeatMsgs; + } else { DBGPRINTF("module is incompatible with RepeatedMsgReduction - turned off\n"); pAction->f_ReduceRepeated = 0; } @@ -1777,7 +1984,7 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringReques if(bSuspended) actionSuspend(pAction, datetime.GetTime(NULL)); /* "good" time call, only during init and unavoidable */ - CHKiRet(actionConstructFinalize(pAction)); + CHKiRet(actionConstructFinalize(pAction, queueParams)); /* TODO: if we exit here, we have a memory leak... */ @@ -1803,11 +2010,92 @@ finalize_it: static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal) { - iActExecOnceInterval = 0; + cs.iActExecOnceInterval = 0; + cs.bActExecWhenPrevSusp = 0; return RS_RET_OK; } +/* initialize (current) config variables. + * Used at program start and when a new scope is created. + */ +static inline void +initConfigVariables(void) +{ + cs.bActionWriteAllMarkMsgs = RSFALSE; + cs.glbliActionResumeRetryCount = 0; + cs.bActExecWhenPrevSusp = 0; + cs.iActExecOnceInterval = 0; + cs.iActExecEveryNthOccur = 0; + cs.iActExecEveryNthOccurTO = 0; + cs.glbliActionResumeInterval = 30; + cs.glbliActionResumeRetryCount = 0; + cs.bActionRepMsgHasMsg = 0; + if(cs.pszActionName != NULL) { + free(cs.pszActionName); + cs.pszActionName = NULL; + } + actionResetQueueParams(); +} + + +rsRetVal +actionNewInst(struct nvlst *lst, action_t **ppAction) +{ + struct cnfparamvals *paramvals; + struct cnfparamvals *queueParams; + modInfo_t *pMod; + uchar *cnfModName = NULL; + omodStringRequest_t *pOMSR; + void *pModData; + action_t *pAction; + int typeIdx; + DEFiRet; + + paramvals = nvlstGetParams(lst, &pblk, NULL); + if(paramvals == NULL) { + ABORT_FINALIZE(RS_RET_ERR); + } + dbgprintf("action param blk after actionNewInst:\n"); + cnfparamsPrint(&pblk, paramvals); + typeIdx = cnfparamGetIdx(&pblk, "type"); + if(paramvals[typeIdx].bUsed == 0) { + errmsg.LogError(0, RS_RET_CONF_RQRD_PARAM_MISSING, "action type missing"); + ABORT_FINALIZE(RS_RET_CONF_RQRD_PARAM_MISSING); // TODO: move this into rainerscript handlers + } + cnfModName = (uchar*)es_str2cstr(paramvals[cnfparamGetIdx(&pblk, ("type"))].val.d.estr, NULL); + if((pMod = module.FindWithCnfName(loadConf, cnfModName, eMOD_OUT)) == NULL) { + errmsg.LogError(0, RS_RET_MOD_UNKNOWN, "module name '%s' is unknown", cnfModName); + ABORT_FINALIZE(RS_RET_MOD_UNKNOWN); + } + iRet = pMod->mod.om.newActInst(cnfModName, lst, &pModData, &pOMSR); + // TODO: check if RS_RET_SUSPENDED is still valid in v6! + if(iRet != RS_RET_OK && iRet != RS_RET_SUSPENDED) { + FINALIZE; /* iRet is already set to error state */ + } + + qqueueDoCnfParams(lst, &queueParams); + + if((iRet = addAction(&pAction, pMod, pModData, pOMSR, paramvals, queueParams, + (iRet == RS_RET_SUSPENDED)? 1 : 0)) == RS_RET_OK) { + /* now check if the module is compatible with select features */ + if(pMod->isCompatibleWithFeature(sFEATURERepeatedMsgReduction) == RS_RET_OK) + pAction->f_ReduceRepeated = loadConf->globals.bReduceRepeatMsgs; + else { + DBGPRINTF("module is incompatible with RepeatedMsgReduction - turned off\n"); + pAction->f_ReduceRepeated = 0; + } + pAction->eState = ACT_STATE_RDY; /* action is enabled */ + loadConf->actions.nbrActions++; /* one more active action! */ + } + *ppAction = pAction; + +finalize_it: + free(cnfModName); + cnfparamvalsDestruct(paramvals, &pblk); + RETiRet; +} + /* TODO: we are not yet a real object, the ClassInit here just looks like it is.. */ rsRetVal actionClassInit(void) @@ -1818,37 +2106,43 @@ rsRetVal actionClassInit(void) CHKiRet(objUse(datetime, CORE_COMPONENT)); CHKiRet(objUse(module, CORE_COMPONENT)); CHKiRet(objUse(errmsg, CORE_COMPONENT)); - - CHKiRet(regCfSysLineHdlr((uchar *)"actionname", 0, eCmdHdlrGetWord, NULL, &pszActionName, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuefilename", 0, eCmdHdlrGetWord, NULL, &pszActionQFName, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesize", 0, eCmdHdlrInt, NULL, &iActionQueueSize, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionwriteallmarkmessages", 0, eCmdHdlrBinary, NULL, &bActionWriteAllMarkMsgs, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuebatchsize", 0, eCmdHdlrInt, NULL, &iActionQueueDeqBatchSize, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuemaxdiskspace", 0, eCmdHdlrSize, NULL, &iActionQueMaxDiskSpace, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuehighwatermark", 0, eCmdHdlrInt, NULL, &iActionQHighWtrMark, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuelowwatermark", 0, eCmdHdlrInt, NULL, &iActionQLowWtrMark, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardmark", 0, eCmdHdlrInt, NULL, &iActionQDiscardMark, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardseverity", 0, eCmdHdlrInt, NULL, &iActionQDiscardSeverity, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuecheckpointinterval", 0, eCmdHdlrInt, NULL, &iActionQPersistUpdCnt, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesyncqueuefiles", 0, eCmdHdlrBinary, NULL, &bActionQSyncQeueFiles, NULL)); + CHKiRet(objUse(statsobj, CORE_COMPONENT)); + CHKiRet(objUse(ruleset, CORE_COMPONENT)); + + CHKiRet(regCfSysLineHdlr((uchar *)"actionname", 0, eCmdHdlrGetWord, NULL, &cs.pszActionName, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuefilename", 0, eCmdHdlrGetWord, NULL, &cs.pszActionQFName, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesize", 0, eCmdHdlrInt, NULL, &cs.iActionQueueSize, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionwriteallmarkmessages", 0, eCmdHdlrBinary, NULL, &cs.bActionWriteAllMarkMsgs, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuebatchsize", 0, eCmdHdlrInt, NULL, &cs.iActionQueueDeqBatchSize, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuemaxdiskspace", 0, eCmdHdlrSize, NULL, &cs.iActionQueMaxDiskSpace, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuehighwatermark", 0, eCmdHdlrInt, NULL, &cs.iActionQHighWtrMark, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuelowwatermark", 0, eCmdHdlrInt, NULL, &cs.iActionQLowWtrMark, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardmark", 0, eCmdHdlrInt, NULL, &cs.iActionQDiscardMark, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardseverity", 0, eCmdHdlrInt, NULL, &cs.iActionQDiscardSeverity, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuecheckpointinterval", 0, eCmdHdlrInt, NULL, &cs.iActionQPersistUpdCnt, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesyncqueuefiles", 0, eCmdHdlrBinary, NULL, &cs.bActionQSyncQeueFiles, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetype", 0, eCmdHdlrGetWord, setActionQueType, NULL, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkerthreads", 0, eCmdHdlrInt, NULL, &iActionQueueNumWorkers, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutshutdown", 0, eCmdHdlrInt, NULL, &iActionQtoQShutdown, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutactioncompletion", 0, eCmdHdlrInt, NULL, &iActionQtoActShutdown, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutenqueue", 0, eCmdHdlrInt, NULL, &iActionQtoEnq, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkertimeoutthreadshutdown", 0, eCmdHdlrInt, NULL, &iActionQtoWrkShutdown, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkerthreadminimummessages", 0, eCmdHdlrInt, NULL, &iActionQWrkMinMsgs, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuemaxfilesize", 0, eCmdHdlrSize, NULL, &iActionQueMaxFileSize, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesaveonshutdown", 0, eCmdHdlrBinary, NULL, &bActionQSaveOnShutdown, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeueslowdown", 0, eCmdHdlrInt, NULL, &iActionQueueDeqSlowdown, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuetimebegin", 0, eCmdHdlrInt, NULL, &iActionQueueDeqtWinFromHr, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuetimeend", 0, eCmdHdlrInt, NULL, &iActionQueueDeqtWinToHr, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlyeverynthtime", 0, eCmdHdlrInt, NULL, &iActExecEveryNthOccur, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlyeverynthtimetimeout", 0, eCmdHdlrInt, NULL, &iActExecEveryNthOccurTO, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlyonceeveryinterval", 0, eCmdHdlrInt, NULL, &iActExecOnceInterval, NULL)); - CHKiRet(regCfSysLineHdlr((uchar *)"repeatedmsgcontainsoriginalmsg", 0, eCmdHdlrBinary, NULL, &bActionRepMsgHasMsg, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkerthreads", 0, eCmdHdlrInt, NULL, &cs.iActionQueueNumWorkers, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutshutdown", 0, eCmdHdlrInt, NULL, &cs.iActionQtoQShutdown, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutactioncompletion", 0, eCmdHdlrInt, NULL, &cs.iActionQtoActShutdown, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutenqueue", 0, eCmdHdlrInt, NULL, &cs.iActionQtoEnq, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkertimeoutthreadshutdown", 0, eCmdHdlrInt, NULL, &cs.iActionQtoWrkShutdown, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkerthreadminimummessages", 0, eCmdHdlrInt, NULL, &cs.iActionQWrkMinMsgs, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuemaxfilesize", 0, eCmdHdlrSize, NULL, &cs.iActionQueMaxFileSize, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesaveonshutdown", 0, eCmdHdlrBinary, NULL, &cs.bActionQSaveOnShutdown, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeueslowdown", 0, eCmdHdlrInt, NULL, &cs.iActionQueueDeqSlowdown, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuetimebegin", 0, eCmdHdlrInt, NULL, &cs.iActionQueueDeqtWinFromHr, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuetimeend", 0, eCmdHdlrInt, NULL, &cs.iActionQueueDeqtWinToHr, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlyeverynthtime", 0, eCmdHdlrInt, NULL, &cs.iActExecEveryNthOccur, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlyeverynthtimetimeout", 0, eCmdHdlrInt, NULL, &cs.iActExecEveryNthOccurTO, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlyonceeveryinterval", 0, eCmdHdlrInt, NULL, &cs.iActExecOnceInterval, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"repeatedmsgcontainsoriginalmsg", 0, eCmdHdlrBinary, NULL, &cs.bActionRepMsgHasMsg, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlywhenpreviousissuspended", 0, eCmdHdlrBinary, NULL, &cs.bActExecWhenPrevSusp, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"actionresumeretrycount", 0, eCmdHdlrInt, NULL, &cs.glbliActionResumeRetryCount, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, NULL)); + initConfigVariables(); /* first-time init of config setings */ + finalize_it: RETiRet; } |