bugfix: failover did not work correctly if repeated msg reduction was on

  affected directive was: $ActionExecOnlyWhenPreviousIsSuspended on

13 files changed, 223 insertions, 49 deletions

diff --git a/ChangeLog b/ChangeLog
index 74fd299f..c9bf22b9 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,5 +1,7 @@
 ---------------------------------------------------------------------------
 Version 5.6.5  [V5-STABLE] (rgerhards), 2011-03-??
+- bugfix: failover did not work correctly if repeated msg reduction was on
+  affected directive was: $ActionExecOnlyWhenPreviousIsSuspended on
 - bugfix: omlibdbi did not use password from rsyslog.con
   closes: http://bugzilla.adiscon.com/show_bug.cgi?id=203
 - bugfix(kind of): tell users that config graph can currently not be
diff --git a/action.c b/action.c
index d41449d0..4312c377 100644
--- a/action.c
+++ b/action.c
@@ -4,7 +4,44 @@
  *
  * File begun on 2007-08-06 by RGerhards (extracted from syslogd.c)
  *
- * Copyright 2007-2010 Rainer Gerhards and Adiscon GmbH.
+ * 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 performanse-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 || f_ReduceRepeated || iSecsExecOnceInterval
+ * - doSubmitToActionQComplexBatch
+ * - helperSubmitToActionQComplexBatch
+ * - doActionCallAction
+ *   handles duplicate message processing, but in essence calls
+ * - actionWriteToAction
+ * - qqueueEnqObj
+ *   (now queue engine processing)
+ * if(pThis->bWriteAllMarkMsgs == FALSE) - this is the DEFAULT
+ * - doSubmitToActionQNotAllMarkBatch
+ * - doSubmitToActionQBatch (and from here like in the else case below!)
+ * else
+ * - doSubmitToActionQBatch
+ * - 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
+ * - processAction
+ * - submitBatch
+ * - tryDoAction
+ * - 
+ *
+ * Copyright 2007-2011 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of rsyslog.
  *
@@ -710,6 +747,7 @@ static rsRetVal releaseBatch(action_t *pAction, batch_t *pBatch)
 	ASSERT(pAction != NULL);
 
 	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
+dbgprintf("XXXX: releaseBatch %d: bPrevWasSuspended %d\n", i, pBatch->pElem[i].bPrevWasSuspended);
 		pElem = &(pBatch->pElem[i]);
 		if(pElem->bFilterOK && pElem->state != BATCH_STATE_DISC) {
 			switch(pAction->eParamPassing) {
@@ -897,8 +935,9 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem)
 	while(iElemProcessed <= *pnElem && i < pBatch->nElem) {
 		if(*(pBatch->pbShutdownImmediate))
 			ABORT_FINALIZE(RS_RET_FORCE_TERM);
+dbgprintf("XXXX:tryDoAction %d: bExecWhenPrevSusp: %d, bPrevWasSuspended %d ptr %p\n", i, pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended, &(pBatch->pElem[i].bPrevWasSuspended));
 		if(   pBatch->pElem[i].bFilterOK
-		   && pBatch->pElem[i].state != BATCH_STATE_DISC
+		   && pBatch->pElem[i].state != BATCH_STATE_DISC//) {
 		   && ((pAction->bExecWhenPrevSusp  == 0) || pBatch->pElem[i].bPrevWasSuspended) ) {
 			pMsg = (msg_t*) pBatch->pElem[i].pUsrp;
 			localRet = actionProcessMessage(pAction, pMsg, pBatch->pElem[i].staticActParams);
@@ -962,6 +1001,7 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem)
 	bDone = 0;
 	do {
 		localRet = tryDoAction(pAction, pBatch, &nElem);
+dbgprintf("XXXX: tryDoAction returns %d\n", localRet);
 		if(localRet == RS_RET_FORCE_TERM) {
 			ABORT_FINALIZE(RS_RET_FORCE_TERM);
 		}
@@ -981,12 +1021,14 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem)
 		} else if(localRet == RS_RET_SUSPENDED) {
 			; /* do nothing, this will retry the full batch */
 		} else if(localRet == RS_RET_ACTION_FAILED) {
+dbgprintf("XXXX: in ACTION_FAILED branch, doneUpTo %d\n", pBatch->iDoneUpTo);
 			/* in this case, everything not yet committed is BAD */
 			for(i = pBatch->iDoneUpTo ; i < wasDoneTo + nElem ; ++i) {
 				if(   pBatch->pElem[i].state != BATCH_STATE_DISC
 				   && pBatch->pElem[i].state != BATCH_STATE_COMM ) {
 					pBatch->pElem[i].state = BATCH_STATE_BAD;
 					pBatch->pElem[i].bPrevWasSuspended = 1;
+dbgprintf("XXXX: setting susps for item %d ptr %p\n", i, &(pBatch->pElem[i].bPrevWasSuspended));
 				}
 			}
 			bDone = 1;
@@ -1029,6 +1071,7 @@ prepareBatch(action_t *pAction, batch_t *pBatch)
 
 	pBatch->iDoneUpTo = 0;
 	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
+dbgprintf("XXXX: prepareBatch %d: bPrevWasSuspended %d\n", i, pBatch->pElem[i].bPrevWasSuspended);
 		pElem = &(pBatch->pElem[i]);
 		if(pElem->bFilterOK && pElem->state != BATCH_STATE_DISC) {
 			pElem->state = BATCH_STATE_RDY;
@@ -1069,6 +1112,7 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate)
 	DEFiRet;
 
 	assert(pBatch != NULL);
+dbgprintf("XXXX: running processBatchMain\n");
 
 	pbShutdownImmdtSave = pBatch->pbShutdownImmediate;
 	pBatch->pbShutdownImmediate = pbShutdownImmediate;
@@ -1098,6 +1142,7 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate)
 		iRet = localRet;
 
 finalize_it:
+dbgprintf("XXXX: exiting processBatchMain\n");
 	pBatch->pbShutdownImmediate = pbShutdownImmdtSave;
 	RETiRet;
 }
@@ -1163,12 +1208,33 @@ static rsRetVal setActionQueType(void __attribute__((unused)) *pVal, uchar *pszT
 }
 
 
+/* 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 complex case once
+ * the complex logic has been applied ;)
+ * rgerhards, 2010-06-08
+ */
+static inline rsRetVal
+doSubmitToActionQ(action_t *pAction, msg_t *pMsg)
+{
+	DEFiRet;
+
+	if(pAction->pQueue->qType == QUEUETYPE_DIRECT)
+		iRet = qqueueEnqObjDirect(pAction->pQueue, (void*) MsgAddRef(pMsg));
+	else
+		iRet = qqueueEnqObj(pAction->pQueue, pMsg->flowCtlType, (void*) MsgAddRef(pMsg));
+
+	RETiRet;
+}
+
+
 /* This function builds up a batch of messages to be (later)
  * submitted to the action queue.
  */
 rsRetVal
-actionWriteToAction(action_t *pAction)
+actionWriteToAction(action_t *pAction, batch_t *pBatch, int idxBtch)
 {
+dbgprintf("XXXX: enter actionWriteToAction idx %d\n", idxBtch);
 	msg_t *pMsgSave;	/* to save current message pointer, necessary to restore
 				   it in case it needs to be updated (e.g. repeated msgs) */
 	DEFiRet;
@@ -1238,7 +1304,7 @@ actionWriteToAction(action_t *pAction)
 		pAction->f_pMsg = pMsg;	/* use the new msg (pointer will be restored below) */
 	}
 
-	DBGPRINTF("Called action, logging to %s\n", module.GetStateName(pAction->pMod));
+	DBGPRINTF("Called action(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
@@ -1256,15 +1322,45 @@ actionWriteToAction(action_t *pAction)
 		FINALIZE;
 	}
 
-	/* we use reception time, not dequeue time - this is considered more appropriate and also faster ;) -- rgerhards, 2008-09-17 */
+	/* we use reception time, not dequeue time - this is considered more appropriate and also faster ;)
+	 * rgerhards, 2008-09-17 */
 	pAction->tLastExec = getActNow(pAction); /* re-init time flags */
 	pAction->f_time = pAction->f_pMsg->ttGenTime;
 
+dbgprintf("XXXX:actionWriteToAction %d: bExecWhenPrevSusp: %d, bPrevWasSuspended %d ptr %p\n", idxBtch, pAction->bExecWhenPrevSusp, pBatch->pElem[idxBtch].bPrevWasSuspended, &(pBatch->pElem[idxBtch].bPrevWasSuspended));
 	/* 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 = qqueueEnqObj(pAction->pQueue, pAction->f_pMsg->flowCtlType, (void*) MsgAddRef(pAction->f_pMsg));
+	if(pAction->bExecWhenPrevSusp  == 1 && pBatch->pElem[idxBtch].bPrevWasSuspended) {
+		/* in that case, we need to create a special batch which reflects the
+		 * suspended state. Otherwise, that information would be dropped inside
+		 * the queue engine. TODO: in later releases (v6?) create a better
+		 * solution than what we do here. However, for v5 this sounds much too
+		 * intrusive. -- rgerhardsm, 2011-03-16
+		 * (Code is copied over from queue.c and slightly modified)
+		 */
+		batch_t singleBatch;
+		batch_obj_t batchObj;
+		int i;
+		memset(&batchObj, 0, sizeof(batch_obj_t));
+		memset(&singleBatch, 0, sizeof(batch_t));
+		batchObj.state = BATCH_STATE_RDY;
+		batchObj.pUsrp = (obj_t*) pAction->f_pMsg;
+		batchObj.bPrevWasSuspended = 1;
+		batchObj.bFilterOK = 1;
+		singleBatch.nElem = 1; /* there always is only one in direct mode */
+		singleBatch.pElem = &batchObj;
+
+		iRet = qqueueEnqObjDirectBatch(pAction->pQueue, &singleBatch);
+
+		for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) {
+			free(batchObj.staticActStrings[i]);
+		}
+	} else { /* standard case, just submit */
+		iRet = doSubmitToActionQ(pAction, pAction->f_pMsg);
+	}
 
+dbgprintf("XXXX: action iRet %d\n", iRet);
 	if(iRet == RS_RET_OK)
 		pAction->f_prevcount = 0; /* message processed, so we start a new cycle */
 
@@ -1291,10 +1387,13 @@ finalize_it:
  * pthread_cleanup_push() POSIX macro...
  */
 static inline rsRetVal
-doActionCallAction(action_t *pAction, msg_t *pMsg)
+doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch)
 {
+	msg_t *pMsg;
 	DEFiRet;
 
+dbgprintf("XXXX: enter doActionCallAction\n");
+	pMsg = (msg_t*)(pBatch->pElem[idxBtch].pUsrp);
 	pAction->tActNow = -1; /* we do not yet know our current time (clear prev. value) */
 
 	/* don't output marks to recently written outputs */
@@ -1321,7 +1420,7 @@ doActionCallAction(action_t *pAction, msg_t *pMsg)
 		 * isolated messages), but back off so we'll flush less often in the future.
 		 */
 		if(getActNow(pAction) > REPEATTIME(pAction)) {
-			iRet = actionWriteToAction(pAction);
+			iRet = actionWriteToAction(pAction, pBatch, idxBtch);
 			BACKOFF(pAction);
 		}
 	} else {/* new message, save it */
@@ -1330,7 +1429,7 @@ doActionCallAction(action_t *pAction, msg_t *pMsg)
 		 */
 		if(pAction->f_pMsg != NULL) {
 			if(pAction->f_prevcount > 0)
-				actionWriteToAction(pAction);
+				actionWriteToAction(pAction, pBatch, idxBtch);
 				/* 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
 				 */
@@ -1338,33 +1437,21 @@ doActionCallAction(action_t *pAction, msg_t *pMsg)
 		}
 		pAction->f_pMsg = MsgAddRef(pMsg);
 		/* call the output driver */
-		iRet = actionWriteToAction(pAction);
+		iRet = actionWriteToAction(pAction, pBatch, idxBtch);
 	}
 
 finalize_it:
-	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.
- * rgerhards, 2010-06-08
- */
-static inline rsRetVal
-doSubmitToActionQ(action_t *pAction, msg_t *pMsg)
-{
-	DEFiRet;
-
-	if(pAction->pQueue->qType == QUEUETYPE_DIRECT)
-		iRet = qqueueEnqObjDirect(pAction->pQueue, (void*) MsgAddRef(pMsg));
-	else
-		iRet = qqueueEnqObj(pAction->pQueue, pMsg->flowCtlType, (void*) MsgAddRef(pMsg));
+	/* we need to update the batch to handle failover processing correctly */
+	if(iRet == RS_RET_OK) {
+		pBatch->pElem[idxBtch].bPrevWasSuspended = 0;
+	} else if(iRet == RS_RET_ACTION_FAILED) {
+		pBatch->pElem[idxBtch].bPrevWasSuspended = 1;
+	}
 
 	RETiRet;
 }
 
 
-
 /* This submits the message to the action queue in case where we need to handle
  * bWriteAllMarkMessage == FALSE only. Note that we use a non-blocking CAS loop
  * for the synchronization. Here, we just modify the filter condition to be false when
@@ -1480,9 +1567,11 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch)
 
 	DBGPRINTF("Called action(complex case), logging to %s\n", module.GetStateName(pAction->pMod));
 	for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) {
+dbgprintf("XXXX: helper complex %d: bExecWhenPrevSusp: %d, bPrevWasSuspended %d ptr %p\n", i, pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended, &(pBatch->pElem[i].bPrevWasSuspended));
 		if(   pBatch->pElem[i].bFilterOK
-                   && pBatch->pElem[i].state != BATCH_STATE_DISC) {
-			doActionCallAction(pAction, (msg_t*)(pBatch->pElem[i].pUsrp));
+                   && pBatch->pElem[i].state != BATCH_STATE_DISC
+		   && ((pAction->bExecWhenPrevSusp  == 0) || pBatch->pElem[i].bPrevWasSuspended) ) {
+			doActionCallAction(pAction, pBatch, i);
 		}
 	}
 
diff --git a/action.h b/action.h
index 0c86ef88..3e5d5266 100644
--- a/action.h
+++ b/action.h
@@ -101,26 +101,9 @@ rsRetVal actionDestruct(action_t *pThis);
 rsRetVal actionDbgPrint(action_t *pThis);
 rsRetVal actionSetGlobalResumeInterval(int iNewVal);
 rsRetVal actionDoAction(action_t *pAction);
-rsRetVal actionWriteToAction(action_t *pAction);
+//rsRetVal actionWriteToAction(action_t *pAction);
 rsRetVal actionCallHUPHdlr(action_t *pAction);
 rsRetVal actionClassInit(void);
 rsRetVal addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringRequest_t *pOMSR, int bSuspended);
 
-#if 1
-#define actionIsSuspended(pThis) ((pThis)->bSuspended == 1)
-#else
-/* The function is a debugging aid */
-inline int actionIsSuspended(action_t *pThis)
-{
-	int i;
-	ASSERT(pThis != NULL);
-	i =  pThis->bSuspended == 1;
-	dbgprintf("in IsSuspend(), returns %d\n", i);
-	return i;
-}
-#endif
-
 #endif /* #ifndef ACTION_H_INCLUDED */
-/*
- * vi:set ai:
- */
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 03a4985a..3ca8efca 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -48,6 +48,9 @@ TESTS = $(TESTRUNS) cfg.sh \
 	dircreate_off.sh \
 	discard-rptdmsg.sh \
 	discard.sh \
+	failover-basic.sh \
+	failover-rptd.sh \
+	failover-no-rptd.sh \
 	queue-persist.sh 
 
 if ENABLE_IMPTCP
@@ -217,6 +220,12 @@ EXTRA_DIST= 1.rstest 2.rstest 3.rstest err1.rstest \
 	   omod-if-array.sh \
 	   discard.sh \
 	   testsuites/discard.conf \
+	   failover-no-rptd.sh \
+	   testsuites/failover-no-rptd.conf \
+	   failover-rptd.sh \
+	   testsuites/failover-rptd.conf \
+	   failover-basic.sh \
+	   testsuites/failover-basic.conf \
 	   discard-rptdmsg.sh \
 	   testsuites/discard-rptdmsg.conf \
 	   diag.sh \
diff --git a/tests/diag.sh b/tests/diag.sh
index bf12fbda..fc0d9000 100755
--- a/tests/diag.sh
+++ b/tests/diag.sh
@@ -10,7 +10,7 @@
 #valgrind="valgrind --tool=helgrind --log-fd=1"
 #valgrind="valgrind --tool=exp-ptrcheck --log-fd=1"
 #set -o xtrace
-#export RSYSLOG_DEBUG="debug nostdout"
+#export RSYSLOG_DEBUG="debug nologfuncflow nostdout"
 #export RSYSLOG_DEBUGLOG="log"
 case $1 in
    'init')	$srcdir/killrsyslog.sh # kill rsyslogd if it runs for some reason
diff --git a/tests/failover-async.sh b/tests/failover-async.sh
new file mode 100755
index 00000000..f17bc0ff
--- /dev/null
+++ b/tests/failover-async.sh
@@ -0,0 +1,12 @@
+# This file is part of the rsyslog project, released under GPLv3
+echo ===============================================================================
+echo \[failover-async.sh\]: async test for failover functionality
+source $srcdir/diag.sh init
+source $srcdir/diag.sh startup failover-async.conf
+source $srcdir/diag.sh injectmsg  0 5
+echo doing shutdown
+source $srcdir/diag.sh shutdown-when-empty
+echo wait on shutdown
+source $srcdir/diag.sh wait-shutdown 
+source $srcdir/diag.sh seq-check  0 4999
+source $srcdir/diag.sh exit
diff --git a/tests/failover-basic.sh b/tests/failover-basic.sh
new file mode 100755
index 00000000..031ea25b
--- /dev/null
+++ b/tests/failover-basic.sh
@@ -0,0 +1,12 @@
+# This file is part of the rsyslog project, released under GPLv3
+echo ===============================================================================
+echo \[failover-basic.sh\]: basic test for failover functionality
+source $srcdir/diag.sh init
+source $srcdir/diag.sh startup failover-basic.conf
+source $srcdir/diag.sh injectmsg  0 5000
+echo doing shutdown
+source $srcdir/diag.sh shutdown-when-empty
+echo wait on shutdown
+source $srcdir/diag.sh wait-shutdown 
+source $srcdir/diag.sh seq-check  0 4999
+source $srcdir/diag.sh exit
diff --git a/tests/failover-no-rptd.sh b/tests/failover-no-rptd.sh
new file mode 100755
index 00000000..6abeba44
--- /dev/null
+++ b/tests/failover-no-rptd.sh
@@ -0,0 +1,19 @@
+# This file is part of the rsyslog project, released under GPLv3
+echo ===============================================================================
+echo \[failover-no-rptd.sh\]: rptd test for failover functionality - no failover
+source $srcdir/diag.sh init
+source $srcdir/diag.sh startup failover-no-rptd.conf
+source $srcdir/diag.sh injectmsg  0 5000
+echo doing shutdown
+source $srcdir/diag.sh shutdown-when-empty
+echo wait on shutdown
+source $srcdir/diag.sh wait-shutdown 
+# now we need our custom logic to see if the result file is empty
+# (what it should be!)
+cmp rsyslog.out.log /dev/null
+if [ $? -eq 1 ]
+then
+	echo "ERROR, output file not empty"
+	exit 1
+fi
+source $srcdir/diag.sh exit
diff --git a/tests/failover-rptd.sh b/tests/failover-rptd.sh
new file mode 100755
index 00000000..8a313e9e
--- /dev/null
+++ b/tests/failover-rptd.sh
@@ -0,0 +1,12 @@
+# This file is part of the rsyslog project, released under GPLv3
+echo ===============================================================================
+echo \[failover-rptd.sh\]: rptd test for failover functionality
+source $srcdir/diag.sh init
+source $srcdir/diag.sh startup failover-rptd.conf
+source $srcdir/diag.sh injectmsg  0 5000
+echo doing shutdown
+source $srcdir/diag.sh shutdown-when-empty
+echo wait on shutdown
+source $srcdir/diag.sh wait-shutdown 
+source $srcdir/diag.sh seq-check  0 4999
+source $srcdir/diag.sh exit
diff --git a/tests/testsuites/failover-async.conf b/tests/testsuites/failover-async.conf
new file mode 100644
index 00000000..76445de3
--- /dev/null
+++ b/tests/testsuites/failover-async.conf
@@ -0,0 +1,9 @@
+# see the equally-named .sh file for details
+$IncludeConfig diag-common.conf
+
+$template outfmt,"%msg:F,58:2%\n"
+# note: the target server shall not be available!
+
+$ActionQueueType LinkedList
+:msg, contains, "msgnum:" @@127.0.0.1:13514
+& ./rsyslog.out.log;outfmt
diff --git a/tests/testsuites/failover-basic.conf b/tests/testsuites/failover-basic.conf
new file mode 100644
index 00000000..a858769c
--- /dev/null
+++ b/tests/testsuites/failover-basic.conf
@@ -0,0 +1,8 @@
+# see the equally-named .sh file for details
+$IncludeConfig diag-common.conf
+
+$template outfmt,"%msg:F,58:2%\n"
+# note: the target server shall not be available!
+:msg, contains, "msgnum:" @@127.0.0.1:13514
+$ActionExecOnlyWhenPreviousIsSuspended on
+& ./rsyslog.out.log;outfmt
diff --git a/tests/testsuites/failover-no-rptd.conf b/tests/testsuites/failover-no-rptd.conf
new file mode 100644
index 00000000..a46ce116
--- /dev/null
+++ b/tests/testsuites/failover-no-rptd.conf
@@ -0,0 +1,9 @@
+# see the equally-named .sh file for details
+$IncludeConfig diag-common.conf
+
+$RepeatedMsgReduction on
+
+# second action should never execute
+:msg, contains, "msgnum:" /dev/null
+$ActionExecOnlyWhenPreviousIsSuspended on
+& ./rsyslog.out.log
diff --git a/tests/testsuites/failover-rptd.conf b/tests/testsuites/failover-rptd.conf
new file mode 100644
index 00000000..d3553dbb
--- /dev/null
+++ b/tests/testsuites/failover-rptd.conf
@@ -0,0 +1,10 @@
+# see the equally-named .sh file for details
+$IncludeConfig diag-common.conf
+
+$RepeatedMsgReduction on
+
+$template outfmt,"%msg:F,58:2%\n"
+# note: the target server shall not be available!
+:msg, contains, "msgnum:" @@127.0.0.1:13514
+$ActionExecOnlyWhenPreviousIsSuspended on
+& ./rsyslog.out.log;outfmt