diff options
Diffstat (limited to 'plugins/ompgsql/ompgsql.c')
-rw-r--r-- | plugins/ompgsql/ompgsql.c | 580 |
1 files changed, 580 insertions, 0 deletions
diff --git a/plugins/ompgsql/ompgsql.c b/plugins/ompgsql/ompgsql.c new file mode 100644 index 0000000..27248ff --- /dev/null +++ b/plugins/ompgsql/ompgsql.c @@ -0,0 +1,580 @@ +/* ompgsql.c + * This is the implementation of the build-in output module for PgSQL. + * + * NOTE: read comments in module-template.h to understand how this file + * works! + * + * File begun on 2007-10-18 by sur5r (converted from ommysql.c) + * + * Copyright 2007-2018 Rainer Gerhards and Adiscon GmbH. + * + * The following link my be useful for the not-so-postgres literate + * when setting up a test environment (on Fedora): + * http://www.jboss.org/community/wiki/InstallPostgreSQLonFedora + * + * 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 <stdarg.h> +#include <stdlib.h> +#include <string.h> +#include <assert.h> +#include <signal.h> +#include <errno.h> +#include <time.h> +#include <netdb.h> +#include <libpq-fe.h> +#include "conf.h" +#include "syslogd-types.h" +#include "srUtils.h" +#include "template.h" +#include "module-template.h" +#include "errmsg.h" +#include "parserif.h" + +MODULE_TYPE_OUTPUT +MODULE_TYPE_NOKEEP +MODULE_CNFNAME("ompgsql") + + +/* internal structures + */ +DEF_OMOD_STATIC_DATA + +typedef struct _instanceData { + char srv[MAXHOSTNAMELEN+1]; /* IP or hostname of DB server*/ + char dbname[_DB_MAXDBLEN+1]; /* DB name */ + char user[_DB_MAXUNAMELEN+1]; /* DB user */ + char pass[_DB_MAXPWDLEN+1]; /* DB user's password */ + char conninfo[_DB_MAXCONNINFOLEN+1]; /* Connection parameters or URI */ + unsigned int trans_age; + unsigned int trans_commit; + unsigned short multi_row; + int port; + uchar *tpl; /* format template to use */ +} instanceData; + +typedef struct wrkrInstanceData { + instanceData *pData; + PGconn *f_hpgsql; /* handle to PgSQL */ + ConnStatusType eLastPgSQLStatus; /* last status from postgres */ +} wrkrInstanceData_t; + +/* action (instance) parameters */ +static struct cnfparamdescr actpdescr[] = { + { "server", eCmdHdlrGetWord, 0 }, + { "db", eCmdHdlrGetWord, 0 }, + { "user", eCmdHdlrGetWord, 0 }, + { "uid", eCmdHdlrGetWord, 0 }, + { "pass", eCmdHdlrGetWord, 0 }, + { "pwd", eCmdHdlrGetWord, 0 }, + { "multirows", eCmdHdlrInt, 0 }, + { "trans_size", eCmdHdlrInt, 0 }, + { "trans_age", eCmdHdlrInt, 0 }, + { "serverport", eCmdHdlrInt, 0 }, + { "port", eCmdHdlrInt, 0 }, + { "template", eCmdHdlrGetWord, 0 }, + { "conninfo", eCmdHdlrGetWord, 0 } +}; + + +static struct cnfparamblk actpblk = + { CNFPARAMBLK_VERSION, + sizeof(actpdescr)/sizeof(struct cnfparamdescr), + actpdescr + }; + + +BEGINinitConfVars /* (re)set config variables to default values */ +CODESTARTinitConfVars +ENDinitConfVars + + +static rsRetVal writePgSQL(uchar *psz, wrkrInstanceData_t *pData); + +BEGINcreateInstance +CODESTARTcreateInstance +ENDcreateInstance + +BEGINcreateWrkrInstance +CODESTARTcreateWrkrInstance + pWrkrData->f_hpgsql = NULL; +ENDcreateWrkrInstance + + +BEGINisCompatibleWithFeature +CODESTARTisCompatibleWithFeature + if (eFeat == sFEATURERepeatedMsgReduction) + iRet = RS_RET_OK; +ENDisCompatibleWithFeature + + +/* The following function is responsible for closing a + * PgSQL connection. + */ +static void closePgSQL(wrkrInstanceData_t *pWrkrData) +{ + assert(pWrkrData != NULL); + + if (pWrkrData->f_hpgsql != NULL) { /* just to be on the safe side... */ + PQfinish(pWrkrData->f_hpgsql); + pWrkrData->f_hpgsql = NULL; + } +} + +BEGINfreeInstance +CODESTARTfreeInstance + free(pData->tpl); +ENDfreeInstance + +BEGINfreeWrkrInstance +CODESTARTfreeWrkrInstance + closePgSQL(pWrkrData); +ENDfreeWrkrInstance + +BEGINdbgPrintInstInfo +CODESTARTdbgPrintInstInfo + /* nothing special here */ +ENDdbgPrintInstInfo + + +/* log a database error with descriptive message. + * We check if we have a valid handle. If not, we simply + * report an error, but can not be specific. RGerhards, 2007-01-30 + */ +static void reportDBError(wrkrInstanceData_t *pWrkrData, int bSilent) +{ + char errMsg[512]; + ConnStatusType ePgSQLStatus; + + assert(pWrkrData != NULL); + bSilent = 0; + + /* output log message */ + errno = 0; + if (pWrkrData->f_hpgsql == NULL) { + LogError(0, NO_ERRCODE, "unknown DB error occurred - could not obtain PgSQL handle"); + } else { /* we can ask pgsql for the error description... */ + ePgSQLStatus = PQstatus(pWrkrData->f_hpgsql); + snprintf(errMsg, sizeof(errMsg), "db error (%d): %s\n", ePgSQLStatus, + PQerrorMessage(pWrkrData->f_hpgsql)); + if (bSilent || ePgSQLStatus == pWrkrData->eLastPgSQLStatus) + dbgprintf("pgsql, DBError(silent): %s\n", errMsg); + else { + pWrkrData->eLastPgSQLStatus = ePgSQLStatus; + LogError(0, NO_ERRCODE, "%s", errMsg); + } + } + + return; +} + + +/* The following function is responsible for initializing a + * PgSQL connection. + */ +static rsRetVal initPgSQL(wrkrInstanceData_t *pWrkrData, int bSilent) +{ + int sslStatus; + instanceData *pData; + DEFiRet; + + pData = pWrkrData->pData; + assert(pData != NULL); + assert(pWrkrData->f_hpgsql == NULL); + + if (strlen(pData->conninfo) > 0) { + /* Don't log the whole connection string, because it contains the DB password */ + dbgprintf("initPgSQL: using connection string provided by conninfo\n"); + pWrkrData->f_hpgsql = PQconnectdb(pData->conninfo); + } else { + dbgprintf("initPgSQL: host=%s port=%d dbname=%s uid=%s\n", pData->srv, pData->port, + pData->dbname, pData->user); + + /* Force PostgreSQL to use ANSI-SQL conforming strings, otherwise we may + * get all sorts of side effects (e.g.: backslash escapes) and warnings + * + * Note: PostgreSQL versions since 9.3 have this already on by default. + */ + const char *PgConnectionOptions = "-c standard_conforming_strings=on"; + + /* Connect to database */ + char port[6]; + snprintf(port, sizeof(port), "%d", pData->port); + + pWrkrData->f_hpgsql = PQsetdbLogin(pData->srv, port, PgConnectionOptions, NULL, + pData->dbname, pData->user, pData->pass); + } + + if (pWrkrData->f_hpgsql == NULL) { + reportDBError(pWrkrData, bSilent); + closePgSQL(pWrkrData); /* ignore any error we may get */ + iRet = RS_RET_SUSPENDED; + } + +#ifdef HAVE_PGSSLINUSE + sslStatus = PQsslInUse(pWrkrData->f_hpgsql); +#else + sslStatus = PQgetssl(pWrkrData->f_hpgsql) == NULL ? 0 : 1; +#endif + dbgprintf("initPgSQL: ssl status: %d\n", sslStatus); + + RETiRet; +} + + +/* try the insert into postgres and return if that failed or not + * (1 = had error, 0=ok). We do not use the standard IRET calling convention + * rgerhards, 2009-04-17 + */ +static int +tryExec(uchar *pszCmd, wrkrInstanceData_t *pWrkrData) +{ + PGresult *pgRet; + ExecStatusType execState; + int bHadError = 0; + + /* try insert */ + pgRet = PQexec(pWrkrData->f_hpgsql, (char*)pszCmd); + execState = PQresultStatus(pgRet); + if (execState != PGRES_COMMAND_OK && execState != PGRES_TUPLES_OK) { + dbgprintf("postgres query execution failed: %s\n", PQresStatus(PQresultStatus(pgRet))); + bHadError = 1; + } + PQclear(pgRet); + + return(bHadError); +} + + +/* The following function writes the current log entry + * to an established PgSQL session. + * Enhanced function to take care of the returned error + * value (if there is such). Note that this may happen due to + * a sql format error - connection aborts were properly handled + * before my patch. -- rgerhards, 2009-04-17 + */ +static rsRetVal +writePgSQL(uchar *psz, wrkrInstanceData_t *pWrkrData) +{ + int bHadError = 0; + DEFiRet; + + assert(psz != NULL); + assert(pWrkrData != NULL); + + dbgprintf("writePgSQL: %s\n", psz); + + bHadError = tryExec(psz, pWrkrData); /* try insert */ + + if (bHadError || (PQstatus(pWrkrData->f_hpgsql) != CONNECTION_OK)) { +#if 0 /* re-enable once we have transaction support */ + /* error occurred, try to re-init connection and retry */ + int inTransaction = 0; + if(pData->f_hpgsql != NULL) { + PGTransactionStatusType xactStatus = PQtransactionStatus(pData->f_hpgsql); + if((xactStatus == PQTRANS_INTRANS) || (xactStatus == PQTRANS_ACTIVE)) { + inTransaction = 1; + } + } + if ( inTransaction == 0 ) +#endif + { + closePgSQL(pWrkrData); /* close the current handle */ + CHKiRet(initPgSQL(pWrkrData, 0)); /* try to re-open */ + bHadError = tryExec(psz, pWrkrData); /* retry */ + } + if(bHadError || (PQstatus(pWrkrData->f_hpgsql) != CONNECTION_OK)) { + /* we failed, giving up for now */ + reportDBError(pWrkrData, 0); + closePgSQL(pWrkrData); /* free ressources */ + ABORT_FINALIZE(RS_RET_SUSPENDED); + } + } + +finalize_it: + if (iRet == RS_RET_OK) { + pWrkrData->eLastPgSQLStatus = CONNECTION_OK; /* reset error for error supression */ + } + + RETiRet; +} + + +BEGINtryResume +CODESTARTtryResume + if (pWrkrData->f_hpgsql == NULL) { + iRet = initPgSQL(pWrkrData, 1); + if (iRet == RS_RET_OK) { + /* the code above seems not to actually connect to the database. As such, we do a + * dummy statement (a pointless select...) to verify the connection and return + * success only when that statemetn succeeds. Note that I am far from being a + * PostgreSQL expert, so any patch that does the desired result in a more + * intelligent way is highly welcome. -- rgerhards, 2009-12-16 + */ + iRet = writePgSQL((uchar*)"select 'a' as a", pWrkrData); + } + } +ENDtryResume + + +BEGINbeginTransaction +CODESTARTbeginTransaction +ENDbeginTransaction + + +BEGINcommitTransaction +CODESTARTcommitTransaction + dbgprintf("ompgsql: beginTransaction\n"); + if (pWrkrData->f_hpgsql == NULL) + initPgSQL(pWrkrData, 0); + CHKiRet(writePgSQL((uchar*) "BEGIN", pWrkrData)); /* TODO: make user-configurable */ + + for (unsigned i = 0 ; i < nParams ; ++i) { + iRet = writePgSQL(actParam(pParams, 1, i, 0).param, pWrkrData); + if (iRet != RS_RET_OK + && iRet != RS_RET_DEFER_COMMIT + && iRet != RS_RET_PREVIOUS_COMMITTED) { + /*if(mysql_rollback(pWrkrData->hmysql) != 0) { + DBGPRINTF("ommysql: server error: transaction could not be rolled back\n"); + }*/ + // closeMySQL(pWrkrData); + // FINALIZE; + } + } + + CHKiRet(writePgSQL((uchar*) "COMMIT", pWrkrData)); /* TODO: make user-configurable */ + +finalize_it: + if (iRet == RS_RET_OK) { + pWrkrData->eLastPgSQLStatus = CONNECTION_OK; /* reset error for error supression */ + } + +ENDcommitTransaction + + +static inline void +setInstParamDefaults(instanceData *pData) +{ + pData->tpl = NULL; + pData->multi_row = 100; + pData->trans_commit = 100; + pData->trans_age = 60; + pData->port = 5432; + strcpy(pData->user, "postgres"); + strcpy(pData->pass, "postgres"); +} + +BEGINnewActInst + struct cnfparamvals *pvals; + int i; + char *cstr; + size_t len; +CODESTARTnewActInst + if ((pvals = nvlstGetParams(lst, &actpblk, NULL)) == NULL) { + ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); + } + + CHKiRet(createInstance(&pData)); + setInstParamDefaults(pData); + + CODE_STD_STRING_REQUESTparseSelectorAct(1) + for (i = 0 ; i < actpblk.nParams ; ++i) { + if (!pvals[i].bUsed) + continue; + if (!strcmp(actpblk.descr[i].name, "server")) { + cstr = es_str2cstr(pvals[i].val.d.estr, NULL); + len = es_strlen(pvals[i].val.d.estr); + if(len >= sizeof(pData->srv)-1) { + parser_errmsg("ompgsql: srv parameter longer than supported " + "maximum of %d characters", (int)sizeof(pData->srv)-1); + ABORT_FINALIZE(RS_RET_PARAM_ERROR); + } + memcpy(pData->srv, cstr, len+1); + free(cstr); + } else if (!strcmp(actpblk.descr[i].name, "port")) { + pData->port = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "serverport")) { + pData->port = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "multirows")) { + pData->multi_row = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "trans_size")) { + pData->trans_commit = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "trans_age")) { + pData->trans_age = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "db")) { + cstr = es_str2cstr(pvals[i].val.d.estr, NULL); + len = es_strlen(pvals[i].val.d.estr); + if(len >= sizeof(pData->dbname)-1) { + parser_errmsg("ompgsql: db parameter longer than supported " + "maximum of %d characters", (int)sizeof(pData->dbname)-1); + ABORT_FINALIZE(RS_RET_PARAM_ERROR); + } + memcpy(pData->dbname, cstr, len+1); + free(cstr); + } else if ( !strcmp(actpblk.descr[i].name, "user") + || !strcmp(actpblk.descr[i].name, "uid")) { + cstr = es_str2cstr(pvals[i].val.d.estr, NULL); + len = es_strlen(pvals[i].val.d.estr); + if(len >= sizeof(pData->user)-1) { + parser_errmsg("ompgsql: user/uid parameter longer than supported " + "maximum of %d characters", (int)sizeof(pData->user)-1); + ABORT_FINALIZE(RS_RET_PARAM_ERROR); + } + memcpy(pData->user, cstr, len+1); + free(cstr); + } else if ( !strcmp(actpblk.descr[i].name, "pass") + || !strcmp(actpblk.descr[i].name, "pwd")) { + cstr = es_str2cstr(pvals[i].val.d.estr, NULL); + len = es_strlen(pvals[i].val.d.estr); + if(len >= sizeof(pData->pass)-1) { + parser_errmsg("ompgsql: pass/pwd parameter longer than supported " + "maximum of %d characters", (int)sizeof(pData->pass)-1); + ABORT_FINALIZE(RS_RET_PARAM_ERROR); + } + memcpy(pData->pass, cstr, len+1); + free(cstr); + } else if (!strcmp(actpblk.descr[i].name, "template")) { + pData->tpl = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + } else if (!strcmp(actpblk.descr[i].name, "conninfo")) { + cstr = es_str2cstr(pvals[i].val.d.estr, NULL); + len = es_strlen(pvals[i].val.d.estr); + if(len >= sizeof(pData->conninfo)-1) { + parser_errmsg("ompgsql: conninfo parameter longer than supported " + "maximum of %d characters", (int)sizeof(pData->conninfo)-1); + ABORT_FINALIZE(RS_RET_PARAM_ERROR); + } + memcpy(pData->conninfo, cstr, len+1); + free(cstr); + } else { + dbgprintf("ompgsql: program error, non-handled " + "param '%s'\n", actpblk.descr[i].name); + } + } + + if (strlen(pData->conninfo) == 0 && (strlen(pData->srv) == 0 || strlen(pData->dbname) == 0)) { + parser_errmsg("ompgsql: must provide conninfo or server and dbname"); + ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); + } + + if (pData->tpl == NULL) { + CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*) strdup(" StdPgSQLFmt"), OMSR_RQD_TPL_OPT_SQL)); + } else { + CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*) strdup((char*) pData->tpl), OMSR_RQD_TPL_OPT_SQL)); + } + +CODE_STD_FINALIZERnewActInst + cnfparamvalsDestruct(pvals, &actpblk); +ENDnewActInst + +BEGINparseSelectorAct + int iPgSQLPropErr = 0; +CODESTARTparseSelectorAct +CODE_STD_STRING_REQUESTparseSelectorAct(1) + /* first check if this config line is actually for us + * The first test [*p == '>'] can be skipped if a module shall only + * support the newer slection syntax [:modname:]. This is in fact + * recommended for new modules. Please note that over time this part + * will be handled by rsyslogd itself, but for the time being it is + * a good compromise to do it at the module level. + * rgerhards, 2007-10-15 + */ + + if (!strncmp((char*) p, ":ompgsql:", sizeof(":ompgsql:") - 1)) + p += sizeof(":ompgsql:") - 1; /* eat indicator sequence (-1 because of '\0'!) */ + else + ABORT_FINALIZE(RS_RET_CONFLINE_UNPROCESSED); + + /* ok, if we reach this point, we have something for us */ + if ((iRet = createInstance(&pData)) != RS_RET_OK) + goto finalize_it; + setInstParamDefaults(pData); + + /* sur5r 2007-10-18: added support for PgSQL + * :ompgsql:server,dbname,userid,password + * Now we read the PgSQL connection properties + * and verify that the properties are valid. + */ + if (getSubString(&p, pData->srv, MAXHOSTNAMELEN+1, ',')) + iPgSQLPropErr++; + dbgprintf("%p:%s\n",p,p); + if (*pData->srv == '\0') + iPgSQLPropErr++; + if (getSubString(&p, pData->dbname, _DB_MAXDBLEN+1, ',')) + iPgSQLPropErr++; + if (*pData->dbname == '\0') + iPgSQLPropErr++; + if (getSubString(&p, pData->user, _DB_MAXUNAMELEN+1, ',')) + iPgSQLPropErr++; + if (*pData->user == '\0') + iPgSQLPropErr++; + if (getSubString(&p, pData->pass, _DB_MAXPWDLEN+1, ';')) + iPgSQLPropErr++; + /* now check for template + * We specify that the SQL option must be present in the template. + * This is for your own protection (prevent sql injection). + */ + if (*(p - 1) == ';') { + p--; + CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, OMSR_RQD_TPL_OPT_SQL, (uchar*) pData->tpl)); + } else { + CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, OMSR_RQD_TPL_OPT_SQL, (uchar*)" StdPgSQLFmt")); + } + + /* If we detect invalid properties, we disable logging, + * because right properties are vital at this place. + * Retries make no sense. + */ + if (iPgSQLPropErr) { + LogError(0, RS_RET_INVALID_PARAMS, "Trouble with PgSQL connection properties. " + "-PgSQL logging disabled"); + ABORT_FINALIZE(RS_RET_INVALID_PARAMS); + } + +CODE_STD_FINALIZERparseSelectorAct +ENDparseSelectorAct + +BEGINmodExit +CODESTARTmodExit +ENDmodExit + + +BEGINqueryEtryPt +CODESTARTqueryEtryPt +CODEqueryEtryPt_STD_OMODTX_QUERIES +CODEqueryEtryPt_STD_OMOD8_QUERIES +CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES +/* CODEqueryEtryPt_TXIF_OMOD_QUERIES currently no TX support! */ /* we support the transactional interface! */ +ENDqueryEtryPt + + +BEGINmodInit() +CODESTARTmodInit +INITLegCnfVars + *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */ +CODEmodInit_QueryRegCFSLineHdlr + INITChkCoreFeature(bCoreSupportsBatching, CORE_FEATURE_BATCHING); + if (!bCoreSupportsBatching) { + LogError(0, NO_ERRCODE, "ompgsql: rsyslog core too old"); + ABORT_FINALIZE(RS_RET_ERR); + } +ENDmodInit + +/* vi:set ai: */ |