diff options
Diffstat (limited to 'action.c')
-rw-r--r-- | action.c | 173 |
1 files changed, 32 insertions, 141 deletions
@@ -12,11 +12,11 @@ * necessary to triple-check that everything works well in *all* modes. * The different modes (and calling sequence) are: * - * if set iExecEveryNthOccur > 1 || f_ReduceRepeated || iSecsExecOnceInterval + * if set iExecEveryNthOccur > 1 || iSecsExecOnceInterval * - doSubmitToActionQComplexBatch * - helperSubmitToActionQComplexBatch * - doActionCallAction - * handles duplicate message processing, but in essence calls + * handles mark message reduction, but in essence calls * - actionWriteToAction * - qqueueEnqObj * (now queue engine processing) @@ -307,9 +307,6 @@ rsRetVal actionDestruct(action_t *pThis) if(pThis->pMod != NULL) pThis->pMod->freeInstance(pThis->pModData); - if(pThis->f_pMsg != NULL) - msgDestruct(&pThis->f_pMsg); - pthread_mutex_destroy(&pThis->mutAction); pthread_mutex_destroy(&pThis->mutActExec); d_free(pThis->pszName); @@ -410,16 +407,11 @@ actionConstructFinalize(action_t *pThis, struct cnfparamvals *queueParams) * mode is much faster processing (and simpler code) -- rgerhards, 2010-06-08 */ if( pThis->iExecEveryNthOccur > 1 - || pThis->f_ReduceRepeated || pThis->iSecsExecOnceInterval ) { DBGPRINTF("info: firehose mode disabled for action because " - "iExecEveryNthOccur=%d, " - "ReduceRepeated=%d, " - "iSecsExecOnceInterval=%d\n", - pThis->iExecEveryNthOccur, pThis->f_ReduceRepeated, - pThis->iSecsExecOnceInterval - ); + "iExecEveryNthOccur=%d, iSecsExecOnceInterval=%d\n", + pThis->iExecEveryNthOccur, pThis->iSecsExecOnceInterval); pThis->submitToActQ = doSubmitToActionQComplexBatch; } else if(pThis->bWriteAllMarkMsgs == RSFALSE) { /* nearly full-speed submission mode, default case */ @@ -438,7 +430,7 @@ actionConstructFinalize(action_t *pThis, struct cnfparamvals *queueParams) CHKiRet(qqueueConstruct(&pThis->pQueue, cs.ActionQueType, 1, cs.iActionQueueSize, (rsRetVal (*)(void*, batch_t*, int*))processBatchMain)); obj.SetName((obj_t*) pThis->pQueue, pszAName); - qqueueSetpUsr(pThis->pQueue, pThis); + qqueueSetpAction(pThis->pQueue, pThis); if(queueParams == NULL) { /* use legacy params? */ /* ... set some properties ... */ @@ -774,7 +766,6 @@ rsRetVal actionDbgPrint(action_t *pThis) pThis->pMod->dbgPrintInstInfo(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) { dbgprintf("\tresume next retry: %u, number retries: %d", @@ -801,7 +792,8 @@ rsRetVal actionDbgPrint(action_t *pThis) /* prepare the calling parameters for doAction() * rgerhards, 2009-05-07 */ -static rsRetVal prepareDoActionParams(action_t *pAction, batch_obj_t *pElem) +static rsRetVal +prepareDoActionParams(action_t *pAction, batch_obj_t *pElem, struct syslogTime *ttNow) { int i; msg_t *pMsg; @@ -811,23 +803,23 @@ static rsRetVal prepareDoActionParams(action_t *pAction, batch_obj_t *pElem) ASSERT(pAction != NULL); ASSERT(pElem != NULL); - pMsg = (msg_t*) pElem->pUsrp; + pMsg = pElem->pMsg; /* 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, &(pElem->staticActStrings[i]), - &pElem->staticLenStrings[i])); + &pElem->staticLenStrings[i], ttNow)); pElem->staticActParams[i] = pElem->staticActStrings[i]; break; case ACT_ARRAY_PASSING: - CHKiRet(tplToArray(pAction->ppTpl[i], pMsg, (uchar***) &(pElem->staticActParams[i]))); + CHKiRet(tplToArray(pAction->ppTpl[i], pMsg, (uchar***) &(pElem->staticActParams[i]), ttNow)); break; case ACT_MSG_PASSING: pElem->staticActParams[i] = (void*) pMsg; break; case ACT_JSON_PASSING: - CHKiRet(tplToJSON(pAction->ppTpl[i], pMsg, &json)); + CHKiRet(tplToJSON(pAction->ppTpl[i], pMsg, &json, ttNow)); pElem->staticActParams[i] = (void*) json; break; default:dbgprintf("software bug/error: unknown pAction->eParamPassing %d in prepareDoActionParams\n", @@ -1061,7 +1053,7 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem) * enq side of the queue (see file header comment)! -- rgerhards, 2011-06-15 */ if(batchIsValidElem(pBatch, i)) { - pMsg = (msg_t*) pBatch->pElem[i].pUsrp; + pMsg = pBatch->pElem[i].pMsg; localRet = actionProcessMessage(pAction, pMsg, pBatch->pElem[i].staticActParams, pBatch->pbShutdownImmediate); DBGPRINTF("action %p call returned %d\n", pAction, localRet); @@ -1218,14 +1210,18 @@ prepareBatch(action_t *pAction, batch_t *pBatch, sbool **activeSave, int *bMustR { int i; batch_obj_t *pElem; + struct syslogTime ttNow; DEFiRet; + /* indicate we have not yet read the date */ + ttNow.year = 0; + pBatch->iDoneUpTo = 0; for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { pElem = &(pBatch->pElem[i]); if(batchIsValidElem(pBatch, i)) { pElem->state = BATCH_STATE_RDY; - if(prepareDoActionParams(pAction, pElem) != RS_RET_OK) { + if(prepareDoActionParams(pAction, pElem, &ttNow) != RS_RET_OK) { /* make sure we have our copy of "active" array */ if(!*bMustRestoreActivePtr) { *activeSave = pBatch->active; @@ -1387,9 +1383,9 @@ doSubmitToActionQ(action_t *pAction, msg_t *pMsg) STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed); if(pAction->pQueue->qType == QUEUETYPE_DIRECT) - iRet = qqueueEnqObjDirect(pAction->pQueue, (void*) MsgAddRef(pMsg)); + iRet = qqueueEnqMsgDirect(pAction->pQueue, MsgAddRef(pMsg)); else - iRet = qqueueEnqObj(pAction->pQueue, eFLOWCTL_NO_DELAY, (void*) MsgAddRef(pMsg)); + iRet = qqueueEnqMsg(pAction->pQueue, eFLOWCTL_NO_DELAY, MsgAddRef(pMsg)); finalize_it: RETiRet; @@ -1404,14 +1400,10 @@ finalize_it: * be filtered out before calling us (what is done currently!). */ rsRetVal -actionWriteToAction(action_t *pAction) +actionWriteToAction(action_t *pAction, msg_t *pMsg) { - msg_t *pMsgSave; /* to save current message pointer, necessary to restore - it in case it needs to be updated (e.g. repeated msgs) */ DEFiRet; - pMsgSave = NULL; /* indicate message poiner not saved */ - /* first, we check if the action should actually be called. The action-specific * $ActionExecOnlyEveryNthTime permits us to execute an action only every Nth * time. So we need to check if we need to drop the (otherwise perfectly executable) @@ -1438,43 +1430,6 @@ actionWriteToAction(action_t *pAction) } } - /* then check if this is a regular message or the repeation of - * a previous message. If so, we need to change the message text - * to "last message repeated n times" and then go ahead and write - * it. Please note that we can not modify the message object, because - * that would update it in other selectors as well. As such, we first - * need to create a local copy of the message, which we than can update. - * rgerhards, 2007-07-10 - */ - if(pAction->f_prevcount > 1) { - msg_t *pMsg; - 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"); - ABORT_FINALIZE(RS_RET_ERR); - } - - 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), &(pMsg->ttGenTime)); - memcpy(&pMsg->tTIMESTAMP, &pMsg->tRcvdAt, sizeof(struct syslogTime)); - 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(complex case), logging to %s\n", module.GetStateName(pAction->pMod)); /* now check if we need to drop the message because otherwise the action would be too @@ -1495,31 +1450,14 @@ actionWriteToAction(action_t *pAction) /* 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; + pAction->f_time = 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 = doSubmitToActionQ(pAction, pAction->f_pMsg); - - if(iRet == RS_RET_OK) - pAction->f_prevcount = 0; /* message processed, so we start a new cycle */ + iRet = doSubmitToActionQ(pAction, pMsg); finalize_it: - if(pMsgSave != NULL) { - /* we had saved the original message pointer. That was - * done because we needed to create a temporary one - * (most often for "message repeated n time" handling). If so, - * we need to restore the original one now, so that procesing - * can continue as normal. We also need to discard the temporary - * one, as we do not like memory leaks ;) Please note that the original - * message object will be discarded by our callers, so this is nothing - * of our business. rgerhards, 2007-07-10 - */ - msgDestruct(&pAction->f_pMsg); - pAction->f_pMsg = pMsgSave; /* restore it */ - } - RETiRet; } @@ -1533,7 +1471,7 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch) msg_t *pMsg; DEFiRet; - pMsg = (msg_t*)(pBatch->pElem[idxBtch].pUsrp); + pMsg = pBatch->pElem[idxBtch].pMsg; pAction->tActNow = -1; /* we do not yet know our current time (clear prev. value) */ /* don't output marks to recently written outputs */ @@ -1542,43 +1480,8 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch) ABORT_FINALIZE(RS_RET_OK); } - /* suppress duplicate messages */ - if ((pAction->f_ReduceRepeated == 1) && pAction->f_pMsg != NULL && - (pMsg->msgFlags & MARK) == 0 && getMSGLen(pMsg) == getMSGLen(pAction->f_pMsg) && - !ustrcmp(getMSG(pMsg), getMSG(pAction->f_pMsg)) && - !strcmp(getHOSTNAME(pMsg), getHOSTNAME(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", - 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) */ - msgDestruct(&pAction->f_pMsg); - pAction->f_pMsg = MsgAddRef(pMsg); - /* If domark would have logged this by now, flush it now (so we don't hold - * isolated messages), but back off so we'll flush less often in the future. - */ - if(getActNow(pAction) > REPEATTIME(pAction)) { - iRet = actionWriteToAction(pAction); - BACKOFF(pAction); - } - } else {/* new message, save it */ - /* first check if we have a previous message stored - * if so, emit and then discard it first - */ - if(pAction->f_pMsg != NULL) { - if(pAction->f_prevcount > 0) - actionWriteToAction(pAction); - /* we do not care about iRet above - I think it's right but if we have - * some troubles, you know where to look at ;) -- rgerhards, 2007-08-01 - */ - msgDestruct(&pAction->f_pMsg); - } - pAction->f_pMsg = MsgAddRef(pMsg); - /* call the output driver */ - iRet = actionWriteToAction(pAction); - } + /* call the output driver */ + iRet = actionWriteToAction(pAction, pMsg); finalize_it: /* we need to update the batch to handle failover processing correctly */ @@ -1660,7 +1563,7 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) * also faster ;) -- rgerhards, 2008-09-17 */ do { lastAct = pAction->f_time; - if(((msg_t*)(pBatch->pElem[i].pUsrp))->msgFlags & MARK) { + if(pBatch->pElem[i].pMsg->msgFlags & MARK) { if((now - lastAct) < MarkInterval / 2) { pBatch->active[i] = 0; DBGPRINTF("batch item %d: action was recently called, ignoring " @@ -1669,7 +1572,7 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) } } } while(ATOMIC_CAS_time_t(&pAction->f_time, lastAct, - ((msg_t*)(pBatch->pElem[i].pUsrp))->ttGenTime, &pAction->mutCAS) == 0); + pBatch->pElem[i].pMsg->ttGenTime, &pAction->mutCAS) == 0); if(pBatch->active[i]) { DBGPRINTF("Called action(NotAllMark), processing batch[%d] via '%s'\n", i, module.GetStateName(pAction->pMod)); @@ -1772,7 +1675,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, (msg_t*)(pBatch->pElem[i].pUsrp)); + doSubmitToActionQ(pAction, pBatch->pElem[i].pMsg); } } } @@ -1868,7 +1771,6 @@ actionApplyCnfParam(action_t *pAction, struct cnfparamvals *pvals) } - /* 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. @@ -1933,7 +1835,7 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, && (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", + " Could not find template '%s' - action disabled", pTplName); errno = 0; errmsg.LogError(0, RS_RET_NOT_FOUND, "%s", errMsg); @@ -1964,13 +1866,7 @@ addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, 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 = loadConf->globals.bReduceRepeatMsgs; - } else { - DBGPRINTF("module is incompatible with RepeatedMsgReduction - turned off\n"); - pAction->f_ReduceRepeated = 0; - } + /* check if the module is compatible with select features (currently no such features exist) */ pAction->eState = ACT_STATE_RDY; /* action is enabled */ if(bSuspended) @@ -2070,13 +1966,8 @@ actionNewInst(struct nvlst *lst, action_t **ppAction) 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; - } + /* check if the module is compatible with select features + * (currently no such features exist) */ pAction->eState = ACT_STATE_RDY; /* action is enabled */ loadConf->actions.nbrActions++; /* one more active action! */ } |