diff options
35 files changed, 644 insertions, 695 deletions
@@ -1,3 +1,5 @@ +- bugfix: unset statement always worked on message var, even if local + var was given --------------------------------------------------------------------------- Version 7.5.6 [devel] 2013-10-?? - RainerScript: make use of 64 bit for numbers where available @@ -119,10 +119,10 @@ #define NO_TIME_PROVIDED 0 /* indicate we do not provide any cached time */ /* forward definitions */ -static rsRetVal processBatchMain(action_t *pAction, batch_t *pBatch, int*); -static rsRetVal doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch); -static rsRetVal doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch); -static rsRetVal doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch); +static rsRetVal processBatchMain(void *pVoid, batch_t *pBatch, wti_t *pWti, int*); +static rsRetVal doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti); +static rsRetVal doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti); +static rsRetVal doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti); /* object static data (once for all instances) */ /* TODO: make this an object! DEFobjStaticHelpers -- rgerhards, 2008-03-05 */ @@ -175,10 +175,9 @@ configSettings_t cs_save; /* our saved (scope!) config settings */ /* the counter below counts actions created. It is used to obtain unique IDs for the action. They * should not be relied on for any long-term activity (e.g. disk queue names!), but they are nice * to have during one instance of an rsyslogd run. For example, I use them to name actions when there - * is no better name available. Note that I do NOT recover previous numbers on HUP - we simply keep - * counting. -- rgerhards, 2008-01-29 + * is no better name available. */ -static int iActionNbr = 0; +int iActionNbr = 0; /* tables for interfacing with the v6 config system */ static struct cnfparamdescr cnfparamdescr[] = { @@ -341,6 +340,7 @@ rsRetVal actionConstruct(action_t **ppThis) pThis->bExecWhenPrevSusp = 0; pThis->bRepMsgHasMsg = 0; pThis->tLastOccur = datetime.GetTime(NULL); /* done once per action on startup only */ + pThis->iActionNbr = iActionNbr; pthread_mutex_init(&pThis->mutActExec, NULL); pthread_mutex_init(&pThis->mutAction, NULL); INIT_ATOMIC_HELPER_MUT(pThis->mutCAS); @@ -428,7 +428,7 @@ actionConstructFinalize(action_t *pThis, struct nvlst *lst) * spec. -- rgerhards, 2008-01-30 */ CHKiRet(qqueueConstruct(&pThis->pQueue, cs.ActionQueType, 1, cs.iActionQueueSize, - (rsRetVal (*)(void*, batch_t*, int*))processBatchMain)); + processBatchMain)); obj.SetName((obj_t*) pThis->pQueue, pszAName); qqueueSetpAction(pThis->pQueue, pThis); @@ -905,17 +905,25 @@ done: RETiRet; * rgerhards, 2008-01-28 */ rsRetVal -actionCallDoAction(action_t *pThis, msg_t *pMsg, void *actParams) +actionCallDoAction(action_t *pThis, msg_t *pMsg, void *actParams, wti_t *pWti) { DEFiRet; ASSERT(pThis != NULL); ISOBJ_TYPE_assert(pMsg, msg); - DBGPRINTF("entering actionCalldoAction(), state: %s\n", getActStateName(pThis)); + DBGPRINTF("entering actionCalldoAction(), state: %s, actionNbr %d\n", + getActStateName(pThis), pThis->iActionNbr); + + if(pWti->actWrkrData[pThis->iActionNbr] == NULL) { + DBGPRINTF("we need to create a new action worker instance for " + "action %d\n", pThis->iActionNbr); + CHKiRet(pThis->pMod->mod.om.createWrkrInstance(&(pWti->actWrkrData[pThis->iActionNbr]), pThis->pModData)); + } pThis->bHadAutoCommit = 0; - iRet = pThis->pMod->mod.om.doAction(actParams, pMsg->msgFlags, pThis->pModData); + iRet = pThis->pMod->mod.om.doAction(actParams, pMsg->msgFlags, + pWti->actWrkrData[pThis->iActionNbr]); switch(iRet) { case RS_RET_OK: actionCommitted(pThis); @@ -952,8 +960,8 @@ finalize_it: * this readies the action and then calls doAction() * rgerhards, 2008-01-28 */ -static inline rsRetVal -actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams, int *pbShutdownImmediate) +static rsRetVal +actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams, int *pbShutdownImmediate, wti_t *pWti) { DEFiRet; @@ -964,7 +972,7 @@ actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams, int *pbShutd if(pThis->pMod->mod.om.SetShutdownImmdtPtr != NULL) pThis->pMod->mod.om.SetShutdownImmdtPtr(pThis->pModData, pbShutdownImmediate); if(pThis->eState == ACT_STATE_ITX) - CHKiRet(actionCallDoAction(pThis, pMsg, actParams)); + CHKiRet(actionCallDoAction(pThis, pMsg, actParams, pWti)); iRet = getReturnCode(pThis); finalize_it: @@ -977,7 +985,7 @@ finalize_it: * rgerhards, 2008-01-28 */ static rsRetVal -finishBatch(action_t *pThis, batch_t *pBatch) +finishBatch(action_t *pThis, batch_t *pBatch, wti_t *pWti) { int i; DEFiRet; @@ -991,7 +999,7 @@ finishBatch(action_t *pThis, batch_t *pBatch) CHKiRet(actionPrepare(pThis, pBatch->pbShutdownImmediate)); if(pThis->eState == ACT_STATE_ITX) { - iRet = pThis->pMod->mod.om.endTransaction(pThis->pModData); + iRet = pThis->pMod->mod.om.endTransaction(pWti->actWrkrData[pThis->iActionNbr]); switch(iRet) { case RS_RET_OK: actionCommitted(pThis); @@ -1034,7 +1042,7 @@ finalize_it: * rgerhards, 2009-05-12 */ static inline rsRetVal -tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem) +tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem, wti_t *pWti) { int i; int iElemProcessed; @@ -1059,7 +1067,7 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem) if(batchIsValidElem(pBatch, i)) { pMsg = pBatch->pElem[i].pMsg; localRet = actionProcessMessage(pAction, pMsg, pBatch->pElem[i].staticActParams, - pBatch->pbShutdownImmediate); + pBatch->pbShutdownImmediate, pWti); DBGPRINTF("action %p call returned %d\n", pAction, localRet); /* Note: we directly modify the batch object state, because we know that * wo do not overwrite BATCH_STATE_DISC indicators! @@ -1111,7 +1119,7 @@ finalize_it: * rgerhards, 2009-05-12 */ static rsRetVal -submitBatch(action_t *pAction, batch_t *pBatch, int nElem) +submitBatch(action_t *pAction, batch_t *pBatch, int nElem, wti_t *pWti) { int i; int bDone; @@ -1125,7 +1133,7 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem) wasDoneTo = pBatch->iDoneUpTo; bDone = 0; do { - localRet = tryDoAction(pAction, pBatch, &nElem); + localRet = tryDoAction(pAction, pBatch, &nElem, pWti); if(localRet == RS_RET_FORCE_TERM) { ABORT_FINALIZE(RS_RET_FORCE_TERM); } @@ -1135,7 +1143,7 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem) /* try commit transaction, once done, we can simply do so as if * that return state was returned from tryDoAction(). */ - localRet = finishBatch(pAction, pBatch); + localRet = finishBatch(pAction, pBatch, pWti); } if( localRet == RS_RET_OK @@ -1164,8 +1172,8 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem) /* retry with half as much. Depth is log_2 batchsize, so recursion is not too deep */ DBGPRINTF("submitBatch recursing trying to find and exclude the culprit " "for iRet %d\n", localRet); - submitBatch(pAction, pBatch, nElem / 2); - submitBatch(pAction, pBatch, nElem - (nElem / 2)); + submitBatch(pAction, pBatch, nElem / 2, pWti); + submitBatch(pAction, pBatch, nElem - (nElem / 2), pWti); bDone = 1; } } @@ -1245,13 +1253,13 @@ prepareBatch(action_t *pAction, batch_t *pBatch, sbool **activeSave, int *bMustR * rgerhards, 2009-05-12 */ static inline rsRetVal -processAction(action_t *pAction, batch_t *pBatch) +processAction(action_t *pAction, batch_t *pBatch, wti_t* pWti) { DEFiRet; assert(pBatch != NULL); - CHKiRet(submitBatch(pAction, pBatch, pBatch->nElem)); - iRet = finishBatch(pAction, pBatch); + CHKiRet(submitBatch(pAction, pBatch, pBatch->nElem, pWti)); + iRet = finishBatch(pAction, pBatch, pWti); finalize_it: RETiRet; @@ -1264,12 +1272,13 @@ finalize_it: * rgerhards, 2009-04-22 */ static rsRetVal -processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) +processBatchMain(void *pVoid, batch_t *pBatch, wti_t *pWti, int *pbShutdownImmediate) { int *pbShutdownImmdtSave; sbool *activeSave; int bMustRestoreActivePtr = 0; rsRetVal localRet; + action_t *pAction = (action_t*) pVoid; DEFiRet; assert(pBatch != NULL); @@ -1288,7 +1297,7 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) d_pthread_mutex_lock(&pAction->mutActExec); pthread_cleanup_push(mutexCancelCleanup, &pAction->mutActExec); - iRet = processAction(pAction, pBatch); + iRet = processAction(pAction, pBatch, pWti); pthread_cleanup_pop(1); /* unlock mutex */ @@ -1381,7 +1390,7 @@ static rsRetVal setActionQueType(void __attribute__((unused)) *pVal, uchar *pszT * rgerhards, 2010-06-08 */ static inline rsRetVal -doSubmitToActionQ(action_t *pAction, msg_t *pMsg) +doSubmitToActionQ(action_t *pAction, msg_t *pMsg, wti_t *pWti) { DEFiRet; @@ -1392,7 +1401,7 @@ doSubmitToActionQ(action_t *pAction, msg_t *pMsg) STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed); if(pAction->pQueue->qType == QUEUETYPE_DIRECT) - iRet = qqueueEnqMsgDirect(pAction->pQueue, MsgAddRef(pMsg)); + iRet = qqueueEnqMsgDirect(pAction->pQueue, MsgAddRef(pMsg), pWti); else iRet = qqueueEnqMsg(pAction->pQueue, eFLOWCTL_NO_DELAY, MsgAddRef(pMsg)); @@ -1409,7 +1418,7 @@ finalize_it: * be filtered out before calling us (what is done currently!). */ rsRetVal -actionWriteToAction(action_t *pAction, msg_t *pMsg) +actionWriteToAction(action_t *pAction, msg_t *pMsg, wti_t *pWti) { DEFiRet; @@ -1464,7 +1473,7 @@ actionWriteToAction(action_t *pAction, msg_t *pMsg) /* When we reach this point, we have a valid, non-disabled action. * So let's enqueue our message for execution. -- rgerhards, 2007-07-24 */ - iRet = doSubmitToActionQ(pAction, pMsg); + iRet = doSubmitToActionQ(pAction, pMsg, pWti); finalize_it: RETiRet; @@ -1475,7 +1484,7 @@ finalize_it: * pthread_cleanup_push() POSIX macro... */ static inline rsRetVal -doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch) +doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch, wti_t *pWti) { msg_t *pMsg; DEFiRet; @@ -1490,7 +1499,7 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch) } /* call the output driver */ - iRet = actionWriteToAction(pAction, pMsg); + iRet = actionWriteToAction(pAction, pMsg, pWti); finalize_it: /* we need to update the batch to handle failover processing correctly */ @@ -1550,7 +1559,7 @@ activateActions(void) * rgerhards, 2010-06-08 */ static rsRetVal -doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) +doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti) { time_t now = 0; time_t lastAct; @@ -1588,7 +1597,7 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) } } - iRet = doSubmitToActionQBatch(pAction, pBatch); + iRet = doSubmitToActionQBatch(pAction, pBatch, pWti); free(pBatch->active); pBatch->active = activeSave; @@ -1613,7 +1622,7 @@ countStatsBatchEnq(action_t *pAction, batch_t *pBatch) * rgerhards, 2011-06-16 */ static inline rsRetVal -doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch) +doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti) { sbool bNeedSubmit; sbool *activeSave; @@ -1645,14 +1654,14 @@ doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch) } if(bNeedSubmit) { /* note: stats were already computed above */ - iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch); + iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch, pWti); } else { DBGPRINTF("no need to submit batch, all invalid\n"); } } else { if(GatherStats) countStatsBatchEnq(pAction, pBatch); - iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch); + iRet = qqueueEnqObjDirectBatch(pAction->pQueue, pBatch, pWti); } free(pBatch->active); @@ -1666,7 +1675,7 @@ doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch) * rgerhards, 2010-06-08 */ static rsRetVal -doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch) +doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti) { int i; DEFiRet; @@ -1674,7 +1683,7 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch) DBGPRINTF("Called action(Batch), logging to %s\n", module.GetStateName(pAction->pMod)); if(pAction->pQueue->qType == QUEUETYPE_DIRECT) { - iRet = doQueueEnqObjDirectBatch(pAction, pBatch); + iRet = doQueueEnqObjDirectBatch(pAction, pBatch, pWti); } else {/* in this case, we do single submits to the queue. * TODO: optimize this, we may do at least a multi-submit! */ @@ -1684,7 +1693,7 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch) pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended); if( batchIsValidElem(pBatch, i) && (pAction->bExecWhenPrevSusp == 0 || pBatch->pElem[i].bPrevWasSuspended == 1)) { - doSubmitToActionQ(pAction, pBatch->pElem[i].pMsg); + doSubmitToActionQ(pAction, pBatch->pElem[i].pMsg, pWti); } } } @@ -1699,7 +1708,7 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch) * rgerhards, 2010-06-23 */ static inline rsRetVal -helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) +helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti) { int i; DEFiRet; @@ -1712,7 +1721,7 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended); if( batchIsValidElem(pBatch, i) && ((pAction->bExecWhenPrevSusp == 0) || pBatch->pElem[i].bPrevWasSuspended) ) { - doActionCallAction(pAction, pBatch, i); + doActionCallAction(pAction, pBatch, i, pWti); } } @@ -1724,13 +1733,13 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) */ #pragma GCC diagnostic ignored "-Wempty-body" static rsRetVal -doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) +doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti) { DEFiRet; d_pthread_mutex_lock(&pAction->mutAction); pthread_cleanup_push(mutexCancelCleanup, &pAction->mutAction); - iRet = helperSubmitToActionQComplexBatch(pAction, pBatch); + iRet = helperSubmitToActionQComplexBatch(pAction, pBatch, pWti); d_pthread_mutex_unlock(&pAction->mutAction); pthread_cleanup_pop(0); /* remove mutex cleanup handler */ @@ -51,6 +51,7 @@ struct action_s { time_t tActNow; /* the current time for an action execution. Initially set to -1 and populated on an as-needed basis. This is a performance optimization. */ time_t tLastExec; /* time this action was last executed */ + int iActionNbr; /* this action's number (ID) */ sbool bExecWhenPrevSusp;/* execute only when previous action is suspended? */ sbool bWriteAllMarkMsgs;/* should all mark msgs be written (not matter how recent the action was executed)? */ int iSecsExecOnceInterval; /* if non-zero, minimum seconds to wait until action is executed again */ @@ -68,7 +69,7 @@ struct action_s { struct modInfo_s *pMod;/* pointer to output module handling this selector */ void *pModData; /* pointer to module data - content is module-specific */ sbool bRepMsgHasMsg; /* "message repeated..." has msg fragment in it (0-no, 1-yes) */ - rsRetVal (*submitToActQ)(action_t *, batch_t *);/* function submit message to action queue */ + rsRetVal (*submitToActQ)(action_t *, batch_t *, wti_t*);/* function submit message to action queue */ rsRetVal (*qConstruct)(struct queue_s *pThis); enum { ACT_STRING_PASSING = 0, ACT_ARRAY_PASSING = 1, ACT_MSG_PASSING = 2, ACT_JSON_PASSING = 3} @@ -96,7 +97,7 @@ rsRetVal actionDestruct(action_t *pThis); rsRetVal actionDbgPrint(action_t *pThis); rsRetVal actionSetGlobalResumeInterval(int iNewVal); rsRetVal actionDoAction(action_t *pAction); -rsRetVal actionWriteToAction(action_t *pAction, msg_t *pMsg); +rsRetVal actionWriteToAction(action_t *pAction, msg_t *pMsg, wti_t*); rsRetVal actionCallHUPHdlr(action_t *pAction); rsRetVal actionClassInit(void); rsRetVal addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringRequest_t *pOMSR, struct cnfparamvals *actParams, struct nvlst *lst, int bSuspended); @@ -104,4 +105,7 @@ rsRetVal activateActions(void); rsRetVal actionNewInst(struct nvlst *lst, action_t **ppAction); rsRetVal actionProcessCnf(struct cnfobj *o); +/* external data */ +extern int iActionNbr; + #endif /* #ifndef ACTION_H_INCLUDED */ diff --git a/grammar/lexer.l b/grammar/lexer.l index 36c23c7d..7fdb68af 100644 --- a/grammar/lexer.l +++ b/grammar/lexer.l @@ -129,7 +129,7 @@ int fileno(FILE *stream); <EXPR>0[0-7]+ | /* octal number */ <EXPR>0x[0-7a-f] | /* hex number, following rule is dec; strtoll handles all! */ <EXPR>([1-9][0-9]*|0) { yylval.n = strtoll(yytext, NULL, 0); return NUMBER; } -<EXPR>\$[$!./]{0,1}[a-z][!a-z0-9\-_\.]* { yylval.s = strdup(yytext); return VAR; } +<EXPR>\$[$!./]{0,1}[a-z][!a-z0-9\-_\.]* { yylval.s = strdup(yytext+1); return VAR; } <EXPR>\'([^'\\]|\\['"\\$bntr]|\\x[0-9a-f][0-9a-f]|\\[0-7][0-7][0-7])*\' { yytext[yyleng-1] = '\0'; unescapeStr((uchar*)yytext+1, yyleng-2); diff --git a/grammar/parserif.h b/grammar/parserif.h index aa271ec4..6c2c1365 100644 --- a/grammar/parserif.h +++ b/grammar/parserif.h @@ -19,5 +19,4 @@ void cnfDoScript(struct cnfstmt *script); void cnfDoCfsysline(char *ln); void cnfDoBSDTag(char *ln); void cnfDoBSDHost(char *ln); -es_str_t *cnfGetVar(char *name, void *usrptr); #endif diff --git a/grammar/rainerscript.c b/grammar/rainerscript.c index a2bed2bf..30c36201 100644 --- a/grammar/rainerscript.c +++ b/grammar/rainerscript.c @@ -47,6 +47,8 @@ #include "obj.h" #include "modules.h" #include "ruleset.h" +#include "msg.h" +#include "unicode-helper.h" DEFobjCurrIf(obj) DEFobjCurrIf(regexp) @@ -176,20 +178,8 @@ DecodePropFilter(uchar *pline, struct cnfstmt *stmt) rsParsDestruct(pPars); ABORT_FINALIZE(iRet); } - iRet = propNameToID(pCSPropName, &stmt->d.s_propfilt.propID); - if(iRet != RS_RET_OK) { - parser_errmsg("invalid property name '%s' in filter", - cstrGetSzStrNoNULL(pCSPropName)); - rsParsDestruct(pPars); - ABORT_FINALIZE(iRet); - } - if(stmt->d.s_propfilt.propID == PROP_CEE) { - /* in CEE case, we need to preserve the actual property name */ - if((stmt->d.s_propfilt.propName = - es_newStrFromBuf((char*)cstrGetSzStrNoNULL(pCSPropName)+2, cstrLen(pCSPropName)-2)) == NULL) { - ABORT_FINALIZE(RS_RET_ERR); - } - } + CHKiRet(msgPropDescrFill(&stmt->d.s_propfilt.prop, cstrGetSzStrNoNULL(pCSPropName), + cstrLen(pCSPropName))); /* read operation */ iRet = parsDelimCStr(pPars, &pCSCompOp, ',', 1, 1, 1); @@ -1257,6 +1247,23 @@ var2CString(struct var *r, int *bMustFree) return cstr; } +/* frees struct var members, but not the struct itself. This is because + * it usually is allocated on the stack. Callers why dynamically allocate + * struct var need to free the struct themselfes! + */ +static void +varFreeMembers(struct var *r) +{ + /* Note: we do NOT need to free JSON objects, as we use + * json_object_object_get() to obtain the values, which does not + * increment the reference count. So json_object_put() [free] is + * neither required nor permitted (would free the original object!). + * So for the time being the string data type is the only one that + * we currently need to free. + */ + if(r->datatype == 'S') es_deleteStr(r->d.estr); +} + static rsRetVal doExtractFieldByChar(uchar *str, uchar delim, int matchnbr, uchar **resstr) { @@ -1424,9 +1431,9 @@ doFunc_re_extract(struct cnffunc *func, struct var *ret, void* usrptr) finalize_it: if(bMustFree) free(str); - if(r[0].datatype == 'S') es_deleteStr(r[0].d.estr); - if(r[2].datatype == 'S') es_deleteStr(r[2].d.estr); - if(r[3].datatype == 'S') es_deleteStr(r[3].d.estr); + varFreeMembers(&r[0]); + varFreeMembers(&r[2]); + varFreeMembers(&r[3]); if(bHadNoMatch) { cnfexprEval(func->expr[4], &r[4], usrptr); @@ -1495,7 +1502,7 @@ doFuncCall(struct cnffunc *func, struct var *ret, void* usrptr) } ret->datatype = 'S'; if(bMustFree) es_deleteStr(estr); - if(r[0].datatype == 'S') es_deleteStr(r[0].d.estr); + varFreeMembers(&r[0]); free(str); break; case CNFFUNC_TOLOWER: @@ -1506,7 +1513,7 @@ doFuncCall(struct cnffunc *func, struct var *ret, void* usrptr) es_tolower(estr); ret->datatype = 'S'; ret->d.estr = estr; - if(r[0].datatype == 'S') es_deleteStr(r[0].d.estr); + varFreeMembers(&r[0]); break; case CNFFUNC_CSTR: cnfexprEval(func->expr[0], &r[0], usrptr); @@ -1515,7 +1522,7 @@ doFuncCall(struct cnffunc *func, struct var *ret, void* usrptr) estr = es_strdup(estr); ret->datatype = 'S'; ret->d.estr = estr; - if(r[0].datatype == 'S') es_deleteStr(r[0].d.estr); + varFreeMembers(&r[0]); break; case CNFFUNC_CNUM: if(func->expr[0]->nodetype == 'N') { @@ -1526,7 +1533,7 @@ doFuncCall(struct cnffunc *func, struct var *ret, void* usrptr) } else { cnfexprEval(func->expr[0], &r[0], usrptr); ret->d.n = var2Number(&r[0], NULL); - if(r[0].datatype == 'S') es_deleteStr(r[0].d.estr); + varFreeMembers(&r[0]); } ret->datatype = 'N'; break; @@ -1544,7 +1551,7 @@ doFuncCall(struct cnffunc *func, struct var *ret, void* usrptr) } ret->datatype = 'N'; if(bMustFree) free(str); - if(r[0].datatype == 'S') es_deleteStr(r[0].d.estr); + varFreeMembers(&r[0]); break; case CNFFUNC_RE_EXTRACT: doFunc_re_extract(func, ret, usrptr); @@ -1577,9 +1584,9 @@ doFuncCall(struct cnffunc *func, struct var *ret, void* usrptr) } ret->datatype = 'S'; if(bMustFree) free(str); - if(r[0].datatype == 'S') es_deleteStr(r[0].d.estr); - if(r[1].datatype == 'S') es_deleteStr(r[1].d.estr); - if(r[2].datatype == 'S') es_deleteStr(r[2].d.estr); + varFreeMembers(&r[0]); + varFreeMembers(&r[1]); + varFreeMembers(&r[2]); break; case CNFFUNC_PRIFILT: pPrifilt = (struct funcData_prifilt*) func->funcdata; @@ -1619,20 +1626,28 @@ dbgprintf("DDDD: executing lookup\n"); static inline void evalVar(struct cnfvar *var, void *usrptr, struct var *ret) { + rs_size_t propLen; + uchar *pszProp = NULL; + unsigned short bMustBeFreed = 0; rsRetVal localRet; - es_str_t *estr; struct json_object *json; - if(var->name[0] == '$' && var->name[1] == '!') { - /* TODO: unify string libs */ - estr = es_newStrFromBuf(var->name+1, strlen(var->name)-1); - localRet = msgGetCEEPropJSON((msg_t*)usrptr, estr, &json); - es_deleteStr(estr); + if(var->prop.id == PROP_CEE || + var->prop.id == PROP_LOCAL_VAR || + var->prop.id == PROP_GLOBAL_VAR ) { + localRet = msgGetJSONPropJSON((msg_t*)usrptr, &var->prop, &json); ret->datatype = 'J'; ret->d.json = (localRet == RS_RET_OK) ? json : NULL; + + DBGPRINTF("rainerscript: var %d:%s: '%s'\n", var->prop.id, var->prop.name, + (ret->d.json == NULL) ? "" : json_object_get_string(ret->d.json)); } else { ret->datatype = 'S'; - ret->d.estr = cnfGetVar(var->name, usrptr); + pszProp = (uchar*) MsgGetProp((msg_t*)usrptr, NULL, &var->prop, &propLen, &bMustBeFreed, NULL); + ret->d.estr = es_newStrFromCStr((char*)pszProp, propLen); + DBGPRINTF("rainerscript: var %d: '%s'\n", var->prop.id, pszProp); + if(bMustBeFreed) + free(pszProp); } } @@ -1678,8 +1693,8 @@ evalStrArrayCmp(es_str_t *estr_l, struct cnfarray* ar, int cmpop) } #define FREE_BOTH_RET \ - if(r.datatype == 'S') es_deleteStr(r.d.estr); \ - if(l.datatype == 'S') es_deleteStr(l.d.estr) + varFreeMembers(&r); \ + varFreeMembers(&l) #define COMP_NUM_BINOP(x) \ cnfexprEval(expr->l, &l, usrptr); \ @@ -1706,9 +1721,9 @@ evalStrArrayCmp(es_str_t *estr_l, struct cnfarray* ar, int cmpop) #define FREE_TWO_STRINGS \ if(bMustFree) es_deleteStr(estr_r); \ - if(expr->r->nodetype != 'S' && expr->r->nodetype != 'A' && r.datatype == 'S') es_deleteStr(r.d.estr); \ + if(expr->r->nodetype != 'S' && expr->r->nodetype != 'A') varFreeMembers(&r); \ if(bMustFree2) es_deleteStr(estr_l); \ - if(l.datatype == 'S') es_deleteStr(l.d.estr) + varFreeMembers(&l) /* evaluate an expression. * Note that we try to avoid malloc whenever possible (because of @@ -1759,7 +1774,7 @@ cnfexprEval(struct cnfexpr *expr, struct var *ret, void* usrptr) if(bMustFree) es_deleteStr(estr_r); } } - if(r.datatype == 'S') es_deleteStr(r.d.estr); + varFreeMembers(&r); } } else if(l.datatype == 'J') { estr_l = var2String(&l, &bMustFree); @@ -1781,7 +1796,7 @@ cnfexprEval(struct cnfexpr *expr, struct var *ret, void* usrptr) if(bMustFree) es_deleteStr(estr_r); } } - if(r.datatype == 'S') es_deleteStr(r.d.estr); + varFreeMembers(&r); } if(bMustFree) es_deleteStr(estr_l); } else { @@ -1798,9 +1813,9 @@ cnfexprEval(struct cnfexpr *expr, struct var *ret, void* usrptr) } else { ret->d.n = (l.d.n == r.d.n); /*CMP*/ } - if(r.datatype == 'S') es_deleteStr(r.d.estr); + varFreeMembers(&r); } - if(l.datatype == 'S') es_deleteStr(l.d.estr); + varFreeMembers(&l); break; case CMP_NE: cnfexprEval(expr->l, &l, usrptr); @@ -2028,9 +2043,9 @@ cnfexprEval(struct cnfexpr *expr, struct var *ret, void* usrptr) ret->d.n = 1ll; else ret->d.n = 0ll; - if(r.datatype == 'S') es_deleteStr(r.d.estr); + varFreeMembers(&r); } - if(l.datatype == 'S') es_deleteStr(l.d.estr); + varFreeMembers(&l); break; case AND: cnfexprEval(expr->l, &l, usrptr); @@ -2041,17 +2056,17 @@ cnfexprEval(struct cnfexpr *expr, struct var *ret, void* usrptr) ret->d.n = 1ll; else ret->d.n = 0ll; - if(r.datatype == 'S') es_deleteStr(r.d.estr); + varFreeMembers(&r); } else { ret->d.n = 0ll; } - if(l.datatype == 'S') es_deleteStr(l.d.estr); + varFreeMembers(&l); break; case NOT: cnfexprEval(expr->r, &r, usrptr); ret->datatype = 'N'; ret->d.n = !var2Number(&r, &convok_r); - if(r.datatype == 'S') es_deleteStr(r.d.estr); + varFreeMembers(&r); break; case 'N': ret->datatype = 'N'; @@ -2102,7 +2117,7 @@ cnfexprEval(struct cnfexpr *expr, struct var *ret, void* usrptr) cnfexprEval(expr->r, &r, usrptr); ret->datatype = 'N'; ret->d.n = -var2Number(&r, &convok_r); - if(r.datatype == 'S') es_deleteStr(r.d.estr); + varFreeMembers(&r); break; case 'F': doFuncCall((struct cnffunc*) expr, ret, usrptr); @@ -2195,6 +2210,7 @@ cnfexprDestruct(struct cnfexpr *expr) break; case 'V': free(((struct cnfvar*)expr)->name); + msgPropDescrDestruct(&(((struct cnfvar*)expr)->prop)); break; case 'F': cnffuncDestruct((struct cnffunc*)expr); @@ -2411,7 +2427,7 @@ cnfstmtPrintOnly(struct cnfstmt *stmt, int indent, sbool subtree) free(cstr); break; case S_ACT: - doIndent(indent); dbgprintf("ACTION %p [%s:%s]\n", stmt->d.act, + doIndent(indent); dbgprintf("ACTION %d [%s:%s]\n", stmt->d.act->iActionNbr, modGetName(stmt->d.act->pMod), stmt->printable); break; case S_IF: @@ -2452,12 +2468,10 @@ cnfstmtPrintOnly(struct cnfstmt *stmt, int indent, sbool subtree) case S_PROPFILT: doIndent(indent); dbgprintf("PROPFILT\n"); doIndent(indent); dbgprintf("\tProperty.: '%s'\n", - propIDToName(stmt->d.s_propfilt.propID)); - if(stmt->d.s_propfilt.propName != NULL) { - cstr = es_str2cstr(stmt->d.s_propfilt.propName, NULL); + propIDToName(stmt->d.s_propfilt.prop.id)); + if(stmt->d.s_propfilt.prop.name != NULL) { doIndent(indent); - dbgprintf("\tCEE-Prop.: '%s'\n", cstr); - free(cstr); + dbgprintf("\tCEE-Prop.: '%s'\n", stmt->d.s_propfilt.prop.name); } doIndent(indent); dbgprintf("\tOperation: "); if(stmt->d.s_propfilt.isNegated) @@ -2564,6 +2578,7 @@ cnfvarNew(char *name) if((var = malloc(sizeof(struct cnfvar))) != NULL) { var->nodetype = 'V'; var->name = name; + msgPropDescrFill(&var->prop, (uchar*)var->name, strlen(var->name)); } return var; } @@ -2617,8 +2632,7 @@ cnfstmtDestruct(struct cnfstmt *stmt) cnfstmtDestructLst(stmt->d.s_prifilt.t_else); break; case S_PROPFILT: - if(stmt->d.s_propfilt.propName != NULL) - es_deleteStr(stmt->d.s_propfilt.propName); + msgPropDescrDestruct(&stmt->d.s_propfilt.prop); if(stmt->d.s_propfilt.regex_cache != NULL) rsCStrRegexDestruct(&stmt->d.s_propfilt.regex_cache); if(stmt->d.s_propfilt.pCSCompValue != NULL) @@ -2703,7 +2717,6 @@ cnfstmtNewPROPFILT(char *propfilt, struct cnfstmt *t_then) if((cnfstmt = cnfstmtNew(S_PROPFILT)) != NULL) { cnfstmt->printable = (uchar*)propfilt; cnfstmt->d.s_propfilt.t_then = t_then; - cnfstmt->d.s_propfilt.propName = NULL; cnfstmt->d.s_propfilt.regex_cache = NULL; cnfstmt->d.s_propfilt.pCSCompValue = NULL; if(DecodePropFilter((uchar*)propfilt, cnfstmt) != RS_RET_OK) { diff --git a/grammar/rainerscript.h b/grammar/rainerscript.h index 4a508f93..001dff4e 100644 --- a/grammar/rainerscript.h +++ b/grammar/rainerscript.h @@ -6,7 +6,6 @@ #include <sys/types.h> #include <regex.h> - #define LOG_NFACILITIES 24 /* current number of syslog facilities */ #define CNFFUNC_MAX_ARGS 32 /**< maximum number of arguments that any function can have (among @@ -182,8 +181,7 @@ struct cnfstmt { regex_t *regex_cache;/* cache for compiled REs, if used */ struct cstr_s *pCSCompValue;/* value to "compare" against */ sbool isNegated; - uintTiny propID;/* ID of the requested property */ - es_str_t *propName;/* name of property for CEE-based filters */ + msgPropDescr_t prop; /* requested property */ struct cnfstmt *t_then; struct cnfstmt *t_else; } s_propfilt; @@ -210,6 +208,7 @@ struct cnfstringval { struct cnfvar { unsigned nodetype; char *name; + msgPropDescr_t prop; }; struct cnfarray { diff --git a/grammar/testdriver.c b/grammar/testdriver.c index b29626d4..58d204a3 100644 --- a/grammar/testdriver.c +++ b/grammar/testdriver.c @@ -87,15 +87,6 @@ void cnfDoBSDHost(char *ln) dbgprintf("global:BSD host: %s\n", ln); } -es_str_t* -cnfGetVar(char __attribute__((unused)) *name, - void __attribute__((unused)) *usrptr) -{ - es_str_t *estr; - estr = es_newStrFromCStr("", 1); - return estr; -} - int main(int argc, char *argv[]) { diff --git a/plugins/ommongodb/ommongodb.c b/plugins/ommongodb/ommongodb.c index a7c42010..af1f5a37 100644 --- a/plugins/ommongodb/ommongodb.c +++ b/plugins/ommongodb/ommongodb.c @@ -235,12 +235,18 @@ getDefaultBSON(msg_t *pMsg) int severity, facil; gint64 ts_gen, ts_rcv; /* timestamps: generated, received */ int secfrac; - - procid = MsgGetProp(pMsg, NULL, PROP_PROGRAMNAME, NULL, &procid_len, &procid_free, NULL); - tag = MsgGetProp(pMsg, NULL, PROP_SYSLOGTAG, NULL, &tag_len, &tag_free, NULL); - pid = MsgGetProp(pMsg, NULL, PROP_PROCID, NULL, &pid_len, &pid_free, NULL); - sys = MsgGetProp(pMsg, NULL, PROP_HOSTNAME, NULL, &sys_len, &sys_free, NULL); - msg = MsgGetProp(pMsg, NULL, PROP_MSG, NULL, &msg_len, &msg_free, NULL); + msgPropDescr_t cProp; /* we use internal implementation knowledge... */ + + cProp.id = PROP_PROGRAMNAME; + procid = MsgGetProp(pMsg, NULL, &cProp, &procid_len, &procid_free, NULL); + cProp.id = PROP_SYSLOGTAG; + tag = MsgGetProp(pMsg, NULL, &cProp, &tag_len, &tag_free, NULL); + cProp.id = PROP_PROCID; + pid = MsgGetProp(pMsg, NULL, &cProp, &pid_len, &pid_free, NULL); + cProp.id = PROP_HOSTNAME; + sys = MsgGetProp(pMsg, NULL, &cProp, &sys_len, &sys_free, NULL); + cProp.id = PROP_MSG; + msg = MsgGetProp(pMsg, NULL, &cProp, &msg_len, &msg_free, NULL); // TODO: move to datetime? Refactor in any case! rgerhards, 2012-03-30 ts_gen = (gint64) datetime.syslogTime2time_t(&pMsg->tTIMESTAMP) * 1000; /* ms! */ diff --git a/plugins/omruleset/omruleset.c b/plugins/omruleset/omruleset.c index 11765507..2908095e 100644 --- a/plugins/omruleset/omruleset.c +++ b/plugins/omruleset/omruleset.c @@ -70,6 +70,10 @@ typedef struct _instanceData { uchar *pszRulesetName; /* primarily for debugging/display purposes */ } instanceData; +typedef struct wrkrInstanceData { + instanceData *pData; +} wrkrInstanceData_t; + typedef struct configSettings_s { ruleset_t *pRuleset; /* ruleset to enqueue message to (NULL = Default, not recommended) */ uchar *pszRulesetName; diff --git a/plugins/omtesting/omtesting.c b/plugins/omtesting/omtesting.c index c9f1e06b..7c793414 100644 --- a/plugins/omtesting/omtesting.c +++ b/plugins/omtesting/omtesting.c @@ -63,7 +63,6 @@ MODULE_CNFNAME("omtesting") */ DEF_OMOD_STATIC_DATA - typedef struct _instanceData { enum { MD_SLEEP, MD_FAIL, MD_RANDFAIL, MD_ALWAYS_SUSPEND } mode; @@ -76,6 +75,10 @@ typedef struct _instanceData { int iCurrRetries; } instanceData; +typedef struct wrkrInstanceData { + instanceData *pData; +} wrkrInstanceData_t; + typedef struct configSettings_s { int bEchoStdout; /* echo non-failed messages to stdout */ } configSettings_t; diff --git a/runtime/module-template.h b/runtime/module-template.h index 8a958f90..f6afc961 100644 --- a/runtime/module-template.h +++ b/runtime/module-template.h @@ -175,6 +175,44 @@ static rsRetVal freeInstance(void* pModData)\ RETiRet;\ } +/* createWrkrInstance() + */ +#define BEGINcreateWrkrInstance \ +static rsRetVal createWrkrInstance(wrkrInstanceData_t **ppWrkrData, instanceData *pData)\ + {\ + DEFiRet; /* store error code here */\ + wrkrInstanceData_t *pWrkrData; /* use this to point to data elements */ + +#define CODESTARTcreateWrkrInstance \ + if((pWrkrData = calloc(1, sizeof(wrkrInstanceData_t))) == NULL) {\ + *ppWrkrData = NULL;\ + ENDfunc \ + return RS_RET_OUT_OF_MEMORY;\ + } \ + pWrkrData->pData = pData; + +#define ENDcreateWrkrInstance \ + *ppWrkrData = pWrkrData;\ + RETiRet;\ +} + +/* freeWrkrInstance */ +#define BEGINfreeWrkrInstance \ +static rsRetVal freeWrkrInstance(void* pd)\ +{\ + DEFiRet;\ + wrkrInstanceData_t *pWrkrData; + +#define CODESTARTfreeWrkrInstance \ + pWrkrData = (wrkrInstanceData_t*) pModData; + +#define ENDfreeWrkrInstance \ + if(pWrkrData != NULL)\ + free(pWrkrData); /* we need to free this in any case */\ + RETiRet;\ +} + + /* isCompatibleWithFeature() */ #define BEGINisCompatibleWithFeature \ @@ -209,8 +247,9 @@ static rsRetVal beginTransaction(instanceData __attribute__((unused)) *pData)\ * introduced in v4.3.3 -- rgerhards, 2009-04-27 */ #define BEGINendTransaction \ -static rsRetVal endTransaction(instanceData __attribute__((unused)) *pData)\ +static rsRetVal endTransaction(wrkrInstanceData_t __attribute__((unused)) *pWrkrData)\ {\ + instanceData *pData = NULL; /* deliberately make module abort if it does not support new IF */\ DEFiRet; #define CODESTARTendTransaction /* currently empty, but may be extended */ @@ -223,8 +262,9 @@ static rsRetVal endTransaction(instanceData __attribute__((unused)) *pData)\ /* doAction() */ #define BEGINdoAction \ -static rsRetVal doAction(uchar __attribute__((unused)) **ppString, unsigned __attribute__((unused)) iMsgOpts, instanceData __attribute__((unused)) *pData)\ +static rsRetVal doAction(uchar __attribute__((unused)) **ppString, unsigned __attribute__((unused)) iMsgOpts, wrkrInstanceData_t __attribute__((unused)) *pWrkrData)\ {\ + instanceData *pData = NULL; /* deliberately make module abort if it does not support new IF */\ DEFiRet; #define CODESTARTdoAction \ @@ -467,6 +507,11 @@ static rsRetVal queryEtryPt(uchar *name, rsRetVal (**pEtryPoint)())\ *pEtryPoint = tryResume;\ } +/* standard queries for output module interface in rsyslog v8+ */ +#define CODEqueryEtryPt_STD_OMOD8_QUERIES \ + else if(!strcmp((char*) name, "createWrkrInstance")) {\ + *pEtryPoint = createWrkrInstance;\ + } /* the following definition is queryEtryPt block that must be added * if an output module supports the transactional interface. diff --git a/runtime/modules.c b/runtime/modules.c index 56606306..eca7e466 100644 --- a/runtime/modules.c +++ b/runtime/modules.c @@ -652,6 +652,12 @@ doModInit(rsRetVal (*modInit)(int, int*, rsRetVal(**)(), rsRetVal(*)(), modInfo_ CHKiRet((*pNew->modQueryEtryPt)((uchar*)"doAction", &pNew->mod.om.doAction)); CHKiRet((*pNew->modQueryEtryPt)((uchar*)"parseSelectorAct", &pNew->mod.om.parseSelectorAct)); CHKiRet((*pNew->modQueryEtryPt)((uchar*)"tryResume", &pNew->tryResume)); + + /* TODO: 2013-10-28: these will become mandatory but are not to get me started */ + localRet = (*pNew->modQueryEtryPt)((uchar*)"createWrkrInstance", &pNew->mod.om.createWrkrInstance); + if(localRet != RS_RET_OK && localRet != RS_RET_MODULE_ENTRY_POINT_NOT_FOUND) + ABORT_FINALIZE(localRet); + /* try load optional interfaces */ localRet = (*pNew->modQueryEtryPt)((uchar*)"doHUP", &pNew->doHUP); if(localRet != RS_RET_OK && localRet != RS_RET_MODULE_ENTRY_POINT_NOT_FOUND) diff --git a/runtime/modules.h b/runtime/modules.h index 23df22d6..31e87995 100644 --- a/runtime/modules.h +++ b/runtime/modules.h @@ -139,6 +139,7 @@ struct modInfo_s { rsRetVal (*parseSelectorAct)(uchar**, void**,omodStringRequest_t**); rsRetVal (*newActInst)(uchar *modName, struct nvlst *lst, void **, omodStringRequest_t **); rsRetVal (*SetShutdownImmdtPtr)(void *pData, void *pPtr); + rsRetVal (*createWrkrInstance)(void*pWrkrData, void*pData); } om; struct { /* data for library modules */ char dummy; diff --git a/runtime/msg.c b/runtime/msg.c index f795242f..21ab18e5 100644 --- a/runtime/msg.c +++ b/runtime/msg.c @@ -65,6 +65,7 @@ #include "net.h" #include "var.h" #include "rsconf.h" +#include "parserif.h" /* TODO: move the global variable root to the config object - had no time to to it * right now before vacation -- rgerhards, 2013-07-22 @@ -326,7 +327,6 @@ static int getAPPNAMELen(msg_t *pM, sbool bLockMutex); static rsRetVal jsonPathFindParent(struct json_object *jroot, uchar *name, uchar *leaf, struct json_object **parent, int bCreate); static uchar * jsonPathGetLeaf(uchar *name, int lenName); static struct json_object *jsonDeepCopy(struct json_object *src); -static rsRetVal msgAddJSONObj(msg_t *pM, uchar *name, struct json_object *json, struct json_object **pjroot); /* the locking and unlocking implementations: */ @@ -452,14 +452,12 @@ getRcvFromIP(msg_t *pM) } -/* map a property name (C string) to a property ID */ +/* map a property name (string) to a property ID */ rsRetVal -propNameStrToID(uchar *pName, propid_t *pPropID) +propNameToID(uchar *pName, propid_t *pPropID) { DEFiRet; - assert(pName != NULL); - /* sometimes there are aliases to the original MonitoWare * property names. These come after || in the ifs below. */ if(!strcmp((char*) pName, "msg")) { @@ -534,16 +532,16 @@ propNameStrToID(uchar *pName, propid_t *pPropID) *pPropID = PROP_SYS_MYHOSTNAME; } else if(!strcmp((char*) pName, "$!all-json")) { *pPropID = PROP_CEE_ALL_JSON; - } else if(!strncmp((char*) pName, "$!", 2)) { - *pPropID = PROP_CEE; - } else if(!strncmp((char*) pName, "$.", 2)) { - *pPropID = PROP_LOCAL_VAR; - } else if(!strncmp((char*) pName, "$/", 2)) { - *pPropID = PROP_GLOBAL_VAR; } else if(!strcmp((char*) pName, "$bom")) { *pPropID = PROP_SYS_BOM; } else if(!strcmp((char*) pName, "$uptime")) { *pPropID = PROP_SYS_UPTIME; + } else if(!strncmp((char*) pName, "$!", 2) || pName[0] == '!') { + *pPropID = PROP_CEE; + } else if(!strncmp((char*) pName, "$.", 2) || pName[0] == '.') { + *pPropID = PROP_LOCAL_VAR; + } else if(!strncmp((char*) pName, "$/", 2) || pName[0] == '/') { + *pPropID = PROP_GLOBAL_VAR; } else { DBGPRINTF("PROP_INVALID for name '%s'\n", pName); *pPropID = PROP_INVALID; @@ -554,21 +552,6 @@ propNameStrToID(uchar *pName, propid_t *pPropID) } -/* map a property name (string) to a property ID */ -rsRetVal -propNameToID(cstr_t *pCSPropName, propid_t *pPropID) -{ - uchar *pName; - DEFiRet; - - assert(pCSPropName != NULL); - assert(pPropID != NULL); - pName = rsCStrGetSzStrNoNULL(pCSPropName); - iRet = propNameStrToID(pName, pPropID); - RETiRet; -} - - /* map a property ID to a name string (useful for displaying) */ uchar *propIDToName(propid_t propID) { @@ -641,8 +624,6 @@ uchar *propIDToName(propid_t propID) return UCHAR_CONSTANT("*CEE-based property*"); case PROP_LOCAL_VAR: return UCHAR_CONSTANT("*LOCAL_VARIABLE*"); - case PROP_GLOBAL_VAR: - return UCHAR_CONSTANT("*GLOBAL_VARIABLE*"); case PROP_CEE_ALL_JSON: return UCHAR_CONSTANT("$!all-json"); case PROP_SYS_BOM: @@ -2476,11 +2457,17 @@ typedef enum ENOWType { NOW_NOW, NOW_YEAR, NOW_MONTH, NOW_DAY, NOW_HOUR, NOW_HHO static uchar *getNOW(eNOWType eNow, struct syslogTime *t) { uchar *pBuf; + struct syslogTime tt; if((pBuf = (uchar*) MALLOC(sizeof(uchar) * tmpBUFSIZE)) == NULL) { return NULL; } + if(t == NULL) { /* can happen if called via script engine */ + datetime.getCurrTime(&tt, NULL); + t = &tt; + } + if(t->year == 0) { /* not yet set! */ datetime.getCurrTime(t, NULL); } @@ -2524,11 +2511,11 @@ static uchar *getNOW(eNOWType eNow, struct syslogTime *t) /* Get a JSON-Property as string value (used for various types of JSON-based vars) */ -static rsRetVal -getJSONPropVal(struct json_object *jroot, es_str_t *propName, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed) +rsRetVal +getJSONPropVal(msg_t *pMsg, msgPropDescr_t *pProp, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed) { - uchar *name = NULL; uchar *leaf; + struct json_object *jroot; struct json_object *parent; struct json_object *field; DEFiRet; @@ -2536,15 +2523,26 @@ getJSONPropVal(struct json_object *jroot, es_str_t *propName, uchar **pRes, rs_s if(*pbMustBeFreed) free(*pRes); *pRes = NULL; - // TODO: mutex? + + if(pProp->id == PROP_CEE) { + jroot = pMsg->json; + } else if(pProp->id == PROP_LOCAL_VAR) { + jroot = pMsg->localvars; + } else if(pProp->id == PROP_GLOBAL_VAR) { + pthread_rwlock_rdlock(&glblVars_rwlock); + jroot = global_var_root; + } else { + DBGPRINTF("msgGetJSONPropVal; invalid property id %d\n", + pProp->id); + ABORT_FINALIZE(RS_RET_NOT_FOUND); + } if(jroot == NULL) goto finalize_it; - if(!es_strbufcmp(propName, (uchar*)"!", 1)) { + if(!strcmp((char*)pProp->name, "!")) { field = jroot; } else { - name = (uchar*)es_str2cstr(propName, NULL); - leaf = jsonPathGetLeaf(name+1, ustrlen(name-1)); - CHKiRet(jsonPathFindParent(jroot, name+1, leaf, &parent, 1)); + leaf = jsonPathGetLeaf(pProp->name, pProp->nameLen); + CHKiRet(jsonPathFindParent(jroot, pProp->name, leaf, &parent, 1)); field = json_object_object_get(parent, (char*)leaf); } if(field != NULL) { @@ -2554,7 +2552,8 @@ getJSONPropVal(struct json_object *jroot, es_str_t *propName, uchar **pRes, rs_s } finalize_it: - free(name); + if(pProp->id == PROP_GLOBAL_VAR) + pthread_rwlock_unlock(&glblVars_rwlock); if(*pRes == NULL) { /* could not find any value, so set it to empty */ *pRes = (unsigned char*)""; @@ -2563,81 +2562,51 @@ finalize_it: RETiRet; } -rsRetVal -getCEEPropVal(msg_t *pM, es_str_t *propName, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed) -{ - return getJSONPropVal(pM->json, propName, pRes, buflen, pbMustBeFreed); -} - -rsRetVal -getLocalVarPropVal(msg_t *pM, es_str_t *propName, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed) -{ - return getJSONPropVal(pM->localvars, propName, pRes, buflen, pbMustBeFreed); -} - -rsRetVal -getGlobalVarPropVal( es_str_t *propName, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed) -{ - DEFiRet; - pthread_rwlock_rdlock(&glblVars_rwlock); - iRet = getJSONPropVal(global_var_root, propName, pRes, buflen, pbMustBeFreed); - pthread_rwlock_unlock(&glblVars_rwlock); - RETiRet; -} - /* Get a JSON-based-variable as native json object */ rsRetVal -msgGetJSONPropJSON(struct json_object *jroot, es_str_t *propName, struct json_object **pjson) +msgGetJSONPropJSON(msg_t *pMsg, msgPropDescr_t *pProp, struct json_object **pjson) { - uchar *name = NULL; + struct json_object *jroot; uchar *leaf; struct json_object *parent; DEFiRet; - // TODO: mutex? + if(pProp->id == PROP_CEE) { + jroot = pMsg->json; + } else if(pProp->id == PROP_LOCAL_VAR) { + jroot = pMsg->localvars; + } else if(pProp->id == PROP_GLOBAL_VAR) { + pthread_rwlock_rdlock(&glblVars_rwlock); + jroot = global_var_root; + } else { + DBGPRINTF("msgGetJSONPropJSON; invalid property id %d\n", + pProp->id); + ABORT_FINALIZE(RS_RET_NOT_FOUND); + } if(jroot == NULL) { + DBGPRINTF("msgGetJSONPropJSON; jroot empty for property %s\n", + pProp->name); ABORT_FINALIZE(RS_RET_NOT_FOUND); } - if(!es_strbufcmp(propName, (uchar*)"!", 1)) { + if(!strcmp((char*)pProp->name, "!")) { *pjson = jroot; FINALIZE; } - name = (uchar*)es_str2cstr(propName, NULL); - leaf = jsonPathGetLeaf(name, ustrlen(name)); - CHKiRet(jsonPathFindParent(jroot, name, leaf, &parent, 1)); + leaf = jsonPathGetLeaf(pProp->name, pProp->nameLen); + CHKiRet(jsonPathFindParent(jroot, pProp->name, leaf, &parent, 1)); *pjson = json_object_object_get(parent, (char*)leaf); if(*pjson == NULL) { ABORT_FINALIZE(RS_RET_NOT_FOUND); } finalize_it: - free(name); + if(pProp->id == PROP_GLOBAL_VAR) + pthread_rwlock_unlock(&glblVars_rwlock); RETiRet; } -rsRetVal -msgGetCEEPropJSON(msg_t *pM, es_str_t *propName, struct json_object **pjson) -{ - return msgGetJSONPropJSON(pM->json, propName, pjson); -} - -rsRetVal -msgGetLocalVarJSON(msg_t *pM, es_str_t *propName, struct json_object **pjson) -{ - return msgGetJSONPropJSON(pM->localvars, propName, pjson); -} - -rsRetVal -msgGetGlobalVarJSON(es_str_t *propName, struct json_object **pjson) -{ - DEFiRet; - pthread_rwlock_rdlock(&glblVars_rwlock); - iRet = msgGetJSONPropJSON(global_var_root, propName, pjson); - pthread_rwlock_unlock(&glblVars_rwlock); - RETiRet; -} /* Encode a JSON value and add it to provided string. Note that * the string object may be NULL. In this case, it is created @@ -2839,7 +2808,7 @@ finalize_it: *pPropLen = sizeof("**OUT OF MEMORY**") - 1; \ return(UCHAR_CONSTANT("**OUT OF MEMORY**"));} uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, - propid_t propid, es_str_t *propName, rs_size_t *pPropLen, + msgPropDescr_t *pProp, rs_size_t *pPropLen, unsigned short *pbMustBeFreed, struct syslogTime *ttNow) { uchar *pRes; /* result pointer */ @@ -2862,7 +2831,7 @@ uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, *pbMustBeFreed = 0; - switch(propid) { + switch(pProp->id) { case PROP_MSG: pRes = getMSG(pMsg); bufLen = getMSGLen(pMsg); @@ -3034,13 +3003,9 @@ uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, } break; case PROP_CEE: - getCEEPropVal(pMsg, propName, &pRes, &bufLen, pbMustBeFreed); - break; case PROP_LOCAL_VAR: - getLocalVarPropVal(pMsg, propName, &pRes, &bufLen, pbMustBeFreed); - break; case PROP_GLOBAL_VAR: - getGlobalVarPropVal(propName, &pRes, &bufLen, pbMustBeFreed); + getJSONPropVal(pMsg, pProp, &pRes, &bufLen, pbMustBeFreed); break; case PROP_SYS_BOM: if(*pbMustBeFreed == 1) @@ -3101,7 +3066,7 @@ uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, /* there is no point in continuing, we may even otherwise render the * error message unreadable. rgerhards, 2007-07-10 */ - dbgprintf("invalid property id: '%d'\n", propid); + dbgprintf("invalid property id: '%d'\n", pProp->id); *pbMustBeFreed = 0; *pPropLen = sizeof("**INVALID PROPERTY NAME**") - 1; return UCHAR_CONSTANT("**INVALID PROPERTY NAME**"); @@ -3709,80 +3674,6 @@ uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, } -/* The function returns a json variable suitable for use with RainerScript. - * Note: caller must free the returned string. - * Note that we need to do a lot of conversions between es_str_t and cstr -- this will go away once - * we have moved larger parts of rsyslog to es_str_t. Acceptable for the moment, especially as we intend - * to rewrite the script engine as well! - * rgerhards, 2010-12-03 - */ -static es_str_t* -msgGetJSONVarNew(msg_t *pMsg, struct json_object *jroot, char *name) -{ - uchar *leaf; - char *val; - es_str_t *estr = NULL; - struct json_object *json, *parent; - - ISOBJ_TYPE_assert(pMsg, msg); - - if(jroot == NULL) { - estr = es_newStr(1); - goto done; - } - leaf = jsonPathGetLeaf((uchar*)name, strlen(name)); - if(jsonPathFindParent(jroot, (uchar*)name, leaf, &parent, 1) != RS_RET_OK) { - estr = es_newStr(1); - goto done; - } - json = json_object_object_get(parent, (char*)leaf); - val = (char*)json_object_get_string(json); - estr = es_newStrFromCStr(val, strlen(val)); -done: - return estr; -} - -es_str_t* -msgGetCEEVarNew(msg_t *pMsg, char *name) -{ - return msgGetJSONVarNew(pMsg, pMsg->json, name); -} - -es_str_t* -msgGetLocalVarNew(msg_t *pMsg, char *name) -{ - return msgGetJSONVarNew(pMsg, pMsg->localvars, name); -} - -/* Return an es_str_t for given message property. - */ -es_str_t* -msgGetMsgVarNew(msg_t *pThis, uchar *name) -{ - rs_size_t propLen; - uchar *pszProp = NULL; - propid_t propid; - unsigned short bMustBeFreed = 0; - es_str_t *estr; - es_str_t *propName; - - ISOBJ_TYPE_assert(pThis, msg); - - /* always call MsgGetProp() without a template specifier */ - /* TODO: optimize propNameToID() call -- rgerhards, 2009-06-26 */ - propNameStrToID(name, &propid); - propName = es_newStrFromCStr((char*)name, ustrlen(name)); // TODO: optimize! - pszProp = (uchar*) MsgGetProp(pThis, NULL, propid, propName, &propLen, &bMustBeFreed, NULL); - es_deleteStr(propName); - - estr = es_newStrFromCStr((char*)pszProp, propLen); - if(bMustBeFreed) - free(pszProp); - - return estr; -} - - /* This function can be used as a generic way to set properties. * We have to handle a lot of legacy, so our return value is not always * 100% correct (called functions do not always provide one, should @@ -3968,9 +3859,8 @@ DBGPRINTF("AAAA jsonMerge adds '%s'\n", it.key); /* find a JSON structure element (field or container doesn't matter). */ rsRetVal -jsonFind(struct json_object *jroot, es_str_t *propName, struct json_object **jsonres) +jsonFind(struct json_object *jroot, msgPropDescr_t *pProp, struct json_object **jsonres) { - uchar *name = NULL; uchar *leaf; struct json_object *parent; struct json_object *field; @@ -3981,31 +3871,39 @@ jsonFind(struct json_object *jroot, es_str_t *propName, struct json_object **jso goto finalize_it; } - if(!es_strbufcmp(propName, (uchar*)"!", 1)) { + if(!strcmp((char*)pProp->name, "!")) { field = jroot; } else { - name = (uchar*)es_str2cstr(propName, NULL); - leaf = jsonPathGetLeaf(name, ustrlen(name)); - CHKiRet(jsonPathFindParent(jroot, name, leaf, &parent, 0)); + leaf = jsonPathGetLeaf(pProp->name, pProp->nameLen); + CHKiRet(jsonPathFindParent(jroot, pProp->name, leaf, &parent, 0)); field = json_object_object_get(parent, (char*)leaf); } *jsonres = field; finalize_it: - free(name); RETiRet; } -static rsRetVal -msgAddJSONObj(msg_t *pM, uchar *name, struct json_object *json, struct json_object **pjroot) +rsRetVal +msgAddJSON(msg_t *pM, uchar *name, struct json_object *json) { /* TODO: error checks! This is a quick&dirty PoC! */ + struct json_object **pjroot; struct json_object *parent, *leafnode; uchar *leaf; DEFiRet; MsgLock(pM); - if((name[0] == '!' || name[0] == '.' || name[0] == '/') && name[1] == '\0') { + if(name[0] == '!') { + pjroot = &pM->json; + } else if(name[0] == '.') { + pjroot = &pM->localvars; + } else { /* globl var */ + pthread_rwlock_wrlock(&glblVars_rwlock); + pjroot = &global_var_root; + } + + if(name[1] == '\0') { /* full tree? */ if(*pjroot == NULL) *pjroot = json; else @@ -4054,26 +3952,40 @@ msgAddJSONObj(msg_t *pM, uchar *name, struct json_object *json, struct json_obje } finalize_it: + if(name[0] == '/') + pthread_rwlock_unlock(&glblVars_rwlock); MsgUnlock(pM); RETiRet; } -rsRetVal -msgAddJSON(msg_t *pM, uchar *name, struct json_object *json) { - return msgAddJSONObj(pM, name, json, &pM->json); -} rsRetVal -msgDelJSONVar(msg_t *pM, struct json_object **jroot, uchar *name) +msgDelJSON(msg_t *pM, uchar *name) { + struct json_object **jroot; struct json_object *parent, *leafnode; uchar *leaf; DEFiRet; dbgprintf("AAAA: unset variable '%s'\n", name); MsgLock(pM); - if((name[0] == '!' || name[0] == '.' || name[0] == '/') && name[1] == '\0') { - /* strange, but I think we should permit this. After all, + + if(name[0] == '!') { + jroot = &pM->json; + } else if(name[0] == '.') { + jroot = &pM->localvars; + } else { /* globl var */ + pthread_rwlock_wrlock(&glblVars_rwlock); + jroot = &global_var_root; + } + if(jroot == NULL) { + DBGPRINTF("msgDelJSONVar; jroot empty in unset for property %s\n", + name); + FINALIZE; + } + + if(name[1] == '\0') { + /* full tree! Strange, but I think we should permit this. After all, * we trust rsyslog.conf to be written by the admin. */ DBGPRINTF("unsetting JSON root object\n"); @@ -4100,16 +4012,12 @@ DBGPRINTF("AAAA: unset found JSON value path '%s', " "leaf '%s', leafnode %p\n", } finalize_it: + if(name[0] == '/') + pthread_rwlock_unlock(&glblVars_rwlock); MsgUnlock(pM); RETiRet; } -rsRetVal -msgDelJSON(msg_t *pM, uchar *name) -{ - return msgDelJSONVar(pM, &pM->json, name); -} - static struct json_object * jsonDeepCopy(struct json_object *src) { @@ -4187,16 +4095,8 @@ msgSetJSONFromVar(msg_t *pMsg, uchar *varname, struct var *v) v->datatype); ABORT_FINALIZE(RS_RET_ERR); } - /* we always know strlen(varname) > 2 */ - if(varname[1] == '!') - msgAddJSONObj(pMsg, varname+1, json, &pMsg->json); - else if(varname[1] == '.') - msgAddJSONObj(pMsg, varname+1, json, &pMsg->localvars); - else { /* global - '/' */ - pthread_rwlock_wrlock(&glblVars_rwlock); - msgAddJSONObj(pMsg, varname+1, json, &global_var_root); - pthread_rwlock_unlock(&glblVars_rwlock); - } + + msgAddJSON(pMsg, varname, json); finalize_it: RETiRet; } @@ -4222,6 +4122,49 @@ finalize_it: } +/* Fill a message propert description. Space must already be alloced + * by the caller. This is for efficiency, as we expect this to happen + * as part of a larger structure alloc. + * Note that CEE/LOCAL_VAR properties can come in either as + * "$!xx"/"$.xx" or "!xx"/".xx" - we will unify them here. + */ +rsRetVal +msgPropDescrFill(msgPropDescr_t *pProp, uchar *name, int nameLen) +{ + propid_t id; + int offs; + DEFiRet; + if(propNameToID(name, &id) != RS_RET_OK) { + parser_errmsg("invalid property '%s'", name); + ABORT_FINALIZE(RS_RET_INVLD_PROP); + } + if(id == PROP_CEE || id == PROP_LOCAL_VAR || id == PROP_GLOBAL_VAR) { + /* in these cases, we need the field name for later processing */ + /* normalize name: remove $ if present */ + offs = (name[0] == '$') ? 1 : 0; + pProp->name = ustrdup(name + offs); + pProp->nameLen = nameLen - offs; + /* we patch the root name, so that support functions do not need to + * check for different root chars. */ + pProp->name[0] = '!'; + } + pProp->id = id; +finalize_it: + RETiRet; +} + +void +msgPropDescrDestruct(msgPropDescr_t *pProp) +{ + if(pProp != NULL) { + if(pProp->id == PROP_CEE || + pProp->id == PROP_LOCAL_VAR || + pProp->id == PROP_GLOBAL_VAR) + free(pProp->name); + } +} + + /* dummy */ rsRetVal msgQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } diff --git a/runtime/msg.h b/runtime/msg.h index ed15622a..a2392a20 100644 --- a/runtime/msg.h +++ b/runtime/msg.h @@ -3,7 +3,7 @@ * * File begun on 2007-07-13 by RGerhards (extracted from syslogd.c) * - * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH. + * Copyright 2007-2013 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -183,21 +183,14 @@ void MsgSetMSGoffs(msg_t *pMsg, short offs); void MsgSetRawMsgWOSize(msg_t *pMsg, char* pszRawMsg); void MsgSetRawMsg(msg_t *pMsg, char* pszRawMsg, size_t lenMsg); rsRetVal MsgReplaceMSG(msg_t *pThis, uchar* pszMSG, int lenMSG); -uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, - propid_t propid, es_str_t *propName, +uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, msgPropDescr_t *pProp, rs_size_t *pPropLen, unsigned short *pbMustBeFreed, struct syslogTime *ttNow); -rsRetVal msgGetMsgVar(msg_t *pThis, cstr_t *pstrPropName, var_t **ppVar); -es_str_t* msgGetMsgVarNew(msg_t *pThis, uchar *name); uchar *getRcvFrom(msg_t *pM); void getTAG(msg_t *pM, uchar **ppBuf, int *piLen); char *getTimeReported(msg_t *pM, enum tplFormatTypes eFmt); char *getPRI(msg_t *pMsg); void getRawMsg(msg_t *pM, uchar **pBuf, int *piLen); -rsRetVal msgGetCEEVar(msg_t *pThis, cstr_t *propName, var_t **ppVar); -es_str_t* msgGetCEEVarNew(msg_t *pMsg, char *name); -es_str_t* msgGetLocalVarNew(msg_t *pMsg, char *name); rsRetVal msgAddJSON(msg_t *pM, uchar *name, struct json_object *json); -rsRetVal getCEEPropVal(msg_t *pM, es_str_t *propName, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed); rsRetVal MsgGetSeverity(msg_t *pThis, int *piSeverity); rsRetVal MsgDeserialize(msg_t *pMsg, strm_t *pStrm); @@ -213,20 +206,16 @@ char *getHOSTNAME(msg_t *pM); int getHOSTNAMELen(msg_t *pM); uchar *getProgramName(msg_t *pM, sbool bLockMutex); uchar *getRcvFrom(msg_t *pM); -rsRetVal propNameToID(cstr_t *pCSPropName, propid_t *pPropID); +rsRetVal propNameToID(uchar *pName, propid_t *pPropID); uchar *propIDToName(propid_t propID); -rsRetVal msgGetCEEPropJSON(msg_t *pM, es_str_t *propName, struct json_object **pjson); -rsRetVal getGlobalVarPropVal( es_str_t *propName, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed); -rsRetVal msgGetLocalVarJSON(msg_t *pM, es_str_t *propName, struct json_object **pjson); -rsRetVal msgGetGlobalVarJSON(es_str_t *propName, struct json_object **pjson); +rsRetVal msgGetJSONPropJSON(msg_t *pMsg, msgPropDescr_t *pProp, struct json_object **pjson); +rsRetVal getJSONPropVal(msg_t *pMsg, msgPropDescr_t *pProp, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed); rsRetVal msgSetJSONFromVar(msg_t *pMsg, uchar *varname, struct var *var); rsRetVal msgDelJSON(msg_t *pMsg, uchar *varname); -rsRetVal jsonFind(struct json_object *jroot, es_str_t *propName, struct json_object **jsonres); +rsRetVal jsonFind(struct json_object *jroot, msgPropDescr_t *pProp, struct json_object **jsonres); -static inline rsRetVal -msgUnsetJSON(msg_t *pMsg, uchar *varname) { - return msgDelJSON(pMsg, varname+1); -} +rsRetVal msgPropDescrFill(msgPropDescr_t *pProp, uchar *name, int nameLen); +void msgPropDescrDestruct(msgPropDescr_t *pProp); static inline int msgGetProtocolVersion(msg_t *pM) diff --git a/runtime/queue.c b/runtime/queue.c index 66cb7218..968c016e 100644 --- a/runtime/queue.c +++ b/runtime/queue.c @@ -81,8 +81,8 @@ static rsRetVal GetDeqBatchSize(qqueue_t *pThis, int *pVal); static rsRetVal ConsumerDA(qqueue_t *pThis, wti_t *pWti); static rsRetVal batchProcessed(qqueue_t *pThis, wti_t *pWti); static rsRetVal qqueueMultiEnqObjNonDirect(qqueue_t *pThis, multi_submit_t *pMultiSub); -static rsRetVal qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub); -static rsRetVal qAddDirect(qqueue_t *pThis, msg_t *pMsg); +static rsRetVal qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub, wti_t *pWti); +static rsRetVal qAddDirect(qqueue_t *pThis, msg_t *pMsg, wti_t *pWti); static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis); static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis); static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis); @@ -959,7 +959,7 @@ static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis) return RS_RET_OK; } -static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg) +static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg, wti_t *pWti) { batch_t singleBatch; batch_obj_t batchObj; @@ -986,7 +986,7 @@ static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg) singleBatch.pElem = &batchObj; singleBatch.eltState = &batchState; singleBatch.active = &active; - iRet = pThis->pConsumer(pThis->pAction, &singleBatch, &pThis->bShutdownImmediate); + iRet = pThis->pConsumer(pThis->pAction, &singleBatch, pWti, &pThis->bShutdownImmediate); /* delete the batch string params: TODO: create its own "class" for this */ for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) { free(batchObj.staticActStrings[i]); @@ -999,7 +999,7 @@ static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg) /* "enqueue" a batch in direct mode. This is a shortcut which saves all the overhead * otherwise incured. -- rgerhards, ~2010-06-23 */ -rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch) +rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch, wti_t *pWti) { DEFiRet; @@ -1013,7 +1013,7 @@ rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch) * We use our knowledge about the batch_t structure below, but without that, we * pay a too-large performance toll... -- rgerhards, 2009-04-22 */ - iRet = pThis->pConsumer(pThis->pAction, pBatch, NULL); + iRet = pThis->pConsumer(pThis->pAction, pBatch, pWti, NULL); RETiRet; } @@ -1319,7 +1319,7 @@ finalize_it: * to modify some parameters before the queue is actually started. */ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads, - int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*,int*)) + int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*, wti_t*, int*)) { DEFiRet; qqueue_t *pThis; @@ -1878,7 +1878,7 @@ ConsumerReg(qqueue_t *pThis, wti_t *pWti) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave); - CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, &pThis->bShutdownImmediate)); + CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, pWti, &pThis->bShutdownImmediate)); /* we now need to check if we should deliberately delay processing a bit * and, if so, do that. -- rgerhards, 2008-01-30 @@ -2675,7 +2675,7 @@ finalize_it: /* now, the same function, but for direct mode */ static rsRetVal -qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub) +qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub, wti_t *pWti) { int i; DEFiRet; @@ -2684,7 +2684,7 @@ qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub) assert(pMultiSub != NULL); for(i = 0 ; i < pMultiSub->nElem ; ++i) { - CHKiRet(qAddDirect(pThis, (void*)pMultiSub->ppMsgs[i])); + CHKiRet(qAddDirect(pThis, (void*)pMultiSub->ppMsgs[i], pWti)); } finalize_it: @@ -2699,16 +2699,16 @@ finalize_it: * Enqueues the new element and awakes worker thread. */ rsRetVal -qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg) +qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg, wti_t *pWti) { DEFiRet; ISOBJ_TYPE_assert(pThis, qqueue); - iRet = qAddDirect(pThis, pMsg); + iRet = qAddDirect(pThis, pMsg, pWti); RETiRet; } -/* enqueue a new user data element +/* enqueue a new user data element * Enqueues the new element and awakes worker thread. */ rsRetVal diff --git a/runtime/queue.h b/runtime/queue.h index 844523ad..91900b30 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -103,7 +103,7 @@ struct queue_s { * the user really wanted...). -- rgerhards, 2008-04-02 */ /* end dequeue time window */ - rsRetVal (*pConsumer)(void *,batch_t*,int*); /* user-supplied consumer function for dequeued messages */ + rsRetVal (*pConsumer)(void *,batch_t*, wti_t*,int*); /* user-supplied consumer function for dequeued messages */ /* calling interface for pConsumer: arg1 is the global user pointer from this structure, arg2 is the * user pointer array that was dequeued (actual sample: for actions, arg1 is the pAction and arg2 * is pointer to an array of message message pointers), arg3 is a pointer to an interger which is zero @@ -195,14 +195,14 @@ struct queue_s { /* prototypes */ rsRetVal qqueueDestruct(qqueue_t **ppThis); -rsRetVal qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg); +rsRetVal qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg, wti_t *pWti); rsRetVal qqueueEnqMsg(qqueue_t *pThis, flowControl_t flwCtlType, msg_t *pMsg); rsRetVal qqueueStart(qqueue_t *pThis); rsRetVal qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize); rsRetVal qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix); rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads, - int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, int*)); -rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch); + int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, wti_t *, int*)); +rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch, wti_t *pWti); int queueCnfParamsSet(struct nvlst *lst); rsRetVal qqueueApplyCnfParam(qqueue_t *pThis, struct nvlst *lst); void qqueueSetDefaultsRulesetQueue(qqueue_t *pThis); diff --git a/runtime/rsconf.c b/runtime/rsconf.c index 6c7c5fd5..abce53b8 100644 --- a/runtime/rsconf.c +++ b/runtime/rsconf.c @@ -254,92 +254,6 @@ CODESTARTobjDebugPrint(rsconf) ENDobjDebugPrint(rsconf) -/* This function returns the current date in different - * variants. It is used to construct the $NOW series of - * system properties. The returned buffer must be freed - * by the caller when no longer needed. If the function - * can not allocate memory, it returns a NULL pointer. - * TODO: this was taken from msg.c and we should consolidate it with the code - * there. This is especially important when we increase the number of system - * variables (what we definitely want to do). - */ -typedef enum ENOWType { NOW_NOW, NOW_YEAR, NOW_MONTH, NOW_DAY, NOW_HOUR, NOW_MINUTE } eNOWType; -static rsRetVal -getNOW(eNOWType eNow, es_str_t **estr) -{ - DEFiRet; - uchar szBuf[16]; - struct syslogTime t; - es_size_t len; - - datetime.getCurrTime(&t, NULL); - switch(eNow) { - case NOW_NOW: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), - "%4.4d-%2.2d-%2.2d", t.year, t.month, t.day); - break; - case NOW_YEAR: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%4.4d", t.year); - break; - case NOW_MONTH: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%2.2d", t.month); - break; - case NOW_DAY: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%2.2d", t.day); - break; - case NOW_HOUR: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%2.2d", t.hour); - break; - case NOW_MINUTE: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%2.2d", t.minute); - break; - default: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "*invld eNow*"); - break; - } - - /* now create a string object out of it and hand that over to the var */ - *estr = es_newStrFromCStr((char*)szBuf, len); - - RETiRet; -} - - - -static inline es_str_t * -getSysVar(char *name) -{ - es_str_t *estr = NULL; - rsRetVal iRet = RS_RET_OK; - - if(!strcmp(name, "now")) { - CHKiRet(getNOW(NOW_NOW, &estr)); - } else if(!strcmp(name, "year")) { - CHKiRet(getNOW(NOW_YEAR, &estr)); - } else if(!strcmp(name, "month")) { - CHKiRet(getNOW(NOW_MONTH, &estr)); - } else if(!strcmp(name, "day")) { - CHKiRet(getNOW(NOW_DAY, &estr)); - } else if(!strcmp(name, "hour")) { - CHKiRet(getNOW(NOW_HOUR, &estr)); - } else if(!strcmp(name, "minute")) { - CHKiRet(getNOW(NOW_MINUTE, &estr)); - } else if(!strcmp(name, "myhostname")) { - char *hn = (char*)glbl.GetLocalHostName(); - estr = es_newStrFromCStr(hn, strlen(hn)); - } else { - ABORT_FINALIZE(RS_RET_SYSVAR_NOT_FOUND); - } -finalize_it: - if(iRet != RS_RET_OK) { - dbgprintf("getSysVar error iRet %d\n", iRet); - if(estr == NULL) - estr = es_newStrFromCStr("*ERROR*", sizeof("*ERROR*") - 1); - } - return estr; -} - - /* Process input() objects */ rsRetVal inputProcessCnf(struct cnfobj *o) @@ -479,30 +393,6 @@ void cnfDoBSDHost(char *ln) "solution (Block '%s')", ln); free(ln); } - -es_str_t* -cnfGetVar(char *name, void *usrptr) -{ - es_str_t *estr; - if(name[0] == '$') { - if(name[1] == '$') - estr = getSysVar(name+2); - else if(name[1] == '!') - estr = msgGetCEEVarNew((msg_t*) usrptr, name+2); - else - estr = msgGetMsgVarNew((msg_t*) usrptr, (uchar*)name); - } else { /* if this happens, we have a program logic error */ - estr = es_newStrFromCStr("err: var must start with $", - sizeof("err: var must start with $")-1); - } - if(Debug) { - char *s; - s = es_str2cstr(estr, NULL); - dbgprintf("rainerscript: var '%s': '%s'\n", name, s); - free(s); - } - return estr; -} /*------------------------------ end interface to flex/bison parser ------------------------------*/ @@ -1315,6 +1205,7 @@ ourConf = loadConf; // TODO: remove, once ourConf is gone! ABORT_FINALIZE(RS_RET_NO_ACTIONS); } tellLexEndParsing(); + DBGPRINTF("Number of actions in this configuration: %d\n", iActionNbr); rulesetOptimizeAll(loadConf); tellCoreConfigLoadDone(); diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h index 85381b8d..edf0c593 100644 --- a/runtime/rsyslog.h +++ b/runtime/rsyslog.h @@ -102,52 +102,6 @@ #define _PATH_CONSOLE "/dev/console" #endif -/* properties are now encoded as (tiny) integers. I do not use an enum as I would like - * to keep the memory footprint small (and thus cache hits high). - * rgerhards, 2009-06-26 - */ -typedef uintTiny propid_t; -#define PROP_INVALID 0 -#define PROP_MSG 1 -#define PROP_TIMESTAMP 2 -#define PROP_HOSTNAME 3 -#define PROP_SYSLOGTAG 4 -#define PROP_RAWMSG 5 -#define PROP_INPUTNAME 6 -#define PROP_FROMHOST 7 -#define PROP_FROMHOST_IP 8 -#define PROP_PRI 9 -#define PROP_PRI_TEXT 10 -#define PROP_IUT 11 -#define PROP_SYSLOGFACILITY 12 -#define PROP_SYSLOGFACILITY_TEXT 13 -#define PROP_SYSLOGSEVERITY 14 -#define PROP_SYSLOGSEVERITY_TEXT 15 -#define PROP_TIMEGENERATED 16 -#define PROP_PROGRAMNAME 17 -#define PROP_PROTOCOL_VERSION 18 -#define PROP_STRUCTURED_DATA 19 -#define PROP_APP_NAME 20 -#define PROP_PROCID 21 -#define PROP_MSGID 22 -#define PROP_PARSESUCCESS 23 -#define PROP_SYS_NOW 150 -#define PROP_SYS_YEAR 151 -#define PROP_SYS_MONTH 152 -#define PROP_SYS_DAY 153 -#define PROP_SYS_HOUR 154 -#define PROP_SYS_HHOUR 155 -#define PROP_SYS_QHOUR 156 -#define PROP_SYS_MINUTE 157 -#define PROP_SYS_MYHOSTNAME 158 -#define PROP_SYS_BOM 159 -#define PROP_SYS_UPTIME 160 -#define PROP_UUID 161 -#define PROP_CEE 200 -#define PROP_CEE_ALL_JSON 201 -#define PROP_LOCAL_VAR 202 -#define PROP_GLOBAL_VAR 203 - /* The error codes below are orginally "borrowed" from * liblogging. As such, we reserve values up to -2999 @@ -401,7 +355,7 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth RS_RET_RULESET_EXISTS = -2306,/**< ruleset already exists */ RS_RET_DEPRECATED = -2307,/**< deprecated functionality is used */ RS_RET_DS_PROP_SEQ_ERR = -2308,/**< property sequence error deserializing object */ - RS_RET_TPL_INVLD_PROP = -2309,/**< property name error in template (unknown name) */ + RS_RET_INVLD_PROP = -2309,/**< property name error (unknown name) */ RS_RET_NO_RULEBASE = -2310,/**< mmnormalize: rulebase can not be found or otherwise invalid */ RS_RET_INVLD_MODE = -2311,/**< invalid mode specified in configuration */ RS_RET_INVLD_ANON_BITS = -2312,/**< mmanon: invalid number of bits to anonymize specified */ diff --git a/runtime/ruleset.c b/runtime/ruleset.c index dae5bbaa..3e5223f9 100644 --- a/runtime/ruleset.c +++ b/runtime/ruleset.c @@ -67,8 +67,8 @@ static struct cnfparamblk rspblk = }; /* forward definitions */ -static rsRetVal processBatch(batch_t *pBatch); -static rsRetVal scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active); +static rsRetVal processBatch(batch_t *pBatch, wti_t *pWti); +static rsRetVal scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active, wti_t *pWti); /* ---------- linked-list key handling functions (ruleset) ---------- */ @@ -168,7 +168,7 @@ finalize_it: * rgerhards, 2010-06-15 */ static inline rsRetVal -processBatchMultiRuleset(batch_t *pBatch) +processBatchMultiRuleset(batch_t *pBatch, wti_t *pWti) { ruleset_t *currRuleset; batch_t snglRuleBatch; @@ -206,7 +206,7 @@ processBatchMultiRuleset(batch_t *pBatch) snglRuleBatch.nElem = iNew; /* was left just right by the for loop */ batchSetSingleRuleset(&snglRuleBatch, 1); /* process temp batch */ - processBatch(&snglRuleBatch); + processBatch(&snglRuleBatch, pWti); batchFree(&snglRuleBatch); } while(bHaveUnprocessed == 1); @@ -226,12 +226,12 @@ static inline void freeActive(sbool *active) { free(active); } /* for details, see scriptExec() header comment! */ /* call action for all messages with filter on */ static rsRetVal -execAct(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execAct(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti) { DEFiRet; dbgprintf("RRRR: execAct [%s]: batch of %d elements, active %p\n", modGetName(stmt->d.act->pMod), batchNumMsgs(pBatch), active); pBatch->active = active; - stmt->d.act->submitToActQ(stmt->d.act, pBatch); + stmt->d.act->submitToActQ(stmt->d.act, pBatch, pWti); RETiRet; } @@ -261,7 +261,7 @@ execUnset(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { if( pBatch->eltState[i] != BATCH_STATE_DISC && (active == NULL || active[i])) { - msgUnsetJSON(pBatch->pElem[i].pMsg, stmt->d.s_unset.varname); + msgDelJSON(pBatch->pElem[i].pMsg, stmt->d.s_unset.varname); } } RETiRet; @@ -292,7 +292,7 @@ execStop(batch_t *pBatch, sbool *active) // set new filter, inverted // perform else (if any messages) static rsRetVal -execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti) { sbool *newAct; int i; @@ -321,7 +321,7 @@ execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) } if(stmt->d.s_if.t_then != NULL) { - scriptExec(stmt->d.s_if.t_then, pBatch, newAct); + scriptExec(stmt->d.s_if.t_then, pBatch, newAct, pWti); } if(stmt->d.s_if.t_else != NULL) { for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { @@ -331,7 +331,7 @@ execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) && (active == NULL || active[i])) newAct[i] = !newAct[i]; } - scriptExec(stmt->d.s_if.t_else, pBatch, newAct); + scriptExec(stmt->d.s_if.t_else, pBatch, newAct, pWti); } freeActive(newAct); finalize_it: @@ -340,7 +340,7 @@ finalize_it: /* for details, see scriptExec() header comment! */ static void -execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti) { sbool *newAct; msg_t *pMsg; @@ -367,7 +367,7 @@ execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) } if(stmt->d.s_prifilt.t_then != NULL) { - scriptExec(stmt->d.s_prifilt.t_then, pBatch, newAct); + scriptExec(stmt->d.s_prifilt.t_then, pBatch, newAct, pWti); } if(stmt->d.s_prifilt.t_else != NULL) { for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { @@ -377,7 +377,7 @@ execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) && (active == NULL || active[i])) newAct[i] = !newAct[i]; } - scriptExec(stmt->d.s_prifilt.t_else, pBatch, newAct); + scriptExec(stmt->d.s_prifilt.t_else, pBatch, newAct, pWti); } freeActive(newAct); } @@ -392,12 +392,11 @@ evalPROPFILT(struct cnfstmt *stmt, msg_t *pMsg) int bRet = 0; rs_size_t propLen; - if(stmt->d.s_propfilt.propID == PROP_INVALID) + if(stmt->d.s_propfilt.prop.id == PROP_INVALID) goto done; - pszPropVal = MsgGetProp(pMsg, NULL, stmt->d.s_propfilt.propID, - stmt->d.s_propfilt.propName, &propLen, - &pbMustBeFreed, NULL); + pszPropVal = MsgGetProp(pMsg, NULL, &stmt->d.s_propfilt.prop, + &propLen, &pbMustBeFreed, NULL); /* Now do the compares (short list currently ;)) */ switch(stmt->d.s_propfilt.operation ) { @@ -441,25 +440,18 @@ evalPROPFILT(struct cnfstmt *stmt, msg_t *pMsg) bRet = (bRet == 1) ? 0 : 1; if(Debug) { - char *cstr; - if(stmt->d.s_propfilt.propID == PROP_CEE) { - cstr = es_str2cstr(stmt->d.s_propfilt.propName, NULL); + if(stmt->d.s_propfilt.prop.id == PROP_CEE) { DBGPRINTF("Filter: check for CEE property '%s' (value '%s') ", - cstr, pszPropVal); - free(cstr); - } else if(stmt->d.s_propfilt.propID == PROP_LOCAL_VAR) { - cstr = es_str2cstr(stmt->d.s_propfilt.propName, NULL); + stmt->d.s_propfilt.prop.name, pszPropVal); + } else if(stmt->d.s_propfilt.prop.id == PROP_LOCAL_VAR) { DBGPRINTF("Filter: check for local var '%s' (value '%s') ", - cstr, pszPropVal); - free(cstr); - } else if(stmt->d.s_propfilt.propID == PROP_GLOBAL_VAR) { - cstr = es_str2cstr(stmt->d.s_propfilt.propName, NULL); - DBGPRINTF("Filter: check for global var '%s' (value '%s') ", - cstr, pszPropVal); - free(cstr); + stmt->d.s_propfilt.prop.name, pszPropVal); + //} else if(stmt->d.s_propfilt.propID == PROP_GLOBAL_VAR) { + //DBGPRINTF("Filter: check for global var '%s' (value '%s') ", + //stmt->d.s_propfilt.propName, pszPropVal); } else { DBGPRINTF("Filter: check for property '%s' (value '%s') ", - propIDToName(stmt->d.s_propfilt.propID), pszPropVal); + propIDToName(stmt->d.s_propfilt.prop.id), pszPropVal); } if(stmt->d.s_propfilt.isNegated) DBGPRINTF("NOT "); @@ -484,7 +476,7 @@ done: /* for details, see scriptExec() header comment! */ static void -execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti) { sbool *thenAct; sbool bRet; @@ -503,7 +495,7 @@ execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) DBGPRINTF("batch: item %d PROPFILT %d\n", i, thenAct[i]); } - scriptExec(stmt->d.s_propfilt.t_then, pBatch, thenAct); + scriptExec(stmt->d.s_propfilt.t_then, pBatch, thenAct, pWti); freeActive(thenAct); } @@ -518,7 +510,7 @@ execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) * rgerhards, 2012-09-04 */ static rsRetVal -scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) +scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active, wti_t *pWti) { DEFiRet; struct cnfstmt *stmt; @@ -536,7 +528,7 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) execStop(pBatch, active); break; case S_ACT: - execAct(stmt, pBatch, active); + execAct(stmt, pBatch, active, pWti); break; case S_SET: execSet(stmt, pBatch, active); @@ -545,16 +537,16 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) execUnset(stmt, pBatch, active); break; case S_CALL: - scriptExec(stmt->d.s_call.stmt, pBatch, active); + scriptExec(stmt->d.s_call.stmt, pBatch, active, pWti); break; case S_IF: - execIf(stmt, pBatch, active); + execIf(stmt, pBatch, active, pWti); break; case S_PRIFILT: - execPRIFILT(stmt, pBatch, active); + execPRIFILT(stmt, pBatch, active, pWti); break; case S_PROPFILT: - execPROPFILT(stmt, pBatch, active); + execPROPFILT(stmt, pBatch, active, pWti); break; default: dbgprintf("error: unknown stmt type %u during exec\n", @@ -573,7 +565,7 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) * rgerhards, 2005-10-13 */ static rsRetVal -processBatch(batch_t *pBatch) +processBatch(batch_t *pBatch, wti_t *pWti) { ruleset_t *pThis; DEFiRet; @@ -585,9 +577,9 @@ processBatch(batch_t *pBatch) if(pThis == NULL) pThis = ourConf->rulesets.pDflt; ISOBJ_TYPE_assert(pThis, ruleset); - CHKiRet(scriptExec(pThis->root, pBatch, NULL)); + CHKiRet(scriptExec(pThis->root, pBatch, NULL, pWti)); } else { - CHKiRet(processBatchMultiRuleset(pBatch)); + CHKiRet(processBatchMultiRuleset(pBatch, pWti)); } finalize_it: diff --git a/runtime/ruleset.h b/runtime/ruleset.h index cbf8243b..05da144d 100644 --- a/runtime/ruleset.h +++ b/runtime/ruleset.h @@ -2,7 +2,7 @@ * * This implements rulesets within rsyslog. * - * Copyright 2009-2012 Rainer Gerhards and Adiscon GmbH. + * Copyright 2009-2013 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -46,7 +46,7 @@ BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */ rsRetVal (*Destruct)(ruleset_t **ppThis); rsRetVal (*DestructAllActions)(rsconf_t *conf); rsRetVal (*SetName)(ruleset_t *pThis, uchar *pszName); - rsRetVal (*ProcessBatch)(batch_t*); + rsRetVal (*ProcessBatch)(batch_t*, wti_t *); rsRetVal (*GetRuleset)(rsconf_t *conf, ruleset_t **ppThis, uchar*); rsRetVal (*SetDefaultRuleset)(rsconf_t *conf, uchar*); rsRetVal (*SetCurrRuleset)(rsconf_t *conf, uchar*); @@ -64,8 +64,9 @@ BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */ /* AddRule() removed */ /*TODO:REMOVE*/rsRetVal (*IterateAllActions)(rsconf_t *conf, rsRetVal (*pFunc)(void*, void*), void* pParam); void (*AddScript)(ruleset_t *pThis, struct cnfstmt *script); + /* v8: changed processBatch interface */ ENDinterface(ruleset) -#define rulesetCURR_IF_VERSION 7 /* increment whenever you change the interface structure! */ +#define rulesetCURR_IF_VERSION 8 /* increment whenever you change the interface structure! */ /* prototypes */ diff --git a/runtime/typedefs.h b/runtime/typedefs.h index a929b30c..2efd7d65 100644 --- a/runtime/typedefs.h +++ b/runtime/typedefs.h @@ -64,6 +64,7 @@ typedef struct nsdsel_ptcp_s nsdsel_ptcp_t; typedef struct nsdsel_gtls_s nsdsel_gtls_t; typedef struct nsdpoll_ptcp_s nsdpoll_ptcp_t; typedef struct wti_s wti_t; +typedef struct msgPropDescr_s msgPropDescr_t; typedef struct msg msg_t; typedef struct queue_s qqueue_t; typedef struct prop_s prop_t; @@ -163,6 +164,53 @@ typedef enum { typedef off_t off64_t; #endif + +/* properties are now encoded as (tiny) integers. I do not use an enum as I would like + * to keep the memory footprint small (and thus cache hits high). + * rgerhards, 2009-06-26 + */ +typedef uintTiny propid_t; +#define PROP_INVALID 0 +#define PROP_MSG 1 +#define PROP_TIMESTAMP 2 +#define PROP_HOSTNAME 3 +#define PROP_SYSLOGTAG 4 +#define PROP_RAWMSG 5 +#define PROP_INPUTNAME 6 +#define PROP_FROMHOST 7 +#define PROP_FROMHOST_IP 8 +#define PROP_PRI 9 +#define PROP_PRI_TEXT 10 +#define PROP_IUT 11 +#define PROP_SYSLOGFACILITY 12 +#define PROP_SYSLOGFACILITY_TEXT 13 +#define PROP_SYSLOGSEVERITY 14 +#define PROP_SYSLOGSEVERITY_TEXT 15 +#define PROP_TIMEGENERATED 16 +#define PROP_PROGRAMNAME 17 +#define PROP_PROTOCOL_VERSION 18 +#define PROP_STRUCTURED_DATA 19 +#define PROP_APP_NAME 20 +#define PROP_PROCID 21 +#define PROP_MSGID 22 +#define PROP_PARSESUCCESS 23 +#define PROP_SYS_NOW 150 +#define PROP_SYS_YEAR 151 +#define PROP_SYS_MONTH 152 +#define PROP_SYS_DAY 153 +#define PROP_SYS_HOUR 154 +#define PROP_SYS_HHOUR 155 +#define PROP_SYS_QHOUR 156 +#define PROP_SYS_MINUTE 157 +#define PROP_SYS_MYHOSTNAME 158 +#define PROP_SYS_BOM 159 +#define PROP_SYS_UPTIME 160 +#define PROP_UUID 161 +#define PROP_CEE 200 +#define PROP_CEE_ALL_JSON 201 +#define PROP_LOCAL_VAR 202 +#define PROP_GLOBAL_VAR 203 + /* types of configuration handlers */ typedef enum cslCmdHdlrType { @@ -213,6 +261,13 @@ struct multi_submit_s { msg_t **ppMsgs; }; +/* the following structure is a helper to describe a message property */ +struct msgPropDescr_s { + propid_t id; + uchar *name; /* name and lenName are only set for dynamic */ + int nameLen; /* properties (JSON) */ +}; + #endif /* multi-include protection */ /* vim:set ai: */ diff --git a/runtime/wti.c b/runtime/wti.c index f91fb5a9..e9d3599d 100644 --- a/runtime/wti.c +++ b/runtime/wti.c @@ -44,6 +44,7 @@ #include "wti.h" #include "obj.h" #include "glbl.h" +#include "action.h" #include "atomic.h" /* static data */ @@ -171,6 +172,7 @@ BEGINobjDestruct(wti) /* be sure to specify the object type also in END and CODE CODESTARTobjDestruct(wti) /* actual destruction */ batchFree(&pThis->batch); + free(pThis->actWrkrData); DESTROY_ATOMIC_HELPER_MUT(pThis->mutIsRunning); free(pThis->pszDbgHdr); @@ -195,11 +197,15 @@ wtiConstructFinalize(wti_t *pThis) ISOBJ_TYPE_assert(pThis, wti); - DBGPRINTF("%s: finalizing construction of worker instance data\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: finalizing construction of worker instance data (for %d actions)\n", + wtiGetDbgHdr(pThis), iActionNbr); /* initialize our thread instance descriptor (no concurrency here) */ pThis->bIsRunning = RSFALSE; + /* must use calloc as we need zeto-init */ + CHKmalloc(pThis->actWrkrData = calloc(iActionNbr, sizeof(void*))); + /* we now alloc the array for user pointers. We obtain the max from the queue itself. */ CHKiRet(pThis->pWtp->pfGetDeqBatchSize(pThis->pWtp->pUsr, &iDeqBatchSize)); CHKiRet(batchInit(&pThis->batch, iDeqBatchSize)); diff --git a/runtime/wti.h b/runtime/wti.h index 014251f0..297fb999 100644 --- a/runtime/wti.h +++ b/runtime/wti.h @@ -37,6 +37,7 @@ struct wti_s { wtp_t *pWtp; /* my worker thread pool (important if only the work thread instance is passed! */ batch_t batch; /* pointer to an object array meaningful for current user pointer (e.g. queue pUsr data elemt) */ uchar *pszDbgHdr; /* header string for debug messages */ + void **actWrkrData; /* *array* of action wrkr data pointers (sized for max nbr of actions in config!) */ DEF_ATOMIC_HELPER_MUT(mutIsRunning); }; diff --git a/runtime/wtp.c b/runtime/wtp.c index 19151e7c..1b960eb9 100644 --- a/runtime/wtp.c +++ b/runtime/wtp.c @@ -8,7 +8,7 @@ * (and in the web doc set on http://www.rsyslog.com/doc). Be sure to read it * if you are getting aquainted to the object. * - * Copyright 2008,2009 Rainer Gerhards and Adiscon GmbH. + * Copyright 2008-2013 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -91,6 +91,7 @@ BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro! pthread_cond_init(&pThis->condThrdTrm, NULL); pthread_attr_init(&pThis->attrThrd); /* Set thread scheduling policy to default */ +#warning do we need this any longer? I think it was a cure for an already fixed bug.. #ifdef HAVE_PTHREAD_SETSCHEDPARAM pthread_attr_setschedpolicy(&pThis->attrThrd, default_thr_sched_policy); pthread_attr_setschedparam(&pThis->attrThrd, &default_sched_param); @@ -121,7 +122,8 @@ wtpConstructFinalize(wtp_t *pThis) ISOBJ_TYPE_assert(pThis, wtp); - DBGPRINTF("%s: finalizing construction of worker thread pool\n", wtpGetDbgHdr(pThis)); + DBGPRINTF("%s: finalizing construction of worker thread pool (numworkerThreads %d)\n", + wtpGetDbgHdr(pThis), pThis->iNumWorkerThreads); /* alloc and construct workers - this can only be done in finalizer as we previously do * not know the max number of workers */ @@ -1,7 +1,7 @@ /* This is the template processing code of rsyslog. * begun 2004-11-17 rgerhards * - * Copyright 2004-2012 Rainer Gerhards and Adiscon + * Copyright 2004-2013 Rainer Gerhards and Adiscon * * This file is part of rsyslog. * @@ -163,13 +163,13 @@ tplToString(struct template *pTpl, msg_t *pMsg, uchar **ppBuf, size_t *pLenBuf, FINALIZE; } - if(pTpl->subtree != NULL) { + if(pTpl->bHaveSubtree) { /* only a single CEE subtree must be provided */ /* note: we could optimize the code below, however, this is * not worth the effort, as this passing mode is not expected * in subtree mode and so most probably only used for debug & test. */ - getCEEPropVal(pMsg, pTpl->subtree, &pVal, &iLenVal, &bMustBeFreed); + getJSONPropVal(pMsg, &pTpl->subtree, &pVal, &iLenVal, &bMustBeFreed); if(iLenVal >= (rs_size_t)*pLenBuf) /* we reserve one char for the final \0! */ CHKiRet(ExtendBuf(ppBuf, pLenBuf, iLenVal + 1)); memcpy(*ppBuf, pVal, iLenVal+1); @@ -193,9 +193,8 @@ tplToString(struct template *pTpl, msg_t *pMsg, uchar **ppBuf, size_t *pLenBuf, iLenVal = pTpe->data.constant.iLenConstant; bMustBeFreed = 0; } else if(pTpe->eEntryType == FIELD) { - pVal = (uchar*) MsgGetProp(pMsg, pTpe, pTpe->data.field.propid, - pTpe->data.field.propName, &iLenVal, - &bMustBeFreed, ttNow); + pVal = (uchar*) MsgGetProp(pMsg, pTpe, &pTpe->data.field.msgProp, + &iLenVal, &bMustBeFreed, ttNow); /* we now need to check if we should use SQL option. In this case, * we must go over the generated string and escape '\'' characters. * rgerhards, 2005-09-22: the option values below look somewhat misplaced, @@ -264,12 +263,12 @@ tplToArray(struct template *pTpl, msg_t *pMsg, uchar*** ppArr, struct syslogTime assert(pMsg != NULL); assert(ppArr != NULL); - if(pTpl->subtree) { + if(pTpl->bHaveSubtree) { /* Note: this mode is untested, as there is no official plugin * using array passing, so I simply could not test it. */ CHKmalloc(pArr = calloc(2, sizeof(uchar*))); - getCEEPropVal(pMsg, pTpl->subtree, &pVal, &propLen, &bMustBeFreed); + getJSONPropVal(pMsg, &pTpl->subtree, &pVal, &propLen, &bMustBeFreed); if(bMustBeFreed) { /* if it must be freed, it is our own private copy... */ pArr[0] = pVal; /* ... so we can use it! */ } else { @@ -290,9 +289,8 @@ tplToArray(struct template *pTpl, msg_t *pMsg, uchar*** ppArr, struct syslogTime if(pTpe->eEntryType == CONSTANT) { CHKmalloc(pArr[iArr] = (uchar*)strdup((char*) pTpe->data.constant.pConstant)); } else if(pTpe->eEntryType == FIELD) { - pVal = (uchar*) MsgGetProp(pMsg, pTpe, pTpe->data.field.propid, - pTpe->data.field.propName, &propLen, - &bMustBeFreed, ttNow); + pVal = (uchar*) MsgGetProp(pMsg, pTpe, &pTpe->data.field.msgProp, + &propLen, &bMustBeFreed, ttNow); if(bMustBeFreed) { /* if it must be freed, it is our own private copy... */ pArr[iArr] = pVal; /* ... so we can use it! */ } else { @@ -326,8 +324,8 @@ tplToJSON(struct template *pTpl, msg_t *pMsg, struct json_object **pjson, struct rsRetVal localRet; DEFiRet; - if(pTpl->subtree != NULL){ - localRet = jsonFind(pMsg->json, pTpl->subtree, pjson); + if(pTpl->bHaveSubtree){ + localRet = jsonFind(pMsg->json, &pTpl->subtree, pjson); if(*pjson == NULL) { /* we need to have a root object! */ *pjson = json_object_new_object(); @@ -345,8 +343,10 @@ tplToJSON(struct template *pTpl, msg_t *pMsg, struct json_object **pjson, struct jsonf = json_object_new_string((char*) pTpe->data.constant.pConstant); json_object_object_add(json, (char*)pTpe->fieldName, jsonf); } else if(pTpe->eEntryType == FIELD) { - if(pTpe->data.field.propid == PROP_CEE) { - localRet = msgGetCEEPropJSON(pMsg, pTpe->data.field.propName, &jsonf); + if(pTpe->data.field.msgProp.id == PROP_CEE || + pTpe->data.field.msgProp.id == PROP_LOCAL_VAR || + pTpe->data.field.msgProp.id == PROP_GLOBAL_VAR ) { + localRet = msgGetJSONPropJSON(pMsg, &pTpe->data.field.msgProp, &jsonf); if(localRet == RS_RET_OK) { json_object_object_add(json, (char*)pTpe->fieldName, json_object_get(jsonf)); } else { @@ -356,32 +356,9 @@ tplToJSON(struct template *pTpl, msg_t *pMsg, struct json_object **pjson, struct json_object_object_add(json, (char*)pTpe->fieldName, NULL); } } - } else if(pTpe->data.field.propid == PROP_LOCAL_VAR) { - localRet = msgGetLocalVarJSON(pMsg, pTpe->data.field.propName, &jsonf); - if(localRet == RS_RET_OK) { - json_object_object_add(json, (char*)pTpe->fieldName, json_object_get(jsonf)); - } else { - DBGPRINTF("tplToJSON: error %d looking up local variable %s\n", - localRet, pTpe->fieldName); - if(pTpe->data.field.options.bMandatory) { - json_object_object_add(json, (char*)pTpe->fieldName, NULL); - } - } - } else if(pTpe->data.field.propid == PROP_GLOBAL_VAR) { - localRet = msgGetGlobalVarJSON(pTpe->data.field.propName, &jsonf); - if(localRet == RS_RET_OK) { - json_object_object_add(json, (char*)pTpe->fieldName, json_object_get(jsonf)); - } else { - DBGPRINTF("tplToJSON: error %d looking up local variable %s\n", - localRet, pTpe->fieldName); - if(pTpe->data.field.options.bMandatory) { - json_object_object_add(json, (char*)pTpe->fieldName, NULL); - } - } } else { - pVal = (uchar*) MsgGetProp(pMsg, pTpe, pTpe->data.field.propid, - pTpe->data.field.propName, &propLen, - &bMustBeFreed, ttNow); + pVal = (uchar*) MsgGetProp(pMsg, pTpe, &pTpe->data.field.msgProp, + &propLen, &bMustBeFreed, ttNow); if(pTpe->data.field.options.bMandatory || propLen > 0) { jsonf = json_object_new_string_len((char*)pVal, propLen); json_object_object_add(json, (char*)pTpe->fieldName, jsonf); @@ -777,7 +754,7 @@ static rsRetVal do_Parameter(uchar **pp, struct template *pTpl) { uchar *p; - cstr_t *pStrProp; + cstr_t *pStrProp = NULL; cstr_t *pStrField = NULL; struct templateEntry *pTpe; int iNum; /* to compute numbers */ @@ -807,26 +784,8 @@ do_Parameter(uchar **pp, struct template *pTpl) /* got the name */ cstrFinalize(pStrProp); - if(propNameToID(pStrProp, &pTpe->data.field.propid) != RS_RET_OK) { - errmsg.LogError(0, RS_RET_TPL_INVLD_PROP, "template '%s': invalid parameter '%s'", - pTpl->pszName, cstrGetSzStrNoNULL(pStrProp)); - cstrDestruct(&pStrProp); - ABORT_FINALIZE(RS_RET_TPL_INVLD_PROP); - } - if(pTpe->data.field.propid == PROP_CEE) { - /* in CEE case, we need to preserve the actual property name */ - if((pTpe->data.field.propName = es_newStrFromCStr((char*)cstrGetSzStrNoNULL(pStrProp)+1, cstrLen(pStrProp)-1)) == NULL) { - cstrDestruct(&pStrProp); - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - } - } else if(pTpe->data.field.propid == PROP_LOCAL_VAR || pTpe->data.field.propid == PROP_GLOBAL_VAR) { - /* in these cases, we need to preserve the actual property name, but correct the root ID (bang vs. dot) */ - if((pTpe->data.field.propName = es_newStrFromCStr((char*)cstrGetSzStrNoNULL(pStrProp)+1, cstrLen(pStrProp)-1)) == NULL) { - cstrDestruct(&pStrProp); - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - } - es_getBufAddr(pTpe->data.field.propName)[0] = '!'; /* patch root name */ - } + CHKiRet(msgPropDescrFill(&pTpe->data.field.msgProp, cstrGetSzStrNoNULL(pStrProp), + cstrLen(pStrProp))); /* Check frompos, if it has an R, then topos should be a regex */ if(*p == ':') { @@ -1123,8 +1082,7 @@ do_Parameter(uchar **pp, struct template *pTpl) /* save field name - if none was given, use the property name instead */ if(pStrField == NULL) { - if(pTpe->data.field.propid == PROP_CEE || pTpe->data.field.propid == PROP_LOCAL_VAR || - pTpe->data.field.propid == PROP_GLOBAL_VAR) { + if(pTpe->data.field.msgProp.id == PROP_CEE || pTpe->data.field.msgProp.id == PROP_LOCAL_VAR) { /* in CEE case, we remove "$!"/"$." from the fieldname - it's just our indicator */ pTpe->fieldName = ustrdup(cstrGetSzStrNoNULL(pStrProp)+2); pTpe->lenFieldName = cstrLen(pStrProp)-2; @@ -1141,10 +1099,11 @@ do_Parameter(uchar **pp, struct template *pTpl) DBGPRINTF("template/do_Parameter: fieldName is NULL!\n"); ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); } - cstrDestruct(&pStrProp); if(*p) ++p; /* eat '%' */ *pp = p; finalize_it: + if(pStrProp != NULL) + cstrDestruct(&pStrProp); RETiRet; } @@ -1605,16 +1564,8 @@ createPropertyTpe(struct template *pTpl, struct cnfobj *o) /* apply */ CHKmalloc(pTpe = tpeConstruct(pTpl)); pTpe->eEntryType = FIELD; - CHKiRet(propNameToID(name, &pTpe->data.field.propid)); - if(pTpe->data.field.propid == PROP_CEE) { - /* in CEE case, we need to preserve the actual property name */ - pTpe->data.field.propName = es_newStrFromCStr((char*)cstrGetSzStrNoNULL(name)+1, - cstrLen(name)-1); - } else if(pTpe->data.field.propid == PROP_LOCAL_VAR || pTpe->data.field.propid == PROP_GLOBAL_VAR) { - /* in these case, we need to preserve the actual property name, but correct the root ID (bang vs. dot) */ - pTpe->data.field.propName = es_newStrFromCStr((char*)cstrGetSzStrNoNULL(name)+1, cstrLen(name)-1); - es_getBufAddr(pTpe->data.field.propName)[0] = '!'; /* patch root name */ - } + CHKiRet(msgPropDescrFill(&pTpe->data.field.msgProp, cstrGetSzStrNoNULL(name), + cstrLen(name))); pTpe->data.field.options.bDropLastLF = droplastlf; pTpe->data.field.options.bSPIffNo1stSP = spifno1stsp; pTpe->data.field.options.bMandatory = mandatory; @@ -1747,8 +1698,9 @@ tplProcessCnf(struct cnfobj *o) char *name = NULL; uchar *tplStr = NULL; uchar *plugin = NULL; - es_str_t *subtree = NULL; uchar *p; + msgPropDescr_t subtree; + sbool bHaveSubtree = 0; enum { T_STRING, T_PLUGIN, T_LIST, T_SUBTREE } tplType = T_STRING; /* init just to keep compiler happy: mandatory parameter */ int i; @@ -1795,10 +1747,11 @@ tplProcessCnf(struct cnfobj *o) free(name); /* overall assigned */ ABORT_FINALIZE(RS_RET_ERR); } else { - /* TODO: unify strings! */ - char *cstr = es_str2cstr(pvals[i].val.d.estr, NULL); - subtree = es_newStrFromBuf(cstr+1, es_strlen(pvals[i].val.d.estr)-1); + uchar *cstr; + cstr = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + CHKiRet(msgPropDescrFill(&subtree, cstr, ustrlen(cstr))); free(cstr); + bHaveSubtree = 1; } } else if(!strcmp(pblk.descr[i].name, "plugin")) { plugin = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL); @@ -1841,7 +1794,7 @@ tplProcessCnf(struct cnfobj *o) } } - if(subtree == NULL) { + if(bHaveSubtree) { if(tplType == T_SUBTREE) { errmsg.LogError(0, RS_RET_ERR, "template '%s' of type subtree needs " "subtree parameter", name); @@ -1911,7 +1864,8 @@ tplProcessCnf(struct cnfobj *o) break; case T_LIST: createListTpl(pTpl, o); break; - case T_SUBTREE: pTpl->subtree = subtree; + case T_SUBTREE: memcpy(&pTpl->subtree, &subtree, sizeof(msgPropDescr_t)); + pTpl->bHaveSubtree = 1; break; } @@ -2003,9 +1957,8 @@ void tplDeleteAll(rsconf_t *conf) regexp.regfree(&(pTpeDel->data.field.re)); } } - if(pTpeDel->data.field.propName != NULL) - es_deleteStr(pTpeDel->data.field.propName); #endif + msgPropDescrDestruct(&pTpeDel->data.field.msgProp); break; } free(pTpeDel->fieldName); @@ -2015,8 +1968,8 @@ void tplDeleteAll(rsconf_t *conf) pTplDel = pTpl; pTpl = pTpl->pNext; free(pTplDel->pszName); - if(pTplDel->subtree != NULL) - es_deleteStr(pTplDel->subtree); + if(pTplDel->bHaveSubtree) + msgPropDescrDestruct(&pTplDel->subtree); free(pTplDel); } ENDfunc @@ -2063,9 +2016,8 @@ void tplDeleteNew(rsconf_t *conf) regexp.regfree(&(pTpeDel->data.field.re)); } } - if(pTpeDel->data.field.propName != NULL) - es_deleteStr(pTpeDel->data.field.propName); #endif + msgPropDescrDestruct(&pTpeDel->data.field.msgProp); break; } /*dbgprintf("\n");*/ @@ -2074,8 +2026,8 @@ void tplDeleteNew(rsconf_t *conf) pTplDel = pTpl; pTpl = pTpl->pNext; free(pTplDel->pszName); - if(pTplDel->subtree != NULL) - es_deleteStr(pTplDel->subtree); + if(pTplDel->bHaveSubtree) + msgPropDescrDestruct(&pTplDel->subtree); free(pTplDel); } ENDfunc @@ -2117,19 +2069,13 @@ void tplPrintList(rsconf_t *conf) pTpe->data.constant.pConstant); break; case FIELD: - dbgprintf("(FIELD), value: '%d' ", pTpe->data.field.propid); - if(pTpe->data.field.propid == PROP_CEE) { - char *cstr = es_str2cstr(pTpe->data.field.propName, NULL); - dbgprintf("[EE-Property: '%s'] ", cstr); - free(cstr); - } else if(pTpe->data.field.propid == PROP_LOCAL_VAR) { - char *cstr = es_str2cstr(pTpe->data.field.propName, NULL); - dbgprintf("[Local Var: '%s'] ", cstr); - free(cstr); - } else if(pTpe->data.field.propid == PROP_GLOBAL_VAR) { - char *cstr = es_str2cstr(pTpe->data.field.propName, NULL); - dbgprintf("[Global Var: '%s'] ", cstr); - free(cstr); + dbgprintf("(FIELD), value: '%d' ", pTpe->data.field.msgProp.id); + if(pTpe->data.field.msgProp.id == PROP_CEE) { + dbgprintf("[EE-Property: '%s'] ", pTpe->data.field.msgProp.name); + } else if(pTpe->data.field.msgProp.id == PROP_LOCAL_VAR) { + dbgprintf("[Local Var: '%s'] ", pTpe->data.field.msgProp.name); + //} else if(pTpe->data.field.propid == PROP_GLOBAL_VAR) { + // dbgprintf("[Global Var: '%s'] ", pTpe->data.field.propName); } switch(pTpe->data.field.eDateFormat) { case tplFmtDefault: @@ -40,7 +40,8 @@ struct template { char *pszName; int iLenName; rsRetVal (*pStrgen)(msg_t*, uchar**, size_t *); - es_str_t *subtree; /* subtree name for subtree-type templates */ + sbool bHaveSubtree; + msgPropDescr_t subtree; /* subtree property name for subtree-type templates */ int tpenElements; /* number of elements in templateEntry list */ struct templateEntry *pEntryRoot; struct templateEntry *pEntryLast; @@ -79,7 +80,7 @@ struct templateEntry { int iLenConstant; /* its length */ } constant; struct { - propid_t propid; /* property to be used */ + msgPropDescr_t msgProp; /* property to be used */ unsigned iFromPos; /* for partial strings only chars from this position ... */ unsigned iToPos; /* up to that one... */ unsigned iFieldNr; /* for field extraction: field to extract */ @@ -103,7 +104,6 @@ struct templateEntry { int field_expand; /* use multiple instances of the field delimiter as a single one? */ #endif - es_str_t *propName; /**< property name (currently being used for CEE only) */ enum tplFormatTypes eDateFormat; enum tplFormatCaseConvTypes eCaseConv; diff --git a/tools/omdiscard.c b/tools/omdiscard.c index 15c6ea82..9bc3093a 100644 --- a/tools/omdiscard.c +++ b/tools/omdiscard.c @@ -49,6 +49,10 @@ typedef struct _instanceData { EMPTY_STRUCT } instanceData; +typedef struct wrkrInstanceData { + instanceData *pData; +} wrkrInstanceData_t; + /* we do not need a createInstance()! BEGINcreateInstance CODESTARTcreateInstance diff --git a/tools/omfile.c b/tools/omfile.c index fdcf355a..77f10cb8 100644 --- a/tools/omfile.c +++ b/tools/omfile.c @@ -133,6 +133,7 @@ typedef struct s_dynaFileCacheEntry dynaFileCacheEntry; typedef struct _instanceData { + pthread_mutex_t mutWrite; /* guard against multiple instances writing to single file */ uchar *f_fname; /* file or template name (display only) */ uchar *tplName; /* name of assigned template */ strm_t *pStrm; /* our output stream */ @@ -181,6 +182,20 @@ typedef struct _instanceData { STATSCOUNTER_DEF(ctrMax, mutCtrMax); } instanceData; +/* to build a linked list for temporary storage of lines while we cannot commit */ +typedef struct linebuf { + uchar *filename; /* for dynafiles, make go away */ + uchar *ln; + unsigned iMsgOpts; + struct linebuf *pNext; +} linebuf_t; + +typedef struct wrkrInstanceData { + instanceData *pData; + linebuf_t *pRoot; + linebuf_t *pLast; +} wrkrInstanceData_t; + typedef struct configSettings_s { int iDynaFileCacheSize; /* max cache for dynamic files */ @@ -786,17 +801,19 @@ finalize_it: /* rgerhards 2004-11-11: write to a file output. */ static rsRetVal -writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pData) +writeFile(instanceData *pData, linebuf_t *linebuf) { DEFiRet; ASSERT(pData != NULL); + pthread_mutex_lock(&pData->mutWrite); + /* first check if we have a dynamic file name and, if so, * check if it still is ok or a new file needs to be created */ if(pData->bDynamicName) { - CHKiRet(prepareDynFile(pData, ppString[1], iMsgOpts)); + CHKiRet(prepareDynFile(pData, linebuf->filename, linebuf->iMsgOpts)); } else { /* "regular", non-dynafile */ if(pData->pStrm == NULL) { CHKiRet(prepareFile(pData, pData->f_fname)); @@ -806,9 +823,10 @@ writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pData) } } - CHKiRet(doWrite(pData, ppString[0], strlen(CHAR_CONVERT(ppString[0])))); + CHKiRet(doWrite(pData, linebuf->ln, ustrlen(linebuf->ln))); finalize_it: + pthread_mutex_unlock(&pData->mutWrite); RETiRet; } @@ -888,9 +906,16 @@ ENDfreeCnf BEGINcreateInstance CODESTARTcreateInstance pData->pStrm = NULL; + pthread_mutex_init(&pData->mutWrite, NULL); ENDcreateInstance +BEGINcreateWrkrInstance +CODESTARTcreateWrkrInstance +ENDcreateWrkrInstance + + + BEGINfreeInstance CODESTARTfreeInstance free(pData->tplName); @@ -913,6 +938,7 @@ CODESTARTfreeInstance free(pData->cryprovName); free(pData->cryprovNameFull); } + pthread_mutex_destroy(&pData->mutWrite); ENDfreeInstance @@ -926,8 +952,61 @@ CODESTARTbeginTransaction ENDbeginTransaction +static rsRetVal +bufferLine(wrkrInstanceData_t *pWrkrData, uchar *filename, uchar *line) +{ + linebuf_t *lb; + DEFiRet; + + dbgprintf("DDDD: buffering root %p, line %s\n", pWrkrData->pRoot, line); + CHKmalloc(lb = (linebuf_t*) malloc(sizeof(linebuf_t))); + CHKmalloc(lb->filename = ustrdup(filename)); + CHKmalloc(lb->ln = ustrdup(line)); + lb->pNext = NULL; + if(pWrkrData->pRoot == NULL) { + pWrkrData->pRoot = pWrkrData->pLast = lb; + } else { + pWrkrData->pLast->pNext = lb; + pWrkrData->pLast = lb; + } +finalize_it: + RETiRet; +} + +static void +submitCachedLines(wrkrInstanceData_t *pWrkrData, instanceData *pData) +{ + linebuf_t *curr, *todel; + + for(curr = pWrkrData->pRoot ; curr != NULL ; ) { + DBGPRINTF("omfile: file to log to: %s\n", curr->filename); + DBGPRINTF("omfile: start of data: '%.128s'\n", curr->ln); + STATSCOUNTER_INC(pData->ctrRequests, pData->mutCtrRequests); + writeFile(pData, curr); + + todel = curr; + curr = curr->pNext; + free(todel->filename); + free(todel->ln); + free(todel); + } + pWrkrData->pRoot = NULL; +} + + +BEGINdoAction +CODESTARTdoAction + pData = pWrkrData->pData; + iRet = bufferLine(pWrkrData, (pData->bDynamicName) ? ppString[1] : pData->f_fname, + ppString[0]); + if(iRet == RS_RET_OK) + iRet = RS_RET_DEFER_COMMIT; +ENDdoAction + BEGINendTransaction CODESTARTendTransaction + pData = pWrkrData->pData; + submitCachedLines(pWrkrData, pData); /* Note: pStrm may be NULL if there was an error opening the stream */ if(pData->bFlushOnTXEnd && pData->pStrm != NULL) { /* if we have an async writer, it controls the flush via @@ -941,21 +1020,6 @@ finalize_it: ENDendTransaction -BEGINdoAction -CODESTARTdoAction - DBGPRINTF("file to log to: %s\n", - (pData->bDynamicName) ? ppString[1] : pData->f_fname); - DBGPRINTF("omfile: start of data: '%.128s'\n", ppString[0]); - STATSCOUNTER_INC(pData->ctrRequests, pData->mutCtrRequests); - CHKiRet(writeFile(ppString, iMsgOpts, pData)); - if(!bCoreSupportsBatching && pData->bFlushOnTXEnd) { - CHKiRet(strm.Flush(pData->pStrm)); - } -finalize_it: - if(iRet == RS_RET_OK) - iRet = RS_RET_DEFER_COMMIT; -ENDdoAction - static inline void setInstParamDefaults(instanceData *pData) @@ -1336,6 +1400,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a BEGINdoHUP CODESTARTdoHUP + pthread_mutex_lock(&pData->mutWrite); if(pData->bDynamicName) { dynaFileFreeCacheEntries(pData); } else { @@ -1343,6 +1408,7 @@ CODESTARTdoHUP closeFile(pData); } } + pthread_mutex_unlock(&pData->mutWrite); ENDdoHUP @@ -1358,6 +1424,7 @@ ENDmodExit BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_OMOD_QUERIES +CODEqueryEtryPt_STD_OMOD8_QUERIES CODEqueryEtryPt_STD_CONF2_QUERIES CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES diff --git a/tools/omfwd.c b/tools/omfwd.c index 6e5cf809..b86c995c 100644 --- a/tools/omfwd.c +++ b/tools/omfwd.c @@ -112,6 +112,10 @@ typedef struct _instanceData { int errsToReport; /* (remaining) number of errors to report */ } instanceData; +typedef struct wrkrInstanceData { + instanceData *pData; +} wrkrInstanceData_t; + /* config data */ typedef struct configSettings_s { uchar *pszTplName; /* name of the default template to use */ diff --git a/tools/ompipe.c b/tools/ompipe.c index df8066b1..6d44cd00 100644 --- a/tools/ompipe.c +++ b/tools/ompipe.c @@ -72,6 +72,10 @@ typedef struct _instanceData { sbool bHadError; /* did we already have/report an error on this pipe? */ } instanceData; +typedef struct wrkrInstanceData { + instanceData *pData; +} wrkrInstanceData_t; + typedef struct configSettings_s { EMPTY_STRUCT } configSettings_t; diff --git a/tools/omshell.c b/tools/omshell.c index ac62fa62..eb2d0a57 100644 --- a/tools/omshell.c +++ b/tools/omshell.c @@ -63,6 +63,10 @@ typedef struct _instanceData { uchar progName[MAXFNAME]; /* program to execute */ } instanceData; +typedef struct wrkrInstanceData { + instanceData *pData; +} wrkrInstanceData_t; + BEGINcreateInstance CODESTARTcreateInstance ENDcreateInstance diff --git a/tools/omusrmsg.c b/tools/omusrmsg.c index f4cc4094..3a268fa5 100644 --- a/tools/omusrmsg.c +++ b/tools/omusrmsg.c @@ -87,6 +87,10 @@ typedef struct _instanceData { uchar *tplName; } instanceData; +typedef struct wrkrInstanceData { + instanceData *pData; +} wrkrInstanceData_t; + typedef struct configSettings_s { EMPTY_STRUCT } configSettings_t; diff --git a/tools/syslogd.c b/tools/syslogd.c index aaeb9866..7597b05d 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -560,13 +560,13 @@ finalize_it: * for the main queue. */ static rsRetVal -msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch, int *pbShutdownImmediate) +msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch, wti_t *pWti, int *pbShutdownImmediate) { DEFiRet; assert(pBatch != NULL); pBatch->pbShutdownImmediate = pbShutdownImmediate; /* TODO: move this to batch creation! */ preprocessBatch(pBatch); - ruleset.ProcessBatch(pBatch); + ruleset.ProcessBatch(pBatch, pWti); //TODO: the BATCH_STATE_COMM must be set somewhere down the road, but we //do not have this yet and so we emulate -- 2010-06-10 int i; |