diff options
| -rw-r--r-- | ChangeLog | 35 | ||||
| -rw-r--r-- | configure.ac | 37 | ||||
| -rw-r--r-- | doc/Makefile.am | 1 | ||||
| -rw-r--r-- | doc/debug.html | 1 | ||||
| -rw-r--r-- | doc/imfile.html | 4 | ||||
| -rw-r--r-- | doc/imjournal.html | 45 | ||||
| -rw-r--r-- | doc/manual.html | 2 | ||||
| -rw-r--r-- | doc/omfile.html | 12 | ||||
| -rw-r--r-- | doc/rsyslog_conf_modules.html | 1 | ||||
| -rw-r--r-- | plugins/imgssapi/imgssapi.c | 21 | ||||
| -rwxr-xr-x[-rw-r--r--] | plugins/imjournal/imjournal.c | 144 | ||||
| -rw-r--r-- | plugins/imzmq3/README | 59 | ||||
| -rw-r--r-- | plugins/imzmq3/imzmq3.c | 735 | ||||
| -rw-r--r-- | plugins/mmaudit/Makefile.am | 4 | ||||
| -rw-r--r-- | plugins/mmaudit/mmaudit.c | 3 | ||||
| -rw-r--r-- | plugins/mmjsonparse/Makefile.am | 4 | ||||
| -rw-r--r-- | plugins/mmjsonparse/mmjsonparse.c | 1 | ||||
| -rw-r--r-- | plugins/omjournal/omjournal.c | 2 | ||||
| -rw-r--r-- | plugins/omzmq3/README | 14 | ||||
| -rw-r--r-- | plugins/omzmq3/omzmq3.c | 202 | ||||
| -rw-r--r-- | runtime/Makefile.am | 6 | ||||
| -rw-r--r-- | runtime/conf.c | 18 | ||||
| -rw-r--r-- | runtime/debug.c | 17 | ||||
| -rw-r--r-- | runtime/librsgt.c | 2 | ||||
| -rw-r--r-- | runtime/librsgt_read.c | 4 | ||||
| -rw-r--r-- | runtime/msg.h | 2 | ||||
| -rw-r--r-- | runtime/stringbuf.c | 2 | ||||
| -rw-r--r-- | template.c | 1 | ||||
| -rw-r--r-- | tools/Makefile.am | 4 | ||||
| -rw-r--r-- | tools/syslogd.c | 1 |
30 files changed, 923 insertions, 461 deletions
@@ -5,7 +5,40 @@ Version 7.5.0 [devel] 2013-0?-?? connections Thanks to Axel Rau for the patch. --------------------------------------------------------------------------- -Version 7.3.15 [beta] 2013-05-0? +Version 7.4.0 [v7.4-stable] 2013-06-06 +This starts a new stable branch based on 7.3.15 plus the following changes: +- add --enable-cached-man-pages ./configure option + permits to build rsyslog on a system where rst2man is not installed. In + that case, cached versions of the man pages are used (they were built + during "make dist", so they should be current for the version in + question. +- doc bugfix: ReadMode wrong in imfile doc, two values were swapped + Thanks to jokajak@gmail.com for mentioning this + closes: http://bugzilla.adiscon.com/show_bug.cgi?id=450 +- imjournal: no longer do periodic wakeup +- bugfix: potential hang *in debug mode* on rsyslogd termination + This ONLY affected rsyslogd if it were running with debug output + enabled. +- bugfix: $template statement with multiple spaces lead to invalid tpl name + If multiple spaces were used in front of the template name, all but one + of them became actually part of the template name. So + $template a,"..." would be name " a", and as such "a" was not + available, e.g. in + *.* /var/log/file;a + This is a legacy config problem. As it was unreported for many years, + no backport of the fix to old versions will happen. + This is a long-standing bug that was only recently reported by forum + user mc-sim. + Reference: http://kb.monitorware.com/post23448.html +- 0mq fixes; credits to Hongfei Cheng and Brian Knox +--------------------------------------------------------------------------- +Version 7.3.15 [beta] 2013-05-15 +- bugfix: problem in build system (especially when cross-compiling) + Thanks to Tomas Heinrich and winfried_mb2@xmsnet.nl for the patch. + closes: http://bugzilla.adiscon.com/show_bug.cgi?id=445 +- bugfix: imjournal had problem with systemd journal API change +- imjournal: now obtain and include PID +- bugfix: .logsig files had tlv16 indicator bit at wrong offset - bugfix: omrelp legacy config parameters set a timeout of zero which lead the legacy config to be unusable. - bugfix: segfault on startup if a disk queue was configure without file diff --git a/configure.ac b/configure.ac index 841b8246..233eba44 100644 --- a/configure.ac +++ b/configure.ac @@ -2,7 +2,7 @@ # Process this file with autoconf to produce a configure script. AC_PREREQ(2.61) -AC_INIT([rsyslog],[7.3.14],[rsyslog@lists.adiscon.com]) +AC_INIT([rsyslog],[7.4.0],[rsyslog@lists.adiscon.com]) AM_INIT_AUTOMAKE m4_ifdef([AM_SILENT_RULES], [AM_SILENT_RULES([yes])]) @@ -28,13 +28,11 @@ fi AC_DISABLE_STATIC AC_PROG_LIBTOOL AC_CANONICAL_HOST -AC_PATH_PROG([RST2MAN], [rst2man]) PKG_PROG_PKG_CONFIG # modules we require PKG_CHECK_MODULES(LIBESTR, libestr >= 0.1.5) -PKG_CHECK_MODULES(LIBEE, libee >= 0.4.0) PKG_CHECK_MODULES([JSON_C], [json]) case "${host}" in @@ -919,6 +917,7 @@ AC_ARG_ENABLE(mmnormalize, [enable_mmnormalize=no] ) if test "x$enable_mmnormalize" = "xyes"; then + PKG_CHECK_MODULES(LIBEE, libee >= 0.4.0) PKG_CHECK_MODULES(LIBLOGNORM, lognorm >= 0.3.1) fi AM_CONDITIONAL(ENABLE_MMNORMALIZE, test x$enable_mmnormalize = xyes) @@ -934,9 +933,6 @@ AC_ARG_ENABLE(mmjsonparse, esac], [enable_mmjsonparse=no] ) -if test "x$enable_mmjsonparse" = "xyes"; then - PKG_CHECK_MODULES(LIBLOGNORM, lognorm >= 0.3.1) -fi AM_CONDITIONAL(ENABLE_MMJSONPARSE, test x$enable_mmjsonparse = xyes) @@ -950,9 +946,6 @@ AC_ARG_ENABLE(mmaudit, esac], [enable_mmaudit=no] ) -if test "x$enable_mmaudit" = "xyes"; then - PKG_CHECK_MODULES(LIBLOGNORM, lognorm >= 0.3.1) -fi AM_CONDITIONAL(ENABLE_MMAUDIT, test x$enable_mmaudit = xyes) @@ -1000,6 +993,31 @@ if test "x$enable_guardtime" = "xyes"; then fi AM_CONDITIONAL(ENABLE_GUARDTIME, test x$enable_guardtime = xyes) + +# Support using cached man file copies, to avoid the need for rst2man +# in the build environment +AC_ARG_ENABLE(cached_man_pages, + [AS_HELP_STRING([--enable-cached-man-pages],[Enable using cached versions of man files (avoid rst2man) @<:@default=no@:>@])], + [case "${enableval}" in + yes) enable_cached_man_pages="yes" ;; + no) enable_cached_man_pages="no" ;; + *) AC_MSG_ERROR(bad value ${enableval} for --enable-cached-man-pages) ;; + esac], + [enable_cached_man_pages=no] +) +if test "x$enable_cached_man_pages" = "xno"; then +# obtain path for rst2man + if test "x$enable_libgcrypt" = "xyes" || \ + "x$enable_guardtime" = "xyes"; then + AC_PATH_PROG([RST2MAN], [rst2man]) + if test "x${RST2MAN}" == "x"; then + AC_MSG_FAILURE([rst2man not found in PATH]) + fi + fi +fi + + + # RFC 3195 support AC_ARG_ENABLE(rfc3195, [AS_HELP_STRING([--enable-rfc3195],[Enable RFC3195 support @<:@default=no@:>@])], @@ -1477,6 +1495,7 @@ echo " Zlib compression support enabled: $enable_zlib" echo " rsyslog runtime will be built: $enable_rsyslogrt" echo " rsyslogd will be built: $enable_rsyslogd" echo " GUI components will be built: $enable_gui" +echo " cached man files will be used: $enable_cached_man_pages" echo " Unlimited select() support enabled: $enable_unlimited_select" echo " uuid support enabled: $enable_uuid" echo " Log file signing support: $enable_guardtime" diff --git a/doc/Makefile.am b/doc/Makefile.am index 46afd900..e1757644 100644 --- a/doc/Makefile.am +++ b/doc/Makefile.am @@ -35,6 +35,7 @@ html_files = \ omfwd.html \ omfile.html \ omjournal.html \ + imjournal.html \ mmanon.html \ omusrmsg.html \ omstdout.html \ diff --git a/doc/debug.html b/doc/debug.html index 996bf5c8..557ca6d3 100644 --- a/doc/debug.html +++ b/doc/debug.html @@ -56,6 +56,7 @@ only available under Linux. This usually does NOT work when privileges have been dropped (that's not a bug, but the way it is). <li><b>help</b> - display a very short list of commands - hopefully a life saver if you can't access the documentation...</li> </ul> +<p>Individual options are separated by spaces.</p> </ul> <h3>Why Environment Variables?</h3> <p>You may ask why we use environment variables for debug-system parameters and not diff --git a/doc/imfile.html b/doc/imfile.html index f37f7055..274d6e60 100644 --- a/doc/imfile.html +++ b/doc/imfile.html @@ -104,8 +104,8 @@ file is very time consuming. <li><b>ReadMode</b> [mode]</b><br> This mode should defined when having multiline messages. The value can range from 0-2 and determines the multiline detection method. <br>0 (default) - line based (Each line is a new message) -<br>1 - indented (New log messages start at the beginning of a line. If a line starts with a space it is part of the log message before it) -<br>2 - paragraph (There is a blank line between log messages) +<br>1 - paragraph (There is a blank line between log messages) +<br>2 - indented (New log messages start at the beginning of a line. If a line starts with a space it is part of the log message before it) <li><b>MaxLinesAtOnce</b> [number]</b> <br> This is useful if multiple files need to be monitored. If set to 0, each file diff --git a/doc/imjournal.html b/doc/imjournal.html new file mode 100644 index 00000000..e3f64502 --- /dev/null +++ b/doc/imjournal.html @@ -0,0 +1,45 @@ +<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN"> +<html><head> +<meta http-equiv="Content-Language" content="en"><title>Text File Input Monitor</title></head> +<body> +<a href="rsyslog_conf_modules.html">back</a> + +<h1>Systemd Journal Input Module</h1> +<p><b>Module Name: imjournal</b></p> +<p><b>Author: </b>Milan Bartos +<mbartos@redhat.com></p> +<p><b>Description</b>:</p> +<p>Provides the ability to import structured log messages from systemd journal +to syslog.</p> + +<p><b>Configuration Directives</b>:</p> +<p><b>Module Directives</b></p> +<ul> +<li><span style="font-weight: bold;">PersistStateInterval +messages</span><br> +This is a global setting. It specifies how often should the journal state be persisted. +This option is useful for rsyslog to start reding from the last journal message it read. + +<li><span style="font-weight: bold;">StateFile +/path/to/file</span><br> +This is a global setting. It specifies where the state file for persisting +journal state is located. +</ul> +<b>Caveats/Known Bugs:</b> +<p> +</p> +<p><b>Sample:</b></p> +<p> +The following example shows pulling structured imjournal messages and saving them into /var/log/ceelog +</p> +<textarea rows="15" cols="60"> +module(load="imjournal" PersistStateInterval="100" StateFile="/path/to/file") #load imjournal module +module(load="mmjsonparse") #load mmjsonparse module for structured logs + +$template CEETemplate,"%TIMESTAMP% %HOSTNAME% %syslogtag% @cee: %$!all-json%\n" #template for messages + +*.* :mmjsonparse: +*.* /var/log/ceelog;CEETemplate +</textarea> +</body> +</html> diff --git a/doc/manual.html b/doc/manual.html index 5a74c8c4..dea0b82d 100644 --- a/doc/manual.html +++ b/doc/manual.html @@ -19,7 +19,7 @@ professional services</a> available directly from the source!</p> <p><b>Please visit the <a href="http://www.rsyslog.com/sponsors">rsyslog sponsor's page</a> to honor the project sponsors or become one yourself!</b> We are very grateful for any help towards the project goals.</p> -<p><b>This documentation is for version 7.3.14 (beta branch) of rsyslog.</b> +<p><b>This documentation is for version 7.4.0 (v7.4-stable branch) of rsyslog.</b> Visit the <i><a href="http://www.rsyslog.com/status">rsyslog status page</a></i></b> to obtain current version information and project status. </p><p><b>If you like rsyslog, you might diff --git a/doc/omfile.html b/doc/omfile.html index 3966ab13..a5c3bb3b 100644 --- a/doc/omfile.html +++ b/doc/omfile.html @@ -57,10 +57,10 @@ Set the group for directories newly created. Please note that this setting does not affect the group of directories already existing. The parameter is a group name, for which the groupid is obtained by rsyslogd on during startup processing. Interim changes to the user mapping are not detected.<br></li><br> <li><strong>FileOwner </strong><br> - Set the file owner for dynaFiles newly created. Please note that this setting does not affect the owner of files already existing. The parameter is a user name, for which the userid is obtained by rsyslogd during startup processing. Interim changes to the user mapping are not detected.<br></li><br> + Set the file owner for files newly created. Please note that this setting does not affect the owner of files already existing. The parameter is a user name, for which the userid is obtained by rsyslogd during startup processing. Interim changes to the user mapping are not detected.<br></li><br> <li><strong>FileGroup </strong><br> - Set the group for dynaFiles newly created. Please note that this setting does not affect the group of files already existing. The parameter is a group name, for which the groupid is obtained by rsyslogd during startup processing. Interim changes to the user mapping are not detected.<br></li><br> + Set the group for files newly created. Please note that this setting does not affect the group of files already existing. The parameter is a group name, for which the groupid is obtained by rsyslogd during startup processing. Interim changes to the user mapping are not detected.<br></li><br> <li><strong>DirCreateMode </strong>[defaul 0700]<br> This is the same as $FileCreateMode, but for directories automatically generated.<br></li><br> @@ -69,7 +69,7 @@ The FileCreateMode directive allows to specify the creation mode with which rsyslogd creates new files. If not specified, the value 0644 is used (which retains backward-compatibility with earlier releases). The value given must always be a 4-digit octal number, with the initial digit being zero. <br>Please note that the actual permission depend on rsyslogd's process umask. If in doubt, use "$umask 0000" right at the beginning of the configuration file to remove any restrictions. <br>FileCreateMode may be specified multiple times. If so, it specifies the creation mode for all selector lines that follow until the next $FileCreateMode directive. Order of lines is vitally important.<br></li><br> <li><strong>FailOnCHOwnFailure </strong>on/off [default on]<br> - This option modifies behaviour of dynaFile creation. If different owners or groups are specified for new files or directories and rsyslogd fails to set these new owners or groups, it will log an error and NOT write to the file in question if that option is set to "on". If it is set to "off", the error will be ignored and processing continues. Keep in mind, that the files in this case may be (in)accessible by people who should not have permission. The default is "on".<br></li><br> + This option modifies behaviour of file creation. If different owners or groups are specified for new files or directories and rsyslogd fails to set these new owners or groups, it will log an error and NOT write to the file in question if that option is set to "on". If it is set to "off", the error will be ignored and processing continues. Keep in mind, that the files in this case may be (in)accessible by people who should not have permission. The default is "on".<br></li><br> <li><strong>CreateDirs </strong>on/off [default on]<br> create directories on an as-needed basis<br></li><br> @@ -136,10 +136,10 @@ File="/var/log/messages") Set the group for directories newly created. Please note that this setting does not affect the group of directories already existing. The parameter is a group name, for which the groupid is obtained by rsyslogd on during startup processing. Interim changes to the user mapping are not detected.<br></li><br> <li><strong>$FileOwner </strong><br> - Set the file owner for dynaFiles newly created. Please note that this setting does not affect the owner of files already existing. The parameter is a user name, for which the userid is obtained by rsyslogd during startup processing. Interim changes to the user mapping are not detected.<br></li><br> + Set the file owner for files newly created. Please note that this setting does not affect the owner of files already existing. The parameter is a user name, for which the userid is obtained by rsyslogd during startup processing. Interim changes to the user mapping are not detected.<br></li><br> <li><strong>$FileGroup </strong><br> - Set the group for dynaFiles newly created. Please note that this setting does not affect the group of files already existing. The parameter is a group name, for which the groupid is obtained by rsyslogd during startup processing. Interim changes to the user mapping are not detected.<br></li><br> + Set the group for files newly created. Please note that this setting does not affect the group of files already existing. The parameter is a group name, for which the groupid is obtained by rsyslogd during startup processing. Interim changes to the user mapping are not detected.<br></li><br> <li><strong>$DirCreateMode </strong>[defaul 0700]<br> This is the same as $FileCreateMode, but for directories automatically generated.<br></li><br> @@ -148,7 +148,7 @@ File="/var/log/messages") The FileCreateMode directive allows to specify the creation mode with which rsyslogd creates new files. If not specified, the value 0644 is used (which retains backward-compatibility with earlier releases). The value given must always be a 4-digit octal number, with the initial digit being zero. <br>Please note that the actual permission depend on rsyslogd's process umask. If in doubt, use "$umask 0000" right at the beginning of the configuration file to remove any restrictions. <br>FileCreateMode may be specified multiple times. If so, it specifies the creation mode for all selector lines that follow until the next $FileCreateMode directive. Order of lines is vitally important.<br></li><br> <li><strong>$FailOnCHOwnFailure </strong>on/off [default on]<br> - This option modifies behaviour of dynaFile creation. If different owners or groups are specified for new files or directories and rsyslogd fails to set these new owners or groups, it will log an error and NOT write to the file in question if that option is set to "on". If it is set to "off", the error will be ignored and processing continues. Keep in mind, that the files in this case may be (in)accessible by people who should not have permission. The default is "on".<br></li><br> + This option modifies behaviour of file creation. If different owners or groups are specified for new files or directories and rsyslogd fails to set these new owners or groups, it will log an error and NOT write to the file in question if that option is set to "on". If it is set to "off", the error will be ignored and processing continues. Keep in mind, that the files in this case may be (in)accessible by people who should not have permission. The default is "on".<br></li><br> <li><strong>$F$OMFileForceCHOwn </strong><br> force ownership change for all files<br></li><br> diff --git a/doc/rsyslog_conf_modules.html b/doc/rsyslog_conf_modules.html index 2668bf2d..18d6b8a1 100644 --- a/doc/rsyslog_conf_modules.html +++ b/doc/rsyslog_conf_modules.html @@ -45,6 +45,7 @@ to message generators. <li><a href="imsolaris.html">imsolaris</a> - input for the Sun Solaris system log source</li> <li><a href="im3195.html">im3195</a> - accepts syslog messages via RFC 3195</li> <li><a href="impstats.html">impstats</a> - provides periodic statistics of rsyslog internal counters</li> +<li><a href="imjournal.html">imjournal</a> - Linux journal inuput module</li> </ul> <a name"om"></a><h2>Output Modules</h2> diff --git a/plugins/imgssapi/imgssapi.c b/plugins/imgssapi/imgssapi.c index 4e3a70ab..15d994cc 100644 --- a/plugins/imgssapi/imgssapi.c +++ b/plugins/imgssapi/imgssapi.c @@ -72,6 +72,7 @@ MODULE_CNFNAME("imgssapi") /* some forward definitions - they may go away when we no longer include imtcp.c */ static rsRetVal addGSSListener(void __attribute__((unused)) *pVal, uchar *pNewVal); +static rsRetVal actGSSListener(uchar *port); static int TCPSessGSSInit(void); static void TCPSessGSSClose(tcps_sess_t* pSess); static rsRetVal TCPSessGSSRecv(tcps_sess_t *pSess, void *buf, size_t buf_len, ssize_t *); @@ -90,6 +91,7 @@ DEFobjCurrIf(glbl) static tcpsrv_t *pOurTcpsrv = NULL; /* our TCP server(listener) TODO: change for multiple instances */ static gss_cred_id_t gss_server_creds = GSS_C_NO_CREDENTIAL; +static uchar *srvPort; /* our usr structure for the tcpsrv object */ typedef struct gsssrv_s { @@ -317,6 +319,16 @@ static rsRetVal addGSSListener(void __attribute__((unused)) *pVal, uchar *pNewVal) { DEFiRet; + + srvPort = pNewVal; + + RETiRet; +} + +static rsRetVal +actGSSListener(uchar *port) +{ + DEFiRet; gsssrv_t *pGSrv; if(pOurTcpsrv == NULL) { @@ -340,7 +352,7 @@ addGSSListener(void __attribute__((unused)) *pVal, uchar *pNewVal) CHKiRet(tcpsrv.SetCBOnRegularClose(pOurTcpsrv, onRegularClose)); CHKiRet(tcpsrv.SetCBOnErrClose(pOurTcpsrv, onErrClose)); CHKiRet(tcpsrv.SetInputName(pOurTcpsrv, UCHAR_CONSTANT("imgssapi"))); - tcpsrv.configureTCPListen(pOurTcpsrv, pNewVal, 1); + tcpsrv.configureTCPListen(pOurTcpsrv, port, 1); CHKiRet(tcpsrv.ConstructFinalize(pOurTcpsrv)); } @@ -676,6 +688,11 @@ ENDfreeCnf */ BEGINrunInput CODESTARTrunInput + /* This will fail if the priviledges are dropped. Should be + * moved to the '*activateCnfPrePrivDrop' section eventually. + */ + actGSSListener(srvPort); + iRet = tcpsrv.Run(pOurTcpsrv); ENDrunInput @@ -683,7 +700,7 @@ ENDrunInput /* initialize and return if will run or not */ BEGINwillRun CODESTARTwillRun - if(pOurTcpsrv == NULL) + if(srvPort == NULL) ABORT_FINALIZE(RS_RET_NO_RUN); net.PrintAllowedSenders(2); /* TCP */ diff --git a/plugins/imjournal/imjournal.c b/plugins/imjournal/imjournal.c index 2af1958e..ae29154d 100644..100755 --- a/plugins/imjournal/imjournal.c +++ b/plugins/imjournal/imjournal.c @@ -30,7 +30,9 @@ #include <ctype.h> #include <stdlib.h> #include <time.h> +#include <sys/poll.h> #include <sys/socket.h> +#include <errno.h> #include "dirty.h" #include "cfsysline.h" @@ -43,6 +45,7 @@ #include "glbl.h" #include "prop.h" #include "errmsg.h" +#include "srUtils.h" #include "unicode-helper.h" #include <systemd/sd-journal.h> @@ -136,16 +139,20 @@ readjournal() { uint64_t timestamp; struct json_object *json = NULL; + int r; /* Information from messages */ char *message; + char *sys_pid; char *sys_iden; char *sys_iden_help; const void *get; + const void *pidget; char *parse; char *get2; size_t length; + size_t pidlength; const void *equal_sign; struct json_object *jval; @@ -158,13 +165,6 @@ readjournal() { int priority = 0; int facility = 0; - /* Get next journal message, if there is none, wait a second */ - if (sd_journal_next(j) == 0) { - sleep(1); - iRet = RS_RET_OK; - goto ret; - } - /* Get message text */ if (sd_journal_get_data(j, "MESSAGE", &get, &length) < 0) { logmsgInternal(NO_ERRCODE, LOG_SYSLOG|LOG_INFO, (uchar *)"log message from journal doesn't have MESSAGE", 0); @@ -203,7 +203,7 @@ readjournal() { goto free_message; } - /* Get message identifier and add ':' */ + /* Get message identifier, client pid and add ':' */ if (sd_journal_get_data(j, "SYSLOG_IDENTIFIER", &get, &length) >= 0) { sys_iden = strndup(get+18, length-18); } else { @@ -214,12 +214,30 @@ readjournal() { goto free_message; } - asprintf(&sys_iden_help, "%s:", sys_iden); - if (sys_iden_help == NULL) { + if (sd_journal_get_data(j, "SYSLOG_PID", &pidget, &pidlength) >= 0) { + sys_pid = strndup(pidget+11, pidlength-11); + if (sys_pid == NULL) { + iRet = RS_RET_OUT_OF_MEMORY; + free (sys_iden); + goto free_message; + } + } else { + sys_pid = NULL; + } + + if (sys_pid) { + r = asprintf(&sys_iden_help, "%s[%s]:", sys_iden, sys_pid); + } else { + r = asprintf(&sys_iden_help, "%s:", sys_iden); + } + + free (sys_iden); + free (sys_pid); + + if (-1 == r) { iRet = RS_RET_OUT_OF_MEMORY; goto finalize_it; } - free (sys_iden); json = json_object_new_object(); @@ -337,7 +355,9 @@ persistJournalState () { char *cursor; int ret = 0; - if ((ret = sd_journal_get_cursor(j, &cursor)) > 0) { + /* On success, sd_journal_get_cursor() returns 1 in systemd + 197 or older and 0 in systemd 198 or newer */ + if ((ret = sd_journal_get_cursor(j, &cursor)) >= 0) { if ((sf = fopen(cs.stateFile, "wb")) != NULL) { if (fprintf(sf, "%s", cursor) < 0) { iRet = RS_RET_IO_ERROR; @@ -345,29 +365,92 @@ persistJournalState () { fclose(sf); free(cursor); } else { + char errStr[256]; + rs_strerror_r(errno, errStr, sizeof(errStr)); + errmsg.LogError(0, RS_RET_FOPEN_FAILURE, "fopen() failed: " + "'%s', path: '%s'\n", errStr, cs.stateFile); iRet = RS_RET_FOPEN_FAILURE; } } else { + char errStr[256]; + rs_strerror_r(-(ret), errStr, sizeof(errStr)); + errmsg.LogError(0, RS_RET_ERR, "sd_journal_get_cursor() failed: '%s'\n", errStr); iRet = RS_RET_ERR; } RETiRet; } +/* Polls the journal for new messages. Similar to sd_journal_wait() + * except for the special handling of EINTR. + */ +static rsRetVal +pollJournal() +{ + DEFiRet; + struct pollfd pollfd; + int r; + + pollfd.fd = sd_journal_get_fd(j); + pollfd.events = sd_journal_get_events(j); + r = poll(&pollfd, 1, -1); + if (r == -1) { + if (errno == EINTR) { + /* EINTR is also received during termination + * so return now to check the term state. + */ + ABORT_FINALIZE(RS_RET_OK); + } else { + char errStr[256]; + + rs_strerror_r(errno, errStr, sizeof(errStr)); + errmsg.LogError(0, RS_RET_ERR, + "poll() failed: '%s'", errStr); + ABORT_FINALIZE(RS_RET_ERR); + } + } + + assert(r == 1); + + r = sd_journal_process(j); + if (r < 0) { + char errStr[256]; + + rs_strerror_r(errno, errStr, sizeof(errStr)); + errmsg.LogError(0, RS_RET_ERR, + "sd_journal_process() failed: '%s'", errStr); + ABORT_FINALIZE(RS_RET_ERR); + } + +finalize_it: + RETiRet; +} + + BEGINrunInput CODESTARTrunInput /* this is an endless loop - it is terminated when the thread is * signalled to do so. This, however, is handled by the framework, * right into the sleep below. */ - int count = 0; + if (cs.stateFile[0] != '/') { + char *new_stateFile; - char readCursor[128 + 1]; - FILE *r_sf; + if (-1 == asprintf(&new_stateFile, "%s/%s", (char *)glbl.GetWorkDir(), cs.stateFile)) { + errmsg.LogError(0, RS_RET_OUT_OF_MEMORY, "imjournal: asprintf failed\n"); + ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + } + free (cs.stateFile); + cs.stateFile = new_stateFile; + } /* if state file exists, set cursor to appropriate position */ if (access(cs.stateFile, F_OK|R_OK) != -1) { + FILE *r_sf; + if ((r_sf = fopen(cs.stateFile, "rb")) != NULL) { + char readCursor[128 + 1]; + if (fscanf(r_sf, "%128s\n", readCursor) != EOF) { if (sd_journal_seek_cursor(j, readCursor) != 0) { errmsg.LogError(0, RS_RET_ERR, "imjournal: " @@ -390,14 +473,32 @@ CODESTARTrunInput } while (glbl.GetGlobalInputTermState() == 0) { + int count = 0, r; + + r = sd_journal_next(j); + if (r < 0) { + char errStr[256]; + + rs_strerror_r(errno, errStr, sizeof(errStr)); + errmsg.LogError(0, RS_RET_ERR, + "sd_journal_next() failed: '%s'", errStr); + ABORT_FINALIZE(RS_RET_ERR); + } + + if (r == 0) { + /* No new messages, wait for activity. */ + CHKiRet(pollJournal()); + continue; + } + CHKiRet(readjournal()); + /* TODO: This could use some finer metric. */ count++; if (count == cs.iPersistStateInterval) { count = 0; persistJournalState(); } } - persistJournalState(); finalize_it: ENDrunInput @@ -444,6 +545,7 @@ ENDwillRun /* close journal */ BEGINafterRun CODESTARTafterRun + persistJournalState(); sd_journal_close(j); ENDafterRun @@ -500,16 +602,22 @@ finalize_it: ENDsetModCnf +BEGINisCompatibleWithFeature +CODESTARTisCompatibleWithFeature + if(eFeat == sFEATURENonCancelInputTermination) + iRet = RS_RET_OK; +ENDisCompatibleWithFeature + + BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_IMOD_QUERIES CODEqueryEtryPt_STD_CONF2_QUERIES CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES +CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES ENDqueryEtryPt - - BEGINmodInit() CODESTARTmodInit *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */ diff --git a/plugins/imzmq3/README b/plugins/imzmq3/README index 88653b83..9a108a01 100644 --- a/plugins/imzmq3/README +++ b/plugins/imzmq3/README @@ -1,24 +1,59 @@ ZeroMQ 3.x Input Plugin Building this plugin: -Requires libzmq and libczmq. First, install libzmq from the HEAD on github: -http://github.com/zeromq/libzmq. You can clone the repository, build, then -install it. The directions for doing so are there in the readme. Then, do -the same for libczmq: http://github.com/zeromq/czmq. At some point, the 3.1 -version of libzmq will be released, and a supporting version of libczmq. -At that time, you could simply download and install the tarballs instead of -using git to clone the repositories. Those tarballs (when available) can -be found at http://download.zeromq.org. As of this writing (5/31/2012), the -most recent version of czmq (1.1.0) and libzmq (3.1.0-beta) will not compile -properly. +Requires libzmq and libczmq. First, download the tarballs of both libzmq +and its supporting libczmq from http://download.zeromq.org. As of this +writing (04/23/2013), the most recent versions of libzmq and czmq are +3.2.2 and 1.3.2 respectively. Configure, build, and then install both libs. Imzmq3 allows you to push data into rsyslog from a zeromq socket. The example below binds a SUB socket to port 7172, and then any messages with the topic "foo" will be pushed into rsyslog. +Please note: +This plugin only supports the newer (v7) config format. Legacy config support +was removed. + Example Rsyslog.conf snippet: ------------------------------------------------------------------------------- - -$InputZmq3ServerRun action=BIND,type=SUB,description=tcp://*:7172,subscribe=foo +module(load="imzmq3" ioThreads="1") +input(type="imzmq3" action="CONNECT" socktype="SUB" description="tcp://*:7172" subscribe="foo,bar") ------------------------------------------------------------------------------- +Note you can specify multiple subscriptions with a comma-delimited list, with +no spaces between values. + +The only global parameter for this plugin is ioThreads, which is optional and +probably best left to the zmq default unless you know exactly what you are +doing. + +The instance-level parameters are: + +Required +description +subscribe (required if the sockType is SUB) + +Optional +sockType (defaults to SUB) +action (defaults to BIND +sndHWM +rcvHWM +identity +sndBuf +rcvBuf +linger +backlog +sndTimeout +rcvTimeout +maxMsgSize +rate +recoveryIVL +multicastHops +reconnectIVL +reconnectIVLMax +ipv4Only +affinity + +These all correspond to zmq optional settings. Except where noted, the defaults +are the zmq defaults if not set. See http://api.zeromq.org/3-2:zmq-setsockopt +for info on these. diff --git a/plugins/imzmq3/imzmq3.c b/plugins/imzmq3/imzmq3.c index 52c12a53..52ca5ebe 100644 --- a/plugins/imzmq3/imzmq3.c +++ b/plugins/imzmq3/imzmq3.c @@ -19,20 +19,21 @@ * License along with this program. If not, see * <http://www.gnu.org/licenses/>. * - * Author: David Kelly - * <davidk@talksum.com> + * Authors: + * David Kelly <davidk@talksum.com> + * Hongfei Cheng <hongfeic@talksum.com> */ + +#include "config.h" +#include "rsyslog.h" + #include <assert.h> #include <errno.h> #include <stdlib.h> #include <string.h> #include <unistd.h> - -#include "rsyslog.h" - #include "cfsysline.h" -#include "config.h" #include "dirty.h" #include "errmsg.h" #include "glbl.h" @@ -49,6 +50,7 @@ MODULE_TYPE_INPUT MODULE_TYPE_NOKEEP +MODULE_CNFNAME("imzmq3"); /* convienent symbols to denote a socket we want to bind * vs one we want to just connect to @@ -83,47 +85,68 @@ typedef struct _poller_data { thrdInfo_t* thread; } poller_data; -typedef struct _socket_info { - int type; - int action; - char* description; - int sndHWM; /* if you want more than 2^32 messages, */ - int rcvHWM; /* then pass in 0 (the default). */ - char* identity; - char** subscriptions; - ruleset_t* ruleset; - int sndBuf; - int rcvBuf; - int linger; - int backlog; - int sndTimeout; - int rcvTimeout; - int maxMsgSize; - int rate; - int recoveryIVL; - int multicastHops; - int reconnectIVL; - int reconnectIVLMax; - int ipv4Only; - int affinity; - -} socket_info; +/* a linked-list of subscription topics */ +typedef struct sublist_t { + char* subscribe; + struct sublist_t* next; +} sublist; + +struct instanceConf_s { + int type; + int action; + char* description; + int sndHWM; /* if you want more than 2^32 messages, */ + int rcvHWM; /* then pass in 0 (the default). */ + char* identity; + sublist* subscriptions; + int sndBuf; + int rcvBuf; + int linger; + int backlog; + int sndTimeout; + int rcvTimeout; + int maxMsgSize; + int rate; + int recoveryIVL; + int multicastHops; + int reconnectIVL; + int reconnectIVLMax; + int ipv4Only; + int affinity; + uchar* pszBindRuleset; + ruleset_t* pBindRuleset; + struct instanceConf_s* next; + +}; + +struct modConfData_s { + rsconf_t* pConf; + instanceConf_t* root; + instanceConf_t* tail; + int io_threads; +}; +struct lstn_s { + struct lstn_s* next; + void* sock; + ruleset_t* pRuleset; +}; /* ---------------------------------------------------------------------------- * Static definitions/initializations. */ -static socket_info* s_socketInfo = NULL; -static size_t s_nitems = 0; -static prop_t * s_namep = NULL; +static modConfData_t* loadModConf = NULL; +static modConfData_t* runModConf = NULL; +static struct lstn_s* lcnfRoot = NULL; +static struct lstn_s* lcnfLast = NULL; +static prop_t* s_namep = NULL; static zloop_t* s_zloop = NULL; -static int s_io_threads = 1; static zctx_t* s_context = NULL; -static ruleset_t* s_ruleset = NULL; static socket_type socketTypes[] = { - {"SUB", ZMQ_SUB }, - {"PULL", ZMQ_PULL }, - {"XSUB", ZMQ_XSUB } + {"SUB", ZMQ_SUB }, + {"PULL", ZMQ_PULL }, + {"ROUTER", ZMQ_ROUTER }, + {"XSUB", ZMQ_XSUB } }; static socket_action socketActions[] = { @@ -131,6 +154,48 @@ static socket_action socketActions[] = { {"CONNECT", ACTION_CONNECT}, }; +static struct cnfparamdescr modpdescr[] = { + { "ioThreads", eCmdHdlrInt, 0 }, +}; + +static struct cnfparamblk modpblk = { + CNFPARAMBLK_VERSION, + sizeof(modpdescr)/sizeof(struct cnfparamdescr), + modpdescr +}; + +static struct cnfparamdescr inppdescr[] = { + { "description", eCmdHdlrGetWord, 0 }, + { "sockType", eCmdHdlrGetWord, 0 }, + { "subscribe", eCmdHdlrGetWord, 0 }, + { "ruleset", eCmdHdlrGetWord, 0 }, + { "action", eCmdHdlrGetWord, 0 }, + { "sndHWM", eCmdHdlrInt, 0 }, + { "rcvHWM", eCmdHdlrInt, 0 }, + { "identity", eCmdHdlrGetWord, 0 }, + { "sndBuf", eCmdHdlrInt, 0 }, + { "rcvBuf", eCmdHdlrInt, 0 }, + { "linger", eCmdHdlrInt, 0 }, + { "backlog", eCmdHdlrInt, 0 }, + { "sndTimeout", eCmdHdlrInt, 0 }, + { "rcvTimeout", eCmdHdlrInt, 0 }, + { "maxMsgSize", eCmdHdlrInt, 0 }, + { "rate", eCmdHdlrInt, 0 }, + { "recoveryIVL", eCmdHdlrInt, 0 }, + { "multicastHops", eCmdHdlrInt, 0 }, + { "reconnectIVL", eCmdHdlrInt, 0 }, + { "reconnectIVLMax", eCmdHdlrInt, 0 }, + { "ipv4Only", eCmdHdlrInt, 0 }, + { "affinity", eCmdHdlrInt, 0 } +}; + +static struct cnfparamblk inppblk = { + CNFPARAMBLK_VERSION, + sizeof(inppdescr)/sizeof(struct cnfparamdescr), + inppdescr +}; + +#include "im-helper.h" /* must be included AFTER the type definitions! */ /* ---------------------------------------------------------------------------- * Helper functions @@ -179,15 +244,16 @@ static int getSocketAction(char* name) { } -static void setDefaults(socket_info* info) { - info->type = ZMQ_SUB; - info->action = ACTION_BIND; +static void setDefaults(instanceConf_t* info) { + info->type = -1; + info->action = -1; info->description = NULL; - info->sndHWM = 0; - info->rcvHWM = 0; + info->sndHWM = -1; + info->rcvHWM = -1; info->identity = NULL; info->subscriptions = NULL; - info->ruleset = NULL; + info->pszBindRuleset = NULL; + info->pBindRuleset = NULL; info->sndBuf = -1; info->rcvBuf = -1; info->linger = -1; @@ -205,90 +271,46 @@ static void setDefaults(socket_info* info) { }; - -/* The config string should look like: - * "action=AAA,type=TTT,description=DDD,sndHWM=SSS,rcvHWM=RRR,subscribe='xxx',subscribe='yyy'" - * +/* given a comma separated list of subscriptions, create a char* array of them + * to set later */ -static rsRetVal parseConfig(char* config, socket_info* info) { - int nsubs = 0; - - char* binding; - char* ptr1; - for (binding = strtok_r(config, ",", &ptr1); - binding != NULL; - binding = strtok_r(NULL, ",", &ptr1)) { - - /* Each binding looks like foo=bar */ - char * sep = strchr(binding, '='); - if (sep == NULL) - { - errmsg.LogError(0, NO_ERRCODE, - "Invalid argument format %s, ignoring ...", - binding); - continue; - } +static rsRetVal parseSubscriptions(char* subscribes, sublist** subList){ + char* tok = strtok(subscribes, ","); + sublist* currentSub; + sublist* head; + DEFiRet; - /* Replace '=' with '\0'. */ - *sep = '\0'; - - char * val = sep + 1; - - if (strcmp(binding, "action") == 0) { - info->action = getSocketAction(val); - } else if (strcmp(binding, "type") == 0) { - info->type = getSocketType(val); - } else if (strcmp(binding, "description") == 0) { - info->description = strdup(val); - } else if (strcmp(binding, "sndHWM") == 0) { - info->sndHWM = atoi(val); - } else if (strcmp(binding, "rcvHWM") == 0) { - info->sndHWM = atoi(val); - } else if (strcmp(binding, "subscribe") == 0) { - /* Add the subscription value to the list.*/ - char * substr = NULL; - substr = strdup(val); - info->subscriptions = realloc(info->subscriptions, sizeof(char *) * nsubs + 1); - info->subscriptions[nsubs] = substr; - ++nsubs; - } else if (strcmp(binding, "sndBuf") == 0) { - info->sndBuf = atoi(val); - } else if (strcmp(binding, "rcvBuf") == 0) { - info->rcvBuf = atoi(val); - } else if (strcmp(binding, "linger") == 0) { - info->linger = atoi(val); - } else if (strcmp(binding, "backlog") == 0) { - info->backlog = atoi(val); - } else if (strcmp(binding, "sndTimeout") == 0) { - info->sndTimeout = atoi(val); - } else if (strcmp(binding, "rcvTimeout") == 0) { - info->rcvTimeout = atoi(val); - } else if (strcmp(binding, "maxMsgSize") == 0) { - info->maxMsgSize = atoi(val); - } else if (strcmp(binding, "rate") == 0) { - info->rate = atoi(val); - } else if (strcmp(binding, "recoveryIVL") == 0) { - info->recoveryIVL = atoi(val); - } else if (strcmp(binding, "multicastHops") == 0) { - info->multicastHops = atoi(val); - } else if (strcmp(binding, "reconnectIVL") == 0) { - info->reconnectIVL = atoi(val); - } else if (strcmp(binding, "reconnectIVLMax") == 0) { - info->reconnectIVLMax = atoi(val); - } else if (strcmp(binding, "ipv4Only") == 0) { - info->ipv4Only = atoi(val); - } else if (strcmp(binding, "affinity") == 0) { - info->affinity = atoi(val); - } else { - errmsg.LogError(0, NO_ERRCODE, "Unknown argument %s", binding); - return RS_RET_INVALID_PARAMS; + /* create empty list */ + CHKmalloc(*subList = (sublist*)MALLOC(sizeof(sublist))); + head = *subList; + head->next = NULL; + head->subscribe=NULL; + currentSub=head; + + if(tok) { + head->subscribe=strdup(tok); + for(tok=strtok(NULL, ","); tok!=NULL;tok=strtok(NULL, ",")) { + CHKmalloc(currentSub->next = (sublist*)MALLOC(sizeof(sublist))); + currentSub=currentSub->next; + currentSub->subscribe=strdup(tok); + currentSub->next=NULL; } + } else { + /* make empty subscription ie subscribe="" */ + head->subscribe=strdup(""); } - - return RS_RET_OK; + /* TODO: temporary logging */ + currentSub = head; + DBGPRINTF("imzmq3: Subscriptions:"); + for(currentSub = head; currentSub != NULL; currentSub=currentSub->next) { + DBGPRINTF("'%s'", currentSub->subscribe); + } + DBGPRINTF("\n"); +finalize_it: + RETiRet; } -static rsRetVal validateConfig(socket_info* info) { +static rsRetVal validateConfig(instanceConf_t* info) { if (info->type == -1) { errmsg.LogError(0, RS_RET_INVALID_PARAMS, @@ -307,7 +329,7 @@ static rsRetVal validateConfig(socket_info* info) { } if(info->type == ZMQ_SUB && info->subscriptions == NULL) { errmsg.LogError(0, RS_RET_INVALID_PARAMS, - "SUB sockets need at least one subscription"); + "SUB sockets need a subscription"); return RS_RET_INVALID_PARAMS; } if(info->type != ZMQ_SUB && info->subscriptions != NULL) { @@ -320,39 +342,40 @@ static rsRetVal validateConfig(socket_info* info) { static rsRetVal createContext() { if (s_context == NULL) { - errmsg.LogError(0, NO_ERRCODE, "creating zctx."); + DBGPRINTF("imzmq3: creating zctx..."); + zsys_handler_set(NULL); s_context = zctx_new(); if (s_context == NULL) { errmsg.LogError(0, RS_RET_INVALID_PARAMS, "zctx_new failed: %s", - strerror(errno)); + zmq_strerror(errno)); /* DK: really should do better than invalid params...*/ return RS_RET_INVALID_PARAMS; } - - if (s_io_threads > 1) { - errmsg.LogError(0, NO_ERRCODE, "setting io worker threads to %d", s_io_threads); - zctx_set_iothreads(s_context, s_io_threads); + DBGPRINTF("success!\n"); + if (runModConf->io_threads > 1) { + DBGPRINTF("setting io worker threads to %d\n", runModConf->io_threads); + zctx_set_iothreads(s_context, runModConf->io_threads); } } return RS_RET_OK; } -static rsRetVal createSocket(socket_info* info, void** sock) { - size_t ii; +static rsRetVal createSocket(instanceConf_t* info, void** sock) { int rv; + sublist* sub; *sock = zsocket_new(s_context, info->type); if (!sock) { - errmsg.LogError(0, + errmsg.LogError(0, RS_RET_INVALID_PARAMS, "zsocket_new failed: %s, for type %d", - strerror(errno),info->type); - /* DK: invalid params seems right here */ + zmq_strerror(errno),info->type); + /* DK: invalid params seems right here */ return RS_RET_INVALID_PARAMS; } - + DBGPRINTF("imzmq3: socket of type %d created successfully\n", info->type) /* Set options *before* the connect/bind. */ if (info->identity) zsocket_set_identity(*sock, info->identity); if (info->sndBuf > -1) zsocket_set_sndbuf(*sock, info->sndBuf); @@ -369,38 +392,36 @@ static rsRetVal createSocket(socket_info* info, void** sock) { if (info->reconnectIVLMax > -1) zsocket_set_reconnect_ivl_max(*sock, info->reconnectIVLMax); if (info->ipv4Only > -1) zsocket_set_ipv4only(*sock, info->ipv4Only); if (info->affinity > -1) zsocket_set_affinity(*sock, info->affinity); - - /* since HWM have defaults, we always set them. No return codes to check, either.*/ - zsocket_set_sndhwm(*sock, info->sndHWM); - zsocket_set_rcvhwm(*sock, info->rcvHWM); - + if (info->sndHWM > -1 ) zsocket_set_sndhwm(*sock, info->sndHWM); + if (info->rcvHWM > -1 ) zsocket_set_rcvhwm(*sock, info->rcvHWM); /* Set subscriptions.*/ if (info->type == ZMQ_SUB) { - for (ii = 0; ii < sizeof(info->subscriptions)/sizeof(char*); ++ii) - zsocket_set_subscribe(*sock, info->subscriptions[ii]); + for(sub = info->subscriptions; sub!=NULL; sub=sub->next) { + zsocket_set_subscribe(*sock, sub->subscribe); + } } - - /* Do the bind/connect... */ if (info->action==ACTION_CONNECT) { rv = zsocket_connect(*sock, info->description); - if (rv < 0) { + if (rv == -1) { errmsg.LogError(0, RS_RET_INVALID_PARAMS, "zmq_connect using %s failed: %s", - info->description, strerror(errno)); + info->description, zmq_strerror(errno)); return RS_RET_INVALID_PARAMS; } + DBGPRINTF("imzmq3: connect for %s successful\n",info->description); } else { rv = zsocket_bind(*sock, info->description); - if (rv <= 0) { + if (rv == -1) { errmsg.LogError(0, RS_RET_INVALID_PARAMS, "zmq_bind using %s failed: %s", - info->description, strerror(errno)); + info->description, zmq_strerror(errno)); return RS_RET_INVALID_PARAMS; } + DBGPRINTF("imzmq3: bind for %s successful\n",info->description); } return RS_RET_OK; } @@ -409,89 +430,138 @@ static rsRetVal createSocket(socket_info* info, void** sock) { * Module endpoints */ -/* accept a new ruleset to bind. Checks if it exists and complains, if not. Note - * that this makes the assumption that after the bind ruleset is called in the config, - * another call will be made to add an endpoint. -*/ -static rsRetVal -set_ruleset(void __attribute__((unused)) *pVal, uchar *pszName) { - ruleset_t* ruleset_ptr; - rsRetVal localRet; - DEFiRet; - - localRet = ruleset.GetRuleset(ourConf, &ruleset_ptr, pszName); - if(localRet == RS_RET_NOT_FOUND) { - errmsg.LogError(0, NO_ERRCODE, "error: " - "ruleset '%s' not found - ignored", pszName); - } - CHKiRet(localRet); - s_ruleset = ruleset_ptr; - DBGPRINTF("imzmq3 current bind ruleset '%s'\n", pszName); - -finalize_it: - free(pszName); /* no longer needed */ - RETiRet; -} /* add an actual endpoint */ -static rsRetVal add_endpoint(void __attribute__((unused)) * oldp, uchar * valp) { +static rsRetVal createInstance(instanceConf_t** pinst) { DEFiRet; + instanceConf_t* inst; + CHKmalloc(inst = MALLOC(sizeof(instanceConf_t))); - /* increment number of items and store old num items, as it will be handy.*/ - size_t idx = s_nitems++; - - /* allocate a new socket_info array to accomidate this new endpoint*/ - socket_info* tmpSocketInfo; - CHKmalloc(tmpSocketInfo = (socket_info*)MALLOC(sizeof(socket_info) * s_nitems)); + /* set defaults into new instance config struct */ + setDefaults(inst); - /* copy existing socket_info across into new array, if any, and free old storage*/ - if(idx) { - memcpy(tmpSocketInfo, s_socketInfo, sizeof(socket_info) * idx); - free(s_socketInfo); + /* add this to the config */ + if(loadModConf->tail == NULL) { + loadModConf->tail = loadModConf->root = inst; + } else { + loadModConf->tail->next = inst; + loadModConf->tail = inst; } + *pinst = inst; +finalize_it: + RETiRet; +} - /* set the static to hold the new array */ - s_socketInfo = tmpSocketInfo; - - /* point to the new one */ - socket_info* sockInfo = &s_socketInfo[idx]; - - /* set defaults for the new socket info */ - setDefaults(sockInfo); - - /* Make a writeable copy of the string so we can use strtok - in the parseConfig call */ - char * copy = NULL; - CHKmalloc(copy = strdup((char *) valp)); +static rsRetVal createListener(struct cnfparamvals* pvals) { + instanceConf_t* inst; + int i; + DEFiRet; - /* parse the config string */ - CHKiRet(parseConfig(copy, sockInfo)); + CHKiRet(createInstance(&inst)); + for(i = 0 ; i < inppblk.nParams ; ++i) { + if(!pvals[i].bUsed) + continue; + if(!strcmp(inppblk.descr[i].name, "ruleset")) { + inst->pszBindRuleset = (uchar *)es_str2cstr(pvals[i].val.d.estr, NULL); + } else if(!strcmp(inppblk.descr[i].name, "description")) { + inst->description = es_str2cstr(pvals[i].val.d.estr, NULL); + } else if(!strcmp(inppblk.descr[i].name, "sockType")){ + inst->type = getSocketType(es_str2cstr(pvals[i].val.d.estr, NULL)); + } else if(!strcmp(inppblk.descr[i].name, "action")){ + inst->action = getSocketAction(es_str2cstr(pvals[i].val.d.estr, NULL)); + } else if(!strcmp(inppblk.descr[i].name, "sndHWM")) { + inst->sndHWM = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "rcvHWM")) { + inst->rcvHWM = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "subscribe")) { + CHKiRet(parseSubscriptions(es_str2cstr(pvals[i].val.d.estr, NULL), + &inst->subscriptions)); + } else if(!strcmp(inppblk.descr[i].name, "identity")){ + inst->identity = es_str2cstr(pvals[i].val.d.estr, NULL); + } else if(!strcmp(inppblk.descr[i].name, "sndBuf")) { + inst->sndBuf = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "rcvBuf")) { + inst->rcvBuf = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "linger")) { + inst->linger = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "backlog")) { + inst->backlog = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "sndTimeout")) { + inst->sndTimeout = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "rcvTimeout")) { + inst->rcvTimeout = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "maxMsgSize")) { + inst->maxMsgSize = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "rate")) { + inst->rate = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "recoveryIVL")) { + inst->recoveryIVL = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "multicastHops")) { + inst->multicastHops = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "reconnectIVL")) { + inst->reconnectIVL = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "reconnectIVLMax")) { + inst->reconnectIVLMax = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "ipv4Only")) { + inst->ipv4Only = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "affinity")) { + inst->affinity = (int) pvals[i].val.d.n; + } else { + errmsg.LogError(0, NO_ERRCODE, "imzmq3: program error, non-handled " + "param '%s'\n", inppblk.descr[i].name); + } - /* validate it */ - CHKiRet(validateConfig(sockInfo)); + } +finalize_it: + RETiRet; +} - /* bind to the current ruleset (if any)*/ - sockInfo->ruleset = s_ruleset; +static rsRetVal addListener(instanceConf_t* inst){ + /* create the socket */ + void* sock; + struct lstn_s* newcnfinfo; + DEFiRet; + + CHKiRet(createSocket(inst, &sock)); + + /* now create new lstn_s struct */ + CHKmalloc(newcnfinfo=(struct lstn_s*)MALLOC(sizeof(struct lstn_s))); + newcnfinfo->next = NULL; + newcnfinfo->sock = sock; + newcnfinfo->pRuleset = inst->pBindRuleset; + /* add this struct to the global */ + if(lcnfRoot == NULL) { + lcnfRoot = newcnfinfo; + } + if(lcnfLast == NULL) { + lcnfLast = newcnfinfo; + } else { + lcnfLast->next = newcnfinfo; + lcnfLast = newcnfinfo; + } + finalize_it: - free(valp); /* in any case, this is no longer needed */ - RETiRet; + RETiRet; } - static int handlePoll(zloop_t __attribute__((unused)) * loop, zmq_pollitem_t *poller, void* pd) { - msg_t* logmsg; + msg_t* pMsg; poller_data* pollerData = (poller_data*)pd; char* buf = zstr_recv(poller->socket); - if (msgConstruct(&logmsg) == RS_RET_OK) { - MsgSetRawMsg(logmsg, buf, strlen(buf)); - MsgSetInputName(logmsg, s_namep); - MsgSetFlowControlType(logmsg, eFLOWCTL_NO_DELAY); - MsgSetRuleset(logmsg, pollerData->ruleset); - logmsg->msgFlags = NEEDS_PARSING; - submitMsg(logmsg); + if (msgConstruct(&pMsg) == RS_RET_OK) { + MsgSetRawMsg(pMsg, buf, strlen(buf)); + MsgSetInputName(pMsg, s_namep); + MsgSetHOSTNAME(pMsg, glbl.GetLocalHostName(), ustrlen(glbl.GetLocalHostName())); + MsgSetRcvFrom(pMsg, glbl.GetLocalHostNameProp()); + MsgSetRcvFromIP(pMsg, glbl.GetLocalHostIP()); + MsgSetMSGoffs(pMsg, 0); + MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY); + MsgSetRuleset(pMsg, pollerData->ruleset); + pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME; + submitMsg2(pMsg); } /* gotta free the string returned from zstr_recv() */ @@ -510,51 +580,65 @@ static int handlePoll(zloop_t __attribute__((unused)) * loop, zmq_pollitem_t *po /* called when runInput is called by rsyslog */ static rsRetVal rcv_loop(thrdInfo_t* pThrd){ + size_t n_items = 0; size_t i; int rv; - zmq_pollitem_t* items; - poller_data* pollerData; - + zmq_pollitem_t* items = NULL; + poller_data* pollerData = NULL; + struct lstn_s* current; + instanceConf_t* inst; DEFiRet; - - /* create the context*/ - CHKiRet(createContext()); + /* now add listeners. This actually creates the sockets, etc... */ + for (inst = runModConf->root; inst != NULL; inst=inst->next) { + addListener(inst); + } + if (lcnfRoot == NULL) { + errmsg.LogError(0, NO_ERRCODE, "imzmq3: no listeners were " + "started, input not activated.\n"); + ABORT_FINALIZE(RS_RET_NO_RUN); + } + + /* count the # of items first */ + for(current=lcnfRoot;current!=NULL;current=current->next) + n_items++; + + /* make arrays of pollitems, pollerdata so they are easy to delete later */ + /* create the poll items*/ - CHKmalloc(items = (zmq_pollitem_t*)MALLOC(sizeof(zmq_pollitem_t)*s_nitems)); + CHKmalloc(items = (zmq_pollitem_t*)MALLOC(sizeof(zmq_pollitem_t)*n_items)); /* create poller data (stuff to pass into the zmq closure called when we get a message)*/ - CHKmalloc(pollerData = (poller_data*)MALLOC(sizeof(poller_data)*s_nitems)); + CHKmalloc(pollerData = (poller_data*)MALLOC(sizeof(poller_data)*n_items)); /* loop through and initialize the poll items and poller_data arrays...*/ - for(i=0; i<s_nitems;++i) { + for(i=0, current = lcnfRoot; current != NULL; current = current->next, i++) { /* create the socket, update items.*/ - createSocket(&s_socketInfo[i], &items[i].socket); + items[i].socket=current->sock; items[i].events = ZMQ_POLLIN; /* now update the poller_data for this item */ pollerData[i].thread = pThrd; - pollerData[i].ruleset = s_socketInfo[i].ruleset; + pollerData[i].ruleset = current->pRuleset; } - + s_zloop = zloop_new(); - for(i=0; i<s_nitems; ++i) { + for(i=0; i<n_items; ++i) { rv = zloop_poller(s_zloop, &items[i], handlePoll, &pollerData[i]); if (rv) { - errmsg.LogError(0, NO_ERRCODE, "imzmq3: zloop_poller failed for item %zu", i); + errmsg.LogError(0, NO_ERRCODE, "imzmq3: zloop_poller failed for item %zu: %s", i, zmq_strerror(errno)); } } + DBGPRINTF("imzmq3: zloop_poller starting..."); zloop_start(s_zloop); zloop_destroy(&s_zloop); - finalize_it: - for(i=0; i< s_nitems; ++i) { - zsocket_destroy(s_context, items[i].socket); - } - + DBGPRINTF("imzmq3: zloop_poller stopped."); +finalize_it: zctx_destroy(&s_context); free(items); + free(pollerData); RETiRet; } @@ -564,7 +648,8 @@ static rsRetVal rcv_loop(thrdInfo_t* pThrd){ BEGINrunInput CODESTARTrunInput - iRet = rcv_loop(pThrd); + CHKiRet(rcv_loop(pThrd)); +finalize_it: RETiRet; ENDrunInput @@ -572,17 +657,13 @@ ENDrunInput /* initialize and return if will run or not */ BEGINwillRun CODESTARTwillRun - /* we need to create the inputName property (only once during our + /* we need to create the inputName property (only once during our lifetime) */ - CHKiRet(prop.Construct(&s_namep)); - CHKiRet(prop.SetString(s_namep, + CHKiRet(prop.Construct(&s_namep)); + CHKiRet(prop.SetString(s_namep, UCHAR_CONSTANT("imzmq3"), sizeof("imzmq3") - 1)); - CHKiRet(prop.ConstructFinalize(s_namep)); - -/* If there are no endpoints this is pointless ...*/ - if (s_nitems == 0) - ABORT_FINALIZE(RS_RET_NO_RUN); + CHKiRet(prop.ConstructFinalize(s_namep)); finalize_it: ENDwillRun @@ -590,70 +671,162 @@ ENDwillRun BEGINafterRun CODESTARTafterRun - /* do cleanup here */ - if(s_namep != NULL) - prop.Destruct(&s_namep); + /* do cleanup here */ + if (s_namep != NULL) + prop.Destruct(&s_namep); ENDafterRun BEGINmodExit CODESTARTmodExit - /* release what we no longer need */ - objRelease(errmsg, CORE_COMPONENT); - objRelease(glbl, CORE_COMPONENT); - objRelease(prop, CORE_COMPONENT); + /* release what we no longer need */ + objRelease(errmsg, CORE_COMPONENT); + objRelease(glbl, CORE_COMPONENT); + objRelease(prop, CORE_COMPONENT); objRelease(ruleset, CORE_COMPONENT); ENDmodExit BEGINisCompatibleWithFeature CODESTARTisCompatibleWithFeature - if(eFeat == sFEATURENonCancelInputTermination) - iRet = RS_RET_OK; + if (eFeat == sFEATURENonCancelInputTermination) + iRet = RS_RET_OK; ENDisCompatibleWithFeature +BEGINbeginCnfLoad +CODESTARTbeginCnfLoad + loadModConf = pModConf; + pModConf->pConf = pConf; + /* init module config */ + loadModConf->io_threads = 0; /* 0 means don't set it */ +ENDbeginCnfLoad + + +BEGINsetModCnf + struct cnfparamvals* pvals = NULL; + int i; +CODESTARTsetModCnf + pvals = nvlstGetParams(lst, &modpblk, NULL); + if (NULL == pvals) { + errmsg.LogError(0, RS_RET_MISSING_CNFPARAMS, "imzmq3: error processing module " + " config parameters ['module(...)']"); + ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); + } + + for (i=0; i < modpblk.nParams; ++i) { + if (!pvals[i].bUsed) + continue; + if (!strcmp(modpblk.descr[i].name, "ioThreads")) { + loadModConf->io_threads = (int)pvals[i].val.d.n; + } else { + errmsg.LogError(0, RS_RET_INVALID_PARAMS, + "imzmq3: config error, unknown " + "param %s in setModCnf\n", + modpblk.descr[i].name); + } + } + +finalize_it: + if (pvals != NULL) + cnfparamvalsDestruct(pvals, &modpblk); +ENDsetModCnf + + +BEGINendCnfLoad +CODESTARTendCnfLoad + loadModConf = NULL; /* done loading, so it becomes NULL */ +ENDendCnfLoad + + +/* function to generate error message if framework does not find requested ruleset */ +static inline void +std_checkRuleset_genErrMsg(__attribute__((unused)) modConfData_t *modConf, instanceConf_t *inst) +{ + errmsg.LogError(0, NO_ERRCODE, "imzmq3: ruleset '%s' for socket %s not found - " + "using default ruleset instead", inst->pszBindRuleset, + inst->description); +} + + +BEGINcheckCnf +instanceConf_t* inst; +CODESTARTcheckCnf + for(inst = pModConf->root; inst!=NULL; inst=inst->next) { + std_checkRuleset(pModConf, inst); + /* now, validate the instanceConf */ + CHKiRet(validateConfig(inst)); + } +finalize_it: + RETiRet; +ENDcheckCnf + + +BEGINactivateCnfPrePrivDrop +CODESTARTactivateCnfPrePrivDrop + runModConf = pModConf; + /* first create the context */ + createContext(); + + /* could setup context here, and set the global worker threads + and so on... */ +ENDactivateCnfPrePrivDrop + + +BEGINactivateCnf +CODESTARTactivateCnf +ENDactivateCnf + + +/*TODO: Fill this in! */ +BEGINfreeCnf +CODESTARTfreeCnf +ENDfreeCnf + + +BEGINnewInpInst + struct cnfparamvals* pvals; +CODESTARTnewInpInst + + DBGPRINTF("newInpInst (imzmq3)\n"); + pvals = nvlstGetParams(lst, &inppblk, NULL); + if(NULL==pvals) { + errmsg.LogError(0, RS_RET_MISSING_CNFPARAMS, + "imzmq3: required parameters are missing\n"); + ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); + } + DBGPRINTF("imzmq3: input param blk:\n"); + cnfparamsPrint(&inppblk, pvals); + + /* now, parse the config params and so on... */ + CHKiRet(createListener(pvals)); + +finalize_it: +CODE_STD_FINALIZERnewInpInst + cnfparamvalsDestruct(pvals, &inppblk); +ENDnewInpInst + + BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_IMOD_QUERIES +CODEqueryEtryPt_STD_CONF2_QUERIES +CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES +CODEqueryEtryPt_STD_CONF2_PREPRIVDROP_QUERIES +CODEqueryEtryPt_STD_CONF2_IMOD_QUERIES CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES ENDqueryEtryPt -static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, - void __attribute__((unused)) *pVal) { - return RS_RET_OK; -} -static rsRetVal setGlobalWorkerThreads(uchar __attribute__((unused)) *pp, int val) { - errmsg.LogError(0, NO_ERRCODE, "setGlobalWorkerThreads called with %d",val); - s_io_threads = val; - return RS_RET_OK; -} BEGINmodInit() CODESTARTmodInit /* we only support the current interface specification */ - *ipIFVersProvided = CURR_MOD_IF_VERSION; + *ipIFVersProvided = CURR_MOD_IF_VERSION; CODEmodInit_QueryRegCFSLineHdlr - CHKiRet(objUse(errmsg, CORE_COMPONENT)); - CHKiRet(objUse(glbl, CORE_COMPONENT)); - CHKiRet(objUse(prop, CORE_COMPONENT)); + CHKiRet(objUse(errmsg, CORE_COMPONENT)); + CHKiRet(objUse(glbl, CORE_COMPONENT)); + CHKiRet(objUse(prop, CORE_COMPONENT)); CHKiRet(objUse(ruleset, CORE_COMPONENT)); - - /* register config file handlers */ - CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputzmq3serverbindruleset", - 0, eCmdHdlrGetWord, - set_ruleset, NULL, - STD_LOADABLE_MODULE_ID)); - CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputzmq3serverrun", - 0, eCmdHdlrGetWord, - add_endpoint, NULL, - STD_LOADABLE_MODULE_ID)); - CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", - 1, eCmdHdlrCustomHandler, - resetConfigVariables, NULL, - STD_LOADABLE_MODULE_ID)); - CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputzmq3globalWorkerThreads", - 1, eCmdHdlrInt, - setGlobalWorkerThreads, NULL, - STD_LOADABLE_MODULE_ID)); ENDmodInit + + diff --git a/plugins/mmaudit/Makefile.am b/plugins/mmaudit/Makefile.am index c64d0822..77b2e85f 100644 --- a/plugins/mmaudit/Makefile.am +++ b/plugins/mmaudit/Makefile.am @@ -1,8 +1,8 @@ pkglib_LTLIBRARIES = mmaudit.la mmaudit_la_SOURCES = mmaudit.c -mmaudit_la_CPPFLAGS = $(RSRT_CFLAGS) $(PTHREADS_CFLAGS) $(LIBLOGNORM_CFLAGS) $(LIBEE_CFLAGS) -mmaudit_la_LDFLAGS = -module -avoid-version $(LIBLOGNORM_LIBS) $(LIBEE_LIBS) +mmaudit_la_CPPFLAGS = $(RSRT_CFLAGS) $(PTHREADS_CFLAGS) +mmaudit_la_LDFLAGS = -module -avoid-version mmaudit_la_LIBADD = EXTRA_DIST = diff --git a/plugins/mmaudit/mmaudit.c b/plugins/mmaudit/mmaudit.c index 018e1771..6b6b804c 100644 --- a/plugins/mmaudit/mmaudit.c +++ b/plugins/mmaudit/mmaudit.c @@ -43,8 +43,7 @@ #include <errno.h> #include <unistd.h> #include <ctype.h> -#include <libestr.h> -#include <libee/libee.h> +#include <json/json.h> #include "conf.h" #include "syslogd-types.h" #include "template.h" diff --git a/plugins/mmjsonparse/Makefile.am b/plugins/mmjsonparse/Makefile.am index 5175fe81..ef39163e 100644 --- a/plugins/mmjsonparse/Makefile.am +++ b/plugins/mmjsonparse/Makefile.am @@ -1,8 +1,8 @@ pkglib_LTLIBRARIES = mmjsonparse.la mmjsonparse_la_SOURCES = mmjsonparse.c -mmjsonparse_la_CPPFLAGS = $(RSRT_CFLAGS) $(PTHREADS_CFLAGS) $(LIBLOGNORM_CFLAGS) $(LIBEE_CFLAGS) -mmjsonparse_la_LDFLAGS = -module -avoid-version $(LIBLOGNORM_LIBS) $(LIBEE_LIBS) +mmjsonparse_la_CPPFLAGS = $(RSRT_CFLAGS) $(PTHREADS_CFLAGS) +mmjsonparse_la_LDFLAGS = -module -avoid-version mmjsonparse_la_LIBADD = EXTRA_DIST = diff --git a/plugins/mmjsonparse/mmjsonparse.c b/plugins/mmjsonparse/mmjsonparse.c index c47aceb6..35f69aab 100644 --- a/plugins/mmjsonparse/mmjsonparse.c +++ b/plugins/mmjsonparse/mmjsonparse.c @@ -35,7 +35,6 @@ #include <errno.h> #include <unistd.h> #include <ctype.h> -#include <libestr.h> #include <json/json.h> #include "conf.h" #include "syslogd-types.h" diff --git a/plugins/omjournal/omjournal.c b/plugins/omjournal/omjournal.c index c340287f..160c369d 100644 --- a/plugins/omjournal/omjournal.c +++ b/plugins/omjournal/omjournal.c @@ -107,6 +107,7 @@ CODESTARTnewActInst * the lst ptr. However, we will most probably need params in the * future. */ + (void) lst; /* prevent compiler warning */ DBGPRINTF("newActInst (mmjournal)\n"); CODE_STD_STRING_REQUESTnewActInst(1) CHKiRet(OMSRsetEntry(*ppOMSR, 0, NULL, OMSR_TPL_AS_MSG)); @@ -145,6 +146,7 @@ CODESTARTdoAction "SYSLOG_IDENTIFIER=%s", tag, NULL); /* FIXME: think about what to do with errors ;) */ + (void) r; /* prevent compiler warning */ ENDdoAction diff --git a/plugins/omzmq3/README b/plugins/omzmq3/README index ccc96c74..c2a33555 100644 --- a/plugins/omzmq3/README +++ b/plugins/omzmq3/README @@ -1,16 +1,10 @@ ZeroMQ 3.x Output Plugin Building this plugin: -Requires libzmq and libczmq. First, install libzmq from the HEAD on github: -http://github.com/zeromq/libzmq. You can clone the repository, build, then -install it. The directions for doing so are there in the readme. Then, do -the same for libczmq: http://github.com/zeromq/czmq. At some point, the 3.1 -version of libzmq will be released, and a supporting version of libczmq. -At that time, you could simply download and install the tarballs instead of -using git to clone the repositories. Those tarballs (when available) can -be found at http://download.zeromq.org. As of this writing (5/31/2012), the -most recent version of czmq (1.1.0) and libzmq (3.1.0-beta) will not compile -properly. +Requires libzmq and libczmq. First, download the tarballs of both libzmq +and its supporting libczmq from http://download.zeromq.org. As of this +writing (04/23/2013), the most recent versions of libzmq and czmq are +3.2.2 and 1.3.2 respectively. Configure, build, and then install both libs. Omzmq3 allows you to push data out of rsyslog from a zeromq socket. The example below binds a PUB socket to port 7171, and any message fitting the criteria will diff --git a/plugins/omzmq3/omzmq3.c b/plugins/omzmq3/omzmq3.c index ee6756b9..c8552f11 100644 --- a/plugins/omzmq3/omzmq3.c +++ b/plugins/omzmq3/omzmq3.c @@ -110,9 +110,10 @@ static zctx_t* s_context = NULL; static int s_workerThreads = -1; static struct socket_type types[] = { - {"PUB", ZMQ_PUB }, - {"PUSH", ZMQ_PUSH }, - {"XPUB", ZMQ_XPUB } + {"PUB", ZMQ_PUB }, + {"PUSH", ZMQ_PUSH }, + {"DEALER", ZMQ_DEALER }, + {"XPUB", ZMQ_XPUB } }; static struct socket_action actions[] = { @@ -201,17 +202,18 @@ static rsRetVal initZMQ(instanceData* pData) { /* create the context if necessary. */ if (NULL == s_context) { + zsys_handler_set(NULL); s_context = zctx_new(); if (s_workerThreads > 0) zctx_set_iothreads(s_context, s_workerThreads); } pData->socket = zsocket_new(s_context, pData->type); - - /* ALWAYS set the HWM as the zmq3 default is 1000 and we default - to 0 (infinity) */ - zsocket_set_rcvhwm(pData->socket, pData->rcvHWM); - zsocket_set_sndhwm(pData->socket, pData->sndHWM); - + if (NULL == pData->socket) { + errmsg.LogError(0, RS_RET_NO_ERRCODE, + "omzmq3: zsocket_new failed for %s: %s", + pData->description, zmq_strerror(errno)); + ABORT_FINALIZE(RS_RET_NO_ERRCODE); + } /* use czmq defaults for these, unless set to non-default values */ if(pData->identity) zsocket_set_identity(pData->socket, (char*)pData->identity); if(pData->sndBuf > -1) zsocket_set_sndbuf(pData->socket, pData->sndBuf); @@ -228,17 +230,26 @@ static rsRetVal initZMQ(instanceData* pData) { if(pData->reconnectIVLMax > -1) zsocket_set_reconnect_ivl_max(pData->socket, pData->reconnectIVLMax); if(pData->ipv4Only > -1) zsocket_set_ipv4only(pData->socket, pData->ipv4Only); if(pData->affinity != 1) zsocket_set_affinity(pData->socket, pData->affinity); - + if(pData->rcvHWM > -1) zsocket_set_rcvhwm(pData->socket, pData->rcvHWM); + if(pData->sndHWM > -1) zsocket_set_sndhwm(pData->socket, pData->sndHWM); + /* bind or connect to it */ if (pData->action == ACTION_BIND) { /* bind asserts, so no need to test return val here which isn't the greatest api -- oh well */ - zsocket_bind(pData->socket, (char*)pData->description); + if(-1 == zsocket_bind(pData->socket, (char*)pData->description)) { + errmsg.LogError(0, RS_RET_NO_ERRCODE, "omzmq3: bind failed for %s: %s", + pData->description, zmq_strerror(errno)); + ABORT_FINALIZE(RS_RET_NO_ERRCODE); + } + DBGPRINTF("omzmq3: bind to %s successful\n",pData->description); } else { - if(zsocket_connect(pData->socket, (char*)pData->description) == -1) { - errmsg.LogError(0, RS_RET_SUSPENDED, "omzmq3: connect failed!"); - ABORT_FINALIZE(RS_RET_SUSPENDED); + if(-1 == zsocket_connect(pData->socket, (char*)pData->description)) { + errmsg.LogError(0, RS_RET_NO_ERRCODE, "omzmq3: connect failed for %s: %s", + pData->description, zmq_strerror(errno)); + ABORT_FINALIZE(RS_RET_NO_ERRCODE); } + DBGPRINTF("omzmq3: connect to %s successful", pData->description); } finalize_it: RETiRet; @@ -256,7 +267,7 @@ rsRetVal writeZMQ(uchar* msg, instanceData* pData) { /* whine if things went wrong */ if (result == -1) { - errmsg.LogError(0, NO_ERRCODE, "omzmq3: send of %s failed with return %d", msg, result); + errmsg.LogError(0, NO_ERRCODE, "omzmq3: send of %s failed: %s", msg, zmq_strerror(errno)); ABORT_FINALIZE(RS_RET_ERR); } finalize_it: @@ -265,13 +276,13 @@ rsRetVal writeZMQ(uchar* msg, instanceData* pData) { static inline void setInstParamDefaults(instanceData* pData) { - pData->description = (uchar*)"tcp://*:7171"; + pData->description = NULL; pData->socket = NULL; pData->tplName = NULL; pData->type = ZMQ_PUB; pData->action = ACTION_BIND; - pData->sndHWM = 0; /*unlimited*/ - pData->rcvHWM = 0; /*unlimited*/ + pData->sndHWM = -1; + pData->rcvHWM = -1; pData->identity = NULL; pData->sndBuf = -1; pData->rcvBuf = -1; @@ -314,6 +325,7 @@ CODESTARTfreeInstance closeZMQ(pData); free(pData->description); free(pData->tplName); + free(pData->identity); ENDfreeInstance BEGINtryResume @@ -329,88 +341,90 @@ ENDdoAction BEGINnewActInst - struct cnfparamvals *pvals; - int i; + struct cnfparamvals *pvals; + int i; CODESTARTnewActInst -if((pvals = nvlstGetParams(lst, &actpblk, NULL)) == NULL) { - ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); - } + 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, "description")) { - pData->description = (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, "sockType")){ - pData->type = getSocketType(es_str2cstr(pvals[i].val.d.estr, NULL)); - } else if(!strcmp(actpblk.descr[i].name, "action")){ - pData->action = getSocketAction(es_str2cstr(pvals[i].val.d.estr, NULL)); - } else if(!strcmp(actpblk.descr[i].name, "sndHWM")) { - pData->sndHWM = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "rcvHWM")) { - pData->rcvHWM = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "identity")){ - pData->identity = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); - } else if(!strcmp(actpblk.descr[i].name, "sndBuf")) { - pData->sndBuf = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "rcvBuf")) { - pData->rcvBuf = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "linger")) { - pData->linger = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "backlog")) { - pData->backlog = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "sndTimeout")) { - pData->sndTimeout = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "rcvTimeout")) { - pData->rcvTimeout = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "maxMsgSize")) { - pData->maxMsgSize = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "rate")) { - pData->rate = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "recoveryIVL")) { - pData->recoveryIVL = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "multicastHops")) { - pData->multicastHops = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "reconnectIVL")) { - pData->reconnectIVL = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "reconnectIVLMax")) { - pData->reconnectIVLMax = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "ipv4Only")) { - pData->ipv4Only = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "affinity")) { - pData->affinity = (int) pvals[i].val.d.n; - } else if(!strcmp(actpblk.descr[i].name, "globalWorkerThreads")) { - s_workerThreads = (int) pvals[i].val.d.n; - } else { - errmsg.LogError(0, NO_ERRCODE, "omzmq3: program error, non-handled " - "param '%s'\n", actpblk.descr[i].name); + for (i = 0; i < actpblk.nParams; ++i) { + if (!pvals[i].bUsed) + continue; + if (!strcmp(actpblk.descr[i].name, "description")) { + pData->description = (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, "sockType")){ + pData->type = getSocketType(es_str2cstr(pvals[i].val.d.estr, NULL)); + } else if (!strcmp(actpblk.descr[i].name, "action")){ + pData->action = getSocketAction(es_str2cstr(pvals[i].val.d.estr, NULL)); + } else if (!strcmp(actpblk.descr[i].name, "sndHWM")) { + pData->sndHWM = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "rcvHWM")) { + pData->rcvHWM = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "identity")){ + pData->identity = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + } else if (!strcmp(actpblk.descr[i].name, "sndBuf")) { + pData->sndBuf = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "rcvBuf")) { + pData->rcvBuf = (int) pvals[i].val.d.n; + } else if(!strcmp(actpblk.descr[i].name, "linger")) { + pData->linger = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "backlog")) { + pData->backlog = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "sndTimeout")) { + pData->sndTimeout = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "rcvTimeout")) { + pData->rcvTimeout = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "maxMsgSize")) { + pData->maxMsgSize = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "rate")) { + pData->rate = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "recoveryIVL")) { + pData->recoveryIVL = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "multicastHops")) { + pData->multicastHops = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "reconnectIVL")) { + pData->reconnectIVL = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "reconnectIVLMax")) { + pData->reconnectIVLMax = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "ipv4Only")) { + pData->ipv4Only = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "affinity")) { + pData->affinity = (int) pvals[i].val.d.n; + } else if (!strcmp(actpblk.descr[i].name, "globalWorkerThreads")) { + s_workerThreads = (int) pvals[i].val.d.n; + } else { + errmsg.LogError(0, NO_ERRCODE, "omzmq3: program error, non-handled " + "param '%s'\n", actpblk.descr[i].name); + } } - } - -if(pData->tplName == NULL) { - CHKiRet(OMSRsetEntry(*ppOMSR, 0, NULL, OMSR_TPL_AS_MSG)); - } else { - CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*)pData->tplName, OMSR_NO_RQD_TPL_OPTS)); - } - -if(pData->type == -1) { - errmsg.LogError(0, RS_RET_CONFIG_ERROR, "omzmq3: unknown socket type."); - ABORT_FINALIZE(RS_RET_CONFIG_ERROR); - } -if(pData->action == -1) { - errmsg.LogError(0, RS_RET_CONFIG_ERROR, "omzmq3: unknown socket action"); - ABORT_FINALIZE(RS_RET_CONFIG_ERROR); - } + if (pData->tplName == NULL) { + CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*)strdup("RSYSLOG_ForwardFormat"), OMSR_NO_RQD_TPL_OPTS)); + } else { + CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*)pData->tplName, OMSR_NO_RQD_TPL_OPTS)); + } + if (NULL == pData->description) { + errmsg.LogError(0, RS_RET_CONFIG_ERROR, "omzmq3: you didn't enter a description"); + ABORT_FINALIZE(RS_RET_CONFIG_ERROR); + } + if (pData->type == -1) { + errmsg.LogError(0, RS_RET_CONFIG_ERROR, "omzmq3: unknown socket type."); + ABORT_FINALIZE(RS_RET_CONFIG_ERROR); + } + if (pData->action == -1) { + errmsg.LogError(0, RS_RET_CONFIG_ERROR, "omzmq3: unknown socket action"); + ABORT_FINALIZE(RS_RET_CONFIG_ERROR); + } CODE_STD_FINALIZERnewActInst - cnfparamvalsDestruct(pvals, &actpblk); + cnfparamvalsDestruct(pvals, &actpblk); ENDnewActInst BEGINparseSelectorAct @@ -433,10 +447,10 @@ ENDinitConfVars BEGINmodExit CODESTARTmodExit -if(NULL != s_context) { - zctx_destroy(&s_context); - s_context=NULL; - } + if (NULL != s_context) { + zctx_destroy(&s_context); + s_context=NULL; + } ENDmodExit diff --git a/runtime/Makefile.am b/runtime/Makefile.am index fe4afb09..dea06fe0 100644 --- a/runtime/Makefile.am +++ b/runtime/Makefile.am @@ -97,12 +97,12 @@ librsyslog_la_SOURCES = \ # if WITH_MODDIRS -librsyslog_la_CPPFLAGS = -DSD_EXPORT_SYMBOLS -D_PATH_MODDIR=\"$(pkglibdir)/:$(moddirs)\" $(PTHREADS_CFLAGS) $(LIBEE_CFLAGS) $(LIBGCRYPT_CFLAGS) -I\$(top_srcdir)/tools +librsyslog_la_CPPFLAGS = -DSD_EXPORT_SYMBOLS -D_PATH_MODDIR=\"$(pkglibdir)/:$(moddirs)\" $(PTHREADS_CFLAGS) -I\$(top_srcdir)/tools else -librsyslog_la_CPPFLAGS = -DSD_EXPORT_SYMBOLS -D_PATH_MODDIR=\"$(pkglibdir)/\" -I$(top_srcdir) $(PTHREADS_CFLAGS) $(LIBEE_CFLAGS) $(LIBGCRYPT_CFLAGS) -I\$(top_srcdir)/tools -I\$(top_srcdir)/grammar +librsyslog_la_CPPFLAGS = -DSD_EXPORT_SYMBOLS -D_PATH_MODDIR=\"$(pkglibdir)/\" -I$(top_srcdir) $(PTHREADS_CFLAGS) -I\$(top_srcdir)/tools -I\$(top_srcdir)/grammar endif #librsyslog_la_LDFLAGS = -module -avoid-version -librsyslog_la_LIBADD = $(DL_LIBS) $(RT_LIBS) $(LIBGCRYPT_LIBS) $(LIBEE_LIBS) +librsyslog_la_LIBADD = $(DL_LIBS) $(RT_LIBS) # # regular expression support diff --git a/runtime/conf.c b/runtime/conf.c index c97391c6..c3c7e447 100644 --- a/runtime/conf.c +++ b/runtime/conf.c @@ -129,6 +129,23 @@ finalize_it: } +/* remove leading spaces from name; this "fixes" some anomalies in + * getSubString(), but I was not brave enough to fix the former as + * it has many other callers... -- rgerhards, 2013-05-27 + */ +static inline void +ltrim(char *src) +{ + char *dst = src; + while(isspace(*src)) + ++src; /*SKIP*/; + if(dst != src) { + while(*src != '\0') + *dst++ = *src++; + *dst = '\0'; + } +} + /* parse and interpret a $-config line that starts with * a name (this is common code). It is parsed to the name * and then the proper sub-function is called to handle @@ -155,6 +172,7 @@ doNameLine(uchar **pp, void* pVal) errmsg.LogError(0, RS_RET_NOT_FOUND, "Invalid config line: could not extract name - line ignored"); ABORT_FINALIZE(RS_RET_NOT_FOUND); } + ltrim(szName); if(*p == ',') ++p; /* comma was eaten */ diff --git a/runtime/debug.c b/runtime/debug.c index 876f61d0..68474989 100644 --- a/runtime/debug.c +++ b/runtime/debug.c @@ -1430,6 +1430,7 @@ dbgSetDebugFile(uchar *fn) rsRetVal dbgClassInit(void) { + pthread_mutexattr_t mutAttr; rsRetVal iRet; /* do not use DEFiRet, as this makes calls into the debug system! */ struct sigaction sigAct; @@ -1437,14 +1438,16 @@ rsRetVal dbgClassInit(void) (void) pthread_key_create(&keyCallStack, dbgCallStackDestruct); /* MUST be the first action done! */ - /* we initialize all Mutexes with code, as some platforms seem to have - * bugs in the static initializer macros. So better be on the safe side... - * rgerhards, 2008-03-06 + /* the mutexes must be recursive, because it may be called from within + * signal handlers, which can lead to a hang if the signal interrupted dbgprintf + * (yes, we have really seen that situation in practice!). -- rgerhards, 2013-05-17 */ - pthread_mutex_init(&mutFuncDBList, NULL); - pthread_mutex_init(&mutMutLog, NULL); - pthread_mutex_init(&mutCallStack, NULL); - pthread_mutex_init(&mutdbgprint, NULL); + pthread_mutexattr_init(&mutAttr); + pthread_mutexattr_settype(&mutAttr, PTHREAD_MUTEX_RECURSIVE); + pthread_mutex_init(&mutFuncDBList, &mutAttr); + pthread_mutex_init(&mutMutLog, &mutAttr); + pthread_mutex_init(&mutCallStack, &mutAttr); + pthread_mutex_init(&mutdbgprint, &mutAttr); /* while we try not to use any of the real rsyslog code (to avoid infinite loops), we * need to have the ability to query object names. Thus, we need to obtain a pointer to diff --git a/runtime/librsgt.c b/runtime/librsgt.c index ae0b0df6..85fc7742 100644 --- a/runtime/librsgt.c +++ b/runtime/librsgt.c @@ -296,7 +296,7 @@ tlv16Write(gtfile gf, int flags, int tlvtype, uint16_t len) { uint16_t typ; int r; - typ = ((flags|1) << 13)|tlvtype; + typ = ((flags|1) << 15)|tlvtype; r = tlvbufAddOctet(gf, typ >> 8); if(r != 0) goto done; r = tlvbufAddOctet(gf, typ & 0xff); diff --git a/runtime/librsgt_read.c b/runtime/librsgt_read.c index 25c0db4d..a6e33160 100644 --- a/runtime/librsgt_read.c +++ b/runtime/librsgt_read.c @@ -249,7 +249,7 @@ rsgt_tlvRecRead(FILE *fp, tlvrecord_t *rec) NEXTC; rec->hdr[0] = c; rec->tlvtype = c & 0x1f; - if(c & 0x20) { /* tlv16? */ + if(c & 0x80) { /* tlv16? */ rec->lenHdr = 4; NEXTC; rec->hdr[1] = c; @@ -290,7 +290,7 @@ rsgt_tlvDecodeSUBREC(tlvrecord_t *rec, uint16_t *stridx, tlvrecord_t *newrec) c = rec->data[(*stridx)++]; newrec->hdr[0] = c; newrec->tlvtype = c & 0x1f; - if(c & 0x20) { /* tlv16? */ + if(c & 0x80) { /* tlv16? */ newrec->lenHdr = 4; if(rec->tlvlen == *stridx) {r=RSGTE_LEN; goto done;} c = rec->data[(*stridx)++]; diff --git a/runtime/msg.h b/runtime/msg.h index edf5ed98..6faf066a 100644 --- a/runtime/msg.h +++ b/runtime/msg.h @@ -35,8 +35,6 @@ #include "syslogd-types.h" #include "template.h" #include "atomic.h" -#include "libee/libee.h" - /* rgerhards 2004-11-08: The following structure represents a * syslog message. diff --git a/runtime/stringbuf.c b/runtime/stringbuf.c index 0d8e7e83..cb4f0457 100644 --- a/runtime/stringbuf.c +++ b/runtime/stringbuf.c @@ -312,7 +312,7 @@ rsRetVal rsCStrAppendStrf(cstr_t *pThis, uchar *fmt, ...) { DEFiRet; va_list ap; - cstr_t *pStr; + cstr_t *pStr = NULL; va_start(ap, fmt); iRet = rsCStrConstructFromszStrv(&pStr, fmt, ap); @@ -1184,6 +1184,7 @@ struct template *tplAddLine(rsconf_t *conf, char* pName, uchar** ppRestOfConfLin if((pTpl = tplConstruct(conf)) == NULL) return NULL; + DBGPRINTF("tplAddLine processing template '%s'\n", pName); pTpl->iLenName = strlen(pName); pTpl->pszName = (char*) MALLOC(sizeof(char) * (pTpl->iLenName + 1)); if(pTpl->pszName == NULL) { diff --git a/tools/Makefile.am b/tools/Makefile.am index 9a1497cd..6832494e 100644 --- a/tools/Makefile.am +++ b/tools/Makefile.am @@ -42,12 +42,14 @@ rsyslogd_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS) # note: it looks like librsyslog.la must be explicitely given on LDDADD, # otherwise dependencies are not properly calculated (resulting in a # potentially incomplete build, a problem we had several times...) -rsyslogd_LDADD = ../grammar/libgrammar.la ../runtime/librsyslog.la $(ZLIB_LIBS) $(PTHREADS_LIBS) $(RSRT_LIBS) $(SOL_LIBS) $(LIBEE_LIBS) $(LIBLOGNORM_LIBS) $(LIBUUID_LIBS) $(LIBGCRYPT_LIBS) +rsyslogd_LDADD = ../grammar/libgrammar.la ../runtime/librsyslog.la $(ZLIB_LIBS) $(PTHREADS_LIBS) $(RSRT_LIBS) $(SOL_LIBS) $(LIBUUID_LIBS) rsyslogd_LDFLAGS = -export-dynamic EXTRA_DIST = $(man_MANS) \ rsgtutil.rst \ + rsgtutil.1 \ rscryutil.rst \ + rscryutil.1 \ recover_qi.pl if ENABLE_DIAGTOOLS diff --git a/tools/syslogd.c b/tools/syslogd.c index d2f249cf..2bc80e44 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -51,7 +51,6 @@ #include <stddef.h> #include <ctype.h> #include <limits.h> -#define GNU_SOURCE #include <string.h> #include <stdarg.h> #include <time.h> |