diff options
Diffstat (limited to 'action.c')
| -rw-r--r-- | action.c | 330 |
1 files changed, 220 insertions, 110 deletions
@@ -4,7 +4,7 @@ * * File begun on 2007-08-06 by RGerhards (extracted from syslogd.c) * - * Copyright 2007 Rainer Gerhards and Adiscon GmbH. + * Copyright 2007-2010 Rainer Gerhards and Adiscon GmbH. * * This file is part of rsyslog. * @@ -43,6 +43,7 @@ #include "srUtils.h" #include "errmsg.h" #include "datetime.h" +#include "unicode-helper.h" /* forward definitions */ rsRetVal actionCallDoAction(action_t *pAction, msg_t *pMsg); @@ -58,7 +59,9 @@ static int iActExecEveryNthOccur = 0; /* execute action every n-th occurence (0, 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 uchar *pszActionName; /* short name for the action */ /* main message 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 */ @@ -70,6 +73,7 @@ static int iActionQueueNumWorkers = 1; /* number of worker threads for the mm 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 */ @@ -149,6 +153,7 @@ actionResetQueueParams(void) 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 */ @@ -162,8 +167,7 @@ actionResetQueueParams(void) glbliActionResumeRetryCount = 0; /* I guess it is smart to reset this one, too */ - if(pszActionQFName != NULL) - d_free(pszActionQFName); + d_free(pszActionQFName); pszActionQFName = NULL; /* prefix for the main message queue file */ RETiRet; @@ -175,11 +179,12 @@ actionResetQueueParams(void) */ rsRetVal actionDestruct(action_t *pThis) { + int i; DEFiRet; ASSERT(pThis != NULL); if(pThis->pQueue != NULL) { - queueDestruct(&pThis->pQueue); + qqueueDestruct(&pThis->pQueue); } if(pThis->pMod != NULL) @@ -190,8 +195,35 @@ rsRetVal actionDestruct(action_t *pThis) SYNC_OBJ_TOOL_EXIT(pThis); pthread_mutex_destroy(&pThis->mutActExec); - if(pThis->ppTpl != NULL) - d_free(pThis->ppTpl); + d_free(pThis->pszName); + d_free(pThis->ppTpl); + + /* message ptr cleanup */ + for(i = 0 ; i < pThis->iNumTpls ; ++i) { + if(((uchar**)pThis->ppMsgs)[i] != NULL) { + switch(pThis->eParamPassing) { + case ACT_ARRAY_PASSING: +#if 0 /* later! */ + iArr = 0; + while(((char **)pThis->ppMsgs[i])[iArr] != NULL) { + d_free(((char **)pThis->ppMsgs[i])[iArr++]); + ((char **)pThis->ppMsgs[i])[iArr++] = NULL; + } + d_free(pThis->ppMsgs[i]); + pThis->ppMsgs[i] = NULL; +#endif + break; + case ACT_STRING_PASSING: + d_free(((uchar**)pThis->ppMsgs)[i]); + break; + default: + assert(0); + } + } + } + d_free(pThis->ppMsgs); + d_free(pThis->lenMsgs); + d_free(pThis); RETiRet; @@ -208,10 +240,7 @@ rsRetVal actionConstruct(action_t **ppThis) ASSERT(ppThis != NULL); - if((pThis = (action_t*) calloc(1, sizeof(action_t))) == NULL) { - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - } - + CHKmalloc(pThis = (action_t*) calloc(1, sizeof(action_t))); pThis->iResumeInterval = glbliActionResumeInterval; pThis->iResumeRetryCount = glbliActionResumeRetryCount; pThis->tLastOccur = time(NULL); /* done once per action on startup only */ @@ -254,7 +283,7 @@ actionConstructFinalize(action_t *pThis) * to be run on multiple threads. So far, this is forbidden by the interface * spec. -- rgerhards, 2008-01-30 */ - CHKiRet(queueConstruct(&pThis->pQueue, ActionQueType, 1, iActionQueueSize, (rsRetVal (*)(void*,void*))actionCallDoAction)); + CHKiRet(qqueueConstruct(&pThis->pQueue, ActionQueType, 1, iActionQueueSize, (rsRetVal (*)(void*,void*))actionCallDoAction)); obj.SetName((obj_t*) pThis->pQueue, pszQName); /* ... set some properties ... */ @@ -267,24 +296,25 @@ actionConstructFinalize(action_t *pThis) errmsg.LogError(0, NO_ERRCODE, "Invalid " #directive ", error %d. Ignored, running with default setting", iRet); \ } - queueSetpUsr(pThis->pQueue, pThis); - setQPROP(queueSetsizeOnDiskMax, "$ActionQueueMaxDiskSpace", iActionQueMaxDiskSpace); - setQPROP(queueSetMaxFileSize, "$ActionQueueFileSize", iActionQueMaxFileSize); - setQPROPstr(queueSetFilePrefix, "$ActionQueueFileName", pszActionQFName); - setQPROP(queueSetiPersistUpdCnt, "$ActionQueueCheckpointInterval", iActionQPersistUpdCnt); - setQPROP(queueSettoQShutdown, "$ActionQueueTimeoutShutdown", iActionQtoQShutdown ); - setQPROP(queueSettoActShutdown, "$ActionQueueTimeoutActionCompletion", iActionQtoActShutdown); - setQPROP(queueSettoWrkShutdown, "$ActionQueueWorkerTimeoutThreadShutdown", iActionQtoWrkShutdown); - setQPROP(queueSettoEnq, "$ActionQueueTimeoutEnqueue", iActionQtoEnq); - setQPROP(queueSetiHighWtrMrk, "$ActionQueueHighWaterMark", iActionQHighWtrMark); - setQPROP(queueSetiLowWtrMrk, "$ActionQueueLowWaterMark", iActionQLowWtrMark); - setQPROP(queueSetiDiscardMrk, "$ActionQueueDiscardMark", iActionQDiscardMark); - setQPROP(queueSetiDiscardSeverity, "$ActionQueueDiscardSeverity", iActionQDiscardSeverity); - setQPROP(queueSetiMinMsgsPerWrkr, "$ActionQueueWorkerThreadMinimumMessages", iActionQWrkMinMsgs); - setQPROP(queueSetbSaveOnShutdown, "$ActionQueueSaveOnShutdown", bActionQSaveOnShutdown); - setQPROP(queueSetiDeqSlowdown, "$ActionQueueDequeueSlowdown", iActionQueueDeqSlowdown); - setQPROP(queueSetiDeqtWinFromHr, "$ActionQueueDequeueTimeBegin", iActionQueueDeqtWinFromHr); - setQPROP(queueSetiDeqtWinToHr, "$ActionQueueDequeueTimeEnd", iActionQueueDeqtWinToHr); + qqueueSetpUsr(pThis->pQueue, pThis); + setQPROP(qqueueSetsizeOnDiskMax, "$ActionQueueMaxDiskSpace", iActionQueMaxDiskSpace); + 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 @@ -293,8 +323,8 @@ actionConstructFinalize(action_t *pThis) bActionQSaveOnShutdown, iActionQueMaxDiskSpace); - CHKiRet(queueStart(pThis->pQueue)); - dbgprintf("Action %p: queue %p created\n", pThis, 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!) */ actionResetQueueParams(); @@ -352,7 +382,13 @@ static rsRetVal actionTryResume(action_t *pThis) ASSERT(pThis != NULL); - ttNow = getActNow(pThis); /* cache "now" */ + /* for resume handling, we must always obtain a fresh timestamp. We used + * to use the action timestamp, but in this case we will never reach a + * point where a resumption is actually tried, because the action timestamp + * is always in the past. So we can not avoid doing a fresh time() call + * here. -- rgerhards, 2009-03-18 + */ + time(&ttNow); /* cache "now" */ /* first check if it is time for a re-try */ if(ttNow >= pThis->ttResumeRtry) { @@ -376,7 +412,7 @@ static rsRetVal actionTryResume(action_t *pThis) if(iRet == RS_RET_OK) actionResume(pThis); - dbgprintf("actionTryResume: iRet: %d, next retry (if applicable): %u [now %u]\n", + DBGPRINTF("actionTryResume: iRet: %d, next retry (if applicable): %u [now %u]\n", iRet, (unsigned) pThis->ttResumeRtry, (unsigned) ttNow); RETiRet; @@ -418,23 +454,13 @@ actionCallDoAction(action_t *pAction, msg_t *pMsg) DEFiRet; int iRetries; int i; + int iArr; int iSleepPeriod; int bCallAction; int iCancelStateSave; - uchar **ppMsgs; /* array of message pointers for doAction */ ASSERT(pAction != NULL); - /* create the array for doAction() message pointers */ - if((ppMsgs = calloc(pAction->iNumTpls, sizeof(uchar *))) == NULL) { - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - } - - /* here we must loop to process all requested strings */ - for(i = 0 ; i < pAction->iNumTpls ; ++i) { - CHKiRet(tplToString(pAction->ppTpl[i], pMsg, &(ppMsgs[i]))); - } - iRetries = 0; /* 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 * if they notify us they are - functionality not yet implemented...). @@ -444,6 +470,21 @@ actionCallDoAction(action_t *pAction, msg_t *pMsg) d_pthread_mutex_lock(&pAction->mutActExec); pthread_cleanup_push(mutexCancelCleanup, &pAction->mutActExec); pthread_setcancelstate(iCancelStateSave, NULL); + + /* here we must loop to process all requested strings */ + for(i = 0 ; i < pAction->iNumTpls ; ++i) { + switch(pAction->eParamPassing) { + case ACT_STRING_PASSING: + CHKiRet(tplToString(pAction->ppTpl[i], pMsg, &(((uchar**)pAction->ppMsgs)[i]), &(pAction->lenMsgs[i]))); + break; + case ACT_ARRAY_PASSING: + CHKiRet(tplToArray(pAction->ppTpl[i], pMsg, (uchar***) &(((uchar**)pAction->ppMsgs)[i]))); + break; + default:assert(0); /* software bug if this happens! */ + } + } + + iRetries = 0; do { /* on first invocation, this if should never be true. We just put it at the top * of the loop so that processing (and code) is simplified. This code is actually @@ -468,9 +509,9 @@ actionCallDoAction(action_t *pAction, msg_t *pMsg) if(bCallAction) { /* call configured action */ - iRet = pAction->pMod->mod.om.doAction(ppMsgs, pMsg->msgFlags, pAction->pModData); + iRet = pAction->pMod->mod.om.doAction(pAction->ppMsgs, pMsg->msgFlags, pAction->pModData); if(iRet == RS_RET_SUSPENDED) { - dbgprintf("Action requested to be suspended, done that.\n"); + DBGPRINTF("Action requested to be suspended, done that.\n"); actionSuspend(pAction, getActNow(pAction)); } } @@ -478,26 +519,72 @@ actionCallDoAction(action_t *pAction, msg_t *pMsg) } while(iRet == RS_RET_SUSPENDED && (pAction->iResumeRetryCount == -1 || iRetries < pAction->iResumeRetryCount)); /* do...while! */ if(iRet == RS_RET_DISABLE_ACTION) { - dbgprintf("Action requested to be disabled, done that.\n"); + DBGPRINTF("Action requested to be disabled, done that.\n"); pAction->bEnabled = 0; /* that's it... */ } - pthread_cleanup_pop(1); /* unlock mutex */ - finalize_it: /* cleanup */ for(i = 0 ; i < pAction->iNumTpls ; ++i) { - if(ppMsgs[i] != NULL) { - d_free(ppMsgs[i]); + if(((uchar**)pAction->ppMsgs)[i] != NULL) { + switch(pAction->eParamPassing) { + case ACT_ARRAY_PASSING: + iArr = 0; + while((((uchar***)pAction->ppMsgs)[i][iArr]) != NULL) { + d_free(((uchar ***)pAction->ppMsgs)[i][iArr++]); + ((uchar ***)pAction->ppMsgs)[i][iArr++] = NULL; + } + d_free(((uchar**)pAction->ppMsgs)[i]); + ((uchar**)pAction->ppMsgs)[i] = NULL; + break; + case ACT_STRING_PASSING: + break; + default: + assert(0); + } } } - d_free(ppMsgs); + + pthread_cleanup_pop(1); /* unlock mutex */ + msgDestruct(&pMsg); /* we are now finished with the message */ + RETiRet; +} +#pragma GCC diagnostic warning "-Wempty-body" + + +/* call the HUP handler for a given action, if such a handler is defined. The + * action mutex is locked, because the HUP handler most probably needs to modify + * some internal state information. + * rgerhards, 2008-10-22 + */ +#pragma GCC diagnostic ignored "-Wempty-body" +rsRetVal +actionCallHUPHdlr(action_t *pAction) +{ + DEFiRet; + int iCancelStateSave; + + ASSERT(pAction != NULL); + DBGPRINTF("Action %p checks HUP hdlr: %p\n", pAction, pAction->pMod->doHUP); + + if(pAction->pMod->doHUP == NULL) { + FINALIZE; /* no HUP handler, so we are done ;) */ + } + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave); + d_pthread_mutex_lock(&pAction->mutActExec); + pthread_cleanup_push(mutexCancelCleanup, &pAction->mutActExec); + pthread_setcancelstate(iCancelStateSave, NULL); + CHKiRet(pAction->pMod->doHUP(pAction->pModData)); + pthread_cleanup_pop(1); /* unlock mutex */ + +finalize_it: RETiRet; } #pragma GCC diagnostic warning "-Wempty-body" + /* set the action message queue mode * TODO: probably move this into queue object, merge with MainMsgQueue! * rgerhards, 2008-01-28 @@ -508,16 +595,16 @@ static rsRetVal setActionQueType(void __attribute__((unused)) *pVal, uchar *pszT if (!strcasecmp((char *) pszType, "fixedarray")) { ActionQueType = QUEUETYPE_FIXED_ARRAY; - dbgprintf("action queue type set to FIXED_ARRAY\n"); + DBGPRINTF("action queue type set to FIXED_ARRAY\n"); } else if (!strcasecmp((char *) pszType, "linkedlist")) { ActionQueType = QUEUETYPE_LINKEDLIST; - dbgprintf("action queue type set to LINKEDLIST\n"); + DBGPRINTF("action queue type set to LINKEDLIST\n"); } else if (!strcasecmp((char *) pszType, "disk")) { ActionQueType = QUEUETYPE_DISK; - dbgprintf("action queue type set to DISK\n"); + DBGPRINTF("action queue type set to DISK\n"); } else if (!strcasecmp((char *) pszType, "direct")) { ActionQueType = QUEUETYPE_DIRECT; - dbgprintf("action queue type set to DIRECT (no queueing at all)\n"); + 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); iRet = RS_RET_INVALID_PARAMS; @@ -563,15 +650,15 @@ actionWriteToAction(action_t *pAction) /* we need to care about multiple occurences */ if( pAction->iExecEveryNthOccurTO > 0 && (getActNow(pAction) - pAction->tLastOccur) > pAction->iExecEveryNthOccurTO) { - dbgprintf("n-th occurence handling timed out (%d sec), restarting from 0\n", + DBGPRINTF("n-th occurence handling timed out (%d sec), restarting from 0\n", (int) (getActNow(pAction) - pAction->tLastOccur)); pAction->iNbrNoExec = 0; pAction->tLastOccur = getActNow(pAction); } if(pAction->iNbrNoExec < pAction->iExecEveryNthOccur - 1) { ++pAction->iNbrNoExec; - dbgprintf("action %p passed %d times to execution - less than neded - discarding\n", - pAction, pAction->iNbrNoExec); + DBGPRINTF("action %p passed %d times to execution - less than neded - discarding\n", + pAction, pAction->iNbrNoExec); FINALIZE; } else { pAction->iNbrNoExec = 0; /* we execute the action now, so the number of no execs is down to */ @@ -588,30 +675,34 @@ actionWriteToAction(action_t *pAction) */ if(pAction->f_prevcount > 1) { msg_t *pMsg; - uchar szRepMsg[64]; - snprintf((char*)szRepMsg, sizeof(szRepMsg), "last message repeated %d times", - pAction->f_prevcount); + size_t lenRepMsg; + uchar szRepMsg[1024]; if((pMsg = MsgDup(pAction->f_pMsg)) == NULL) { /* it failed - nothing we can do against it... */ - dbgprintf("Message duplication failed, dropping repeat message.\n"); + DBGPRINTF("Message duplication failed, dropping repeat message.\n"); ABORT_FINALIZE(RS_RET_ERR); } - /* We now need to update the other message properties. - * ... RAWMSG is a problem ... Please note that digital + if(pAction->bRepMsgHasMsg == 0) { /* old format repeat message? */ + lenRepMsg = snprintf((char*)szRepMsg, sizeof(szRepMsg), " last message repeated %d times", + pAction->f_prevcount); + } else { + lenRepMsg = snprintf((char*)szRepMsg, sizeof(szRepMsg), " message repeated %d times: [%.800s]", + pAction->f_prevcount, getMSG(pAction->f_pMsg)); + } + + /* We now need to update the other message properties. Please note that digital * signatures inside the message are also invalidated. */ - datetime.getCurrTime(&(pMsg->tRcvdAt)); + datetime.getCurrTime(&(pMsg->tRcvdAt), &(pMsg->ttGenTime)); memcpy(&pMsg->tTIMESTAMP, &pMsg->tRcvdAt, sizeof(struct syslogTime)); - MsgSetMSG(pMsg, (char*)szRepMsg); - MsgSetRawMsg(pMsg, (char*)szRepMsg); - + MsgReplaceMSG(pMsg, szRepMsg, lenRepMsg); pMsgSave = pAction->f_pMsg; /* save message pointer for later restoration */ pAction->f_pMsg = pMsg; /* use the new msg (pointer will be restored below) */ } - dbgprintf("Called action, logging to %s\n", module.GetStateName(pAction->pMod)); + DBGPRINTF("Called action, logging to %s\n", module.GetStateName(pAction->pMod)); /* now check if we need to drop the message because otherwise the action would be too * frequently called. -- rgerhards, 2008-04-08 @@ -619,24 +710,23 @@ actionWriteToAction(action_t *pAction) * a purely logical point of view. However, if safes us to check the system time in * (those common) cases where ExecOnceInterval is not used. -- rgerhards, 2008-09-16 */ - if(pAction->f_time != 0 && pAction->iSecsExecOnceInterval > 0 && + if(pAction->iSecsExecOnceInterval > 0 && pAction->iSecsExecOnceInterval + pAction->tLastExec > getActNow(pAction)) { /* in this case we need to discard the message - its not yet time to exec the action */ - dbgprintf("action not yet ready again to be executed, onceInterval %d, tCurr %d, tNext %d\n", + DBGPRINTF("action not yet ready again to be executed, onceInterval %d, tCurr %d, tNext %d\n", (int) pAction->iSecsExecOnceInterval, (int) getActNow(pAction), (int) (pAction->iSecsExecOnceInterval + pAction->tLastExec)); FINALIZE; } - pAction->f_time = pAction->tLastExec = getActNow(pAction); /* re-init time flags */ - /* Note: tLastExec could be set in the if block above, but f_time causes us a hard time - * so far, I do not see a solution to getting rid of it. -- rgerhards, 2008-09-16 - */ + /* we use reception time, not dequeue time - this is considered more appropriate and also faster ;) -- rgerhards, 2008-09-17 */ + pAction->tLastExec = getActNow(pAction); /* re-init time flags */ + pAction->f_time = pAction->f_pMsg->ttGenTime; /* 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 = queueEnqObj(pAction->pQueue, pAction->f_pMsg->flowCtlType, (void*) MsgAddRef(pAction->f_pMsg)); + iRet = qqueueEnqObj(pAction->pQueue, pAction->f_pMsg->flowCtlType, (void*) MsgAddRef(pAction->f_pMsg)); if(iRet == RS_RET_OK) pAction->f_prevcount = 0; /* message processed, so we start a new cycle */ @@ -660,32 +750,13 @@ finalize_it: } -/* call the configured action. Does all necessary housekeeping. - * rgerhards, 2007-08-01 - * FYI: currently, this function is only called from the queue - * consumer. So we (conceptually) run detached from the input - * threads (which also means we may run much later than when the - * message was generated). +/* helper to actonCallAction, mostly needed because of this damn + * pthread_cleanup_push() POSIX macro... */ -#pragma GCC diagnostic ignored "-Wempty-body" -rsRetVal -actionCallAction(action_t *pAction, msg_t *pMsg) +static rsRetVal +doActionCallAction(action_t *pAction, msg_t *pMsg) { DEFiRet; - int iCancelStateSave; - - ISOBJ_TYPE_assert(pMsg, msg); - ASSERT(pAction != NULL); - - /* Make sure nodbody else modifies/uses this action object. Right now, this - * is important because of "message repeated n times" processing and potentially - * multiple worker threads. -- rgerhards, 2007-12-11 - */ - pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave); - LockObj(pAction); - pthread_cleanup_push(mutexCancelCleanup, pAction->Sync_mut); - pthread_setcancelstate(iCancelStateSave, NULL); - /* first, we need to check if this is a disabled * entry. If so, we must not further process it. * rgerhards 2005-09-26 @@ -710,12 +781,12 @@ actionCallAction(action_t *pAction, msg_t *pMsg) /* suppress duplicate messages */ if ((pAction->f_ReduceRepeated == 1) && pAction->f_pMsg != NULL && (pMsg->msgFlags & MARK) == 0 && getMSGLen(pMsg) == getMSGLen(pAction->f_pMsg) && - !strcmp(getMSG(pMsg), getMSG(pAction->f_pMsg)) && + !ustrcmp(getMSG(pMsg), getMSG(pAction->f_pMsg)) && !strcmp(getHOSTNAME(pMsg), getHOSTNAME(pAction->f_pMsg)) && - !strcmp(getPROCID(pMsg), getPROCID(pAction->f_pMsg)) && - !strcmp(getAPPNAME(pMsg), getAPPNAME(pAction->f_pMsg))) { + !strcmp(getPROCID(pMsg, LOCK_MUTEX), getPROCID(pAction->f_pMsg, LOCK_MUTEX)) && + !strcmp(getAPPNAME(pMsg, LOCK_MUTEX), getAPPNAME(pAction->f_pMsg, LOCK_MUTEX))) { pAction->f_prevcount++; - dbgprintf("msg repeated %d times, %ld sec of %d.\n", + DBGPRINTF("msg repeated %d times, %ld sec of %d.\n", pAction->f_prevcount, (long) getActNow(pAction) - pAction->f_time, repeatinterval[pAction->f_repeatcount]); /* use current message, so we have the new timestamp (means we need to discard previous one) */ @@ -746,10 +817,36 @@ actionCallAction(action_t *pAction, msg_t *pMsg) } finalize_it: + RETiRet; +} + +/* call the configured action. Does all necessary housekeeping. + * rgerhards, 2007-08-01 + * FYI: currently, this function is only called from the queue + * consumer. So we (conceptually) run detached from the input + * threads (which also means we may run much later than when the + * message was generated). + */ +#pragma GCC diagnostic ignored "-Wempty-body" +rsRetVal +actionCallAction(action_t *pAction, msg_t *pMsg) +{ + DEFiRet; + int iCancelStateSave; + + ISOBJ_TYPE_assert(pMsg, msg); + ASSERT(pAction != NULL); + + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave); + LockObj(pAction); + pthread_cleanup_push(mutexCancelCleanup, pAction->Sync_mut); + pthread_setcancelstate(iCancelStateSave, NULL); + iRet = doActionCallAction(pAction, pMsg); pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave); UnlockObj(pAction); pthread_cleanup_pop(0); /* remove mutex cleanup handler */ pthread_setcancelstate(iCancelStateSave, NULL); + RETiRet; } #pragma GCC diagnostic warning "-Wempty-body" @@ -763,6 +860,7 @@ actionAddCfSysLineHdrl(void) { DEFiRet; + 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 *)"actionqueuemaxdiskspace", 0, eCmdHdlrSize, NULL, &iActionQueMaxDiskSpace, NULL)); @@ -771,6 +869,7 @@ actionAddCfSysLineHdrl(void) 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(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)); @@ -785,6 +884,7 @@ actionAddCfSysLineHdrl(void) 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 *)"repeatedmsgcontainsoriginalmsg", 0, eCmdHdlrBinary, NULL, &bActionRepMsgHasMsg, NULL)); finalize_it: RETiRet; @@ -809,15 +909,18 @@ 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 processed this config line.\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->bExecWhenPrevSusp = bActExecWhenPrevSusp; pAction->iSecsExecOnceInterval = iActExecOnceInterval; pAction->iExecEveryNthOccur = iActExecEveryNthOccur; pAction->iExecEveryNthOccurTO = iActExecEveryNthOccurTO; + pAction->bRepMsgHasMsg = bActionRepMsgHasMsg; iActExecEveryNthOccur = 0; /* auto-reset */ iActExecEveryNthOccurTO = 0; /* auto-reset */ @@ -830,9 +933,9 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringReques */ if(pAction->iNumTpls > 0) { /* we first need to create the template pointer array */ - if((pAction->ppTpl = calloc(pAction->iNumTpls, sizeof(struct template *))) == NULL) { - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - } + CHKmalloc(pAction->ppTpl = (struct template **)calloc(pAction->iNumTpls, sizeof(struct template *))); + CHKmalloc(pAction->ppMsgs = (uchar**) calloc(pAction->iNumTpls, sizeof(uchar *))); + CHKmalloc(pAction->lenMsgs = (size_t*) calloc(pAction->iNumTpls, sizeof(size_t))); } for(i = 0 ; i < pAction->iNumTpls ; ++i) { @@ -858,7 +961,14 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringReques ABORT_FINALIZE(RS_RET_RQD_TPLOPT_MISSING); } - dbgprintf("template: '%s' assigned\n", pTplName); + /* set parameter-passing mode */ + if(iTplOpts & OMSR_TPL_AS_ARRAY) { + pAction->eParamPassing = ACT_ARRAY_PASSING; + } else { + pAction->eParamPassing = ACT_STRING_PASSING; + } + + DBGPRINTF("template: '%s' assigned\n", pTplName); } pAction->pMod = pMod; @@ -867,7 +977,7 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringReques if(pMod->isCompatibleWithFeature(sFEATURERepeatedMsgReduction) == RS_RET_OK) pAction->f_ReduceRepeated = bReduceRepeatMsgs; else { - dbgprintf("module is incompatible with RepeatedMsgReduction - turned off\n"); + DBGPRINTF("module is incompatible with RepeatedMsgReduction - turned off\n"); pAction->f_ReduceRepeated = 0; } pAction->bEnabled = 1; /* action is enabled */ |