path: root/action.c

/* action.c
 *
 * Implementation of the action object.
 *
 * File begun on 2007-08-06 by RGerhards (extracted from syslogd.c)
 *
 * Copyright 2007 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 "rsyslog.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 "dirty.h"
#include "template.h"
#include "action.h"
#include "modules.h"
#include "sync.h"
#include "cfsysline.h"
#include "srUtils.h"
#include "errmsg.h"
#include "datetime.h"

/* forward definitions */
rsRetVal actionCallDoAction(action_t *pAction, msg_t *pMsg);

/* object static data (once for all instances) */
/* TODO: make this an object! DEFobjStaticHelpers -- rgerhards, 2008-03-05 */
DEFobjCurrIf(obj)
DEFobjCurrIf(datetime)
DEFobjCurrIf(module)
DEFobjCurrIf(errmsg)

static int glbliActionResumeInterval = 30;
int glbliActionResumeRetryCount = 0;		/* how often should suspended actions be retried? */

/* main message queue and its configuration parameters */
static queueType_t ActionQueType = QUEUETYPE_DIRECT;		/* type of the main message queue above */
static int iActionQueueSize = 1000;				/* size of the main message queue above */
static int iActionQHighWtrMark = 800;				/* high water mark for disk-assisted queues */
static int iActionQLowWtrMark = 200;				/* low water mark for disk-assisted queues */
static int iActionQDiscardMark = 9800;				/* begin to discard messages */
static int iActionQDiscardSeverity = 8;				/* by default, discard nothing to prevent unintentional loss */
static int iActionQueueNumWorkers = 1;				/* number of worker threads for the mm queue above */
static uchar *pszActionQFName = NULL;				/* prefix for the main message queue file */
static int64 iActionQueMaxFileSize = 1024*1024;
static int iActionQPersistUpdCnt = 0;				/* persist queue info every n updates */
static int iActionQtoQShutdown = 0;				/* queue shutdown */ 
static int iActionQtoActShutdown = 1000;			/* action shutdown (in phase 2) */ 
static int iActionQtoEnq = 2000;				/* timeout for queue enque */ 
static int iActionQtoWrkShutdown = 60000;			/* timeout for worker thread shutdown */
static int iActionQWrkMinMsgs = 100;				/* minimum messages per worker needed to start a new one */
static int bActionQSaveOnShutdown = 1;				/* save queue on shutdown (when DA enabled)? */
static int64 iActionQueMaxDiskSpace = 0;			/* max disk space allocated 0 ==> unlimited */
static int iActionQueueDeqSlowdown = 0;				/* dequeue slowdown (simple rate limiting) */
static int iActionQueueDeqtWinFromHr = 0;			/* hour begin of time frame when queue is to be dequeued */
static int iActionQueueDeqtWinToHr = 25;			/* hour begin of time frame when queue is to be dequeued */

/* 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
 */
static int iActionNbr = 0;

/* ------------------------------ methods ------------------------------ */ 

/* 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;

	ActionQueType = QUEUETYPE_DIRECT;		/* type of the main message queue above */
	iActionQueueSize = 1000;			/* size of the main message queue above */
	iActionQHighWtrMark = 800;			/* high water mark for disk-assisted queues */
	iActionQLowWtrMark = 200;			/* low water mark for disk-assisted queues */
	iActionQDiscardMark = 9800;			/* begin to discard messages */
	iActionQDiscardSeverity = 4;			/* discard warning and above */
	iActionQueueNumWorkers = 1;			/* number of worker threads for the mm queue above */
	iActionQueMaxFileSize = 1024*1024;
	iActionQPersistUpdCnt = 0;			/* persist queue info every n updates */
	iActionQtoQShutdown = 0;			/* queue shutdown */ 
	iActionQtoActShutdown = 1000;			/* action shutdown (in phase 2) */ 
	iActionQtoEnq = 2000;				/* timeout for queue enque */ 
	iActionQtoWrkShutdown = 60000;			/* timeout for worker thread shutdown */
	iActionQWrkMinMsgs = 100;			/* minimum messages per worker needed to start a new one */
	bActionQSaveOnShutdown = 1;			/* save queue on shutdown (when DA enabled)? */
	iActionQueMaxDiskSpace = 0;
	iActionQueueDeqSlowdown = 0;
	iActionQueueDeqtWinFromHr = 0;
	iActionQueueDeqtWinToHr = 25;			/* 25 disables time windowed dequeuing */

	glbliActionResumeRetryCount = 0;		/* I guess it is smart to reset this one, too */

	if(pszActionQFName != NULL)
		d_free(pszActionQFName);
	pszActionQFName = NULL;				/* prefix for the main message queue file */

	RETiRet;
}


/* destructs an action descriptor object
 * rgerhards, 2007-08-01
 */
rsRetVal actionDestruct(action_t *pThis)
{
	DEFiRet;
	ASSERT(pThis != NULL);

	if(pThis->pQueue != NULL) {
		queueDestruct(&pThis->pQueue);
	}

	if(pThis->pMod != NULL)
		pThis->pMod->freeInstance(pThis->pModData);

	if(pThis->f_pMsg != NULL)
		msgDestruct(&pThis->f_pMsg);

	SYNC_OBJ_TOOL_EXIT(pThis);
	pthread_mutex_destroy(&pThis->mutActExec);
	if(pThis->ppTpl != NULL)
		d_free(pThis->ppTpl);
	d_free(pThis);
	
	RETiRet;
}


/* create a new action descriptor object
 * rgerhards, 2007-08-01
 */
rsRetVal actionConstruct(action_t **ppThis)
{
	DEFiRet;
	action_t *pThis;

	ASSERT(ppThis != NULL);
	
	if((pThis = (action_t*) calloc(1, sizeof(action_t))) == NULL) {
		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
	}

	pThis->iResumeInterval = glbliActionResumeInterval;
	pThis->iResumeRetryCount = glbliActionResumeRetryCount;
	pthread_mutex_init(&pThis->mutActExec, NULL);
	SYNC_OBJ_TOOL_INIT(pThis);

	/* indicate we have a new action */
	++iActionNbr;

finalize_it:
	*ppThis = pThis;
	RETiRet;
}


/* action construction finalizer
 */
rsRetVal
actionConstructFinalize(action_t *pThis)
{
	DEFiRet;
	uchar pszQName[64]; /* friendly name of our queue */

	ASSERT(pThis != NULL);

	/* find a name for our queue */
	snprintf((char*) pszQName, sizeof(pszQName)/sizeof(uchar), "action %d queue", iActionNbr);

	/* we need to make a safety check: if the queue is NOT in direct mode, a single 
	 * message object may be accessed by multiple threads. As such, we need to enable
	 * msg object thread safety in this case (this costs a bit performance and thus
	 * is not enabled by default. -- rgerhards, 2008-02-20
	 */
	if(ActionQueType != QUEUETYPE_DIRECT)
		MsgEnableThreadSafety();

	/* 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(queueConstruct(&pThis->pQueue, ActionQueType, 1, iActionQueueSize, (rsRetVal (*)(void*,void*))actionCallDoAction));
	obj.SetName((obj_t*) pThis->pQueue, pszQName);

	/* ... set some properties ... */
#	define setQPROP(func, directive, data) \
	CHKiRet_Hdlr(func(pThis->pQueue, data)) { \
		errmsg.LogError(NO_ERRCODE, "Invalid " #directive ", error %d. Ignored, running with default setting", iRet); \
	}
#	define setQPROPstr(func, directive, data) \
	CHKiRet_Hdlr(func(pThis->pQueue, data, (data == NULL)? 0 : strlen((char*) data))) { \
		errmsg.LogError(NO_ERRCODE, "Invalid " #directive ", error %d. Ignored, running with default setting", iRet); \
	}

	queueSetpUsr(pThis->pQueue, pThis);
	setQPROP(queueSetsizeOnDiskMax, "$ActionQueueMaxDiskSpace", iActionQueMaxDiskSpace);
	setQPROP(queueSetMaxFileSize, "$ActionQueueFileSize", iActionQueMaxFileSize);
	setQPROPstr(queueSetFilePrefix, "$ActionQueueFileName", pszActionQFName);
	setQPROP(queueSetiPersistUpdCnt, "$ActionQueueCheckpointInterval", iActionQPersistUpdCnt);
	setQPROP(queueSettoQShutdown, "$ActionQueueTimeoutShutdown", iActionQtoQShutdown );
	setQPROP(queueSettoActShutdown, "$ActionQueueTimeoutActionCompletion", iActionQtoActShutdown);
	setQPROP(queueSettoWrkShutdown, "$ActionQueueWorkerTimeoutThreadShutdown", iActionQtoWrkShutdown);
	setQPROP(queueSettoEnq, "$ActionQueueTimeoutEnqueue", iActionQtoEnq);
	setQPROP(queueSetiHighWtrMrk, "$ActionQueueHighWaterMark", iActionQHighWtrMark);
	setQPROP(queueSetiLowWtrMrk, "$ActionQueueLowWaterMark", iActionQLowWtrMark);
	setQPROP(queueSetiDiscardMrk, "$ActionQueueDiscardMark", iActionQDiscardMark);
	setQPROP(queueSetiDiscardSeverity, "$ActionQueueDiscardSeverity", iActionQDiscardSeverity);
	setQPROP(queueSetiMinMsgsPerWrkr, "$ActionQueueWorkerThreadMinimumMessages", iActionQWrkMinMsgs);
	setQPROP(queueSetbSaveOnShutdown, "$ActionQueueSaveOnShutdown", bActionQSaveOnShutdown);
	setQPROP(queueSetiDeqSlowdown,    "$ActionQueueDequeueSlowdown", iActionQueueDeqSlowdown);
	setQPROP(queueSetiDeqtWinFromHr,  "$ActionQueueDequeueTimeBegin", iActionQueueDeqtWinFromHr);
	setQPROP(queueSetiDeqtWinToHr,    "$ActionQueueDequeueTimeEnd", iActionQueueDeqtWinToHr);

#	undef setQPROP
#	undef setQPROPstr

	dbgoprint((obj_t*) pThis->pQueue, "save on shutdown %d, max disk space allowed %lld\n",
		   bActionQSaveOnShutdown, iActionQueMaxDiskSpace);
 	

	CHKiRet(queueStart(pThis->pQueue));
	dbgprintf("Action %p: queue %p created\n", pThis, pThis->pQueue);
	
	/* and now reset the queue params (see comment in its function header!) */
	actionResetQueueParams();

finalize_it:
	RETiRet;
}


/* set an action back to active state -- rgerhards, 2007-08-02
 */
static rsRetVal actionResume(action_t *pThis)
{
	DEFiRet;

	ASSERT(pThis != NULL);
	pThis->bSuspended = 0;

	RETiRet;
}


/* set the global resume interval
 */
rsRetVal actionSetGlobalResumeInterval(int iNewVal)
{
	glbliActionResumeInterval = iNewVal;
	return RS_RET_OK;
}


/* suspend an action -- rgerhards, 2007-08-02
 */
rsRetVal actionSuspend(action_t *pThis)
{
	DEFiRet;

	ASSERT(pThis != NULL);
	pThis->bSuspended = 1;
	pThis->ttResumeRtry = time(NULL) + pThis->iResumeInterval;
	pThis->iNbrResRtry = 0; /* tell that we did not yet retry to resume */

	RETiRet;
}

/* try to resume an action -- rgerhards, 2007-08-02
 * returns RS_RET_OK if resumption worked, RS_RET_SUSPEND if the
 * action is still suspended.
 */
rsRetVal actionTryResume(action_t *pThis)
{
	DEFiRet;
	time_t ttNow;

	ASSERT(pThis != NULL);

	ttNow = time(NULL); /* do the system call just once */

	/* first check if it is time for a re-try */
	if(ttNow > pThis->ttResumeRtry) {
		iRet = pThis->pMod->tryResume(pThis->pModData);
		if(iRet == RS_RET_SUSPENDED) {
			/* set new tryResume time */
			++pThis->iNbrResRtry;
			/* 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
			 */
			pThis->ttResumeRtry = ttNow + pThis->iResumeInterval * (pThis->iNbrResRtry / 10 + 1);
		}
	} else {
		/* it's too early, we are still suspended --> indicate this */
		iRet = RS_RET_SUSPENDED;
	}

	if(iRet == RS_RET_OK)
		actionResume(pThis);

	dbgprintf("actionTryResume: iRet: %d, next retry (if applicable): %u [now %u]\n",
		iRet, (unsigned) pThis->ttResumeRtry, (unsigned) ttNow);

	RETiRet;
}


/* debug-print the contents of an action object
 * rgerhards, 2007-08-02
 */
rsRetVal actionDbgPrint(action_t *pThis)
{
	DEFiRet;

	dbgprintf("%s: ", module.GetStateName(pThis->pMod));
	pThis->pMod->dbgPrintInstInfo(pThis->pModData);
	dbgprintf("\n\tInstance data: 0x%lx\n", (unsigned long) pThis->pModData);
	dbgprintf("\tRepeatedMsgReduction: %d\n", pThis->f_ReduceRepeated);
	dbgprintf("\tResume Interval: %d\n", pThis->iResumeInterval);
	dbgprintf("\tSuspended: %d", pThis->bSuspended);
	if(pThis->bSuspended) {
		dbgprintf(" next retry: %u, number retries: %d", (unsigned) pThis->ttResumeRtry, pThis->iNbrResRtry);
	}
	dbgprintf("\n");
	dbgprintf("\tDisabled: %d\n", !pThis->bEnabled);
	dbgprintf("\tExec only when previous is suspended: %d\n", pThis->bExecWhenPrevSusp);
	dbgprintf("\n");

	RETiRet;
}


/* call the DoAction output plugin entry point
 * rgerhards, 2008-01-28
 */
rsRetVal
actionCallDoAction(action_t *pAction, msg_t *pMsg)
{
	DEFiRet;
	int iRetries;
	int i;
	int iSleepPeriod;
	int bCallAction;
	int iCancelStateSave;
	uchar **ppMsgs;		/* array of message pointers for doAction */

	ASSERT(pAction != NULL);

	/* create the array for doAction() message pointers */
	if((ppMsgs = calloc(pAction->iNumTpls, sizeof(uchar *))) == NULL) {
		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
	}

	/* here we must loop to process all requested strings */
	for(i = 0 ; i < pAction->iNumTpls ; ++i) {
		CHKiRet(tplToString(pAction->ppTpl[i], pMsg, &(ppMsgs[i])));
	}
	iRetries = 0;
	/* We now must guard the output module against execution by multiple threads. The
	 * plugin interface specifies that output modules must not be thread-safe (except
	 * if they notify us they are - functionality not yet implemented...).
	 * rgerhards, 2008-01-30
	 */
	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
	d_pthread_mutex_lock(&pAction->mutActExec);
	pthread_cleanup_push(mutexCancelCleanup, &pAction->mutActExec);
	pthread_setcancelstate(iCancelStateSave, NULL);
	do {
		/* on first invocation, this if should never be true. We just put it at the top
		 * of the loop so that processing (and code) is simplified. This code is actually
		 * triggered on the 2nd+ invocation. -- rgerhards, 2008-01-30
		 */
		if(iRet == RS_RET_SUSPENDED) {
			/* ok, this calls for our retry logic... */
			++iRetries;
			iSleepPeriod = pAction->iResumeInterval;
			srSleep(iSleepPeriod, 0);
		}
		/* first check if we are suspended and, if so, retry */
		if(actionIsSuspended(pAction)) {
			iRet = actionTryResume(pAction);
			if(iRet == RS_RET_OK)
				bCallAction = 1;
			else
				bCallAction = 0;
		} else {
			bCallAction = 1;
		}

		if(bCallAction) {
			/* call configured action */
			iRet = pAction->pMod->mod.om.doAction(ppMsgs, pMsg->msgFlags, pAction->pModData);
			if(iRet == RS_RET_SUSPENDED) {
				dbgprintf("Action requested to be suspended, done that.\n");
				actionSuspend(pAction);
			}
		}

	} while(iRet == RS_RET_SUSPENDED && (pAction->iResumeRetryCount == -1 || iRetries < pAction->iResumeRetryCount)); /* do...while! */

	if(iRet == RS_RET_DISABLE_ACTION) {
		dbgprintf("Action requested to be disabled, done that.\n");
		pAction->bEnabled = 0; /* that's it... */
	}

	pthread_cleanup_pop(1); /* unlock mutex */

finalize_it:
	/* cleanup */
	for(i = 0 ; i < pAction->iNumTpls ; ++i) {
		if(ppMsgs[i] != NULL) {
			d_free(ppMsgs[i]);
		}
	}
	d_free(ppMsgs);
	msgDestruct(&pMsg); /* we are now finished with the message */

	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")) {
		ActionQueType = QUEUETYPE_FIXED_ARRAY;
		dbgprintf("action queue type set to FIXED_ARRAY\n");
	} else if (!strcasecmp((char *) pszType, "linkedlist")) {
		ActionQueType = QUEUETYPE_LINKEDLIST;
		dbgprintf("action queue type set to LINKEDLIST\n");
	} else if (!strcasecmp((char *) pszType, "disk")) {
		ActionQueType = QUEUETYPE_DISK;
		dbgprintf("action queue type set to DISK\n");
	} else if (!strcasecmp((char *) pszType, "direct")) {
		ActionQueType = QUEUETYPE_DIRECT;
		dbgprintf("action queue type set to DIRECT (no queueing at all)\n");
	} else {
		errmsg.LogError(NO_ERRCODE, "unknown actionqueue parameter: %s", (char *) pszType);
		iRet = RS_RET_INVALID_PARAMS;
	}
	d_free(pszType); /* no longer needed */

	RETiRet;
}


/* rgerhards 2004-11-09: fprintlog() is the actual driver for
 * the output channel. It receives the channel description (f) as
 * well as the message and outputs them according to the channel
 * semantics. The message is typically already contained in the
 * channel save buffer (f->f_prevline). This is not only the case
 * when a message was already repeated but also when a new message
 * arrived.
 * rgerhards 2007-08-01: interface changed to use action_t
 * rgerhards, 2007-12-11: please note: THIS METHOD MUST ONLY BE
 * CALLED AFTER THE CALLER HAS LOCKED THE pAction OBJECT! We do
 * not do this here. Failing to do so results in all kinds of
 * "interesting" problems!
 * RGERHARDS, 2008-01-29:
 * This is now the action caller and has been renamed.
 */
rsRetVal
actionWriteToAction(action_t *pAction)
{
	msg_t *pMsgSave;	/* to save current message pointer, necessary to restore
				   it in case it needs to be updated (e.g. repeated msgs) */
	time_t now;
	DEFiRet;

	pMsgSave = NULL;	/* indicate message poiner not saved */
	/* first check if this is a regular message or the repeation of
	 * a previous message. If so, we need to change the message text
	 * to "last message repeated n times" and then go ahead and write
	 * it. Please note that we can not modify the message object, because
	 * that would update it in other selectors as well. As such, we first
	 * need to create a local copy of the message, which we than can update.
	 * rgerhards, 2007-07-10
	 */
	if(pAction->f_prevcount > 1) {
		msg_t *pMsg;
		uchar szRepMsg[64];
		snprintf((char*)szRepMsg, sizeof(szRepMsg), "last message repeated %d times",
		    pAction->f_prevcount);

		if((pMsg = MsgDup(pAction->f_pMsg)) == NULL) {
			/* it failed - nothing we can do against it... */
			dbgprintf("Message duplication failed, dropping repeat message.\n");
			ABORT_FINALIZE(RS_RET_ERR);
		}

		/* We now need to update the other message properties.
		 * ... RAWMSG is a problem ... Please note that digital
		 * signatures inside the message are also invalidated.
		 */
		datetime.getCurrTime(&(pMsg->tRcvdAt));
		datetime.getCurrTime(&(pMsg->tTIMESTAMP));
		MsgSetMSG(pMsg, (char*)szRepMsg);
		MsgSetRawMsg(pMsg, (char*)szRepMsg);

		pMsgSave = pAction->f_pMsg;	/* save message pointer for later restoration */
		pAction->f_pMsg = pMsg;	/* use the new msg (pointer will be restored below) */
	}

	dbgprintf("Called action, logging to %s\n", module.GetStateName(pAction->pMod));

	time(&now); /* we need this for message repeation processing AND $ActionExecOnlyOnceEveryInterval */
	/* now check if we need to drop the message because otherwise the action would be too
	 * frequently called. -- rgerhards, 2008-04-08
	 */
	if(pAction->f_time != 0 && pAction->iSecsExecOnceInterval + pAction->tLastExec > now) {
		/* 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) now,
			  (int) (pAction->iSecsExecOnceInterval + pAction->tLastExec));
		FINALIZE;
	}

	pAction->tLastExec = now; /* we need this OnceInterval */
	pAction->f_time = now; /* we need this for message repeation processing */

	/* When we reach this point, we have a valid, non-disabled action.
	 * So let's enqueue our message for execution. -- rgerhards, 2007-07-24
	 */
	iRet = queueEnqObj(pAction->pQueue, pAction->f_pMsg->flowCtlType, (void*) MsgAddRef(pAction->f_pMsg));

	if(iRet == RS_RET_OK)
		pAction->f_prevcount = 0; /* message processed, so we start a new cycle */

finalize_it:
	if(pMsgSave != NULL) {
		/* we had saved the original message pointer. That was
		 * done because we needed to create a temporary one
		 * (most often for "message repeated n time" handling). If so,
		 * we need to restore the original one now, so that procesing
		 * can continue as normal. We also need to discard the temporary
		 * one, as we do not like memory leaks ;) Please note that the original
		 * message object will be discarded by our callers, so this is nothing
		 * of our business. rgerhards, 2007-07-10
		 */
		msgDestruct(&pAction->f_pMsg);
		pAction->f_pMsg = pMsgSave;	/* restore it */
	}

	RETiRet;
}


/* call the configured action. Does all necessary housekeeping.
 * rgerhards, 2007-08-01
 */
rsRetVal
actionCallAction(action_t *pAction, msg_t *pMsg)
{
	DEFiRet;
	int iCancelStateSave;

	ISOBJ_TYPE_assert(pMsg, msg);
	ASSERT(pAction != NULL);

	/* Make sure nodbody else modifies/uses this action object. Right now, this
         * is important because of "message repeated n times" processing and potentially
	 * multiple worker threads. -- rgerhards, 2007-12-11
 	 */
	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
	LockObj(pAction);
	pthread_cleanup_push(mutexCancelCleanup, pAction->Sync_mut);
	pthread_setcancelstate(iCancelStateSave, NULL);

	/* first, we need to check if this is a disabled
	 * entry. If so, we must not further process it.
	 * rgerhards 2005-09-26
	 * In the future, disabled modules may be re-probed from time
	 * to time. They are in a perfectly legal state, except that the
	 * doAction method indicated that it wanted to be disabled - but
	 * we do not consider this is a solution for eternity... So we
	 * should check from time to time if affairs have improved.
	 * rgerhards, 2007-07-24
	 */
	if(pAction->bEnabled == 0) {
		ABORT_FINALIZE(RS_RET_OK);
	}

	/* don't output marks to recently written files */
	if ((pMsg->msgFlags & MARK) && (time(NULL) - pAction->f_time) < MarkInterval / 2) {
		ABORT_FINALIZE(RS_RET_OK);
	}

	/* suppress duplicate messages
	 */
	if ((pAction->f_ReduceRepeated == 1) && pAction->f_pMsg != NULL &&
	    (pMsg->msgFlags & MARK) == 0 && getMSGLen(pMsg) == getMSGLen(pAction->f_pMsg) &&
	    !strcmp(getMSG(pMsg), getMSG(pAction->f_pMsg)) &&
	    !strcmp(getHOSTNAME(pMsg), getHOSTNAME(pAction->f_pMsg)) &&
	    !strcmp(getPROCID(pMsg), getPROCID(pAction->f_pMsg)) &&
	    !strcmp(getAPPNAME(pMsg), getAPPNAME(pAction->f_pMsg))) {
		pAction->f_prevcount++;
		dbgprintf("msg repeated %d times, %ld sec of %d.\n",
		    pAction->f_prevcount, (long) time(NULL) - pAction->f_time,
		    repeatinterval[pAction->f_repeatcount]);
		/* use current message, so we have the new timestamp (means we need to discard previous one) */
		msgDestruct(&pAction->f_pMsg);
		pAction->f_pMsg = MsgAddRef(pMsg);
		/* If domark would have logged this by now, flush it now (so we don't hold
		 * isolated messages), but back off so we'll flush less often in the future.
		 */
		if(time(NULL) > REPEATTIME(pAction)) {
			iRet = actionWriteToAction(pAction);
			BACKOFF(pAction);
		}
	} else {
		/* new message, save it */
		/* first check if we have a previous message stored
		 * if so, emit and then discard it first
		 */
		if(pAction->f_pMsg != NULL) {
			if(pAction->f_prevcount > 0)
				actionWriteToAction(pAction);
				/* we do not care about iRet above - I think it's right but if we have
				 * some troubles, you know where to look at ;) -- rgerhards, 2007-08-01
				 */
			msgDestruct(&pAction->f_pMsg);
		}
		pAction->f_pMsg = MsgAddRef(pMsg);
		/* call the output driver */
		iRet = actionWriteToAction(pAction);
	}

finalize_it:
	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
	UnlockObj(pAction);
	pthread_cleanup_pop(0); /* remove mutex cleanup handler */
	pthread_setcancelstate(iCancelStateSave, NULL);
	RETiRet;
}


/* add our cfsysline handlers
 * rgerhards, 2008-01-28
 */
rsRetVal
actionAddCfSysLineHdrl(void)
{
	DEFiRet;

	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuefilename", 0, eCmdHdlrGetWord, NULL, &pszActionQFName, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesize", 0, eCmdHdlrInt, NULL, &iActionQueueSize, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuemaxdiskspace", 0, eCmdHdlrSize, NULL, &iActionQueMaxDiskSpace, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuehighwatermark", 0, eCmdHdlrInt, NULL, &iActionQHighWtrMark, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuelowwatermark", 0, eCmdHdlrInt, NULL, &iActionQLowWtrMark, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardmark", 0, eCmdHdlrInt, NULL, &iActionQDiscardMark, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuediscardseverity", 0, eCmdHdlrInt, NULL, &iActionQDiscardSeverity, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuecheckpointinterval", 0, eCmdHdlrInt, NULL, &iActionQPersistUpdCnt, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetype", 0, eCmdHdlrGetWord, setActionQueType, NULL, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkerthreads", 0, eCmdHdlrInt, NULL, &iActionQueueNumWorkers, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutshutdown", 0, eCmdHdlrInt, NULL, &iActionQtoQShutdown, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutactioncompletion", 0, eCmdHdlrInt, NULL, &iActionQtoActShutdown, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuetimeoutenqueue", 0, eCmdHdlrInt, NULL, &iActionQtoEnq, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkertimeoutthreadshutdown", 0, eCmdHdlrInt, NULL, &iActionQtoWrkShutdown, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueueworkerthreadminimummessages", 0, eCmdHdlrInt, NULL, &iActionQWrkMinMsgs, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuemaxfilesize", 0, eCmdHdlrSize, NULL, &iActionQueMaxFileSize, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuesaveonshutdown", 0, eCmdHdlrBinary, NULL, &bActionQSaveOnShutdown, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeueslowdown", 0, eCmdHdlrInt, NULL, &iActionQueueDeqSlowdown, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuetimebegin", 0, eCmdHdlrInt, NULL, &iActionQueueDeqtWinFromHr, NULL));
	CHKiRet(regCfSysLineHdlr((uchar *)"actionqueuedequeuetimeend", 0, eCmdHdlrInt, NULL, &iActionQueueDeqtWinToHr, NULL));
	
finalize_it:
	RETiRet;
}


/* add an Action to the current selector
 * The pOMSR is freed, as it is not needed after this function.
 * Note: this function pulls global data that specifies action config state.
 * rgerhards, 2007-07-27
 */
rsRetVal
addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringRequest_t *pOMSR, int bSuspended)
{
	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 processed this config line.\n", module.GetName(pMod));

	CHKiRet(actionConstruct(&pAction)); /* create action object first */
	pAction->pMod = pMod;
	pAction->pModData = pModData;
	pAction->bExecWhenPrevSusp = bActExecWhenPrevSusp;
	pAction->iSecsExecOnceInterval = iActExecOnceInterval;

	/* 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 pointer array */
		if((pAction->ppTpl = calloc(pAction->iNumTpls, sizeof(struct template *))) == NULL) {
			ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
		}
	}
	
	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((pAction->ppTpl[i] = tplFind((char*)pTplName, strlen((char*)pTplName))) == NULL) {
			snprintf(errMsg, sizeof(errMsg) / sizeof(char),
				 " Could not find template '%s' - action disabled\n",
				 pTplName);
			errno = 0;
			errmsg.LogError(NO_ERRCODE, "%s", errMsg);
			ABORT_FINALIZE(RS_RET_NOT_FOUND);
		}
		/* check required template options */
		if(   (iTplOpts & OMSR_RQD_TPL_OPT_SQL)
		   && (pAction->ppTpl[i]->optFormatForSQL == 0)) {
			errno = 0;
			errmsg.LogError(NO_ERRCODE, "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);
		}

		dbgprintf("template: '%s' assigned\n", pTplName);
	}

	pAction->pMod = pMod;
	pAction->pModData = pModData;
	/* now check if the module is compatible with select features */
	if(pMod->isCompatibleWithFeature(sFEATURERepeatedMsgReduction) == RS_RET_OK)
		pAction->f_ReduceRepeated = bReduceRepeatMsgs;
	else {
		dbgprintf("module is incompatible with RepeatedMsgReduction - turned off\n");
		pAction->f_ReduceRepeated = 0;
	}
	pAction->bEnabled = 1; /* action is enabled */

	if(bSuspended)
		actionSuspend(pAction);

	CHKiRet(actionConstructFinalize(pAction));
	
	/* TODO: if we exit here, we have a memory leak... */

	*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;
}


/* TODO: we are not yet a real object, the ClassInit here just looks like it is..
 */
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(errmsg, CORE_COMPONENT));

finalize_it:
	RETiRet;
}

/* vi:set ai:
 */