Adding upstream version 8.2402.0.upstream/8.2402.0

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 943 insertions, 0 deletions

diff --git a/plugins/imdiag/imdiag.c b/plugins/imdiag/imdiag.c
new file mode 100644
index 0000000..774939a
--- /dev/null
+++ b/plugins/imdiag/imdiag.c
@@ -0,0 +1,943 @@
+/* imdiag.c
+ * This is a testbench tool. It started out with a broader scope,
+ * but we dropped this idea. To learn about rsyslog runtime statistics
+ * have a look at impstats.
+ *
+ * File begun on 2008-07-25 by RGerhards
+ *
+ * Copyright 2008-2023 Adiscon GmbH.
+ *
+ * This file is part of rsyslog.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *       -or-
+ *       see COPYING.ASL20 in the source distribution
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "config.h"
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <string.h>
+#include <errno.h>
+#include <unistd.h>
+#include <stdarg.h>
+#include <ctype.h>
+#include <signal.h>
+#include <netinet/in.h>
+#include <netdb.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <pthread.h>
+#include <semaphore.h>
+#if HAVE_FCNTL_H
+#include <fcntl.h>
+#endif
+#include "rsyslog.h"
+#include "dirty.h"
+#include "cfsysline.h"
+#include "module-template.h"
+#include "unicode-helper.h"
+#include "net.h"
+#include "netstrm.h"
+#include "errmsg.h"
+#include "tcpsrv.h"
+#include "srUtils.h"
+#include "msg.h"
+#include "datetime.h"
+#include "ratelimit.h"
+#include "queue.h"
+#include "lookup.h"
+#include "net.h" /* for permittedPeers, may be removed when this is removed */
+#include "statsobj.h"
+
+
+MODULE_TYPE_INPUT
+MODULE_TYPE_NOKEEP
+
+/* static data */
+DEF_IMOD_STATIC_DATA
+DEFobjCurrIf(tcpsrv)
+DEFobjCurrIf(tcps_sess)
+DEFobjCurrIf(net)
+DEFobjCurrIf(netstrm)
+DEFobjCurrIf(datetime)
+DEFobjCurrIf(prop)
+DEFobjCurrIf(statsobj)
+
+/* Module static data */
+static tcpsrv_t *pOurTcpsrv = NULL;  /* our TCP server(listener) TODO: change for multiple instances */
+static permittedPeers_t *pPermPeersRoot = NULL;
+static prop_t *pInputName = NULL;
+/* there is only one global inputName for all messages generated by this input */
+static prop_t *pRcvDummy = NULL;
+static prop_t *pRcvIPDummy = NULL;
+
+static int max_empty_checks = 3; /* how often check for queue empty during shutdown? */
+
+statsobj_t *diagStats;
+STATSCOUNTER_DEF(potentialArtificialDelayMs, mutPotentialArtificialDelayMs)
+STATSCOUNTER_DEF(actualArtificialDelayMs, mutActualArtificialDelayMs)
+STATSCOUNTER_DEF(delayInvocationCount, mutDelayInvocationCount)
+
+static sem_t statsReportingBlocker;
+static long long statsReportingBlockStartTimeMs = 0;
+static int allowOnlyOnce = 0;
+DEF_ATOMIC_HELPER_MUT(mutAllowOnlyOnce);
+pthread_mutex_t mutStatsReporterWatch;
+pthread_cond_t statsReporterWatch;
+int statsReported = 0;
+static int abortTimeout = -1;		/* for timeoutGuard - if set, abort rsyslogd after that many seconds */
+static pthread_t timeoutGuard_thrd;	/* thread ID for timeoutGuard thread (if active) */
+
+/* config settings */
+struct modConfData_s {
+	EMPTY_STRUCT;
+};
+
+static flowControl_t injectmsgDelayMode = eFLOWCTL_NO_DELAY;
+static int iTCPSessMax = 20; /* max number of sessions */
+static int iStrmDrvrMode = 0; /* mode for stream driver, driver-dependent (0 mostly means plain tcp) */
+static uchar *pszLstnPortFileName = NULL;
+static uchar *pszStrmDrvrAuthMode = NULL; /* authentication mode to use */
+static uchar *pszInputName = NULL; /* value for inputname property, NULL is OK and handled by core engine */
+
+
+/* callbacks */
+/* this shall go into a specific ACL module! */
+static int
+isPermittedHost(struct sockaddr __attribute__((unused)) *addr, char __attribute__((unused)) *fromHostFQDN,
+		void __attribute__((unused)) *pUsrSrv, void __attribute__((unused)) *pUsrSess)
+{
+	return 1;	/* TODO: implement ACLs ... or via some other way? */
+}
+
+
+static rsRetVal
+doOpenLstnSocks(tcpsrv_t *pSrv)
+{
+	ISOBJ_TYPE_assert(pSrv, tcpsrv);
+	dbgprintf("in imdiag doOpenLstnSocks\n");
+	return tcpsrv.create_tcp_socket(pSrv);
+}
+
+
+static rsRetVal
+doRcvData(tcps_sess_t *pSess, char *buf, size_t lenBuf, ssize_t *piLenRcvd, int *oserr)
+{
+	assert(pSess != NULL);
+	assert(piLenRcvd != NULL);
+
+	*piLenRcvd = lenBuf;
+	return netstrm.Rcv(pSess->pStrm, (uchar*) buf, piLenRcvd, oserr);
+}
+
+static rsRetVal
+onRegularClose(tcps_sess_t *pSess)
+{
+	DEFiRet;
+	assert(pSess != NULL);
+
+	/* process any incomplete frames left over */
+	tcps_sess.PrepareClose(pSess);
+	/* Session closed */
+	tcps_sess.Close(pSess);
+	RETiRet;
+}
+
+
+static rsRetVal
+onErrClose(tcps_sess_t *pSess)
+{
+	DEFiRet;
+	assert(pSess != NULL);
+
+	tcps_sess.Close(pSess);
+	RETiRet;
+}
+
+/* ------------------------------ end callbacks ------------------------------ */
+
+
+/* get the first word delimited by space from a given string. The pointer is
+ * advanced to after the word. Any leading spaces are discarded. If the
+ * output buffer is too small, parsing ends on buffer full condition.
+ * An empty buffer is returned if there is no more data inside the string.
+ * rgerhards, 2009-05-27
+ */
+#define TO_LOWERCASE	1
+#define NO_MODIFY	0
+static void
+getFirstWord(uchar **ppszSrc, uchar *pszBuf, size_t lenBuf, int options)
+{
+	uchar c;
+	uchar *pszSrc = *ppszSrc;
+
+	while(*pszSrc && *pszSrc == ' ')
+		++pszSrc; /* skip to first non-space */
+
+	while(*pszSrc && *pszSrc != ' ' && lenBuf > 1) {
+		c = *pszSrc++;
+		if(options & TO_LOWERCASE)
+			c = tolower(c);
+		*pszBuf++ = c;
+		lenBuf--;
+	}
+
+	*pszBuf = '\0';
+	*ppszSrc = pszSrc;
+}
+
+
+/* send a response back to the originator
+ * rgerhards, 2009-05-27
+ */
+static rsRetVal __attribute__((format(printf, 2, 3)))
+sendResponse(tcps_sess_t *pSess, const char *const __restrict__ fmt, ...)
+{
+	va_list ap;
+	ssize_t len;
+	uchar buf[1024];
+	DEFiRet;
+
+	va_start(ap, fmt);
+	len = vsnprintf((char*)buf, sizeof(buf), fmt, ap);
+	va_end(ap);
+	CHKiRet(netstrm.Send(pSess->pStrm, buf, &len));
+
+finalize_it:
+	RETiRet;
+}
+
+/* submit a generated numeric-suffix message to the rsyslog core
+ */
+static rsRetVal
+doInjectMsg(uchar *szMsg, ratelimit_t *ratelimiter)
+{
+	smsg_t *pMsg;
+	struct syslogTime stTime;
+	time_t ttGenTime;
+	DEFiRet;
+
+	datetime.getCurrTime(&stTime, &ttGenTime, TIME_IN_LOCALTIME);
+	/* we now create our own message object and submit it to the queue */
+	CHKiRet(msgConstructWithTime(&pMsg, &stTime, ttGenTime));
+	MsgSetRawMsg(pMsg, (char*) szMsg, ustrlen(szMsg));
+	MsgSetInputName(pMsg, pInputName);
+	MsgSetFlowControlType(pMsg, injectmsgDelayMode);
+	pMsg->msgFlags  = NEEDS_PARSING | PARSE_HOSTNAME;
+	MsgSetRcvFrom(pMsg, pRcvDummy);
+	CHKiRet(MsgSetRcvFromIP(pMsg, pRcvIPDummy));
+	CHKiRet(ratelimitAddMsg(ratelimiter, NULL, pMsg));
+
+finalize_it:
+	RETiRet;
+}
+
+/* submit a generated numeric-suffix message to the rsyslog core
+ */
+static rsRetVal
+doInjectNumericSuffixMsg(int iNum, ratelimit_t *ratelimiter)
+{
+	uchar szMsg[1024];
+	DEFiRet;
+	snprintf((char*)szMsg, sizeof(szMsg)/sizeof(uchar),
+		"<167>Mar  1 01:00:00 172.20.245.8 tag msgnum:%8.8d:", iNum);
+	iRet = doInjectMsg(szMsg, ratelimiter);
+	RETiRet;
+}
+
+/* This function injects messages. Command format:
+ * injectmsg <fromnbr> <number-of-messages>
+ * rgerhards, 2009-05-27
+ */
+static rsRetVal
+injectMsg(uchar *pszCmd, tcps_sess_t *pSess)
+{
+	uchar wordBuf[1024];
+	int iFrom, nMsgs;
+	uchar *literalMsg;
+	int i;
+	ratelimit_t *ratelimit = NULL;
+	DEFiRet;
+
+	literalMsg = NULL;
+
+	memset(wordBuf, 0, sizeof(wordBuf));
+	CHKiRet(ratelimitNew(&ratelimit, "imdiag", "injectmsg"));
+	/* we do not check errors here! */
+	getFirstWord(&pszCmd, wordBuf, sizeof(wordBuf), TO_LOWERCASE);
+	if (ustrcmp(UCHAR_CONSTANT("literal"), wordBuf) == 0) {
+		/* user has provided content for a message */
+		++pszCmd; /* ignore following space */
+		CHKiRet(doInjectMsg(pszCmd, ratelimit));
+		nMsgs = 1;
+	} else { /* assume 2 args, (from_idx, count) */
+		iFrom = atoi((char*)wordBuf);
+		getFirstWord(&pszCmd, wordBuf, sizeof(wordBuf), TO_LOWERCASE);
+		nMsgs = atoi((char*)wordBuf);
+		for(i = 0 ; i < nMsgs ; ++i) {
+			CHKiRet(doInjectNumericSuffixMsg(i + iFrom, ratelimit));
+		}
+	}
+	CHKiRet(sendResponse(pSess, "%d messages injected\n", nMsgs));
+
+	DBGPRINTF("imdiag: %d messages injected\n", nMsgs);
+
+finalize_it:
+	if(ratelimit != NULL)
+		ratelimitDestruct(ratelimit);
+	free(literalMsg);
+	RETiRet;
+}
+
+
+/* This function waits until all queues are drained (size = 0)
+ * To make sure it really is drained, we check multiple times. Otherwise we
+ * may just see races. Note: it is important to ensure that the size
+ * is zero multiple times in succession. Otherwise, we may just accidently
+ * hit a situation where the queue isn't filled for a while (we have seen
+ * this in practice, see https://github.com/rsyslog/rsyslog/issues/688).
+ * Note: until 2014--07-13, this checked just the main queue. However,
+ * the testbench was the sole user and checking all queues makes much more
+ * sense. So we change function semantics instead of carrying the old
+ * semantics over and crafting a new function. -- rgerhards
+ */
+static rsRetVal
+waitMainQEmpty(tcps_sess_t *pSess)
+{
+	int iPrint = 0;
+	int iPrintVerbosity = 500; // 500 default
+	int nempty = 0;
+	static unsigned lastOverallQueueSize = 1;
+	DEFiRet;
+
+	while(1) {
+		processImInternal();
+		const unsigned OverallQueueSize = PREFER_FETCH_32BIT(iOverallQueueSize);
+		if(OverallQueueSize == 0) {
+			++nempty;
+		} else {
+			if(OverallQueueSize > 500) {
+				/* do a bit of extra sleep to not poll too frequently */
+				srSleep(0, (OverallQueueSize > 2000) ? 900000 : 100000);
+			}
+			nempty = 0;
+		}
+		if(dbgTimeoutToStderr) { /* we abuse this setting a bit ;-) */
+			if(OverallQueueSize != lastOverallQueueSize) {
+				fprintf(stderr, "imdiag: wait q_empty: qsize %d nempty %d\n",
+					OverallQueueSize, nempty);
+				lastOverallQueueSize = OverallQueueSize;
+			}
+		}
+		if(nempty > max_empty_checks)
+			break;
+		if(iPrint++ % iPrintVerbosity == 0)
+			DBGPRINTF("imdiag sleeping, wait queues drain, "
+				"curr size %d, nempty %d\n",
+				OverallQueueSize, nempty);
+		srSleep(0,100000);/* wait a little bit */
+	}
+
+	CHKiRet(sendResponse(pSess, "mainqueue empty\n"));
+	DBGPRINTF("imdiag: mainqueue empty\n");
+
+finalize_it:
+	RETiRet;
+}
+
+static rsRetVal
+awaitLookupTableReload(tcps_sess_t *pSess)
+{
+	DEFiRet;
+
+	while(1) {
+		if(lookupPendingReloadCount() == 0) {
+			break;
+		}
+		srSleep(0,500000);
+	}
+
+	CHKiRet(sendResponse(pSess, "no pending lookup-table reloads found\n"));
+	DBGPRINTF("imdiag: no pending lookup-table reloads found\n");
+
+finalize_it:
+	RETiRet;
+}
+
+static rsRetVal
+awaitHUPComplete(tcps_sess_t *pSess)
+{
+	const int max_tries = 10;
+	const int ms_to_sleep = 50;
+	const char *return_msg;
+	int b_saw_HUP = 0;
+	int tries = max_tries;
+	unsigned actual_tries = 0;
+	DEFiRet;
+
+	while(tries > 0) {
+		++actual_tries;
+		if(get_bHadHUP() == 1) {
+			tries = max_tries;
+			b_saw_HUP = 1;
+		} else {
+			--tries;
+		}
+		srSleep(0, ms_to_sleep * 50);
+	}
+
+	if(b_saw_HUP) {
+		return_msg = "seen HUP request, looks like it has completed";
+	} else {
+		return_msg = "timeout - looks like_HUP has completed";
+	}
+	CHKiRet(sendResponse(pSess, "%s [%d tries]\n", return_msg, actual_tries));
+	DBGPRINTF("imdiag: %s\n", return_msg);
+
+finalize_it:
+	RETiRet;
+}
+
+static rsRetVal
+enableDebug(tcps_sess_t *pSess)
+{
+	DEFiRet;
+
+	Debug = DEBUG_FULL;
+	debugging_on = 1;
+	dbgprintf("Note: debug turned on via imdiag\n");
+
+	CHKiRet(sendResponse(pSess, "debug enabled\n"));
+
+finalize_it:
+	RETiRet;
+}
+
+static void
+imdiag_statsReadCallback(statsobj_t __attribute__((unused)) *const ignore_stats,
+	void __attribute__((unused)) *const ignore_ctx)
+{
+	long long waitStartTimeMs = currentTimeMills();
+	sem_wait(&statsReportingBlocker);
+	long delta = currentTimeMills() - waitStartTimeMs;
+	if ((int)ATOMIC_DEC_AND_FETCH(&allowOnlyOnce, &mutAllowOnlyOnce) < 0) {
+		sem_post(&statsReportingBlocker);
+	} else {
+		LogError(0, RS_RET_OK, "imdiag(stats-read-callback): current stats-reporting "
+						"cycle will proceed now, next reporting cycle will again be blocked");
+	}
+
+	if (pthread_mutex_lock(&mutStatsReporterWatch) == 0) {
+		statsReported = 1;
+		pthread_cond_signal(&statsReporterWatch);
+		pthread_mutex_unlock(&mutStatsReporterWatch);
+	}
+
+	if (delta > 0) {
+		STATSCOUNTER_ADD(actualArtificialDelayMs, mutActualArtificialDelayMs, delta);
+	}
+}
+
+static rsRetVal
+blockStatsReporting(tcps_sess_t *pSess) {
+	DEFiRet;
+
+	sem_wait(&statsReportingBlocker);
+	CHKiConcCtrl(pthread_mutex_lock(&mutStatsReporterWatch));
+	statsReported = 0;
+	CHKiConcCtrl(pthread_mutex_unlock(&mutStatsReporterWatch));
+	ATOMIC_STORE_0_TO_INT(&allowOnlyOnce, &mutAllowOnlyOnce);
+	statsReportingBlockStartTimeMs = currentTimeMills();
+	LogError(0, RS_RET_OK, "imdiag: blocked stats reporting");
+	CHKiRet(sendResponse(pSess, "next stats reporting call will be blocked\n"));
+
+finalize_it:
+	if (iRet != RS_RET_OK) {
+		LogError(0, iRet, "imdiag: block-stats-reporting wasn't successful");
+		CHKiRet(sendResponse(pSess, "imdiag::error something went wrong\n"));
+	}
+	RETiRet;
+}
+
+static rsRetVal
+awaitStatsReport(uchar *pszCmd, tcps_sess_t *pSess) {
+	uchar subCmd[1024];
+	int blockAgain = 0;
+	DEFiRet;
+
+	memset(subCmd, 0, sizeof(subCmd));
+	getFirstWord(&pszCmd, subCmd, sizeof(subCmd), TO_LOWERCASE);
+	blockAgain = (ustrcmp(UCHAR_CONSTANT("block_again"), subCmd) == 0);
+	if (statsReportingBlockStartTimeMs > 0) {
+		long delta = currentTimeMills() - statsReportingBlockStartTimeMs;
+		if (blockAgain) {
+			ATOMIC_STORE_1_TO_INT(&allowOnlyOnce, &mutAllowOnlyOnce);
+			LogError(0, RS_RET_OK, "imdiag: un-blocking ONLY the next cycle of stats reporting");
+		} else {
+			statsReportingBlockStartTimeMs = 0;
+			LogError(0, RS_RET_OK, "imdiag: un-blocking stats reporting");
+		}
+		sem_post(&statsReportingBlocker);
+		LogError(0, RS_RET_OK, "imdiag: stats reporting unblocked");
+		STATSCOUNTER_ADD(potentialArtificialDelayMs, mutPotentialArtificialDelayMs, delta);
+		STATSCOUNTER_INC(delayInvocationCount, mutDelayInvocationCount);
+		LogError(0, RS_RET_OK, "imdiag: will now await next reporting cycle");
+		CHKiConcCtrl(pthread_mutex_lock(&mutStatsReporterWatch));
+		while (! statsReported) {
+			CHKiConcCtrl(pthread_cond_wait(&statsReporterWatch, &mutStatsReporterWatch));
+		}
+		statsReported = 0;
+		CHKiConcCtrl(pthread_mutex_unlock(&mutStatsReporterWatch));
+		if (blockAgain) {
+			statsReportingBlockStartTimeMs = currentTimeMills();
+		}
+		LogError(0, RS_RET_OK, "imdiag: stats were reported, wait complete, returning");
+		CHKiRet(sendResponse(pSess, "stats reporting was unblocked\n"));
+	} else {
+		CHKiRet(sendResponse(pSess, "imdiag::error : stats reporting was not blocked, bug?\n"));
+	}
+
+finalize_it:
+	if (iRet != RS_RET_OK) {
+		LogError(0, iRet, "imdiag: stats-reporting unblock + await-run wasn't successfully completed");
+		CHKiRet(sendResponse(pSess, "imdiag::error something went wrong\n"));
+	}
+	RETiRet;
+}
+
+/* Function to handle received messages. This is our core function!
+ * rgerhards, 2009-05-24
+ */
+static rsRetVal ATTR_NONNULL()
+OnMsgReceived(tcps_sess_t *const pSess, uchar *const pRcv, const int iLenMsg)
+{
+	uchar *pszMsg;
+	uchar *pToFree = NULL;
+	uchar cmdBuf[1024];
+	DEFiRet;
+
+	assert(pSess != NULL);
+	assert(pRcv != NULL);
+
+	/* NOTE: pRcv is NOT a C-String but rather an array of characters
+	 * WITHOUT a termination \0 char. So we need to convert it to one
+	 * before proceeding.
+	 */
+	CHKmalloc(pszMsg = calloc(1, iLenMsg + 1));
+	pToFree = pszMsg;
+	memcpy(pszMsg, pRcv, iLenMsg);
+	pszMsg[iLenMsg] = '\0';
+
+	memset(cmdBuf, 0, sizeof(cmdBuf)); /* keep valgrind happy */
+	getFirstWord(&pszMsg, cmdBuf, sizeof(cmdBuf), TO_LOWERCASE);
+
+	dbgprintf("imdiag received command '%s'\n", cmdBuf);
+	if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("getmainmsgqueuesize"))) {
+		CHKiRet(sendResponse(pSess, "%d\n", iOverallQueueSize));
+		DBGPRINTF("imdiag: %d messages in main queue\n", iOverallQueueSize);
+	} else if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("waitmainqueueempty"))) {
+		CHKiRet(waitMainQEmpty(pSess));
+	} else if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("awaitlookuptablereload"))) {
+		CHKiRet(awaitLookupTableReload(pSess));
+	} else if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("injectmsg"))) {
+		CHKiRet(injectMsg(pszMsg, pSess));
+	} else if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("blockstatsreporting"))) {
+		CHKiRet(blockStatsReporting(pSess));
+	} else if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("awaitstatsreport"))) {
+		CHKiRet(awaitStatsReport(pszMsg, pSess));
+	} else if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("awaithupcomplete"))) {
+		CHKiRet(awaitHUPComplete(pSess));
+	} else if(!ustrcmp(cmdBuf, UCHAR_CONSTANT("enabledebug"))) {
+		CHKiRet(enableDebug(pSess));
+	} else {
+		dbgprintf("imdiag unkown command '%s'\n", cmdBuf);
+		CHKiRet(sendResponse(pSess, "unkown command '%s'\n", cmdBuf));
+	}
+
+finalize_it:
+	free(pToFree);
+	RETiRet;
+}
+
+
+/* set permitted peer -- rgerhards, 2008-05-19
+ */
+static rsRetVal
+setPermittedPeer(void __attribute__((unused)) *pVal, uchar *pszID)
+{
+	DEFiRet;
+	CHKiRet(net.AddPermittedPeer(&pPermPeersRoot, pszID));
+	free(pszID); /* no longer needed, but we need to free as of interface def */
+finalize_it:
+	RETiRet;
+}
+
+
+static rsRetVal
+setInjectDelayMode(void __attribute__((unused)) *pVal, uchar *const pszMode)
+{
+	DEFiRet;
+
+	if(!strcasecmp((char*)pszMode, "no")) {
+		injectmsgDelayMode = eFLOWCTL_NO_DELAY;
+	} else if(!strcasecmp((char*)pszMode, "light")) {
+		injectmsgDelayMode = eFLOWCTL_LIGHT_DELAY;
+	} else if(!strcasecmp((char*)pszMode, "full")) {
+		injectmsgDelayMode = eFLOWCTL_FULL_DELAY;
+	} else {
+		LogError(0, RS_RET_PARAM_ERROR,
+			"imdiag: invalid imdiagInjectDelayMode '%s' - ignored", pszMode);
+	}
+	free(pszMode);
+	RETiRet;
+}
+
+
+static rsRetVal
+addTCPListener(void __attribute__((unused)) *pVal, uchar *pNewVal)
+{
+	tcpLstnParams_t *cnf_params = NULL;
+	DEFiRet;
+
+	if(pOurTcpsrv != NULL) {
+		LogError(0, NO_ERRCODE, "imdiag: only a single listener is supported, "
+			"trying to add a second");
+		ABORT_FINALIZE(RS_RET_ERR);
+	}
+	CHKmalloc(cnf_params = (tcpLstnParams_t*) calloc(1, sizeof(tcpLstnParams_t)));
+	CHKiRet(tcpsrv.Construct(&pOurTcpsrv));
+	CHKiRet(tcpsrv.SetSessMax(pOurTcpsrv, iTCPSessMax));
+	CHKiRet(tcpsrv.SetCBIsPermittedHost(pOurTcpsrv, isPermittedHost));
+	CHKiRet(tcpsrv.SetCBRcvData(pOurTcpsrv, doRcvData));
+	CHKiRet(tcpsrv.SetCBOpenLstnSocks(pOurTcpsrv, doOpenLstnSocks));
+	CHKiRet(tcpsrv.SetCBOnRegularClose(pOurTcpsrv, onRegularClose));
+	CHKiRet(tcpsrv.SetCBOnErrClose(pOurTcpsrv, onErrClose));
+	CHKiRet(tcpsrv.SetDrvrMode(pOurTcpsrv, iStrmDrvrMode));
+	CHKiRet(tcpsrv.SetOnMsgReceive(pOurTcpsrv, OnMsgReceived));
+	/* now set optional params, but only if they were actually configured */
+	if(pszStrmDrvrAuthMode != NULL) {
+		CHKiRet(tcpsrv.SetDrvrAuthMode(pOurTcpsrv, pszStrmDrvrAuthMode));
+	}
+	if(pPermPeersRoot != NULL) {
+		CHKiRet(tcpsrv.SetDrvrPermPeers(pOurTcpsrv, pPermPeersRoot));
+	}
+
+	/* initialized, now add socket */
+	CHKiRet(tcpsrv.SetInputName(pOurTcpsrv, cnf_params, pszInputName == NULL ?
+						UCHAR_CONSTANT("imdiag") : pszInputName));
+	CHKiRet(tcpsrv.SetOrigin(pOurTcpsrv, (uchar*)"imdiag"));
+	/* we support octect-counted frame (constant 1 below) */
+	cnf_params->pszPort = pNewVal;
+	cnf_params->bSuppOctetFram = 1;
+	CHKmalloc(cnf_params->pszLstnPortFileName = (const uchar*) strdup((const char*)pszLstnPortFileName));
+	tcpsrv.configureTCPListen(pOurTcpsrv, cnf_params);
+	cnf_params = NULL;
+
+finalize_it:
+	if(iRet != RS_RET_OK) {
+		LogError(0, NO_ERRCODE, "error %d trying to add listener", iRet);
+		if(pOurTcpsrv != NULL)
+			tcpsrv.Destruct(&pOurTcpsrv);
+	}
+	free(cnf_params);
+	RETiRet;
+}
+
+
+static void *
+timeoutGuard(ATTR_UNUSED void *arg)
+{
+	assert(abortTimeout != -1);
+	sigset_t sigSet;
+	time_t strtTO;
+	time_t endTO;
+
+	/* block all signals except SIGTTIN and SIGSEGV */
+	sigfillset(&sigSet);
+	sigdelset(&sigSet, SIGSEGV);
+	pthread_sigmask(SIG_BLOCK, &sigSet, NULL);
+
+	dbgprintf("timeoutGuard: timeout %d seconds, time %lld\n", abortTimeout, (long long) time(NULL));
+
+	time(&strtTO);
+	endTO = strtTO + abortTimeout;
+
+	while(1) {
+		int to = endTO - time(NULL);
+		dbgprintf("timeoutGuard: sleep timeout %d seconds\n", to);
+		if(to > 0) {
+			srSleep(to, 0);
+		}
+		if(time(NULL) < endTO) {
+			dbgprintf("timeoutGuard: spurios wakeup, going back to sleep, time: %lld\n",
+				(long long) time(NULL));
+		} else {
+			break;
+		}
+	}
+	dbgprintf("timeoutGuard: sleep expired, aborting\n");
+	/* note: we use fprintf to stderr intentionally! */
+
+	fprintf(stderr, "timeoutGuard: rsyslog still active after expiry of guard "
+		"period (strtTO %lld, endTO %lld, time now %lld, diff %lld), pid %d - initiating abort()\n",
+	(long long) strtTO, (long long) endTO, (long long) time(NULL), (long long) (time(NULL) - strtTO),
+	(int) glblGetOurPid());
+	fflush(stderr);
+	abort();
+}
+
+
+static rsRetVal
+setAbortTimeout(void __attribute__((unused)) *pVal, int timeout)
+{
+	DEFiRet;
+
+	if(abortTimeout != -1) {
+		LogError(0, NO_ERRCODE, "imdiag: abort timeout already set -"
+			"ignoring 2nd+ request");
+		ABORT_FINALIZE(RS_RET_ERR);
+	}
+	if(timeout <= 0) {
+		LogError(0, NO_ERRCODE, "imdiag: $IMDiagAbortTimeout must be greater "
+			"than 0 - ignored");
+		ABORT_FINALIZE(RS_RET_ERR);
+	}
+	abortTimeout = timeout;
+	const int iState = pthread_create(&timeoutGuard_thrd, NULL, timeoutGuard, NULL);
+	if(iState != 0) {
+		LogError(iState, NO_ERRCODE, "imdiag: error enabling timeoutGuard thread -"
+			"not guarding against system hang");
+		ABORT_FINALIZE(RS_RET_ERR);
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+
+#if 0 /* can be used to integrate into new config system */
+BEGINbeginCnfLoad
+CODESTARTbeginCnfLoad
+ENDbeginCnfLoad
+
+
+BEGINendCnfLoad
+CODESTARTendCnfLoad
+ENDendCnfLoad
+
+
+BEGINcheckCnf
+CODESTARTcheckCnf
+ENDcheckCnf
+
+
+BEGINactivateCnf
+CODESTARTactivateCnf
+ENDactivateCnf
+
+
+BEGINfreeCnf
+CODESTARTfreeCnf
+ENDfreeCnf
+#endif
+
+/* This function is called to gather input.
+ */
+BEGINrunInput
+CODESTARTrunInput
+	CHKiRet(tcpsrv.ConstructFinalize(pOurTcpsrv));
+	iRet = tcpsrv.Run(pOurTcpsrv);
+finalize_it:
+ENDrunInput
+
+
+/* initialize and return if will run or not */
+BEGINwillRun
+CODESTARTwillRun
+	/* first apply some config settings */
+	if(pOurTcpsrv == NULL)
+		ABORT_FINALIZE(RS_RET_NO_RUN);
+	/* we need to create the inputName property (only once during our lifetime) */
+	CHKiRet(prop.Construct(&pInputName));
+	CHKiRet(prop.SetString(pInputName, UCHAR_CONSTANT("imdiag"), sizeof("imdiag") - 1));
+	CHKiRet(prop.ConstructFinalize(pInputName));
+
+	CHKiRet(prop.Construct(&pRcvDummy));
+	CHKiRet(prop.SetString(pRcvDummy, UCHAR_CONSTANT("127.0.0.1"), sizeof("127.0.0.1") - 1));
+	CHKiRet(prop.ConstructFinalize(pRcvDummy));
+
+	CHKiRet(prop.Construct(&pRcvIPDummy));
+	CHKiRet(prop.SetString(pRcvIPDummy, UCHAR_CONSTANT("127.0.0.1"), sizeof("127.0.0.1") - 1));
+	CHKiRet(prop.ConstructFinalize(pRcvIPDummy));
+
+finalize_it:
+ENDwillRun
+
+
+BEGINafterRun
+CODESTARTafterRun
+	if(pInputName != NULL)
+		prop.Destruct(&pInputName);
+	if(pRcvDummy != NULL)
+		prop.Destruct(&pRcvDummy);
+	if(pRcvIPDummy != NULL)
+		prop.Destruct(&pRcvIPDummy);
+ENDafterRun
+
+
+BEGINmodExit
+CODESTARTmodExit
+	if(pOurTcpsrv != NULL)
+		iRet = tcpsrv.Destruct(&pOurTcpsrv);
+
+	if(pPermPeersRoot != NULL) {
+		net.DestructPermittedPeers(&pPermPeersRoot);
+	}
+
+	/* free some globals to keep valgrind happy */
+	free(pszInputName);
+	free(pszLstnPortFileName);
+	free(pszStrmDrvrAuthMode);
+
+	statsobj.Destruct(&diagStats);
+	sem_destroy(&statsReportingBlocker);
+	DESTROY_ATOMIC_HELPER_MUT(mutAllowOnlyOnce);
+	pthread_cond_destroy(&statsReporterWatch);
+	pthread_mutex_destroy(&mutStatsReporterWatch);
+
+	/* release objects we used */
+	objRelease(net, LM_NET_FILENAME);
+	objRelease(netstrm, LM_NETSTRMS_FILENAME);
+	objRelease(tcps_sess, LM_TCPSRV_FILENAME);
+	objRelease(tcpsrv, LM_TCPSRV_FILENAME);
+	objRelease(datetime, CORE_COMPONENT);
+	objRelease(prop, CORE_COMPONENT);
+	objRelease(statsobj, CORE_COMPONENT);
+
+	/* clean up timeoutGuard if active */
+	if(abortTimeout != -1) {
+		int r = pthread_cancel(timeoutGuard_thrd);
+		if(r == 0) {
+			void *dummy;
+			pthread_join(timeoutGuard_thrd, &dummy);
+		}
+	}
+ENDmodExit
+
+
+static rsRetVal
+resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal)
+{
+	iTCPSessMax = 200;
+	iStrmDrvrMode = 0;
+	free(pszInputName);
+	free(pszLstnPortFileName);
+	pszLstnPortFileName = NULL;
+	if(pszStrmDrvrAuthMode != NULL) {
+		free(pszStrmDrvrAuthMode);
+		pszStrmDrvrAuthMode = NULL;
+	}
+	return RS_RET_OK;
+}
+
+
+BEGINisCompatibleWithFeature
+CODESTARTisCompatibleWithFeature
+	if(eFeat == sFEATURENonCancelInputTermination)
+		iRet = RS_RET_OK;
+ENDisCompatibleWithFeature
+
+
+BEGINqueryEtryPt
+CODESTARTqueryEtryPt
+CODEqueryEtryPt_STD_IMOD_QUERIES
+CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES
+ENDqueryEtryPt
+
+
+BEGINmodInit()
+CODESTARTmodInit
+	*ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
+CODEmodInit_QueryRegCFSLineHdlr
+	pOurTcpsrv = NULL;
+	/* request objects we use */
+	CHKiRet(objUse(net, LM_NET_FILENAME));
+	CHKiRet(objUse(netstrm, LM_NETSTRMS_FILENAME));
+	CHKiRet(objUse(tcps_sess, LM_TCPSRV_FILENAME));
+	CHKiRet(objUse(tcpsrv, LM_TCPSRV_FILENAME));
+	CHKiRet(objUse(datetime, CORE_COMPONENT));
+	CHKiRet(objUse(prop, CORE_COMPONENT));
+	CHKiRet(objUse(statsobj, CORE_COMPONENT));
+
+	const char *ci_max_empty_checks = getenv("CI_SHUTDOWN_QUEUE_EMPTY_CHECKS");
+	if(ci_max_empty_checks != NULL) {
+		int n = atoi(ci_max_empty_checks);
+		if(n > 200) {
+			LogError(0, RS_RET_PARAM_ERROR, "env var CI_SHUTDOWN_QUEUE_EMPTY_CHECKS has "
+				"value over 200, which is the maximum - capped to 200");
+			n = 200;
+		}
+		if(n > 0) {
+			max_empty_checks = n;
+		} else {
+			LogError(0, RS_RET_PARAM_ERROR, "env var CI_SHUTDOWN_QUEUE_EMPTY_CHECKS has "
+				"value below 1, ignored; using default instead");
+		}
+		fprintf(stderr, "rsyslogd: info: imdiag does %d empty checks due to "
+			"CI_SHUTDOWN_QUEUE_EMPTY_CHECKS\n", max_empty_checks);
+	}
+
+	/* register config file handlers */
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("imdiagaborttimeout"), 0, eCmdHdlrInt,
+				   setAbortTimeout, NULL, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("imdiagserverrun"), 0, eCmdHdlrGetWord,
+				   addTCPListener, NULL, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("imdiaginjectdelaymode"), 0, eCmdHdlrGetWord,
+				   setInjectDelayMode, NULL, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("imdiagmaxsessions"), 0, eCmdHdlrInt,
+				   NULL, &iTCPSessMax, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("imdiagserverstreamdrivermode"), 0,
+				   eCmdHdlrInt, NULL, &iStrmDrvrMode, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("imdiaglistenportfilename"), 0,
+				   eCmdHdlrGetWord, NULL, &pszLstnPortFileName, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("imdiagserverstreamdriverauthmode"), 0,
+				   eCmdHdlrGetWord, NULL, &pszStrmDrvrAuthMode, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("imdiagserverstreamdriverpermittedpeer"), 0,
+				   eCmdHdlrGetWord, setPermittedPeer, NULL, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("imdiagserverinputname"), 0,
+				   eCmdHdlrGetWord, NULL, &pszInputName, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("resetconfigvariables"), 1, eCmdHdlrCustomHandler,
+							   resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID));
+
+	sem_init(&statsReportingBlocker, 0, 1);
+	INIT_ATOMIC_HELPER_MUT(mutAllowOnlyOnce);
+	CHKiConcCtrl(pthread_mutex_init(&mutStatsReporterWatch, NULL));
+	CHKiConcCtrl(pthread_cond_init(&statsReporterWatch, NULL));
+
+	CHKiRet(statsobj.Construct(&diagStats));
+	CHKiRet(statsobj.SetName(diagStats, UCHAR_CONSTANT("imdiag-stats-reporting-controller")));
+	CHKiRet(statsobj.SetOrigin(diagStats, UCHAR_CONSTANT("imdiag")));
+	statsobj.SetStatsObjFlags(diagStats, STATSOBJ_FLAG_DO_PREPEND);
+	STATSCOUNTER_INIT(potentialArtificialDelayMs, mutPotentialArtificialDelayMs);
+	CHKiRet(statsobj.AddCounter(diagStats, UCHAR_CONSTANT("potentialTotalArtificialDelayInMs"),
+						ctrType_IntCtr, CTR_FLAG_NONE, &potentialArtificialDelayMs));
+	STATSCOUNTER_INIT(actualArtificialDelayMs, mutActualArtificialDelayMs);
+	CHKiRet(statsobj.AddCounter(diagStats, UCHAR_CONSTANT("actualTotalArtificialDelayInMs"),
+							ctrType_IntCtr, CTR_FLAG_NONE, &actualArtificialDelayMs));
+	STATSCOUNTER_INIT(delayInvocationCount, mutDelayInvocationCount);
+	CHKiRet(statsobj.AddCounter(diagStats, UCHAR_CONSTANT("delayInvocationCount"),
+			ctrType_IntCtr, CTR_FLAG_NONE, &delayInvocationCount));
+	CHKiRet(statsobj.SetReadNotifier(diagStats, imdiag_statsReadCallback, NULL));
+	CHKiRet(statsobj.ConstructFinalize(diagStats));
+ENDmodInit