path: root/tools/ompipe.c

/* ompipe.c
 * This is the implementation of the build-in pipe output module.
 * Note that this module stems back to the "old" (4.4.2 and below)
 * omfile. There were some issues with the new omfile code and pipes
 * (namely in regard to xconsole), so we took out the pipe code and moved
 * that to a separate module. That a) immediately solves the issue for a
 * less common use case and probably makes it much easier to enhance
 * file and pipe support (now independently) in the future (we always
 * needed to think about pipes in omfile so far, what we now no longer
 * need to, hopefully resulting in reduction of complexity).
 *
 * NOTE: read comments in module-template.h to understand how this pipe
 *       works!
 *
 * Copyright 2007-2018 Rainer Gerhards and 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 <stdarg.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/file.h>

#include "rsyslog.h"
#include "syslogd.h"
#include "syslogd-types.h"
#include "srUtils.h"
#include "template.h"
#include "ompipe.h"
#include "omfile.h" /* for dirty trick: access to $ActionFileDefaultTemplate value */
#include "cfsysline.h"
#include "module-template.h"
#include "conf.h"
#include "errmsg.h"

MODULE_TYPE_OUTPUT
MODULE_TYPE_NOKEEP
MODULE_CNFNAME("ompipe")

/* internal structures
 */
DEF_OMOD_STATIC_DATA


typedef struct _instanceData {
	uchar	*pipe;	/* pipe or template name (display only) */
	uchar	*tplName;       /* format template to use */
	short	fd;		/* pipe descriptor for (current) pipe */
	pthread_mutex_t mutWrite; /* guard against multiple instances writing to same pipe */
	sbool	bHadError;	/* did we already have/report an error on this pipe? */
	sbool	bTryResumeReopen;	/* should we attempt to reopen the pipe on action resume? */
} instanceData;

typedef struct wrkrInstanceData {
	instanceData *pData;
} wrkrInstanceData_t;

typedef struct configSettings_s {
	EMPTY_STRUCT
} configSettings_t;
static configSettings_t __attribute__((unused)) cs;

/* tables for interfacing with the v6 config system */
/* module-global parameters */
static struct cnfparamdescr modpdescr[] = {
	{ "template", eCmdHdlrGetWord, 0 },
};
static struct cnfparamblk modpblk =
	{ CNFPARAMBLK_VERSION,
	  sizeof(modpdescr)/sizeof(struct cnfparamdescr),
	  modpdescr
	};

/* action (instance) parameters */
static struct cnfparamdescr actpdescr[] = {
	{ "pipe", eCmdHdlrString, CNFPARAM_REQUIRED },
	{ "template", eCmdHdlrGetWord, 0 },
	{ "tryResumeReopen", eCmdHdlrBinary, 0 },
};
static struct cnfparamblk actpblk =
	{ CNFPARAMBLK_VERSION,
	  sizeof(actpdescr)/sizeof(struct cnfparamdescr),
	  actpdescr
	};

struct modConfData_s {
	rsconf_t *pConf;	/* our overall config object */
	uchar 	*tplName;	/* default template */
};

static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */
static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current exec process */

/* this function gets the default template */
static uchar*
getDfltTpl(void)
{
	if(loadModConf != NULL && loadModConf->tplName != NULL)
		return loadModConf->tplName;
	else
		return (uchar*)"RSYSLOG_FileFormat";
}


BEGINinitConfVars		/* (re)set config variables to default values */
CODESTARTinitConfVars
ENDinitConfVars


BEGINisCompatibleWithFeature
CODESTARTisCompatibleWithFeature
	if(eFeat == sFEATURERepeatedMsgReduction)
		iRet = RS_RET_OK;
ENDisCompatibleWithFeature


BEGINdbgPrintInstInfo
CODESTARTdbgPrintInstInfo
	dbgprintf("pipe %s", pData->pipe);
	if (pData->fd == -1)
		dbgprintf(" (unused)");
ENDdbgPrintInstInfo


/* This is now shared code for all types of files. It simply prepares
 * pipe access, which, among others, means the the pipe wil be opened
 * and any directories in between will be created (based on config, of
 * course). -- rgerhards, 2008-10-22
 * changed to iRet interface - 2009-03-19
 */
static rsRetVal
preparePipe(instanceData *pData)
{
	DEFiRet;
	pData->fd = open((char*) pData->pipe, O_RDWR|O_NONBLOCK|O_CLOEXEC);
	if(pData->fd < 0 ) {
		pData->fd = -1;
		if(!pData->bHadError) {
			LogError(errno, RS_RET_NO_FILE_ACCESS, "Could not open output pipe '%s':",
				        pData->pipe);
			pData->bHadError = 1;
		}
		DBGPRINTF("Error opening log pipe: %s\n", pData->pipe);
	}
	RETiRet;
}


/* rgerhards 2004-11-11: write to a pipe output. This
 * will be called for all outputs using pipe semantics,
 * for example also for pipes.
 */
static rsRetVal writePipe(uchar **ppString, instanceData *pData)
{
	int iLenWritten;
	DEFiRet;

	assert(pData != NULL);

	if(pData->fd == -1) {
		rsRetVal iRetLocal;
		iRetLocal = preparePipe(pData);
		if((iRetLocal != RS_RET_OK) || (pData->fd == -1))
			ABORT_FINALIZE(RS_RET_SUSPENDED); /* whatever the failure was, we need to retry */
	}

	/* create the message based on format specified */
	iLenWritten = write(pData->fd, ppString[0], strlen((char*)ppString[0]));
	if(iLenWritten < 0) {
		const int e = errno;
		/* If a named pipe is full, we suspend this action for a while */
		if(e == EAGAIN)
			ABORT_FINALIZE(RS_RET_SUSPENDED);

		close(pData->fd);
		pData->fd = -1; /* tell that fd is no longer open! */
		iRet = RS_RET_SUSPENDED;
		LogError(e, NO_ERRCODE, "write error on pipe %s", pData->pipe);
	}

finalize_it:
	RETiRet;
}


BEGINbeginCnfLoad
CODESTARTbeginCnfLoad
	loadModConf = pModConf;
	pModConf->pConf = pConf;
	pModConf->tplName = NULL;
ENDbeginCnfLoad

BEGINsetModCnf
	struct cnfparamvals *pvals = NULL;
	int i;
CODESTARTsetModCnf
	pvals = nvlstGetParams(lst, &modpblk, NULL);
	if(pvals == NULL) {
		LogError(0, RS_RET_MISSING_CNFPARAMS, "error processing module "
				"config parameters [module(...)]");
		ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS);
	}

	if(Debug) {
		dbgprintf("module (global) param blk for ompipe:\n");
		cnfparamsPrint(&modpblk, pvals);
	}

	for(i = 0 ; i < modpblk.nParams ; ++i) {
		if(!pvals[i].bUsed)
			continue;
		if(!strcmp(modpblk.descr[i].name, "template")) {
			loadModConf->tplName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
			if(pszFileDfltTplName != NULL) {
				LogError(0, RS_RET_DUP_PARAM, "ompipe: warning: default template "
						"was already set via legacy directive - may lead to inconsistent "
						"results.");
			}
		} else {
			dbgprintf("ompipe: program error, non-handled "
			  "param '%s' in beginCnfLoad\n", modpblk.descr[i].name);
		}
	}
finalize_it:
	if(pvals != NULL)
		cnfparamvalsDestruct(pvals, &modpblk);
ENDsetModCnf

BEGINendCnfLoad
CODESTARTendCnfLoad
	loadModConf = NULL; /* done loading */
	/* free legacy config vars */
	free(pszFileDfltTplName);
	pszFileDfltTplName = NULL;
ENDendCnfLoad

BEGINcheckCnf
CODESTARTcheckCnf
ENDcheckCnf

BEGINactivateCnf
CODESTARTactivateCnf
	runModConf = pModConf;
ENDactivateCnf

BEGINfreeCnf
CODESTARTfreeCnf
	free(pModConf->tplName);
ENDfreeCnf

BEGINcreateInstance
CODESTARTcreateInstance
	pData->pipe = NULL;
	pData->fd = -1;
	pData->bHadError = 0;
	pData->bTryResumeReopen = 0;
	pthread_mutex_init(&pData->mutWrite, NULL);
ENDcreateInstance


BEGINcreateWrkrInstance
CODESTARTcreateWrkrInstance
ENDcreateWrkrInstance


BEGINfreeInstance
CODESTARTfreeInstance
	pthread_mutex_destroy(&pData->mutWrite);
	free(pData->pipe);
	if(pData->fd != -1)
		close(pData->fd);
ENDfreeInstance


BEGINfreeWrkrInstance
CODESTARTfreeWrkrInstance
ENDfreeWrkrInstance


BEGINtryResume
	instanceData *__restrict__ const pData = pWrkrData->pData;
	fd_set wrds;
	struct timeval tv;
	int ready;
CODESTARTtryResume
	if(pData->fd == -1) {
		rsRetVal iRetLocal;
		iRetLocal = preparePipe(pData);
		if((iRetLocal != RS_RET_OK) || (pData->fd == -1))
			ABORT_FINALIZE(RS_RET_SUSPENDED);
	} else {
		/* we can reach this if the pipe is full, so we need
		 * to check if we can write again. /dev/xconsole is the
		 * ugly example of why this is necessary.
		 */
		FD_ZERO(&wrds);
		FD_SET(pData->fd, &wrds);
		tv.tv_sec = 0;
		tv.tv_usec = 0;
		ready = select(pData->fd+1, NULL, &wrds, NULL, &tv);
		DBGPRINTF("ompipe: tryResume: ready to write fd %d: %d\n", pData->fd, ready);
		if(ready != 1) {
			if(pData->bTryResumeReopen && pData->fd != -1) {
				close(pData->fd);
				pData->fd = -1;
			}
			ABORT_FINALIZE(RS_RET_SUSPENDED);
		}
	}
finalize_it:
ENDtryResume

BEGINdoAction
	instanceData *pData;
CODESTARTdoAction
	pData = pWrkrData->pData;
	DBGPRINTF("ompipe: writing to %s\n", pData->pipe);
	/* this module is single-threaded by nature */
	pthread_mutex_lock(&pData->mutWrite);
	iRet = writePipe(ppString, pData);
	pthread_mutex_unlock(&pData->mutWrite);
ENDdoAction


static inline void
setInstParamDefaults(instanceData *pData)
{
	pData->tplName = NULL;
}

BEGINnewActInst
	struct cnfparamvals *pvals;
	int i;
CODESTARTnewActInst
	if((pvals = nvlstGetParams(lst, &actpblk, NULL)) == NULL) {
		ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS);
	}

	CHKiRet(createInstance(&pData));
	setInstParamDefaults(pData);

	CODE_STD_STRING_REQUESTnewActInst(1)
	for(i = 0 ; i < actpblk.nParams ; ++i) {
		if(!pvals[i].bUsed)
			continue;
		if(!strcmp(actpblk.descr[i].name, "pipe")) {
			pData->pipe = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
		} else if(!strcmp(actpblk.descr[i].name, "template")) {
			pData->tplName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
		} else if(!strcmp(actpblk.descr[i].name, "tryResumeReopen")) {
			pData->bTryResumeReopen = (int) pvals[i].val.d.n;
		} else {
			dbgprintf("ompipe: program error, non-handled "
			  "param '%s'\n", actpblk.descr[i].name);
		}
	}

	CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*)strdup((pData->tplName == NULL) ?
						"RSYSLOG_FileFormat" : (char*)pData->tplName),
						OMSR_NO_RQD_TPL_OPTS));
CODE_STD_FINALIZERnewActInst
	cnfparamvalsDestruct(pvals, &actpblk);
ENDnewActInst

BEGINparseSelectorAct
CODESTARTparseSelectorAct
	/* yes, the if below is redundant, but I need it now. Will go away as
	 * the code further changes.  -- rgerhards, 2007-07-25
	 */
	if(*p == '|') {
		if((iRet = createInstance(&pData)) != RS_RET_OK) {
			return iRet; /* this can not use RET_iRet! */
		}
	} else {
		/* this is not clean, but we need it for the time being
		 * TODO: remove when cleaning up modularization
		 */
		return RS_RET_CONFLINE_UNPROCESSED;
	}

	CODE_STD_STRING_REQUESTparseSelectorAct(1)
	CHKmalloc(pData->pipe = malloc(512));
	++p;
	CHKiRet(cflineParseFileName(p, (uchar*) pData->pipe, *ppOMSR, 0, OMSR_NO_RQD_TPL_OPTS,
				       getDfltTpl()));

CODE_STD_FINALIZERparseSelectorAct
ENDparseSelectorAct


BEGINdoHUP
CODESTARTdoHUP
	pthread_mutex_lock(&pData->mutWrite);
	if(pData->fd != -1) {
		close(pData->fd);
		pData->fd = -1;
	}
	pthread_mutex_unlock(&pData->mutWrite);
ENDdoHUP


BEGINmodExit
CODESTARTmodExit
ENDmodExit


BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_doHUP
CODEqueryEtryPt_STD_CONF2_QUERIES
CODEqueryEtryPt_STD_CONF2_CNFNAME_QUERIES
CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES
CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
ENDqueryEtryPt


BEGINmodInit(Pipe)
CODESTARTmodInit
INITLegCnfVars
	*ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
CODEmodInit_QueryRegCFSLineHdlr
ENDmodInit
/* vi:set ai:
 */