summaryrefslogtreecommitdiffstats
path: root/action.c
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--action.c2398
1 files changed, 2398 insertions, 0 deletions
diff --git a/action.c b/action.c
new file mode 100644
index 0000000..1f77529
--- /dev/null
+++ b/action.c
@@ -0,0 +1,2398 @@
+/* action.c
+ *
+ * Implementation of the action object.
+ *
+ * File begun on 2007-08-06 by RGerhards (extracted from syslogd.c)
+ *
+ * Some notes on processing (this hopefully makes it easier to find
+ * the right code in question): For performance reasons, this module
+ * uses different methods of message submission based on the user-selected
+ * configuration. This code is similar, but can not be abstracted because
+ * of the performance-affecting differences in it. As such, it is often
+ * necessary to triple-check that everything works well in *all* modes.
+ * The different modes (and calling sequence) are:
+ *
+ * if set iExecEveryNthOccur > 1 || iSecsExecOnceInterval
+ * - doSubmitToActionQComplex
+ * handles mark message reduction, but in essence calls
+ * - actionWriteToAction
+ * - doSubmitToActionQ
+ * (now queue engine processing)
+ * if(pThis->bWriteAllMarkMsgs == RSFALSE)
+ * - doSubmitToActionQNotAllMark
+ * - doSubmitToActionQ (and from here like in the else case below!)
+ * else
+ * - doSubmitToActionQ
+ * - qqueueEnqObj
+ * (now queue engine processing)
+ *
+ * Note that bWriteAllMakrMsgs on or off creates almost the same processing.
+ * The difference ist that if WriteAllMarkMsgs is not set, we need to
+ * preprocess the batch and drop mark messages which are not yet due for
+ * writing.
+ *
+ * After dequeue, processing is as follows:
+ * - processBatchMain
+ * - processMsgMain (direct entry for DIRECT queue!)
+ * - ...
+ *
+ * MORE ON PROCESSING, QUEUES and FILTERING
+ * All filtering needs to be done BEFORE messages are enqueued to an
+ * action. In previous code, part of the filtering was done at the
+ * "remote end" of the action queue, which lead to problems in
+ * non-direct mode (because then things run asynchronously). In order
+ * to solve this problem once and for all, I have changed the code so
+ * that all filtering is done before enq, and processing on the
+ * dequeue side of action processing now always executes whatever is
+ * enqueued. This is the only way to handle things consistently and
+ * (as much as possible) in a queue-type agnostic way. However, it is
+ * a rather radical change, which I unfortunately needed to make from
+ * stable version 5.8.1 to 5.8.2. If new problems pop up, you now know
+ * what may be their cause. In any case, the way it is done now is the
+ * only correct one.
+ * A problem is that, under fortunate conditions, we use the current
+ * batch for the output system as well. This is very good from a performance
+ * point of view, but makes the distinction between enq and deq side of
+ * the queue a bit hard. The current idea is that the filter condition
+ * alone is checked at the deq side of the queue (seems to be unavoidable
+ * to do it that way), but all other complex conditons (like failover
+ * handling) go into the computation of the filter condition. For
+ * non-direct queues, we still enqueue only what is acutally necessary.
+ * Note that in this case the rest of the code must ensure that the filter
+ * is set to "true". While this is not perfect and not as simple as
+ * we would like to see it, it looks like the best way to tackle that
+ * beast.
+ * rgerhards, 2011-06-15
+ *
+ * Copyright 2007-2022 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of rsyslog.
+ *
+ * Rsyslog is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Rsyslog is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Rsyslog. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ */
+#include "config.h"
+#include <stdio.h>
+#include <assert.h>
+#include <stdarg.h>
+#include <stdlib.h>
+#include <string.h>
+#include <strings.h>
+#include <time.h>
+#include <errno.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <json.h>
+
+#include "rsyslog.h"
+#include "dirty.h"
+#include "template.h"
+#include "action.h"
+#include "modules.h"
+#include "cfsysline.h"
+#include "srUtils.h"
+#include "errmsg.h"
+#include "batch.h"
+#include "wti.h"
+#include "rsconf.h"
+#include "datetime.h"
+#include "unicode-helper.h"
+#include "atomic.h"
+#include "ruleset.h"
+#include "parserif.h"
+#include "statsobj.h"
+
+/* AIXPORT : cs renamed to legacy_cs as clashes with libpthreads variable in complete file*/
+#ifdef _AIX
+#define cs legacy_cs
+#endif
+
+PRAGMA_INGORE_Wswitch_enum
+
+#ifndef O_LARGEFILE
+#define O_LARGEFILE 0
+#endif
+
+#define NO_TIME_PROVIDED 0 /* indicate we do not provide any cached time */
+
+/* forward definitions */
+static rsRetVal ATTR_NONNULL() processBatchMain(void *pVoid, batch_t *pBatch, wti_t * const pWti);
+static rsRetVal doSubmitToActionQ(action_t * const pAction, wti_t * const pWti, smsg_t*);
+static rsRetVal doSubmitToActionQComplex(action_t * const pAction, wti_t * const pWti, smsg_t*);
+static rsRetVal doSubmitToActionQNotAllMark(action_t * const pAction, wti_t * const pWti, smsg_t*);
+static void ATTR_NONNULL() actionSuspend(action_t * const pThis, wti_t * const pWti);
+static void ATTR_NONNULL() actionRetry(action_t * const pThis, wti_t * const pWti);
+
+/* object static data (once for all instances) */
+DEFobjCurrIf(obj)
+DEFobjCurrIf(datetime)
+DEFobjCurrIf(module)
+DEFobjCurrIf(statsobj)
+DEFobjCurrIf(ruleset)
+
+
+typedef struct configSettings_s {
+ int bActExecWhenPrevSusp; /* execute action only when previous one was suspended? */
+ int bActionWriteAllMarkMsgs; /* should all mark messages be unconditionally written? */
+ int iActExecOnceInterval; /* execute action once every nn seconds */
+ int iActExecEveryNthOccur; /* execute action every n-th occurrence (0,1=always) */
+ time_t iActExecEveryNthOccurTO; /* timeout for n-occurrence setting (in seconds, 0=never) */
+ int glbliActionResumeInterval;
+ int glbliActionResumeRetryCount; /* how often should suspended actions be retried? */
+ int bActionRepMsgHasMsg; /* last messsage repeated... has msg fragment in it */
+ uchar *pszActionName; /* short name for the action */
+ /* action queue and its configuration parameters */
+ queueType_t ActionQueType; /* type of the main message queue above */
+ int iActionQueueSize; /* size of the main message queue above */
+ int iActionQueueDeqBatchSize; /* batch size for action queues */
+ int iActionQHighWtrMark; /* high water mark for disk-assisted queues */
+ int iActionQLowWtrMark; /* low water mark for disk-assisted queues */
+ int iActionQDiscardMark; /* begin to discard messages */
+ int iActionQDiscardSeverity;
+ /* by default, discard nothing to prevent unintentional loss */
+ int iActionQueueNumWorkers; /* number of worker threads for the mm queue above */
+ uchar *pszActionQFName; /* prefix for the main message queue file */
+ int64 iActionQueMaxFileSize;
+ int iActionQPersistUpdCnt; /* persist queue info every n updates */
+ int bActionQSyncQeueFiles; /* sync queue files */
+ int iActionQtoQShutdown; /* queue shutdown */
+ int iActionQtoActShutdown; /* action shutdown (in phase 2) */
+ int iActionQtoEnq; /* timeout for queue enque */
+ int iActionQtoWrkShutdown; /* timeout for worker thread shutdown */
+ int iActionQWrkMinMsgs; /* minimum messages per worker needed to start a new one */
+ int bActionQSaveOnShutdown; /* save queue on shutdown (when DA enabled)? */
+ int64 iActionQueMaxDiskSpace; /* max disk space allocated 0 ==> unlimited */
+ int iActionQueueDeqSlowdown; /* dequeue slowdown (simple rate limiting) */
+ int iActionQueueDeqtWinFromHr; /* hour begin of time frame when queue is to be dequeued */
+ int iActionQueueDeqtWinToHr; /* hour begin of time frame when queue is to be dequeued */
+} configSettings_t;
+
+
+static configSettings_t cs; /* our current config settings */
+
+/* tables for interfacing with the v6 config system */
+static struct cnfparamdescr cnfparamdescr[] = {
+ { "name", eCmdHdlrGetWord, 0 }, /* legacy: actionname */
+ { "type", eCmdHdlrString, CNFPARAM_REQUIRED }, /* legacy: actionname */
+ { "action.errorfile", eCmdHdlrString, 0 },
+ { "action.errorfile.maxsize", eCmdHdlrInt, 0 },
+ { "action.writeallmarkmessages", eCmdHdlrBinary, 0 }, /* legacy: actionwriteallmarkmessages */
+ { "action.execonlyeverynthtime", eCmdHdlrInt, 0 }, /* legacy: actionexeconlyeverynthtime */
+ { "action.execonlyeverynthtimetimeout", eCmdHdlrInt, 0 }, /* legacy: actionexeconlyeverynthtimetimeout */
+ { "action.execonlyonceeveryinterval", eCmdHdlrInt, 0 }, /* legacy: actionexeconlyonceeveryinterval */
+ { "action.execonlywhenpreviousissuspended", eCmdHdlrBinary, 0 },
+ /* legacy: actionexeconlywhenpreviousissuspended */
+ { "action.repeatedmsgcontainsoriginalmsg", eCmdHdlrBinary, 0 }, /* legacy: repeatedmsgcontainsoriginalmsg */
+ { "action.resumeretrycount", eCmdHdlrInt, 0 }, /* legacy: actionresumeretrycount */
+ { "action.reportsuspension", eCmdHdlrBinary, 0 },
+ { "action.reportsuspensioncontinuation", eCmdHdlrBinary, 0 },
+ { "action.resumeintervalmax", eCmdHdlrPositiveInt, 0 },
+ { "action.resumeinterval", eCmdHdlrInt, 0 },
+ { "action.externalstate.file", eCmdHdlrString, 0 },
+ { "action.copymsg", eCmdHdlrBinary, 0 }
+};
+static struct cnfparamblk pblk =
+ { CNFPARAMBLK_VERSION,
+ sizeof(cnfparamdescr)/sizeof(struct cnfparamdescr),
+ cnfparamdescr
+ };
+
+
+/* primarily a helper for debug purposes, get human-readble name of state */
+/* currently not needed, but may be useful in the future! */
+#if 0
+static const char *
+batchState2String(const batch_state_t state)
+{
+ switch(state) {
+ case BATCH_STATE_RDY:
+ return "BATCH_STATE_RDY";
+ case BATCH_STATE_BAD:
+ return "BATCH_STATE_BAD";
+ case BATCH_STATE_SUB:
+ return "BATCH_STATE_SUB";
+ case BATCH_STATE_COMM:
+ return "BATCH_STATE_COMM";
+ case BATCH_STATE_DISC:
+ return "BATCH_STATE_DISC";
+ default:
+ return "ERROR, batch state not known!";
+ }
+}
+#endif // #if 0
+
+/* ------------------------------ methods ------------------------------ */
+
+/* This function returns the "current" time for this action. Current time
+ * is not necessarily real-time. In order to enhance performance, current
+ * system time is obtained the first time an action needs to know the time
+ * and then kept cached inside the action structure. Later requests will
+ * always return that very same time. Wile not totally accurate, it is far
+ * accurate in most cases and considered "acurate enough" for all cases.
+ * When changing the threading model, please keep in mind that this
+ * logic needs to be changed should we once allow more than one parallel
+ * call into the same action (object). As this is currently not supported,
+ * we simply cache the time inside the action object itself, after it
+ * is under mutex protection.
+ * Side-note: the value -1 is used as tActNow, because it also is the
+ * error return value of time(). So we would do a retry with the next
+ * invocation if time() failed. Then, of course, we would probably already
+ * be in trouble, but for the sake of performance we accept this very,
+ * very slight risk.
+ * This logic has been added as part of an overall performance improvment
+ * effort inspired by David Lang. -- rgerhards, 2008-09-16
+ * Note: this function does not use the usual iRet call conventions
+ * because that would provide little to no benefit but complicate things
+ * a lot. So we simply return the system time.
+ */
+static time_t
+getActNow(action_t * const pThis)
+{
+ assert(pThis != NULL);
+ if(pThis->tActNow == -1) {
+ pThis->tActNow = datetime.GetTime(NULL); /* good time call - the only one done */
+ if(pThis->tLastExec > pThis->tActNow) {
+ /* if we are traveling back in time, reset tLastExec */
+ pThis->tLastExec = (time_t) 0;
+ }
+ }
+
+ return pThis->tActNow;
+}
+
+
+/* resets action queue parameters to their default values. This happens
+ * after each action has been created in order to prevent any wild defaults
+ * to be used. It is somewhat against the original spirit of the config file
+ * reader, but I think it is a good thing to do.
+ * rgerhards, 2008-01-29
+ */
+static rsRetVal
+actionResetQueueParams(void)
+{
+ DEFiRet;
+
+ cs.ActionQueType = QUEUETYPE_DIRECT; /* type of the main message queue above */
+ cs.iActionQueueSize = 1000; /* size of the main message queue above */
+ cs.iActionQueueDeqBatchSize = 16; /* default batch size */
+ cs.iActionQHighWtrMark = -1; /* high water mark for disk-assisted queues */
+ cs.iActionQLowWtrMark = -1; /* low water mark for disk-assisted queues */
+ cs.iActionQDiscardMark = 980; /* begin to discard messages */
+ cs.iActionQDiscardSeverity = 8; /* discard warning and above */
+ cs.iActionQueueNumWorkers = 1; /* number of worker threads for the mm queue above */
+ cs.iActionQueMaxFileSize = 1024*1024;
+ cs.iActionQPersistUpdCnt = 0; /* persist queue info every n updates */
+ cs.bActionQSyncQeueFiles = 0;
+ cs.iActionQtoQShutdown = 0; /* queue shutdown */
+ cs.iActionQtoActShutdown = 1000; /* action shutdown (in phase 2) */
+ cs.iActionQtoEnq = 50; /* timeout for queue enque */
+ cs.iActionQtoWrkShutdown = 60000; /* timeout for worker thread shutdown */
+ cs.iActionQWrkMinMsgs = -1; /* minimum messages per worker needed to start a new one */
+ cs.bActionQSaveOnShutdown = 1; /* save queue on shutdown (when DA enabled)? */
+ cs.iActionQueMaxDiskSpace = 0;
+ cs.iActionQueueDeqSlowdown = 0;
+ cs.iActionQueueDeqtWinFromHr = 0;
+ cs.iActionQueueDeqtWinToHr = 25; /* 25 disables time windowed dequeuing */
+
+ cs.glbliActionResumeRetryCount = 0; /* I guess it is smart to reset this one, too */
+
+ free(cs.pszActionQFName);
+ cs.pszActionQFName = NULL; /* prefix for the main message queue file */
+
+ RETiRet;
+}
+
+/* free action worker data table
+*/
+static void freeWrkrDataTable(action_t * const pThis)
+{
+ int freeSpot;
+ for(freeSpot = 0; freeSpot < pThis->wrkrDataTableSize; ++freeSpot) {
+ if(pThis->wrkrDataTable[freeSpot] != NULL) {
+ pThis->pMod->mod.om.freeWrkrInstance(pThis->wrkrDataTable[freeSpot]);
+ pThis->wrkrDataTable[freeSpot] = NULL;
+ }
+ }
+ free(pThis->wrkrDataTable);
+ return;
+}
+
+/* destructs an action descriptor object
+ * rgerhards, 2007-08-01
+ */
+rsRetVal actionDestruct(action_t * const pThis)
+{
+ DEFiRet;
+ assert(pThis != NULL);
+
+ if(!strcmp((char*)modGetName(pThis->pMod), "builtin:omdiscard")) {
+ /* discard actions will be optimized out */
+ FINALIZE;
+ }
+
+ if(pThis->pQueue != NULL) {
+ qqueueDestruct(&pThis->pQueue);
+ }
+
+ /* destroy stats object, if we have one (may not always be
+ * be the case, e.g. if turned off)
+ */
+ if(pThis->statsobj != NULL)
+ statsobj.Destruct(&pThis->statsobj);
+
+ if(pThis->pModData != NULL)
+ pThis->pMod->freeInstance(pThis->pModData);
+
+ if(pThis->fdErrFile != -1)
+ close(pThis->fdErrFile);
+ pthread_mutex_destroy(&pThis->mutErrFile);
+ pthread_mutex_destroy(&pThis->mutAction);
+ pthread_mutex_destroy(&pThis->mutWrkrDataTable);
+ free((void*)pThis->pszErrFile);
+ free((void*)pThis->pszExternalStateFile);
+ free(pThis->pszName);
+ free(pThis->ppTpl);
+ free(pThis->peParamPassing);
+ freeWrkrDataTable(pThis);
+
+finalize_it:
+ free(pThis);
+ RETiRet;
+}
+
+
+/* Disable action, this means it will never again be usable
+ * until rsyslog is reloaded. Use only as a last resort, but
+ * depends on output module.
+ * rgerhards, 2007-08-02
+ */
+static inline void
+actionDisable(action_t *__restrict__ const pThis)
+{
+ pThis->bDisabled = 1;
+}
+
+
+
+/* create a new action descriptor object
+ * rgerhards, 2007-08-01
+ * Note that it is vital to set proper initial values as the v6 config
+ * system depends on these!
+ */
+rsRetVal actionConstruct(action_t **ppThis)
+{
+ DEFiRet;
+ action_t *pThis;
+
+ assert(ppThis != NULL);
+
+ CHKmalloc(pThis = (action_t*) calloc(1, sizeof(action_t)));
+ pThis->iResumeInterval = 30;
+ pThis->iResumeIntervalMax = 1800; /* max interval default is half an hour */
+ pThis->iResumeRetryCount = 0;
+ pThis->pszName = NULL;
+ pThis->pszErrFile = NULL;
+ pThis->maxErrFileSize = 0;
+ pThis->currentErrFileSize = 0;
+ pThis->pszExternalStateFile = NULL;
+ pThis->fdErrFile = -1;
+ pThis->bWriteAllMarkMsgs = 1;
+ pThis->iExecEveryNthOccur = 0;
+ pThis->iExecEveryNthOccurTO = 0;
+ pThis->iSecsExecOnceInterval = 0;
+ pThis->bExecWhenPrevSusp = 0;
+ pThis->bRepMsgHasMsg = 0;
+ pThis->bDisabled = 0;
+ pThis->isTransactional = 0;
+ pThis->bReportSuspension = -1; /* indicate "not yet set" */
+ pThis->bReportSuspensionCont = -1; /* indicate "not yet set" */
+ pThis->bCopyMsg = 0;
+ pThis->tLastOccur = datetime.GetTime(NULL); /* done once per action on startup only */
+ pThis->iActionNbr = loadConf->actions.iActionNbr;
+ pthread_mutex_init(&pThis->mutErrFile, NULL);
+ pthread_mutex_init(&pThis->mutAction, NULL);
+ pthread_mutex_init(&pThis->mutWrkrDataTable, NULL);
+ INIT_ATOMIC_HELPER_MUT(pThis->mutCAS);
+
+ /* indicate we have a new action */
+ loadConf->actions.iActionNbr++;
+
+finalize_it:
+ *ppThis = pThis;
+ RETiRet;
+}
+
+
+/* action construction finalizer
+ */
+rsRetVal
+actionConstructFinalize(action_t *__restrict__ const pThis, struct nvlst *lst)
+{
+ DEFiRet;
+ uchar pszAName[64]; /* friendly name of our action */
+
+ if(!strcmp((char*)modGetName(pThis->pMod), "builtin:omdiscard")) {
+ /* discard actions will be optimized out */
+ FINALIZE;
+ }
+ /* generate a friendly name for us action stats */
+ if(pThis->pszName == NULL) {
+ snprintf((char*) pszAName, sizeof(pszAName), "action-%d-%s",
+ pThis->iActionNbr, pThis->pMod->pszName);
+ pThis->pszName = ustrdup(pszAName);
+ }
+
+ /* cache transactional attribute */
+ pThis->isTransactional = pThis->pMod->mod.om.supportsTX;
+ if(pThis->isTransactional) {
+ int i;
+ for(i = 0 ; i < pThis->iNumTpls ; ++i) {
+ if(pThis->peParamPassing[i] != ACT_STRING_PASSING) {
+ LogError(0, RS_RET_INVLD_OMOD, "action '%s'(%d) is transactional but "
+ "parameter %d "
+ "uses invalid parameter passing mode -- disabling "
+ "action. This is probably caused by a pre-v7 "
+ "output module that needs upgrade.",
+ pThis->pszName, pThis->iActionNbr, i);
+ actionDisable(pThis);
+ ABORT_FINALIZE(RS_RET_INVLD_OMOD);
+
+ }
+ }
+ }
+
+
+ /* support statistics gathering */
+ CHKiRet(statsobj.Construct(&pThis->statsobj));
+ CHKiRet(statsobj.SetName(pThis->statsobj, pThis->pszName));
+ CHKiRet(statsobj.SetOrigin(pThis->statsobj, (uchar*)"core.action"));
+
+ STATSCOUNTER_INIT(pThis->ctrProcessed, pThis->mutCtrProcessed);
+ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("processed"),
+ ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrProcessed));
+
+ STATSCOUNTER_INIT(pThis->ctrFail, pThis->mutCtrFail);
+ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("failed"),
+ ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrFail));
+
+ STATSCOUNTER_INIT(pThis->ctrSuspend, pThis->mutCtrSuspend);
+ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("suspended"),
+ ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrSuspend));
+ STATSCOUNTER_INIT(pThis->ctrSuspendDuration, pThis->mutCtrSuspendDuration);
+ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("suspended.duration"),
+ ctrType_IntCtr, 0, &pThis->ctrSuspendDuration));
+
+ STATSCOUNTER_INIT(pThis->ctrResume, pThis->mutCtrResume);
+ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("resumed"),
+ ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrResume));
+
+ CHKiRet(statsobj.ConstructFinalize(pThis->statsobj));
+
+ /* create our queue */
+
+ /* generate a friendly name for the queue */
+ snprintf((char*) pszAName, sizeof(pszAName), "%s queue",
+ pThis->pszName);
+
+ /* now check if we can run the action in "firehose mode" during stage one of
+ * its processing (that is before messages are enqueued into the action q).
+ * This is only possible if some features, which require strict sequence, are
+ * not used. Thankfully, that is usually the case. The benefit of firehose
+ * mode is much faster processing (and simpler code) -- rgerhards, 2010-06-08
+ */
+ if( pThis->iExecEveryNthOccur > 1
+ || pThis->iSecsExecOnceInterval
+ ) {
+ DBGPRINTF("info: firehose mode disabled for action because "
+ "iExecEveryNthOccur=%d, iSecsExecOnceInterval=%d\n",
+ pThis->iExecEveryNthOccur, pThis->iSecsExecOnceInterval);
+ pThis->submitToActQ = doSubmitToActionQComplex;
+ } else if(pThis->bWriteAllMarkMsgs) {
+ /* full firehose submission mode, default case*/
+ pThis->submitToActQ = doSubmitToActionQ;
+ } else {
+ /* nearly full-speed submission mode */
+ pThis->submitToActQ = doSubmitToActionQNotAllMark;
+ }
+
+ /* create queue */
+ /* action queues always (for now) have just one worker. This may change when
+ * we begin to implement an interface the enable output modules to request
+ * to be run on multiple threads. So far, this is forbidden by the interface
+ * spec. -- rgerhards, 2008-01-30
+ */
+ CHKiRet(qqueueConstruct(&pThis->pQueue, cs.ActionQueType, 1, cs.iActionQueueSize,
+ processBatchMain));
+ obj.SetName((obj_t*) pThis->pQueue, pszAName);
+ qqueueSetpAction(pThis->pQueue, pThis);
+
+ if(lst == NULL) { /* use legacy params? */
+ /* ... set some properties ... */
+# define setQPROP(func, directive, data) \
+ CHKiRet_Hdlr(func(pThis->pQueue, data)) { \
+ LogError(0, NO_ERRCODE, "Invalid " #directive ", \
+ error %d. Ignored, running with default setting", iRet); \
+ }
+# define setQPROPstr(func, directive, data) \
+ CHKiRet_Hdlr(func(pThis->pQueue, data, (data == NULL)? 0 : strlen((char*) data))) { \
+ LogError(0, NO_ERRCODE, "Invalid " #directive ", \
+ error %d. Ignored, running with default setting", iRet); \
+ }
+ setQPROP(qqueueSetsizeOnDiskMax, "$ActionQueueMaxDiskSpace", cs.iActionQueMaxDiskSpace);
+ setQPROP(qqueueSetiDeqBatchSize, "$ActionQueueDequeueBatchSize", cs.iActionQueueDeqBatchSize);
+ setQPROP(qqueueSetMaxFileSize, "$ActionQueueFileSize", cs.iActionQueMaxFileSize);
+ setQPROPstr(qqueueSetFilePrefix, "$ActionQueueFileName", cs.pszActionQFName);
+ setQPROP(qqueueSetiPersistUpdCnt, "$ActionQueueCheckpointInterval", cs.iActionQPersistUpdCnt);
+ setQPROP(qqueueSetbSyncQueueFiles, "$ActionQueueSyncQueueFiles", cs.bActionQSyncQeueFiles);
+ setQPROP(qqueueSettoQShutdown, "$ActionQueueTimeoutShutdown", cs.iActionQtoQShutdown );
+ setQPROP(qqueueSettoActShutdown, "$ActionQueueTimeoutActionCompletion", cs.iActionQtoActShutdown);
+ setQPROP(qqueueSettoWrkShutdown, "$ActionQueueWorkerTimeoutThreadShutdown", cs.iActionQtoWrkShutdown);
+ setQPROP(qqueueSettoEnq, "$ActionQueueTimeoutEnqueue", cs.iActionQtoEnq);
+ setQPROP(qqueueSetiHighWtrMrk, "$ActionQueueHighWaterMark", cs.iActionQHighWtrMark);
+ setQPROP(qqueueSetiLowWtrMrk, "$ActionQueueLowWaterMark", cs.iActionQLowWtrMark);
+ setQPROP(qqueueSetiDiscardMrk, "$ActionQueueDiscardMark", cs.iActionQDiscardMark);
+ setQPROP(qqueueSetiDiscardSeverity, "$ActionQueueDiscardSeverity", cs.iActionQDiscardSeverity);
+ setQPROP(qqueueSetiMinMsgsPerWrkr, "$ActionQueueWorkerThreadMinimumMessages", cs.iActionQWrkMinMsgs);
+ setQPROP(qqueueSetiNumWorkerThreads, "$ActionQueueWorkerThreads", cs.iActionQueueNumWorkers);
+ setQPROP(qqueueSetbSaveOnShutdown, "$ActionQueueSaveOnShutdown", cs.bActionQSaveOnShutdown);
+ setQPROP(qqueueSetiDeqSlowdown, "$ActionQueueDequeueSlowdown", cs.iActionQueueDeqSlowdown);
+ setQPROP(qqueueSetiDeqtWinFromHr, "$ActionQueueDequeueTimeBegin", cs.iActionQueueDeqtWinFromHr);
+ setQPROP(qqueueSetiDeqtWinToHr, "$ActionQueueDequeueTimeEnd", cs.iActionQueueDeqtWinToHr);
+ } else {
+ /* we have v6-style config params */
+ qqueueSetDefaultsActionQueue(pThis->pQueue);
+ qqueueApplyCnfParam(pThis->pQueue, lst);
+ }
+ qqueueCorrectParams(pThis->pQueue);
+
+# undef setQPROP
+# undef setQPROPstr
+
+ qqueueDbgPrint(pThis->pQueue);
+
+ DBGPRINTF("Action %p: queue %p created\n", pThis, pThis->pQueue);
+
+ if(pThis->bUsesMsgPassingMode && pThis->pQueue->qType != QUEUETYPE_DIRECT) {
+ parser_warnmsg("module %s with message passing mode uses "
+ "non-direct queue. This most probably leads to undesired "
+ "results. For message modificaton modules (mm*), this means "
+ "that they will have no effect - "
+ "see https://www.rsyslog.com/mm-no-queue/", (char*)modGetName(pThis->pMod));
+ }
+
+ /* and now reset the queue params (see comment in its function header!) */
+ actionResetQueueParams();
+
+finalize_it:
+ RETiRet;
+}
+
+
+
+/* set the global resume interval
+ */
+rsRetVal actionSetGlobalResumeInterval(int iNewVal)
+{
+ cs.glbliActionResumeInterval = iNewVal;
+ return RS_RET_OK;
+}
+
+
+/* returns the action state name in human-readable form
+ * returned string must not be modified.
+ * rgerhards, 2009-05-07
+ */
+static uchar *getActStateName(action_t * const pThis, wti_t * const pWti)
+{
+ switch(getActionState(pWti, pThis)) {
+ case ACT_STATE_RDY:
+ return (uchar*) "rdy";
+ case ACT_STATE_ITX:
+ return (uchar*) "itx";
+ case ACT_STATE_RTRY:
+ return (uchar*) "rtry";
+ case ACT_STATE_SUSP:
+ return (uchar*) "susp";
+ case ACT_STATE_DATAFAIL:
+ return (uchar*) "datafail";
+ default:
+ return (uchar*) "ERROR/UNKNWON";
+ }
+}
+
+
+/* returns a suitable return code based on action state
+ * rgerhards, 2009-05-07
+ */
+static rsRetVal getReturnCode(action_t * const pThis, wti_t * const pWti)
+{
+ DEFiRet;
+
+ switch(getActionState(pWti, pThis)) {
+ case ACT_STATE_RDY:
+ iRet = RS_RET_OK;
+ break;
+ case ACT_STATE_ITX:
+ if(pWti->actWrkrInfo[pThis->iActionNbr].bHadAutoCommit) {
+ pWti->actWrkrInfo[pThis->iActionNbr].bHadAutoCommit = 0; /* auto-reset */
+ iRet = RS_RET_PREVIOUS_COMMITTED;
+ } else {
+ iRet = RS_RET_DEFER_COMMIT;
+ }
+ break;
+ case ACT_STATE_RTRY:
+ iRet = RS_RET_SUSPENDED;
+ break;
+ case ACT_STATE_SUSP:
+ iRet = RS_RET_ACTION_FAILED;
+ break;
+ case ACT_STATE_DATAFAIL:
+ iRet = RS_RET_DATAFAIL;
+ break;
+ default:
+ DBGPRINTF("Invalid action engine state %u, program error\n",
+ getActionState(pWti, pThis));
+ iRet = RS_RET_ERR;
+ break;
+ }
+
+ RETiRet;
+}
+
+
+/* set the action to a new state
+ * rgerhards, 2007-08-02
+ */
+static void
+actionSetState(action_t * const pThis, wti_t * const pWti, uint8_t newState)
+{
+ setActionState(pWti, pThis, newState);
+ DBGPRINTF("action[%s] transitioned to state: %s\n",
+ pThis->pszName, getActStateName(pThis, pWti));
+}
+
+/* Handles the transient commit state. So far, this is
+ * mostly a dummy...
+ * rgerhards, 2007-08-02
+ */
+static void actionCommitted(action_t * const pThis, wti_t * const pWti)
+{
+ actionSetState(pThis, pWti, ACT_STATE_RDY);
+}
+
+
+/* set action state according to external state file (if configured)
+*/
+static rsRetVal ATTR_NONNULL()
+checkExternalStateFile(action_t *const pThis, wti_t *const pWti)
+{
+ char filebuf[1024];
+ int fd = -1;
+ int r;
+ DEFiRet;
+
+ DBGPRINTF("checking external state file\n");
+
+ if(pThis->pszExternalStateFile == NULL) {
+ FINALIZE;
+ }
+
+ fd = open(pThis->pszExternalStateFile, O_RDONLY|O_CLOEXEC);
+ if(fd == -1) {
+ dbgprintf("could not read external state file\n");
+ FINALIZE;
+ }
+
+ r = read(fd, filebuf, sizeof(filebuf) - 1);
+ if(r < 1) {
+ dbgprintf("checkExternalStateFile read() returned %d\n", r);
+ FINALIZE;
+ }
+
+ filebuf[r] = '\0';
+ dbgprintf("external state file content: '%s'\n", filebuf);
+ /* trim trailing whitespace */
+ for(int j = r-1 ; j > 0 ; --j) {
+ if(filebuf[j] == '\n' || filebuf[j] == '\t' || filebuf[j] == ' ') {
+ filebuf[j] = '\0';
+ } else {
+ break;
+ }
+ }
+ if(!strcmp(filebuf, "SUSPENDED")) {
+ LogMsg(0, RS_RET_SUSPENDED, LOG_WARNING,
+ "action '%s' suspended (module '%s') by external state file",
+ pThis->pszName, pThis->pMod->pszName);
+ actionRetry(pThis, pWti);
+ ABORT_FINALIZE(RS_RET_SUSPENDED);
+ }
+
+finalize_it:
+ if(fd != -1) {
+ close(fd);
+ }
+ DBGPRINTF("done checking external state file, iRet=%d\n", iRet);
+ RETiRet;
+}
+
+
+/* we need to defer setting the action's own bReportSuspension state until
+ * after the full config has been processed. So the most simple case to do
+ * that is here. It's not a performance problem, as it happens infrequently.
+ * it's not a threading race problem, as always the same value will be written.
+ * As we need to do this in several places, we have moved the code to its own
+ * helper function.
+ */
+static void
+setSuspendMessageConfVars(action_t *__restrict__ const pThis)
+{
+ if(pThis->bReportSuspension == -1)
+ pThis->bReportSuspension = runConf->globals.bActionReportSuspension;
+ if(pThis->bReportSuspensionCont == -1) {
+ pThis->bReportSuspensionCont = runConf->globals.bActionReportSuspensionCont;
+ if(pThis->bReportSuspensionCont == -1)
+ pThis->bReportSuspensionCont = 1;
+ }
+}
+
+
+/* set action to "rtry" state.
+ * rgerhards, 2007-08-02
+ */
+static void ATTR_NONNULL() actionRetry(action_t * const pThis, wti_t * const pWti)
+{
+ setSuspendMessageConfVars(pThis);
+ actionSetState(pThis, pWti, ACT_STATE_RTRY);
+ if(pThis->bReportSuspension) {
+ LogMsg(0, RS_RET_SUSPENDED, LOG_WARNING,
+ "action '%s' suspended (module '%s'), retry %d. There should "
+ "be messages before this one giving the reason for suspension.",
+ pThis->pszName, pThis->pMod->pszName,
+ getActionNbrResRtry(pWti, pThis));
+ }
+ incActionResumeInRow(pWti, pThis);
+}
+
+/* Suspend action, this involves changing the action state as well
+ * as setting the next retry time.
+ * if we have more than 10 retries, we prolong the
+ * retry interval. If something is really stalled, it will
+ * get re-tried only very, very seldom - but that saves
+ * CPU time. TODO: maybe a config option for that?
+ * rgerhards, 2007-08-02
+ */
+static void ATTR_NONNULL()
+actionSuspend(action_t * const pThis, wti_t * const pWti)
+{
+ time_t ttNow;
+ int suspendDuration;
+ char timebuf[32];
+
+ setSuspendMessageConfVars(pThis);
+
+ /* note: we can NOT use a cached timestamp, as time may have evolved
+ * since caching, and this would break logic (and it actually did so!)
+ */
+ datetime.GetTime(&ttNow);
+ suspendDuration = pThis->iResumeInterval * (getActionNbrResRtry(pWti, pThis) / 10 + 1);
+ if(pThis->iResumeIntervalMax > 0 && suspendDuration > pThis->iResumeIntervalMax) {
+ suspendDuration = pThis->iResumeIntervalMax;
+ }
+ pThis->ttResumeRtry = ttNow + suspendDuration;
+ actionSetState(pThis, pWti, ACT_STATE_SUSP);
+ pThis->ctrSuspendDuration += suspendDuration;
+ if(getActionNbrResRtry(pWti, pThis) == 0) {
+ STATSCOUNTER_INC(pThis->ctrSuspend, pThis->mutCtrSuspend);
+ }
+
+ if( pThis->bReportSuspensionCont
+ || (pThis->bReportSuspension && getActionNbrResRtry(pWti, pThis) == 0) ) {
+ ctime_r(&pThis->ttResumeRtry, timebuf);
+ timebuf[strlen(timebuf)-1] = '\0'; /* strip LF */
+ LogMsg(0, RS_RET_SUSPENDED, LOG_WARNING,
+ "action '%s' suspended (module '%s'), next retry is %s, retry nbr %d. "
+ "There should be messages before this one giving the reason for suspension.",
+ pThis->pszName, pThis->pMod->pszName, timebuf,
+ getActionNbrResRtry(pWti, pThis));
+ }
+ DBGPRINTF("action '%s' suspended, earliest retry=%lld (now %lld), iNbrResRtry %d, "
+ "duration %d\n",
+ pThis->pszName, (long long) pThis->ttResumeRtry, (long long) ttNow,
+ getActionNbrResRtry(pWti, pThis), suspendDuration);
+}
+
+
+/* actually do retry processing. Note that the function receives a timestamp so
+ * that we do not need to call the (expensive) time() API.
+ * Note that we do the full retry processing here, doing the configured number of
+ * iterations. -- rgerhards, 2009-05-07
+ * We need to guard against module which always return RS_RET_OK from their tryResume()
+ * entry point. This is invalid, but has harsh consequences: it will cause the rsyslog
+ * engine to go into a tight loop. That obviously is not acceptable. As such, we track the
+ * count of iterations that a tryResume returning RS_RET_OK is immediately followed by
+ * an unsuccessful call to doAction(). If that happens more than 10 times, we assume
+ * the return acutally is a RS_RET_SUSPENDED. In order to go through the various
+ * resumption stages, we do this for every 10 requests. This magic number 10 may
+ * not be the most appropriate, but it should be thought of a "if nothing else helps"
+ * kind of facility: in the first place, the module should return a proper indication
+ * of its inability to recover. -- rgerhards, 2010-04-26.
+ */
+static rsRetVal ATTR_NONNULL()
+actionDoRetry(action_t * const pThis, wti_t * const pWti)
+{
+ int iRetries;
+ int iSleepPeriod;
+ int bTreatOKasSusp;
+ DEFiRet;
+
+ assert(pThis != NULL);
+
+ iRetries = 0;
+ while((*pWti->pbShutdownImmediate == 0) && getActionState(pWti, pThis) == ACT_STATE_RTRY) {
+ DBGPRINTF("actionDoRetry: %s enter loop, iRetries=%d, ResumeInRow %d\n",
+ pThis->pszName, iRetries, getActionResumeInRow(pWti, pThis));
+ iRet = pThis->pMod->tryResume(pWti->actWrkrInfo[pThis->iActionNbr].actWrkrData);
+ DBGPRINTF("actionDoRetry: %s action->tryResume returned %d\n", pThis->pszName, iRet);
+ if((getActionResumeInRow(pWti, pThis) > 9) && (getActionResumeInRow(pWti, pThis) % 10 == 0)) {
+ bTreatOKasSusp = 1;
+ setActionResumeInRow(pWti, pThis, 0);
+ iRet = RS_RET_SUSPENDED;
+ } else {
+ bTreatOKasSusp = 0;
+ }
+ if((iRet == RS_RET_OK) && (!bTreatOKasSusp)) {
+ DBGPRINTF("actionDoRetry: %s had success RDY again (iRet=%d)\n",
+ pThis->pszName, iRet);
+ STATSCOUNTER_INC(pThis->ctrResume, pThis->mutCtrResume);
+ if(pThis->bReportSuspension) {
+ LogMsg(0, RS_RET_RESUMED, LOG_INFO, "action '%s' "
+ "resumed (module '%s')",
+ pThis->pszName, pThis->pMod->pszName);
+ }
+ actionSetState(pThis, pWti, ACT_STATE_RDY);
+ } else if(iRet == RS_RET_SUSPENDED || bTreatOKasSusp) {
+ /* max retries reached? */
+ DBGPRINTF("actionDoRetry: %s check for max retries, iResumeRetryCount "
+ "%d, iRetries %d\n",
+ pThis->pszName, pThis->iResumeRetryCount, iRetries);
+ if((pThis->iResumeRetryCount != -1 && iRetries >= pThis->iResumeRetryCount)) {
+ actionSuspend(pThis, pWti);
+ if(getActionNbrResRtry(pWti, pThis) < 20)
+ incActionNbrResRtry(pWti, pThis);
+ } else {
+ ++iRetries;
+ iSleepPeriod = pThis->iResumeInterval;
+ srSleep(iSleepPeriod, 0);
+ if(*pWti->pbShutdownImmediate) {
+ ABORT_FINALIZE(RS_RET_FORCE_TERM);
+ }
+ }
+ } else if(iRet == RS_RET_DISABLE_ACTION) {
+ actionDisable(pThis);
+ }
+ }
+
+ if(getActionState(pWti, pThis) == ACT_STATE_RDY) {
+ setActionNbrResRtry(pWti, pThis, 0);
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* special retry handling if disabled via file: simply wait for the file
+ * to indicate whether or not it is ready again
+ */
+static rsRetVal ATTR_NONNULL()
+actionDoRetry_extFile(action_t *const pThis, wti_t *const pWti)
+{
+ int iRetries;
+ int iSleepPeriod;
+ DEFiRet;
+
+ assert(pThis != NULL);
+
+ DBGPRINTF("actionDoRetry_extFile: enter, actionState: %d\n",getActionState(pWti, pThis));
+ iRetries = 0;
+ while((*pWti->pbShutdownImmediate == 0) && getActionState(pWti, pThis) == ACT_STATE_RTRY) {
+ DBGPRINTF("actionDoRetry_extFile: %s enter loop, iRetries=%d, ResumeInRow %d\n",
+ pThis->pszName, iRetries, getActionResumeInRow(pWti, pThis));
+ iRet = checkExternalStateFile(pThis, pWti);
+ DBGPRINTF("actionDoRetry_extFile: %s checkExternalStateFile returned %d\n", pThis->pszName, iRet);
+ if(iRet == RS_RET_OK) {
+ DBGPRINTF("actionDoRetry_extFile: %s had success RDY again (iRet=%d)\n",
+ pThis->pszName, iRet);
+ if(pThis->bReportSuspension) {
+ LogMsg(0, RS_RET_RESUMED, LOG_INFO, "action '%s' "
+ "resumed (module '%s') via external state file",
+ pThis->pszName, pThis->pMod->pszName);
+ }
+ actionSetState(pThis, pWti, ACT_STATE_RDY);
+ } else if(iRet == RS_RET_SUSPENDED) {
+ /* max retries reached? */
+ DBGPRINTF("actionDoRetry_extFile: %s check for max retries, iResumeRetryCount "
+ "%d, iRetries %d\n",
+ pThis->pszName, pThis->iResumeRetryCount, iRetries);
+ if((pThis->iResumeRetryCount != -1 && iRetries >= pThis->iResumeRetryCount)) {
+ DBGPRINTF("actionDoRetry_extFile: did not work out, suspending\n");
+ actionSuspend(pThis, pWti);
+ pWti->execState.bPrevWasSuspended = 1;
+ if(getActionNbrResRtry(pWti, pThis) < 20)
+ incActionNbrResRtry(pWti, pThis);
+ } else {
+ ++iRetries;
+ iSleepPeriod = pThis->iResumeInterval;
+ srSleep(iSleepPeriod, 0);
+ if(*pWti->pbShutdownImmediate) {
+ ABORT_FINALIZE(RS_RET_FORCE_TERM);
+ }
+ }
+ } else if(iRet == RS_RET_DISABLE_ACTION) {
+ actionDisable(pThis);
+ }
+ }
+
+ if(getActionState(pWti, pThis) == ACT_STATE_RDY) {
+ setActionNbrResRtry(pWti, pThis, 0);
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+static rsRetVal
+actionCheckAndCreateWrkrInstance(action_t * const pThis, const wti_t *const pWti)
+{
+ int locked = 0;
+ DEFiRet;
+ if(pWti->actWrkrInfo[pThis->iActionNbr].actWrkrData == NULL) {
+ DBGPRINTF("wti %p: we need to create a new action worker instance for "
+ "action %d\n", pWti, pThis->iActionNbr);
+ CHKiRet(pThis->pMod->mod.om.createWrkrInstance(&(pWti->actWrkrInfo[pThis->iActionNbr].actWrkrData),
+ pThis->pModData));
+ pWti->actWrkrInfo[pThis->iActionNbr].pAction = pThis;
+ setActionState(pWti, pThis, ACT_STATE_RDY); /* action is enabled */
+
+ /* maintain worker data table -- only needed if wrkrHUP is requested! */
+
+ pthread_mutex_lock(&pThis->mutWrkrDataTable);
+ locked = 1;
+ int freeSpot;
+ for(freeSpot = 0 ; freeSpot < pThis->wrkrDataTableSize ; ++freeSpot)
+ if(pThis->wrkrDataTable[freeSpot] == NULL)
+ break;
+ if(pThis->nWrkr == pThis->wrkrDataTableSize) {
+ void *const newTable = realloc(pThis->wrkrDataTable,
+ (pThis->wrkrDataTableSize + 1) * sizeof(void*));
+ if(newTable == NULL) {
+ DBGPRINTF("actionCheckAndCreateWrkrInstance: out of "
+ "memory realloc wrkrDataTable\n")
+ ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+ }
+ pThis->wrkrDataTable = newTable;
+ pThis->wrkrDataTableSize++;
+ }
+ pThis->wrkrDataTable[freeSpot] = pWti->actWrkrInfo[pThis->iActionNbr].actWrkrData;
+ pThis->nWrkr++;
+ DBGPRINTF("wti %p: created action worker instance %d for "
+ "action %d\n", pWti, pThis->nWrkr, pThis->iActionNbr);
+ }
+finalize_it:
+ if(locked) {
+ pthread_mutex_unlock(&pThis->mutWrkrDataTable);
+ }
+ RETiRet;
+}
+
+/* try to resume an action -- rgerhards, 2007-08-02
+ * changed to new action state engine -- rgerhards, 2009-05-07
+ */
+static rsRetVal
+actionTryResume(action_t * const pThis, wti_t * const pWti)
+{
+ DEFiRet;
+ time_t ttNow = NO_TIME_PROVIDED;
+
+ if(getActionState(pWti, pThis) == ACT_STATE_SUSP) {
+ /* if we are suspended, we need to check if the timeout expired.
+ * for this handling, we must always obtain a fresh timestamp. We used
+ * to use the action timestamp, but in this case we will never reach a
+ * point where a resumption is actually tried, because the action timestamp
+ * is always in the past. So we can not avoid doing a fresh time() call
+ * here. -- rgerhards, 2009-03-18
+ */
+ datetime.GetTime(&ttNow); /* cache "now" */
+ if(ttNow >= pThis->ttResumeRtry) {
+ actionSetState(pThis, pWti, ACT_STATE_RTRY); /* back to retries */
+ }
+ }
+
+ if(getActionState(pWti, pThis) == ACT_STATE_RTRY) {
+ CHKiRet(actionDoRetry(pThis, pWti));
+ }
+
+ if(Debug && (getActionState(pWti, pThis) == ACT_STATE_RTRY ||
+ getActionState(pWti, pThis) == ACT_STATE_SUSP)) {
+ if(ttNow == NO_TIME_PROVIDED) /* use cached result if we have it */
+ datetime.GetTime(&ttNow);
+ dbgprintf("actionTryResume: action[%s] state: %s, next retry (if applicable): %u [now %u]\n",
+ pThis->pszName, getActStateName(pThis, pWti),
+ (unsigned) pThis->ttResumeRtry, (unsigned) ttNow);
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* prepare an action for performing work. This involves trying to recover it,
+ * depending on its current state.
+ * rgerhards, 2009-05-07
+ */
+static rsRetVal ATTR_NONNULL()
+actionPrepare(action_t *__restrict__ const pThis, wti_t *__restrict__ const pWti)
+{
+ DEFiRet;
+
+DBGPRINTF("actionPrepare[%s]: enter\n", pThis->pszName);
+ CHKiRet(actionCheckAndCreateWrkrInstance(pThis, pWti));
+ CHKiRet(actionTryResume(pThis, pWti));
+
+ /* if we are now ready, we initialize the transaction and advance
+ * action state accordingly
+ */
+ if(getActionState(pWti, pThis) == ACT_STATE_RDY) {
+ iRet = checkExternalStateFile(pThis, pWti);
+ if(iRet == RS_RET_SUSPENDED) {
+ DBGPRINTF("actionPrepare[%s]: SUSPENDED via external state file, "
+ "doing retry processing\n", pThis->pszName);
+ CHKiRet(actionDoRetry_extFile(pThis, pWti));
+ }
+ iRet = pThis->pMod->mod.om.beginTransaction(pWti->actWrkrInfo[pThis->iActionNbr].actWrkrData);
+ switch(iRet) {
+ case RS_RET_OK:
+ actionSetState(pThis, pWti, ACT_STATE_ITX);
+ break;
+ case RS_RET_SUSPENDED:
+ actionRetry(pThis, pWti);
+ break;
+ case RS_RET_DISABLE_ACTION:
+ actionDisable(pThis);
+ break;
+ default:FINALIZE;
+ }
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* prepare the calling parameters for doAction()
+ * rgerhards, 2009-05-07
+ */
+static rsRetVal
+prepareDoActionParams(action_t * __restrict__ const pAction,
+ wti_t * __restrict__ const pWti,
+ smsg_t *__restrict__ const pMsg,
+ struct syslogTime *ttNow)
+{
+ int i;
+ struct json_object *json;
+ actWrkrIParams_t *iparams;
+ actWrkrInfo_t *__restrict__ pWrkrInfo;
+ DEFiRet;
+
+ pWrkrInfo = &(pWti->actWrkrInfo[pAction->iActionNbr]);
+ if(pAction->isTransactional) {
+ CHKiRet(wtiNewIParam(pWti, pAction, &iparams));
+ for(i = 0 ; i < pAction->iNumTpls ; ++i) {
+ CHKiRet(tplToString(pAction->ppTpl[i], pMsg,
+ &actParam(iparams, pAction->iNumTpls, 0, i),
+ ttNow));
+ }
+ } else {
+ for(i = 0 ; i < pAction->iNumTpls ; ++i) {
+ switch(pAction->peParamPassing[i]) {
+ case ACT_STRING_PASSING:
+ CHKiRet(tplToString(pAction->ppTpl[i], pMsg,
+ &(pWrkrInfo->p.nontx.actParams[i]),
+ ttNow));
+ break;
+ /* note: ARRAY_PASSING mode has been removed in 8.26.0; if it
+ * is ever needed again, it can be found in 8.25.0.
+ * rgerhards 2017-03-06
+ */
+ case ACT_MSG_PASSING:
+ pWrkrInfo->p.nontx.actParams[i].param = (void*) pMsg;
+ break;
+ case ACT_JSON_PASSING:
+ CHKiRet(tplToJSON(pAction->ppTpl[i], pMsg, &json, ttNow));
+ pWrkrInfo->p.nontx.actParams[i].param = (void*) json;
+ break;
+ default:dbgprintf("software bug/error: unknown "
+ "pAction->peParamPassing[%d] %d in prepareDoActionParams\n",
+ i, (int) pAction->peParamPassing[i]);
+ break;
+ }
+ }
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+
+void
+releaseDoActionParams(action_t *__restrict__ const pAction, wti_t *__restrict__ const pWti, int action_destruct)
+{
+ int j;
+ actWrkrInfo_t *__restrict__ pWrkrInfo;
+
+ pWrkrInfo = &(pWti->actWrkrInfo[pAction->iActionNbr]);
+ for(j = 0 ; j < pAction->iNumTpls ; ++j) {
+ if (action_destruct) {
+ if (ACT_STRING_PASSING == pAction->peParamPassing[j]) {
+ free(pWrkrInfo->p.nontx.actParams[j].param);
+ pWrkrInfo->p.nontx.actParams[j].param = NULL;
+ pWrkrInfo->p.nontx.actParams[j].lenBuf = 0;
+ pWrkrInfo->p.nontx.actParams[j].lenStr = 0;
+ }
+ } else {
+ switch(pAction->peParamPassing[j]) {
+ case ACT_ARRAY_PASSING:
+ LogError(0, RS_RET_ERR, "plugin error: no longer supported "
+ "ARRAY_PASSING mode is used (see action.c)");
+ return;
+ case ACT_JSON_PASSING:
+ json_object_put((struct json_object*)
+ pWrkrInfo->p.nontx.actParams[j].param);
+ pWrkrInfo->p.nontx.actParams[j].param = NULL;
+ pWrkrInfo->p.nontx.actParams[j].lenBuf = 0;
+ pWrkrInfo->p.nontx.actParams[j].lenStr = 0;
+ break;
+ case ACT_STRING_PASSING:
+ case ACT_MSG_PASSING:
+ /* no need to do anything with these */
+ break;
+ }
+ }
+ }
+
+ return;
+}
+
+
+/* This is used in resume processing. We only finally know that a resume
+ * worked when we have been able to actually process a messages. As such,
+ * we need to do some cleanup and status tracking in that case.
+ */
+static void
+actionSetActionWorked(action_t *__restrict__ const pThis, wti_t *__restrict__ const pWti)
+{
+ setActionResumeInRow(pWti, pThis, 0);
+}
+
+static rsRetVal
+handleActionExecResult(action_t *__restrict__ const pThis,
+ wti_t *__restrict__ const pWti,
+ const rsRetVal ret)
+{
+ DEFiRet;
+ switch(ret) {
+ case RS_RET_OK:
+ actionCommitted(pThis, pWti);
+ actionSetActionWorked(pThis, pWti); /* we had a successful call! */
+ break;
+ case RS_RET_DEFER_COMMIT:
+ actionSetActionWorked(pThis, pWti); /* we had a successful call! */
+ /* we are done, action state remains the same */
+ break;
+ case RS_RET_PREVIOUS_COMMITTED:
+ /* action state remains the same, but we had a commit. */
+ pWti->actWrkrInfo[pThis->iActionNbr].bHadAutoCommit = 1;
+ actionSetActionWorked(pThis, pWti); /* we had a successful call! */
+ break;
+ case RS_RET_DISABLE_ACTION:
+ actionDisable(pThis);
+ break;
+ case RS_RET_SUSPENDED:
+ actionRetry(pThis, pWti);
+ break;
+ default:/* error happened - if it hits us here, we assume the message cannot
+ * be processed but an retry makes no sense. Usually, this should be
+ * return code RS_RET_DATAFAIL. -- rgerhards, 2017-10-06
+ */
+ LogError(0, ret, "action '%s' (module '%s') "
+ "message lost, could not be processed. Check for "
+ "additional error messages before this one.",
+ pThis->pszName, pThis->pMod->pszName);
+ actionSetState(pThis, pWti, ACT_STATE_DATAFAIL);
+ break;
+ }
+ iRet = getReturnCode(pThis, pWti);
+
+ RETiRet;
+}
+
+/* call the DoAction output plugin entry point
+ * rgerhards, 2008-01-28
+ */
+static rsRetVal
+actionCallDoAction(action_t *__restrict__ const pThis,
+ actWrkrIParams_t *__restrict__ const iparams,
+ wti_t *__restrict__ const pWti)
+{
+ void *param[CONF_OMOD_NUMSTRINGS_MAXSIZE];
+ int i;
+ DEFiRet;
+
+ DBGPRINTF("entering actionCalldoAction(), state: %s, actionNbr %d\n",
+ getActStateName(pThis, pWti), pThis->iActionNbr);
+
+ pWti->actWrkrInfo[pThis->iActionNbr].bHadAutoCommit = 0;
+ /* for this interface, we need to emulate the old style way
+ * of parameter passing.
+ */
+ for(i = 0 ; i < pThis->iNumTpls ; ++i) {
+ param[i] = actParam(iparams, pThis->iNumTpls, 0, i).param;
+ }
+
+ iRet = pThis->pMod->mod.om.doAction(param,
+ pWti->actWrkrInfo[pThis->iActionNbr].actWrkrData);
+ iRet = handleActionExecResult(pThis, pWti, iRet);
+ RETiRet;
+}
+
+
+/* call the commitTransaction output plugin entry point */
+static rsRetVal ATTR_NONNULL()
+actionCallCommitTransaction(action_t * const pThis,
+ wti_t *const pWti,
+ actWrkrIParams_t *__restrict__ const iparams, const int nparams)
+{
+ DEFiRet;
+
+ DBGPRINTF("entering actionCallCommitTransaction[%s], state: %s, nMsgs %u\n",
+ pThis->pszName, getActStateName(pThis, pWti), nparams);
+
+ iRet = pThis->pMod->mod.om.commitTransaction(
+ pWti->actWrkrInfo[pThis->iActionNbr].actWrkrData,
+ iparams, nparams);
+ DBGPRINTF("actionCallCommitTransaction[%s] state: %s "
+ "mod commitTransaction returned %d\n",
+ pThis->pszName, getActStateName(pThis, pWti), iRet);
+ iRet = handleActionExecResult(pThis, pWti, iRet);
+ RETiRet;
+}
+
+
+/* process a message
+ * this readies the action and then calls doAction()
+ * rgerhards, 2008-01-28
+ */
+static rsRetVal
+actionProcessMessage(action_t * const pThis, void *actParams, wti_t * const pWti)
+{
+ DEFiRet;
+
+ CHKiRet(actionPrepare(pThis, pWti));
+ if(pThis->pMod->mod.om.SetShutdownImmdtPtr != NULL)
+ pThis->pMod->mod.om.SetShutdownImmdtPtr(pThis->pModData, pWti->pbShutdownImmediate);
+ if(getActionState(pWti, pThis) == ACT_STATE_ITX)
+ CHKiRet(actionCallDoAction(pThis, actParams, pWti));
+
+ iRet = getReturnCode(pThis, pWti);
+finalize_it:
+ RETiRet;
+}
+
+
+/* the following function uses the new-style transactional interface */
+static rsRetVal
+doTransaction(action_t *__restrict__ const pThis, wti_t *__restrict__ const pWti,
+ actWrkrIParams_t *__restrict__ const iparams, const int nparams)
+{
+ actWrkrInfo_t *wrkrInfo;
+ int i;
+ DEFiRet;
+
+ wrkrInfo = &(pWti->actWrkrInfo[pThis->iActionNbr]);
+ if(pThis->pMod->mod.om.commitTransaction != NULL) {
+ DBGPRINTF("doTransaction: have commitTransaction IF, using that, pWrkrInfo %p\n", wrkrInfo);
+ CHKiRet(actionCallCommitTransaction(pThis, pWti, iparams, nparams));
+ } else { /* note: this branch is for compatibility with old TX modules */
+ DBGPRINTF("doTransaction: action '%s', currIParam %d\n",
+ pThis->pszName, wrkrInfo->p.tx.currIParam);
+ for(i = 0 ; i < nparams ; ++i) {
+ /* Note: we provide the message's base iparam - actionProcessMessage()
+ * uses this as *base* address.
+ */
+ iRet = actionProcessMessage(pThis,
+ &actParam(iparams, pThis->iNumTpls, i, 0), pWti);
+ DBGPRINTF("doTransaction: action %d, processing msg %d, result %d\n",
+ pThis->iActionNbr, i,iRet);
+ if(iRet == RS_RET_SUSPENDED) {
+ --i; /* we need to re-submit */
+ /* note: we are suspended and need to retry. In order not to
+ * hammer the CPU, we now do a voluntarly wait of 1 second.
+ * The rest will be handled by the standard retry handler.
+ */
+ srSleep(1, 0);
+ } else if(iRet != RS_RET_DEFER_COMMIT && iRet != RS_RET_PREVIOUS_COMMITTED &&
+ iRet != RS_RET_OK) {
+ FINALIZE; /* let upper peer handle the error condition! */
+ }
+ }
+ }
+finalize_it:
+ if(iRet == RS_RET_DEFER_COMMIT || iRet == RS_RET_PREVIOUS_COMMITTED)
+ iRet = RS_RET_OK; /* this is expected for transactional action! */
+ RETiRet;
+}
+
+
+/* Commit try committing (do not handle retry processing and such) */
+static rsRetVal ATTR_NONNULL()
+actionTryCommit(action_t *__restrict__ const pThis, wti_t *__restrict__ const pWti,
+ actWrkrIParams_t *__restrict__ const iparams, const int nparams)
+{
+ DEFiRet;
+
+ DBGPRINTF("actionTryCommit[%s] enter\n", pThis->pszName);
+ CHKiRet(actionPrepare(pThis, pWti));
+
+ CHKiRet(doTransaction(pThis, pWti, iparams, nparams));
+
+ if(getActionState(pWti, pThis) == ACT_STATE_ITX) {
+ iRet = pThis->pMod->mod.om.endTransaction(pWti->actWrkrInfo[pThis->iActionNbr].actWrkrData);
+ switch(iRet) {
+ case RS_RET_OK:
+ actionCommitted(pThis, pWti);
+ break;
+ case RS_RET_SUSPENDED:
+ actionRetry(pThis, pWti);
+ break;
+ case RS_RET_DISABLE_ACTION:
+ actionDisable(pThis);
+ break;
+ case RS_RET_DEFER_COMMIT:
+ DBGPRINTF("output plugin error: endTransaction() returns RS_RET_DEFER_COMMIT "
+ "- ignored\n");
+ actionCommitted(pThis, pWti);
+ break;
+ case RS_RET_PREVIOUS_COMMITTED:
+ DBGPRINTF("output plugin error: endTransaction() returns RS_RET_PREVIOUS_COMMITTED "
+ "- ignored\n");
+ actionCommitted(pThis, pWti);
+ break;
+ default:/* permanent failure of this message - no sense in retrying. This is
+ * not yet handled (but easy TODO)
+ */
+ DBGPRINTF("action[%s]: actionTryCommit receveived iRet %d\n",
+ pThis->pszName, iRet);
+ FINALIZE;
+ }
+ }
+ iRet = getReturnCode(pThis, pWti);
+
+finalize_it:
+ RETiRet;
+}
+
+/* If a transcation failed, we write the error file (if configured).
+ */
+static void ATTR_NONNULL()
+actionWriteErrorFile(action_t *__restrict__ const pThis, const rsRetVal ret,
+ actWrkrIParams_t *__restrict__ const iparams, const int nparams)
+{
+ fjson_object *etry=NULL;
+ int bNeedUnlock = 0;
+
+ STATSCOUNTER_INC(pThis->ctrFail, pThis->mutCtrFail);
+
+ if(pThis->pszErrFile == NULL) {
+ DBGPRINTF("action %s: commit failed, no error file set, silently "
+ "discarding %d messages\n", pThis->pszName, nparams);
+ goto done;
+ }
+
+ DBGPRINTF("action %d commit failed, writing %u messages (%d tpls) to error file\n",
+ pThis->iActionNbr, nparams, pThis->iNumTpls);
+
+ pthread_mutex_lock(&pThis->mutErrFile);
+ bNeedUnlock = 1;
+
+ if(pThis->fdErrFile == -1) {
+ pThis->fdErrFile = open(pThis->pszErrFile,
+ O_WRONLY|O_CREAT|O_APPEND|O_LARGEFILE|O_CLOEXEC,
+ S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP);
+ if(pThis->fdErrFile == -1) {
+ LogError(errno, RS_RET_ERR, "action %s: error opening error file %s",
+ pThis->pszName, pThis->pszErrFile);
+ goto done;
+ }
+ if (pThis->maxErrFileSize > 0) {
+ struct stat statbuf;
+ if (fstat(pThis->fdErrFile, &statbuf) == -1) {
+ LogError(errno, RS_RET_ERR, "failed to fstat %s", pThis->pszErrFile);
+ goto done;
+ }
+ pThis->currentErrFileSize = statbuf.st_size;
+ }
+ }
+
+ for(int i = 0 ; i < nparams ; ++i) {
+ if((etry = fjson_object_new_object()) == NULL)
+ goto done;
+ fjson_object_object_add(etry, "action", fjson_object_new_string((char*)pThis->pszName));
+ fjson_object_object_add(etry, "status", fjson_object_new_int(ret));
+ for(int j = 0 ; j < pThis->iNumTpls ; ++j) {
+ char tplname[20];
+ snprintf(tplname, sizeof(tplname), "template%d", j);
+ tplname[sizeof(tplname)-1] = '\0';
+ fjson_object_object_add(etry, tplname,
+ fjson_object_new_string((char*)actParam(iparams, 1, i, j).param));
+ }
+
+ char *const rendered = strdup((char*)fjson_object_to_json_string(etry));
+ if(rendered == NULL)
+ goto done;
+
+ size_t toWrite = strlen(rendered) + 1;
+ // Check if need to truncate the amount of bytes to write
+ if (pThis->maxErrFileSize > 0) {
+ if (pThis->currentErrFileSize + toWrite > pThis->maxErrFileSize) {
+ // Truncate to the pending available
+ toWrite = pThis->maxErrFileSize - pThis->currentErrFileSize;
+ }
+ pThis->currentErrFileSize += toWrite;
+ }
+ if(toWrite > 0) {
+ /* note: we use the '\0' inside the string to store a LF - we do not
+ * otherwise need it and it safes us a copy/realloc.
+ */
+ rendered[toWrite-1] = '\n'; /* NO LONGER A STRING! */
+ const ssize_t wrRet = write(pThis->fdErrFile, rendered, toWrite);
+ if(wrRet != (ssize_t) toWrite) {
+ LogError(errno, RS_RET_IO_ERROR,
+ "action %s: error writing errorFile %s, write returned %lld",
+ pThis->pszName, pThis->pszErrFile, (long long) wrRet);
+ }
+ }
+ free(rendered);
+
+ fjson_object_put(etry);
+ etry = NULL;
+ }
+done:
+ if(bNeedUnlock) {
+ pthread_mutex_unlock(&pThis->mutErrFile);
+ }
+ fjson_object_put(etry);
+ return;
+}
+
+
+static rsRetVal
+actionTryRemoveHardErrorsFromBatch(action_t *__restrict__ const pThis, wti_t *__restrict__ const pWti,
+ actWrkrIParams_t *const new_iparams, unsigned *new_nMsgs)
+{
+ actWrkrInfo_t *const wrkrInfo = &(pWti->actWrkrInfo[pThis->iActionNbr]);
+ const unsigned nMsgs = wrkrInfo->p.tx.currIParam;
+ actWrkrIParams_t oneParamSet[CONF_OMOD_NUMSTRINGS_MAXSIZE];
+ rsRetVal ret;
+ DEFiRet;
+
+ *new_nMsgs = 0;
+ for(unsigned i = 0 ; i < nMsgs ; ++i) {
+ setActionResumeInRow(pWti, pThis, 0); // make sure we do not trigger OK-as-SUSPEND handling
+ memcpy(&oneParamSet, &actParam(wrkrInfo->p.tx.iparams, pThis->iNumTpls, i, 0),
+ sizeof(actWrkrIParams_t) * pThis->iNumTpls);
+ ret = actionTryCommit(pThis, pWti, oneParamSet, 1);
+ if(ret == RS_RET_SUSPENDED) {
+ memcpy(new_iparams + (*new_nMsgs * pThis->iNumTpls), &oneParamSet,
+ sizeof(actWrkrIParams_t) * pThis->iNumTpls);
+ ++(*new_nMsgs);
+ } else if(ret != RS_RET_OK) {
+ actionWriteErrorFile(pThis, ret, oneParamSet, 1);
+ }
+ }
+ RETiRet;
+}
+
+/* Note: we currently need to return an iRet, as this is used in
+ * direct mode. TODO: However, it may be worth further investigating this,
+ * as it looks like there is no ultimate consumer of this code.
+ * rgerhards, 2013-11-06
+ */
+static rsRetVal ATTR_NONNULL()
+actionCommit(action_t *__restrict__ const pThis, wti_t *__restrict__ const pWti)
+{
+ actWrkrInfo_t *const wrkrInfo = &(pWti->actWrkrInfo[pThis->iActionNbr]);
+ /* Variables that permit us to override the batch of messages */
+ unsigned nMsgs = 0;
+ actWrkrIParams_t *iparams = NULL;
+ int needfree_iparams = 0; // work-around for clang static analyzer false positive
+ DEFiRet;
+
+ DBGPRINTF("actionCommit[%s]: enter, %d msgs\n", pThis->pszName, wrkrInfo->p.tx.currIParam);
+ if(!pThis->isTransactional || pWti->actWrkrInfo[pThis->iActionNbr].p.tx.currIParam == 0) {
+ FINALIZE;
+ } else if(getActionState(pWti, pThis) == ACT_STATE_SUSP) {
+ /* if we are suspended, we already tried everything to recover the
+ * action - and failed. So all we can do here is write the error file.
+ */
+ actionWriteErrorFile(pThis, iRet, wrkrInfo->p.tx.iparams, wrkrInfo->p.tx.currIParam);
+ FINALIZE;
+ }
+ DBGPRINTF("actionCommit[%s]: processing...\n", pThis->pszName);
+
+ /* we now do one try at commiting the whole batch. Usually, this will
+ * succeed. If so, we are happy and done. If not, we dig into the details
+ * of finding out if we have a non-temporary error and try to handle this
+ * as well as retry processing. Due to this logic we do a bit more retries
+ * than configured (if temporary failure), but this unavoidable and should
+ * do no real harm. - rgerhards, 2017-10-06
+ */
+ iRet = actionTryCommit(pThis, pWti, wrkrInfo->p.tx.iparams, wrkrInfo->p.tx.currIParam);
+DBGPRINTF("actionCommit[%s]: return actionTryCommit %d\n", pThis->pszName, iRet);
+ if(iRet == RS_RET_OK) {
+ FINALIZE;
+ }
+
+ /* check if this was a single-message batch. If it had a datafail error, we
+ * are done. If it is a multi-message batch, we need to sort out the individual
+ * message states.
+ */
+ if(wrkrInfo->p.tx.currIParam == 1) {
+ needfree_iparams = 0;
+ iparams = wrkrInfo->p.tx.iparams;
+ nMsgs = wrkrInfo->p.tx.currIParam;
+ if(iRet == RS_RET_DATAFAIL) {
+ FINALIZE;
+ }
+ } else {
+ DBGPRINTF("actionCommit[%s]: somewhat unhappy, full batch of %d msgs returned "
+ "status %d. Trying messages as individual actions.\n",
+ pThis->pszName, wrkrInfo->p.tx.currIParam, iRet);
+ CHKmalloc(iparams = malloc(sizeof(actWrkrIParams_t) * pThis->iNumTpls
+ * wrkrInfo->p.tx.currIParam));
+ needfree_iparams = 1;
+ actionTryRemoveHardErrorsFromBatch(pThis, pWti, iparams, &nMsgs);
+ }
+
+ if(nMsgs == 0) {
+ ABORT_FINALIZE(RS_RET_OK); // here, we consider everyting OK
+ }
+
+ /* We still have some messages with suspend error. So now let's do our
+ * "regular" retry and suspend processing.
+ */
+ DBGPRINTF("actionCommit[%s]: unhappy, we still have %d uncommitted messages.\n",
+ pThis->pszName, nMsgs);
+ int bDone = 0;
+ do {
+ iRet = actionTryCommit(pThis, pWti, iparams, nMsgs);
+ DBGPRINTF("actionCommit[%s]: in retry loop, iRet %d\n",
+ pThis->pszName, iRet);
+ if(iRet == RS_RET_FORCE_TERM) {
+ ABORT_FINALIZE(RS_RET_FORCE_TERM);
+ } else if(iRet == RS_RET_SUSPENDED) {
+ iRet = actionDoRetry(pThis, pWti);
+ DBGPRINTF("actionCommit[%s]: actionDoRetry returned %d\n",
+ pThis->pszName, iRet);
+ if(iRet == RS_RET_FORCE_TERM) {
+ ABORT_FINALIZE(RS_RET_FORCE_TERM);
+ } else if(iRet != RS_RET_OK) {
+ actionWriteErrorFile(pThis, iRet, iparams, nMsgs);
+ bDone = 1;
+ }
+ continue;
+ } else if(iRet == RS_RET_OK ||
+ iRet == RS_RET_SUSPENDED ||
+ iRet == RS_RET_ACTION_FAILED) {
+ bDone = 1;
+ }
+ if(getActionState(pWti, pThis) == ACT_STATE_RDY ||
+ getActionState(pWti, pThis) == ACT_STATE_SUSP) {
+ bDone = 1;
+ }
+ } while(!bDone);
+finalize_it:
+ DBGPRINTF("actionCommit[%s]: done, iRet %d\n", pThis->pszName, iRet);
+ if(needfree_iparams) {
+ free(iparams);
+ }
+ wrkrInfo->p.tx.currIParam = 0; /* reset to beginning */
+ RETiRet;
+}
+
+/* Commit all active transactions in *DIRECT mode* */
+void ATTR_NONNULL()
+actionCommitAllDirect(wti_t *__restrict__ const pWti)
+{
+ int i;
+ action_t *pAction;
+
+ for(i = 0 ; i < runConf->actions.iActionNbr ; ++i) {
+ pAction = pWti->actWrkrInfo[i].pAction;
+ if(pAction == NULL)
+ continue;
+ DBGPRINTF("actionCommitAllDirect: action %d, state %u, nbr to commit %d "
+ "isTransactional %d\n",
+ i, getActionStateByNbr(pWti, i), pWti->actWrkrInfo->p.tx.currIParam,
+ pAction->isTransactional);
+ if(pAction->pQueue->qType == QUEUETYPE_DIRECT)
+ actionCommit(pAction, pWti);
+ }
+}
+
+/* process a single message. This is both called if we run from the
+ * consumer side of an action queue as well as directly from the main
+ * queue thread if the action queue is set to "direct".
+ */
+static rsRetVal
+processMsgMain(action_t *__restrict__ const pAction,
+ wti_t *__restrict__ const pWti,
+ smsg_t *__restrict__ const pMsg,
+ struct syslogTime *ttNow)
+{
+ DEFiRet;
+
+ CHKiRet(prepareDoActionParams(pAction, pWti, pMsg, ttNow));
+
+ if(pAction->isTransactional) {
+ pWti->actWrkrInfo[pAction->iActionNbr].pAction = pAction;
+ DBGPRINTF("action '%s': is transactional - executing in commit phase\n", pAction->pszName);
+ actionPrepare(pAction, pWti);
+ iRet = getReturnCode(pAction, pWti);
+ FINALIZE;
+ }
+
+ iRet = actionProcessMessage(pAction,
+ pWti->actWrkrInfo[pAction->iActionNbr].p.nontx.actParams,
+ pWti);
+ if(pAction->bNeedReleaseBatch)
+ releaseDoActionParams(pAction, pWti, 0);
+finalize_it:
+ if(iRet == RS_RET_OK) {
+ if(pWti->execState.bDoAutoCommit)
+ iRet = actionCommit(pAction, pWti);
+ }
+ RETiRet;
+}
+
+/* This entry point is called by the ACTION queue (not main queue!)
+ */
+static rsRetVal ATTR_NONNULL()
+processBatchMain(void *__restrict__ const pVoid,
+ batch_t *__restrict__ const pBatch,
+ wti_t *__restrict__ const pWti)
+{
+ action_t *__restrict__ const pAction = (action_t*__restrict__ const) pVoid;
+ int i;
+ struct syslogTime ttNow;
+ DEFiRet;
+
+ wtiResetExecState(pWti, pBatch);
+ /* indicate we have not yet read the date */
+ ttNow.year = 0;
+
+ for(i = 0 ; i < batchNumMsgs(pBatch) && !*pWti->pbShutdownImmediate ; ++i) {
+ if(batchIsValidElem(pBatch, i)) {
+ /* we do not check error state below, because aborting would be
+ * more harmful than continuing.
+ */
+ rsRetVal localRet = processMsgMain(pAction, pWti, pBatch->pElem[i].pMsg, &ttNow);
+ DBGPRINTF("processBatchMain: i %d, processMsgMain iRet %d\n", i, localRet);
+ if( localRet == RS_RET_OK
+ || localRet == RS_RET_DEFER_COMMIT
+ || localRet == RS_RET_ACTION_FAILED
+ || localRet == RS_RET_PREVIOUS_COMMITTED ) {
+ batchSetElemState(pBatch, i, BATCH_STATE_COMM);
+ DBGPRINTF("processBatchMain: i %d, COMM state set\n", i);
+ }
+ }
+ }
+
+ iRet = actionCommit(pAction, pWti);
+ RETiRet;
+}
+
+
+/* remove an action worker instance from our table of
+ * workers. To be called from worker handler (wti).
+ */
+void
+actionRemoveWorker(action_t *const __restrict__ pAction,
+ void *const __restrict__ actWrkrData)
+{
+ pthread_mutex_lock(&pAction->mutWrkrDataTable);
+ pAction->nWrkr--;
+ for(int w = 0 ; w < pAction->wrkrDataTableSize ; ++w) {
+ if(pAction->wrkrDataTable[w] == actWrkrData) {
+ pAction->wrkrDataTable[w] = NULL;
+ break; /* done */
+ }
+ }
+ pthread_mutex_unlock(&pAction->mutWrkrDataTable);
+}
+
+
+/* call the HUP handler for a given action, if such a handler is defined.
+ * Note that the action must be able to service HUP requests concurrently
+ * to any current doAction() processing.
+ */
+rsRetVal
+actionCallHUPHdlr(action_t * const pAction)
+{
+ DEFiRet;
+
+ assert(pAction != NULL);
+ DBGPRINTF("Action %p checks HUP hdlr, act level: %p, wrkr level %p\n",
+ pAction, pAction->pMod->doHUP, pAction->pMod->doHUPWrkr);
+
+ if(pAction->pMod->doHUP != NULL) {
+ CHKiRet(pAction->pMod->doHUP(pAction->pModData));
+ }
+
+ if(pAction->pMod->doHUPWrkr != NULL) {
+ pthread_mutex_lock(&pAction->mutWrkrDataTable);
+ for(int i = 0 ; i < pAction->wrkrDataTableSize ; ++i) {
+ dbgprintf("HUP: table entry %d: %p %s\n", i,
+ pAction->wrkrDataTable[i],
+ pAction->wrkrDataTable[i] == NULL ? "[unused]" : "");
+ if(pAction->wrkrDataTable[i] != NULL) {
+ const rsRetVal localRet
+ = pAction->pMod->doHUPWrkr(pAction->wrkrDataTable[i]);
+ if(localRet != RS_RET_OK) {
+ DBGPRINTF("HUP handler returned error state %d - "
+ "ignored\n", localRet);
+ }
+ }
+ }
+ pthread_mutex_unlock(&pAction->mutWrkrDataTable);
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* set the action message queue mode
+ * TODO: probably move this into queue object, merge with MainMsgQueue!
+ * rgerhards, 2008-01-28
+ */
+static rsRetVal setActionQueType(void __attribute__((unused)) *pVal, uchar *pszType)
+{
+ DEFiRet;
+
+ if (!strcasecmp((char *) pszType, "fixedarray")) {
+ cs.ActionQueType = QUEUETYPE_FIXED_ARRAY;
+ DBGPRINTF("action queue type set to FIXED_ARRAY\n");
+ } else if (!strcasecmp((char *) pszType, "linkedlist")) {
+ cs.ActionQueType = QUEUETYPE_LINKEDLIST;
+ DBGPRINTF("action queue type set to LINKEDLIST\n");
+ } else if (!strcasecmp((char *) pszType, "disk")) {
+ cs.ActionQueType = QUEUETYPE_DISK;
+ DBGPRINTF("action queue type set to DISK\n");
+ } else if (!strcasecmp((char *) pszType, "direct")) {
+ cs.ActionQueType = QUEUETYPE_DIRECT;
+ DBGPRINTF("action queue type set to DIRECT (no queueing at all)\n");
+ } else {
+ LogError(0, RS_RET_INVALID_PARAMS, "unknown actionqueue parameter: %s", (char *) pszType);
+ iRet = RS_RET_INVALID_PARAMS;
+ }
+ free(pszType); /* no longer needed */
+
+ RETiRet;
+}
+
+
+/* This submits the message to the action queue in case we do NOT need to handle repeat
+ * message processing. That case permits us to gain lots of freedom during processing
+ * and thus speed. This is also utilized to submit messages in more complex cases once
+ * the complex logic has been applied ;)
+ * rgerhards, 2010-06-08
+ */
+static rsRetVal ATTR_NONNULL()
+doSubmitToActionQ(action_t * const pAction, wti_t * const pWti, smsg_t *pMsg)
+{
+ struct syslogTime ttNow; // TODO: think if we can buffer this in pWti
+ DEFiRet;
+
+ DBGPRINTF("action '%s': called, logging to %s (susp %d/%d, direct q %d)\n",
+ pAction->pszName, module.GetStateName(pAction->pMod),
+ pAction->bExecWhenPrevSusp, pWti->execState.bPrevWasSuspended,
+ pAction->pQueue->qType == QUEUETYPE_DIRECT);
+
+ if( pAction->bExecWhenPrevSusp
+ && !pWti->execState.bPrevWasSuspended) {
+ DBGPRINTF("action '%s': NOT executing, as previous action was "
+ "not suspended\n", pAction->pszName);
+ FINALIZE;
+ }
+
+ STATSCOUNTER_INC(pAction->ctrProcessed, pAction->mutCtrProcessed);
+ if(pAction->pQueue->qType == QUEUETYPE_DIRECT) {
+ ttNow.year = 0;
+ iRet = processMsgMain(pAction, pWti, pMsg, &ttNow);
+ } else {/* in this case, we do single submits to the queue.
+ * TODO: optimize this, we may do at least a multi-submit!
+ */
+ iRet = qqueueEnqMsg(pAction->pQueue, eFLOWCTL_NO_DELAY,
+ pAction->bCopyMsg ? MsgDup(pMsg) : MsgAddRef(pMsg));
+ }
+ pWti->execState.bPrevWasSuspended =
+ (iRet == RS_RET_SUSPENDED || iRet == RS_RET_ACTION_FAILED);
+
+ if (iRet == RS_RET_ACTION_FAILED) /* Increment failed counter */
+ STATSCOUNTER_INC(pAction->ctrFail, pAction->mutCtrFail);
+
+ DBGPRINTF("action '%s': set suspended state to %d\n",
+ pAction->pszName, pWti->execState.bPrevWasSuspended);
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* This function builds up a batch of messages to be (later)
+ * submitted to the action queue.
+ * Important: this function MUST not be called with messages that are to
+ * be discarded due to their "prevWasSuspended" state. It will not check for
+ * this and submit all messages to the queue for execution. So these must
+ * be filtered out before calling us (what is done currently!).
+ */
+rsRetVal
+actionWriteToAction(action_t * const pAction, smsg_t *pMsg, wti_t * const pWti)
+{
+ DEFiRet;
+
+ /* first, we check if the action should actually be called. The action-specific
+ * $ActionExecOnlyEveryNthTime permits us to execute an action only every Nth
+ * time. So we need to check if we need to drop the (otherwise perfectly executable)
+ * action for this reason. Note that in case we need to drop it, we return RS_RET_OK
+ * as the action was properly "passed to execution" from the upper layer's point
+ * of view. -- rgerhards, 2008-08-07.
+ */
+ if(pAction->iExecEveryNthOccur > 1) {
+ /* we need to care about multiple occurrences */
+ if( pAction->iExecEveryNthOccurTO > 0
+ && (getActNow(pAction) - pAction->tLastOccur) > pAction->iExecEveryNthOccurTO) {
+ DBGPRINTF("n-th occurrence handling timed out (%d sec), restarting from 0\n",
+ (int) (getActNow(pAction) - pAction->tLastOccur));
+ pAction->iNbrNoExec = 0;
+ pAction->tLastOccur = getActNow(pAction);
+ }
+ if(pAction->iNbrNoExec < pAction->iExecEveryNthOccur - 1) {
+ ++pAction->iNbrNoExec;
+ DBGPRINTF("action %p passed %d times to execution - less than configured - discarding\n",
+ pAction, pAction->iNbrNoExec);
+ FINALIZE;
+ } else {
+ pAction->iNbrNoExec = 0; /* we execute the action now, so the number of no execs is down to */
+ }
+ }
+
+ DBGPRINTF("Called action(complex case), logging to %s\n", module.GetStateName(pAction->pMod));
+
+ /* now check if we need to drop the message because otherwise the action would be too
+ * frequently called. -- rgerhards, 2008-04-08
+ * Note that the check for "pAction->iSecsExecOnceInterval > 0" is not necessary from
+ * a purely logical point of view. However, if safes us to check the system time in
+ * (those common) cases where ExecOnceInterval is not used. -- rgerhards, 2008-09-16
+ */
+ if(pAction->iSecsExecOnceInterval > 0 &&
+ pAction->iSecsExecOnceInterval + pAction->tLastExec > getActNow(pAction)) {
+ /* in this case we need to discard the message - its not yet time to exec the action */
+ DBGPRINTF("action not yet ready again to be executed, onceInterval %d, tCurr %d, tNext %d\n",
+ (int) pAction->iSecsExecOnceInterval, (int) getActNow(pAction),
+ (int) (pAction->iSecsExecOnceInterval + pAction->tLastExec));
+ FINALIZE;
+ }
+
+ /* we use reception time, not dequeue time - this is considered more appropriate and also faster ;)
+ * rgerhards, 2008-09-17 */
+ pAction->tLastExec = getActNow(pAction); /* re-init time flags */
+ pAction->f_time = pMsg->ttGenTime;
+
+ /* When we reach this point, we have a valid, non-disabled action.
+ * So let's enqueue our message for execution. -- rgerhards, 2007-07-24
+ */
+ iRet = doSubmitToActionQ(pAction, pWti, pMsg);
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* Call configured action, most complex case with all features supported (and thus slow).
+ * rgerhards, 2010-06-08
+ */
+
+PRAGMA_DIAGNOSTIC_PUSH;
+PRAGMA_IGNORE_Wempty_body;
+static rsRetVal
+doSubmitToActionQComplex(action_t * const pAction, wti_t * const pWti, smsg_t *pMsg)
+{
+ DEFiRet;
+
+ d_pthread_mutex_lock(&pAction->mutAction);
+ pthread_cleanup_push(mutexCancelCleanup, &pAction->mutAction);
+ DBGPRINTF("Called action %p (complex case), logging to %s\n",
+ pAction, module.GetStateName(pAction->pMod));
+
+ pAction->tActNow = -1; /* we do not yet know our current time (clear prev. value) */
+ // TODO: can we optimize the "now" handling again (was batch, I guess...)?
+
+ /* don't output marks to recently written outputs */
+ if(pAction->bWriteAllMarkMsgs == 0
+ && (pMsg->msgFlags & MARK) && (getActNow(pAction) - pAction->f_time) < MarkInterval / 2) {
+ ABORT_FINALIZE(RS_RET_OK);
+ }
+
+ /* call the output driver */
+ iRet = actionWriteToAction(pAction, pMsg, pWti);
+
+finalize_it:
+ d_pthread_mutex_unlock(&pAction->mutAction);
+ pthread_cleanup_pop(0); /* remove mutex cleanup handler */
+
+ RETiRet;
+}
+PRAGMA_DIAGNOSTIC_POP
+
+
+/* helper to activateActions, it activates a specific action.
+ */
+DEFFUNC_llExecFunc(doActivateActions)
+{
+ rsRetVal localRet;
+ action_t * const pThis = (action_t*) pData;
+ localRet = qqueueStart(runConf, pThis->pQueue);
+ if(localRet != RS_RET_OK) {
+ if(runConf->globals.bAbortOnFailedQueueStartup) {
+ fprintf(stderr, "rsyslogd: error %d starting up action queue, "
+ "abortOnFailedQueueStartup is set, so we abort rsyslog now.", localRet);
+ fflush(stderr);
+ exit(1); /* "good" exit, this is intended here */
+ }
+ LogError(0, localRet, "error starting up action queue");
+ if(localRet == RS_RET_FILE_PREFIX_MISSING) {
+ LogError(0, localRet, "file prefix (work directory?) "
+ "is missing");
+ }
+ actionDisable(pThis);
+ }
+ DBGPRINTF("Action %s[%p]: queue %p started\n", modGetName(pThis->pMod),
+ pThis, pThis->pQueue);
+ return RS_RET_OK; /* we ignore errors, we can not do anything either way */
+}
+
+
+/* This function "activates" the action after privileges have been dropped. Currently,
+ * this means that the queues are started.
+ * rgerhards, 2011-05-02
+ */
+rsRetVal
+activateActions(void)
+{
+ DEFiRet;
+ iRet = ruleset.IterateAllActions(runConf, doActivateActions, NULL);
+ RETiRet;
+}
+
+
+
+/* This submits the message to the action queue in case where we need to handle
+ * bWriteAllMarkMessage == RSFALSE only. Note that we use a non-blocking CAS loop
+ * for the synchronization. Here, we just modify the filter condition to be false when
+ * a mark message must not be written. However, in this case we must save the previous
+ * filter as we may need it in the next action (potential future optimization: check if this is
+ * the last action TODO).
+ * rgerhards, 2010-06-08
+ */
+static rsRetVal
+doSubmitToActionQNotAllMark(action_t * const pAction, wti_t * const pWti, smsg_t * const pMsg)
+{
+ int doProcess = 1;
+ time_t lastAct;
+ DEFiRet;
+
+ /* TODO: think about the whole logic. If messages come in out of order, things
+ * tend to become a bit unreliable. On the other hand, this only happens if we have
+ * very high traffic, in which this use case here is not really affected (as the
+ * MarkInterval is pretty corase).
+ */
+ /* CAS loop, we write back a bit early, but that's OK... */
+ /* we use reception time, not dequeue time - this is considered more appropriate and
+ * also faster ;) -- rgerhards, 2008-09-17 */
+ do {
+ lastAct = pAction->f_time;
+ if(pMsg->msgFlags & MARK) {
+ if((pMsg->ttGenTime - lastAct) < MarkInterval / 2) {
+ doProcess = 0;
+ DBGPRINTF("action was recently called, ignoring mark message\n");
+ break; /* do not update timestamp for non-written mark messages */
+ }
+ }
+ } while(ATOMIC_CAS_time_t(&pAction->f_time, lastAct,
+ pMsg->ttGenTime, &pAction->mutCAS) == 0);
+
+ if(doProcess) {
+ DBGPRINTF("Called action(NotAllMark), processing via '%s'\n",
+ module.GetStateName(pAction->pMod));
+ iRet = doSubmitToActionQ(pAction, pWti, pMsg);
+ }
+
+ RETiRet;
+}
+
+
+/* apply all params from param block to action. This supports the v6 config system.
+ * Defaults must have been set appropriately during action construct!
+ * rgerhards, 2011-08-01
+ */
+static rsRetVal
+actionApplyCnfParam(action_t * const pAction, struct cnfparamvals * const pvals)
+{
+ int i;
+
+ for(i = 0 ; i < pblk.nParams ; ++i) {
+ if(!pvals[i].bUsed)
+ continue;
+ if(!strcmp(pblk.descr[i].name, "name")) {
+ pAction->pszName = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL);
+ } else if(!strcmp(pblk.descr[i].name, "type")) {
+ continue; /* this is handled seperately during module select! */
+ } else if(!strcmp(pblk.descr[i].name, "action.errorfile")) {
+ pAction->pszErrFile = es_str2cstr(pvals[i].val.d.estr, NULL);
+ } else if(!strcmp(pblk.descr[i].name, "action.errorfile.maxsize")) {
+ pAction->maxErrFileSize = pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.externalstate.file")) {
+ pAction->pszExternalStateFile = es_str2cstr(pvals[i].val.d.estr, NULL);
+ } else if(!strcmp(pblk.descr[i].name, "action.writeallmarkmessages")) {
+ pAction->bWriteAllMarkMsgs = pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.execonlyeverynthtime")) {
+ pAction->iExecEveryNthOccur = pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.execonlyeverynthtimetimeout")) {
+ pAction->iExecEveryNthOccurTO = pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.execonlyonceeveryinterval")) {
+ pAction->iSecsExecOnceInterval = pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.execonlywhenpreviousissuspended")) {
+ pAction->bExecWhenPrevSusp = pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.repeatedmsgcontainsoriginalmsg")) {
+ pAction->bRepMsgHasMsg = pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.resumeretrycount")) {
+ pAction->iResumeRetryCount = pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.reportsuspension")) {
+ pAction->bReportSuspension = (int) pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.reportsuspensioncontinuation")) {
+ pAction->bReportSuspensionCont = (int) pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.copymsg")) {
+ pAction->bCopyMsg = (int) pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.resumeinterval")) {
+ pAction->iResumeInterval = pvals[i].val.d.n;
+ } else if(!strcmp(pblk.descr[i].name, "action.resumeintervalmax")) {
+ pAction->iResumeIntervalMax = pvals[i].val.d.n;
+ } else {
+ dbgprintf("action: program error, non-handled "
+ "param '%s'\n", pblk.descr[i].name);
+ }
+ }
+ return RS_RET_OK;
+}
+
+
+/* add an Action to the current selector
+ * The pOMSR is freed, as it is not needed after this function.
+ * Note: this function pulls global data that specifies action config state.
+ * rgerhards, 2007-07-27
+ */
+rsRetVal
+addAction(action_t **ppAction, modInfo_t *pMod, void *pModData,
+ omodStringRequest_t *pOMSR, struct cnfparamvals *actParams,
+ struct nvlst * const lst)
+{
+ DEFiRet;
+ int i;
+ int iTplOpts;
+ uchar *pTplName;
+ action_t *pAction;
+ char errMsg[512];
+
+ assert(ppAction != NULL);
+ assert(pMod != NULL);
+ assert(pOMSR != NULL);
+ DBGPRINTF("Module %s processes this action.\n", module.GetName(pMod));
+
+ CHKiRet(actionConstruct(&pAction)); /* create action object first */
+ pAction->pMod = pMod;
+ pAction->pModData = pModData;
+ if(actParams == NULL) { /* use legacy systemn */
+ pAction->pszName = cs.pszActionName;
+ pAction->iResumeInterval = cs.glbliActionResumeInterval;
+ pAction->iResumeRetryCount = cs.glbliActionResumeRetryCount;
+ pAction->bWriteAllMarkMsgs = cs.bActionWriteAllMarkMsgs;
+ pAction->bExecWhenPrevSusp = cs.bActExecWhenPrevSusp;
+ pAction->iSecsExecOnceInterval = cs.iActExecOnceInterval;
+ pAction->iExecEveryNthOccur = cs.iActExecEveryNthOccur;
+ pAction->iExecEveryNthOccurTO = cs.iActExecEveryNthOccurTO;
+ pAction->bRepMsgHasMsg = cs.bActionRepMsgHasMsg;
+ cs.iActExecEveryNthOccur = 0; /* auto-reset */
+ cs.iActExecEveryNthOccurTO = 0; /* auto-reset */
+ cs.bActionWriteAllMarkMsgs = 1; /* auto-reset */
+ cs.pszActionName = NULL; /* free again! */
+ } else {
+ actionApplyCnfParam(pAction, actParams);
+ }
+
+ /* check if we can obtain the template pointers - TODO: move to separate function? */
+ pAction->iNumTpls = OMSRgetEntryCount(pOMSR);
+ assert(pAction->iNumTpls >= 0); /* only debug check because this "can not happen" */
+ /* please note: iNumTpls may validly be zero. This is the case if the module
+ * does not request any templates. This sounds unlikely, but an actual example is
+ * the discard action, which does not require a string. -- rgerhards, 2007-07-30
+ */
+ if(pAction->iNumTpls > 0) {
+ /* we first need to create the template arrays */
+ CHKmalloc(pAction->ppTpl = (struct template **)calloc(pAction->iNumTpls, sizeof(struct template *)));
+ CHKmalloc(pAction->peParamPassing = (paramPassing_t*)calloc(pAction->iNumTpls,
+ sizeof(paramPassing_t)));
+ }
+
+ pAction->bUsesMsgPassingMode = 0;
+ pAction->bNeedReleaseBatch = 0;
+ for(i = 0 ; i < pAction->iNumTpls ; ++i) {
+ CHKiRet(OMSRgetEntry(pOMSR, i, &pTplName, &iTplOpts));
+ /* Ok, we got everything, so it now is time to look up the template
+ * (Hint: templates MUST be defined before they are used!)
+ */
+ if(!(iTplOpts & OMSR_TPL_AS_MSG)) {
+ if((pAction->ppTpl[i] =
+ tplFind(loadConf, (char*)pTplName, strlen((char*)pTplName))) == NULL) {
+ snprintf(errMsg, sizeof(errMsg),
+ " Could not find template %d '%s' - action disabled",
+ i, pTplName);
+ errno = 0;
+ LogError(0, RS_RET_NOT_FOUND, "%s", errMsg);
+ ABORT_FINALIZE(RS_RET_NOT_FOUND);
+ }
+ /* check required template options */
+ if( (iTplOpts & OMSR_RQD_TPL_OPT_SQL)
+ && (pAction->ppTpl[i]->optFormatEscape == 0)) {
+ errno = 0;
+ LogError(0, RS_RET_RQD_TPLOPT_MISSING, "Action disabled."
+ " To use this action, you have to specify "
+ "the SQL or stdSQL option in your template!\n");
+ ABORT_FINALIZE(RS_RET_RQD_TPLOPT_MISSING);
+ }
+ }
+
+ /* set parameter-passing mode */
+ if(iTplOpts & OMSR_TPL_AS_ARRAY) {
+ ABORT_FINALIZE(RS_RET_ERR);
+ } else if(iTplOpts & OMSR_TPL_AS_MSG) {
+ pAction->peParamPassing[i] = ACT_MSG_PASSING;
+ pAction->bUsesMsgPassingMode = 1;
+ } else if(iTplOpts & OMSR_TPL_AS_JSON) {
+ pAction->peParamPassing[i] = ACT_JSON_PASSING;
+ pAction->bNeedReleaseBatch = 1;
+ } else {
+ pAction->peParamPassing[i] = ACT_STRING_PASSING;
+ }
+
+ DBGPRINTF("template: '%s' assigned\n", pTplName);
+ }
+
+ pAction->pMod = pMod;
+ pAction->pModData = pModData;
+
+ CHKiRet(actionConstructFinalize(pAction, lst));
+
+ *ppAction = pAction; /* finally store the action pointer */
+
+finalize_it:
+ if(iRet == RS_RET_OK)
+ iRet = OMSRdestruct(pOMSR);
+ else {
+ /* do not overwrite error state! */
+ OMSRdestruct(pOMSR);
+ if(pAction != NULL)
+ actionDestruct(pAction);
+ }
+
+ RETiRet;
+}
+
+
+/* Reset config variables to default values.
+ * rgerhards, 2009-11-12
+ */
+static rsRetVal
+resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal)
+{
+ cs.iActExecOnceInterval = 0;
+ cs.bActExecWhenPrevSusp = 0;
+ return RS_RET_OK;
+}
+
+
+/* initialize (current) config variables.
+ * Used at program start and when a new scope is created.
+ */
+static void
+initConfigVariables(void)
+{
+ cs.bActionWriteAllMarkMsgs = 1;
+ cs.glbliActionResumeRetryCount = 0;
+ cs.bActExecWhenPrevSusp = 0;
+ cs.iActExecOnceInterval = 0;
+ cs.iActExecEveryNthOccur = 0;
+ cs.iActExecEveryNthOccurTO = 0;
+ cs.glbliActionResumeInterval = 30;
+ cs.glbliActionResumeRetryCount = 0;
+ cs.bActionRepMsgHasMsg = 0;
+ if(cs.pszActionName != NULL) {
+ free(cs.pszActionName);
+ cs.pszActionName = NULL;
+ }
+ actionResetQueueParams();
+}
+
+
+rsRetVal
+actionNewInst(struct nvlst *lst, action_t **ppAction)
+{
+ struct cnfparamvals *paramvals;
+ modInfo_t *pMod;
+ uchar *cnfModName = NULL;
+ omodStringRequest_t *pOMSR;
+ void *pModData;
+ action_t *pAction;
+ DEFiRet;
+
+ paramvals = nvlstGetParams(lst, &pblk, NULL);
+ if(paramvals == NULL) {
+ ABORT_FINALIZE(RS_RET_PARAM_ERROR);
+ }
+ dbgprintf("action param blk after actionNewInst:\n");
+ cnfparamsPrint(&pblk, paramvals);
+ cnfModName = (uchar*)es_str2cstr(paramvals[cnfparamGetIdx(&pblk, ("type"))].val.d.estr, NULL);
+ if((pMod = module.FindWithCnfName(loadConf, cnfModName, eMOD_OUT)) == NULL) {
+ LogError(0, RS_RET_MOD_UNKNOWN, "module name '%s' is unknown", cnfModName);
+ ABORT_FINALIZE(RS_RET_MOD_UNKNOWN);
+ }
+ CHKiRet(pMod->mod.om.newActInst(cnfModName, lst, &pModData, &pOMSR));
+
+ if((iRet = addAction(&pAction, pMod, pModData, pOMSR, paramvals, lst)) == RS_RET_OK) {
+ /* check if the module is compatible with select features
+ * (currently no such features exist) */
+ loadConf->actions.nbrActions++; /* one more active action! */
+ *ppAction = pAction;
+ } else {
+ // TODO: cleanup
+ }
+
+finalize_it:
+ free(cnfModName);
+ cnfparamvalsDestruct(paramvals, &pblk);
+ RETiRet;
+}
+
+rsRetVal actionClassInit(void)
+{
+ DEFiRet;
+ /* request objects we use */
+ CHKiRet(objGetObjInterface(&obj)); /* this provides the root pointer for all other queries */
+ CHKiRet(objUse(datetime, CORE_COMPONENT));
+ CHKiRet(objUse(module, CORE_COMPONENT));
+ CHKiRet(objUse(statsobj, CORE_COMPONENT));
+ CHKiRet(objUse(ruleset, CORE_COMPONENT));
+
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionname", 0, eCmdHdlrGetWord, NULL, &cs.pszActionName, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuefilename", 0, eCmdHdlrGetWord, NULL,
+ &cs.pszActionQFName, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesize", 0, eCmdHdlrInt, NULL, &cs.iActionQueueSize,
+ NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionwriteallmarkmessages", 0, eCmdHdlrBinary, NULL,
+ &cs.bActionWriteAllMarkMsgs, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuebatchsize", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQueueDeqBatchSize, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuemaxdiskspace", 0, eCmdHdlrSize, NULL,
+ &cs.iActionQueMaxDiskSpace, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuehighwatermark", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQHighWtrMark, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuelowwatermark", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQLowWtrMark, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardmark", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQDiscardMark, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardseverity", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQDiscardSeverity, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuecheckpointinterval", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQPersistUpdCnt, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesyncqueuefiles", 0, eCmdHdlrBinary, NULL,
+ &cs.bActionQSyncQeueFiles, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetype", 0, eCmdHdlrGetWord, setActionQueType, NULL, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkerthreads", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQueueNumWorkers, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutshutdown", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQtoQShutdown, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutactioncompletion", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQtoActShutdown, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutenqueue", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQtoEnq, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkertimeoutthreadshutdown", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQtoWrkShutdown, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkerthreadminimummessages", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQWrkMinMsgs, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuemaxfilesize", 0, eCmdHdlrSize, NULL,
+ &cs.iActionQueMaxFileSize, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesaveonshutdown", 0, eCmdHdlrBinary, NULL,
+ &cs.bActionQSaveOnShutdown, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeueslowdown", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQueueDeqSlowdown, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuetimebegin", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQueueDeqtWinFromHr, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuetimeend", 0, eCmdHdlrInt, NULL,
+ &cs.iActionQueueDeqtWinToHr, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlyeverynthtime", 0, eCmdHdlrInt, NULL,
+ &cs.iActExecEveryNthOccur, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlyeverynthtimetimeout", 0, eCmdHdlrInt, NULL,
+ &cs.iActExecEveryNthOccurTO, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlyonceeveryinterval", 0, eCmdHdlrInt, NULL,
+ &cs.iActExecOnceInterval, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"repeatedmsgcontainsoriginalmsg", 0, eCmdHdlrBinary, NULL,
+ &cs.bActionRepMsgHasMsg, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionexeconlywhenpreviousissuspended", 0, eCmdHdlrBinary, NULL,
+ &cs.bActExecWhenPrevSusp, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionresumeretrycount", 0, eCmdHdlrInt, NULL,
+ &cs.glbliActionResumeRetryCount, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler,
+ resetConfigVariables, NULL, NULL));
+
+ initConfigVariables(); /* first-time init of config setings */
+
+finalize_it:
+ RETiRet;
+}
+
+/* vi:set ai:
+ */