diff --git a/SPECS/rsyslog/add-network-namespace-APIs.patch b/SPECS/rsyslog/add-network-namespace-APIs.patch new file mode 100644 index 00000000000..5c28db388b4 --- /dev/null +++ b/SPECS/rsyslog/add-network-namespace-APIs.patch @@ -0,0 +1,504 @@ +From c78c587cafd4d495f7e2bc3c0df5d8c18db48d51 Mon Sep 17 00:00:00 2001 +From: Billie Alsup +Date: Thu, 11 Sep 2025 08:54:49 -0700 +Subject: [PATCH] net: Add NetworkNamespace APIs + +This moves NetworkNamespace functionality into +the net module. This allows the same code to +be reused across multiple tools and plugins. +The first usage is with omfwd, which is changed +to use the common net implementation. Note +the net implementation is based on the original +omfwd implementation. Subsequent PRs will be +opened for integrating this into omuxsock and +imtcp. + +The original test case tcp_forwarding_ns_tpl.sh +was broken due to use of single quotes rather +than double quotes, thus preventing the proper +port number argument to be passed to the +listener. Note this test must be run as +root. + +Development has occurred across Fedora41 and +Fedora42, which uses glibc 2.40 and glibc2.41 +respectively. The valgrind suppressions are +updated to handle new glibc issues +accordingly. + +Four new functions are callable through the +net module. These allow one to save a handle +to the current namespace, switch to a new +namespace by name, and restore the namespace +from the saved handle. A fourth higher +level wrapper is used to open a socket in +a named network namespace, and handles the +invocation of the lower level functions. +Ideally this would be the only public +function, however it simplifies integration +into imtcp in the future (which doesn't +directly open sockets). This may change +in the future as network namespaces are +integrated into more modules and plugins. + +Signed-off-by: Billie Alsup + +Upstream Patch reference: https://patch-diff.githubusercontent.com/raw/rsyslog/rsyslog/pull/6121.patch +--- + runtime/Makefile.am | 2 +- + runtime/net.c | 5 + + runtime/net.h | 60 +++++++++++- + runtime/netns_socket.c | 166 +++++++++++++++++++++++++++++++++ + runtime/netns_socket.h | 40 ++++++++ + tests/known_issues.supp | 18 ++++ + tests/tcp_forwarding_ns_tpl.sh | 2 +- + tools/omfwd.c | 50 +--------- + 8 files changed, 293 insertions(+), 50 deletions(-) + create mode 100644 runtime/netns_socket.c + create mode 100644 runtime/netns_socket.h + +diff --git a/runtime/Makefile.am b/runtime/Makefile.am +index 8a25f21..cc3480c 100644 +--- a/runtime/Makefile.am ++++ b/runtime/Makefile.am +@@ -164,7 +164,7 @@ endif + # basic network support, needed for rsyslog startup (e.g. our own system name) + # + pkglib_LTLIBRARIES += lmnet.la +-lmnet_la_SOURCES = net.c net.h ++lmnet_la_SOURCES = net.c net.h netns_socket.c netns_socket.h + lmnet_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS) + lmnet_la_LDFLAGS = -module -avoid-version ../compat/compat_la-getifaddrs.lo + lmnet_la_LIBADD = +diff --git a/runtime/net.c b/runtime/net.c +index 8611125..758a0c6 100644 +--- a/runtime/net.c ++++ b/runtime/net.c +@@ -1687,6 +1687,11 @@ CODESTARTobjQueryInterface(net) + pIf->CmpHost = CmpHost; + pIf->HasRestrictions = HasRestrictions; + pIf->GetIFIPAddr = getIFIPAddr; ++ ++ pIf->netns_save = netns_save; ++ pIf->netns_restore = netns_restore; ++ pIf->netns_switch = netns_switch; ++ pIf->netns_socket = netns_socket; + finalize_it: + ENDobjQueryInterface(net) + +diff --git a/runtime/net.h b/runtime/net.h +index c2847f6..6de8ef9 100644 +--- a/runtime/net.h ++++ b/runtime/net.h +@@ -27,6 +27,7 @@ + #include + #include + #include /* this is needed on HP UX -- rgerhards, 2008-03-04 */ ++#include "netns_socket.h" + + typedef enum _TCPFRAMINGMODE { + TCP_FRAMING_OCTET_STUFFING = 0, /* traditional LF-delimited */ +@@ -168,8 +169,65 @@ BEGINinterface(net) /* name must also be changed in ENDinterface macro! */ + /* v8 cvthname() signature change -- rgerhards, 2013-01-18 */ + /* v9 create_udp_socket() signature change -- dsahern, 2016-11-11 */ + /* v10 moved data members to rsconf_t -- alakatos, 2021-12-29 */ ++ ++ /* v11 netns functions -- balsup, 2025-09-11 */ ++ /* ++ * @brief Open a socket on the given namespace ++ * @param fdp A place to store the descriptor. This must not be NULL. ++ * A failure will store -1 here. ++ * @param domain The communication domain argument to the underlying socket call ++ * @param type The type argument to the underlying socket call ++ * @param protocol The protocol argument to the underlying socket call ++ * @param ns The desired namespace. This may be NULL or the empty string if ++ * the current namespace is desired. ++ * @return RS_RET_OK on success, otherwise a failure code. ++ * @details This is a wrapper to socket, allowing one to open a socket in ++ * a given namespace. For platforms that do not support network ++ * namespaces, an error will be returned if the ns parameter ++ * is not NULL or the empty string. ++ */ ++ rsRetVal (*netns_socket)(int *fdp, int domain, int type, int protocol, const char *ns); ++ ++ /* ++ * @brief Switch to the given network namespace ++ * @param ns The desired namespace. If this is NULL or the empty string, then ++ * this function is a no-op. ++ * @return RS_RET_OK on success, otherwise a failure code. ++ * @details For platforms that do not support network namespaces, an error will ++ * be returned if the ns parameter is not NULL or the empty string. ++ */ ++ rsRetVal (*netns_switch)(const char *ns); ++ ++ /* ++ * @brief Save a descriptor to the current network namespace ++ * @param fd The location to store the descriptor for the current ++ * namespace. This must not not be NULL, and the ++ * descriptor must be pre-initialized to -1, i.e. *fd == -1 ++ * is a precondition. This style is to prevent inadvertent ++ * descriptor leaks that might arise by overwriting a valid ++ * descriptor. ++ * @return RS_RET_OK on success, otherwise a failure code. ++ * @details For platforms that do not support network namespaces, this ++ * function is a no-op and will always return RS_RET_OK. ++ */ ++ rsRetVal (*netns_save)(int *fd); ++ ++ /* ++ * @brief Restore the original network namespace ++ * @param fd A pointer to a descriptor associated with the original ++ * namespace. This must not not be NULL. If the descriptor ++ * is -1, then this function is a no-op. A valid descriptor ++ * is always closed as a side-effect of this function, ++ * with the descriptor being updated to -1. ++ * @return RS_RET_OK on success, otherwise a failure code. ++ * @details For platforms that do not support network namespaces, this ++ * function cannot change the network namespace. However, if ++ * presented with an fd that is not -1, it will still close ++ * that fd and reset the value to -1. ++ */ ++ rsRetVal (*netns_restore)(int *fd); + ENDinterface(net) +-#define netCURR_IF_VERSION 10 /* increment whenever you change the interface structure! */ ++#define netCURR_IF_VERSION 11 /* increment whenever you change the interface structure! */ + + /* prototypes */ + PROTOTYPEObj(net); +diff --git a/runtime/netns_socket.c b/runtime/netns_socket.c +new file mode 100644 +index 0000000..14fad27 +--- /dev/null ++++ b/runtime/netns_socket.c +@@ -0,0 +1,166 @@ ++/* Implementation for netns_socket API ++ * ++ * This file is part of rsyslog. ++ * ++ * Licensed under the Apache License, Version 2.0 (the "License"); ++ * you may not use this file except in compliance with the License. ++ * You may obtain a copy of the License at ++ * ++ * http://www.apache.org/licenses/LICENSE-2.0 ++ * -or- ++ * see COPYING.ASL20 in the source distribution ++ * ++ * Unless required by applicable law or agreed to in writing, software ++ * distributed under the License is distributed on an "AS IS" BASIS, ++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. ++ * See the License for the specific language governing permissions and ++ * limitations under the License. ++ */ ++#include "config.h" ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++ ++#include "rsyslog.h" ++#include "debug.h" ++#include "errmsg.h" ++#include "netns_socket.h" ++ ++ ++/* Change to the given network namespace. ++ * This function based on previous implementation ++ * of tools/omfwd.c function changeToNs. ++ */ ++rsRetVal netns_switch(const char *ns) { ++ DEFiRet; ++#ifdef HAVE_SETNS ++ int ns_fd = -1; ++ char *nsPath = NULL; ++ ++ if (ns && *ns) { ++ /* Build network namespace path */ ++ if (asprintf(&nsPath, "/var/run/netns/%s", ns) == -1) { ++ // Some implementations say nsPath would be undefined on failure ++ nsPath = NULL; ++ LogError(0, RS_RET_OUT_OF_MEMORY, "%s: asprintf failed", __func__); ++ ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); ++ } ++ ++ /* Open file descriptor of destination network namespace */ ++ ns_fd = open(nsPath, O_RDONLY); ++ if (ns_fd < 0) { ++ LogError(errno, RS_RET_IO_ERROR, "%s: could not open namespace '%s'", __func__, ns); ++ ABORT_FINALIZE(RS_RET_IO_ERROR); ++ } ++ /* Change to the destination network namespace */ ++ if (setns(ns_fd, CLONE_NEWNET) != 0) { ++ LogError(errno, RS_RET_IO_ERROR, "%s: could not change to namespace '%s'", __func__, ns); ++ ABORT_FINALIZE(RS_RET_IO_ERROR); ++ } ++ dbgprintf("%s: changed to network namespace '%s'\n", __func__, ns); ++ } ++finalize_it: ++ free(nsPath); ++ if ((ns_fd >= 0) && (close(ns_fd) != 0)) { ++ LogError(errno, RS_RET_IO_ERROR, "%s: failed to close namespace '%s'", __func__, ns); ++ } ++#else // ndef HAVE_SETNS ++ if (ns && *ns) { ++ LogError(ENOSYS, RS_RET_VALUE_NOT_SUPPORTED, "%s: could not change to namespace '%s'", __func__, ns); ++ ABORT_FINALIZE(RS_RET_VALUE_NOT_SUPPORTED); ++ } ++finalize_it: ++#endif // ndef HAVE_SETNS ++ RETiRet; ++} ++ ++ ++/* Return to the startup network namespace. ++ * This function based on code in tools/omfwd.c ++ */ ++rsRetVal ATTR_NONNULL() netns_restore(int *fd) { ++ DEFiRet; ++ ++#ifdef HAVE_SETNS ++ if (*fd >= 0) { ++ if (setns(*fd, CLONE_NEWNET) != 0) { ++ LogError(errno, RS_RET_IO_ERROR, "%s: could not return to startup namespace", __func__); ++ ABORT_FINALIZE(RS_RET_IO_ERROR); ++ } ++ dbgprintf("%s: returned to startup network namespace\n", __func__); ++ } ++ ++finalize_it: ++#endif // def HAVE_SETNS ++ if (*fd >= 0 && close(*fd) != 0) { ++ LogError(errno, RS_RET_IO_ERROR, "%s: could not close startup namespace fd", __func__); ++ } ++ *fd = -1; ++ RETiRet; ++} ++ ++/* Save the current network namespace fd ++ */ ++rsRetVal ATTR_NONNULL() netns_save(int *fd) { ++ DEFiRet; ++ ++ /* ++ * The fd must always point to either a valid fd ++ * or to -1. We expect it to be -1 on entry here. ++ * To avoid bugs, or possible descriptor leaks, ++ * check that it is always -1 on entry. ++ */ ++#ifdef HAVE_SETNS ++ if (*fd != -1) { ++ LogError(0, RS_RET_CODE_ERR, "%s: called with uninitialized descriptor", __func__); ++ ABORT_FINALIZE(RS_RET_CODE_ERR); ++ } ++ *fd = open("/proc/self/ns/net", O_RDONLY); ++ if (*fd == -1) { ++ LogError(errno, RS_RET_IO_ERROR, "%s: could not access startup namespace", __func__); ++ ABORT_FINALIZE(RS_RET_IO_ERROR); ++ } ++ dbgprintf("%s: saved startup network namespace\n", __func__); ++finalize_it: ++#endif // def HAVE_SETNS ++ RETiRet; ++} ++ ++rsRetVal netns_socket(int *fdp, int domain, int type, int protocol, const char *ns) { ++ DEFiRet; ++ int fd = -1; ++ ++#ifndef HAVE_SETNS ++ if (ns && *ns) { ++ LogError(0, RS_RET_VALUE_NOT_SUPPORTED, "Network namespaces are not supported"); ++ ABORT_FINALIZE(RS_RET_VALUE_NOT_SUPPORTED); ++ } ++#else /* def HAVE_SETNS */ ++ rsRetVal iRet_restore; ++ int ns_fd = -1; ++ ++ if (ns && *ns) { ++ CHKiRet(netns_save(&ns_fd)); ++ CHKiRet(netns_switch(ns)); ++ } ++#endif /* def HAVE_SETNS */ ++ *fdp = fd = socket(domain, type, protocol); ++ if (fd == -1) { ++ LogError(errno, RS_RET_NO_SOCKET, "%s: socket(%d, %d, %d) failed", __func__, domain, type, protocol); ++ ABORT_FINALIZE(RS_RET_NO_SOCKET); ++ } ++finalize_it: ++#ifdef HAVE_SETNS ++ iRet_restore = netns_restore(&ns_fd); ++ if (iRet == RS_RET_OK) iRet = iRet_restore; ++#endif /* def HAVE_SETNS */ ++ if (iRet != RS_RET_OK && fd != -1) { ++ (void)close(fd); ++ *fdp = -1; ++ } ++ RETiRet; ++} +diff --git a/runtime/netns_socket.h b/runtime/netns_socket.h +new file mode 100644 +index 0000000..223d106 +--- /dev/null ++++ b/runtime/netns_socket.h +@@ -0,0 +1,40 @@ ++/* Definitions for netns_socket API ++ * ++ * This file is part of rsyslog. ++ * ++ * Licensed under the Apache License, Version 2.0 (the "License"); ++ * you may not use this file except in compliance with the License. ++ * You may obtain a copy of the License at ++ * ++ * http://www.apache.org/licenses/LICENSE-2.0 ++ * -or- ++ * see COPYING.ASL20 in the source distribution ++ * ++ * Unless required by applicable law or agreed to in writing, software ++ * distributed under the License is distributed on an "AS IS" BASIS, ++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. ++ * See the License for the specific language governing permissions and ++ * limitations under the License. ++ */ ++#ifndef INCLUDED_NETNS_SOCKET_H ++#define INCLUDED_NETNS_SOCKET_H ++ ++#include "rsyslog.h" ++ ++/* ++ * Open a socket in the named network namespace ++ */ ++rsRetVal netns_socket(int *fdp, int domain, int type, int protocol, const char *ns); ++ ++/* ++ * Switch to the named networknamespace ++ */ ++rsRetVal netns_switch(const char *ns); ++ ++/* ++ * Save and restore our current network namespace ++ */ ++rsRetVal ATTR_NONNULL() netns_save(int *fd); ++rsRetVal ATTR_NONNULL() netns_restore(int *fd); ++ ++#endif /* #ifndef INCLUDED_NETNS_SOCKET_H */ +diff --git a/tests/known_issues.supp b/tests/known_issues.supp +index 9bb8fa3..80a6198 100644 +--- a/tests/known_issues.supp ++++ b/tests/known_issues.supp +@@ -60,3 +60,21 @@ + fun:exit + fun:(below main) + } ++{ ++ ++ Memcheck:Leak ++ match-leak-kinds: definite ++ fun:malloc ++ fun:UnknownInlinedFun ++ fun:_dl_map_object_deps ++ fun:dl_open_worker_begin ++ fun:_dl_catch_exception ++ fun:dl_open_worker ++ fun:_dl_catch_exception ++ fun:_dl_open ++ fun:do_dlopen ++ fun:_dl_catch_exception ++ fun:_dl_catch_error ++ fun:dlerror_run ++ fun:__libc_dlopen_mode ++} +diff --git a/tests/tcp_forwarding_ns_tpl.sh b/tests/tcp_forwarding_ns_tpl.sh +index 89a45e3..2bf52fe 100755 +--- a/tests/tcp_forwarding_ns_tpl.sh ++++ b/tests/tcp_forwarding_ns_tpl.sh +@@ -26,7 +26,7 @@ ip netns add rsyslog_test_ns + ip netns exec rsyslog_test_ns ip link set dev lo up + + # run server in namespace +-ip netns exec rsyslog_test_ns ./minitcpsrv -t127.0.0.1 -p'$TCPFLOOD_PORT' -f $RSYSLOG_OUT_LOG & ++ip netns exec rsyslog_test_ns ./minitcpsrv -t127.0.0.1 -p"$TCPFLOOD_PORT" -f $RSYSLOG_OUT_LOG & + BGPROCESS=$! + echo background minitcpsrvr process id is $BGPROCESS + +diff --git a/tools/omfwd.c b/tools/omfwd.c +index d1eeeff..4c23ee9 100644 +--- a/tools/omfwd.c ++++ b/tools/omfwd.c +@@ -869,46 +869,12 @@ static rsRetVal changeToNs(instanceData *const pData __attribute__((unused))) + { + DEFiRet; + #ifdef HAVE_SETNS +- int iErr; +- int destinationNs = -1; +- char *nsPath = NULL; + + if(pData->networkNamespace) { +- /* keep file descriptor of original network namespace */ +- pData->originalNamespace = open("/proc/self/ns/net", O_RDONLY); +- if (pData->originalNamespace < 0) { +- LogError(0, RS_RET_IO_ERROR, "omfwd: could not read /proc/self/ns/net"); +- ABORT_FINALIZE(RS_RET_IO_ERROR); +- } +- +- /* build network namespace path */ +- if (asprintf(&nsPath, "/var/run/netns/%s", pData->networkNamespace) == -1) { +- LogError(0, RS_RET_OUT_OF_MEMORY, "omfwd: asprintf failed"); +- ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); +- } +- +- /* keep file descriptor of destination network namespace */ +- destinationNs = open(nsPath, 0); +- if (destinationNs < 0) { +- LogError(0, RS_RET_IO_ERROR, "omfwd: could not change to namespace '%s'", +- pData->networkNamespace); +- ABORT_FINALIZE(RS_RET_IO_ERROR); +- } +- +- /* actually change in the destination network namespace */ +- if((iErr = (setns(destinationNs, CLONE_NEWNET))) != 0) { +- LogError(0, RS_RET_IO_ERROR, "could not change to namespace '%s': %s", +- pData->networkNamespace, gai_strerror(iErr)); +- ABORT_FINALIZE(RS_RET_IO_ERROR); +- } +- dbgprintf("omfwd: changed to network namespace '%s'\n", pData->networkNamespace); ++ CHKiRet(net.netns_save(&pData->originalNamespace)); ++ CHKiRet(net.netns_switch(pData->networkNamespace)); + } +- + finalize_it: +- free(nsPath); +- if(destinationNs >= 0) { +- close(destinationNs); +- } + #else /* #ifdef HAVE_SETNS */ + dbgprintf("omfwd: OS does not support network namespaces\n"); + #endif /* #ifdef HAVE_SETNS */ +@@ -923,22 +889,12 @@ static rsRetVal returnToOriginalNs(instanceData *const pData __attribute__((unus + { + DEFiRet; + #ifdef HAVE_SETNS +- int iErr; +- + /* only in case a network namespace is given and a file descriptor to + * the original namespace exists */ + if(pData->networkNamespace && pData->originalNamespace >= 0) { +- /* actually change to the original network namespace */ +- if((iErr = (setns(pData->originalNamespace, CLONE_NEWNET))) != 0) { +- LogError(0, RS_RET_IO_ERROR, "could not return to original namespace: %s", +- gai_strerror(iErr)); +- ABORT_FINALIZE(RS_RET_IO_ERROR); +- } +- +- close(pData->originalNamespace); ++ CHKiRet(net.netns_restore(&pData->originalNamespace)); + dbgprintf("omfwd: returned to original network namespace\n"); + } +- + finalize_it: + #endif /* #ifdef HAVE_SETNS */ + RETiRet; +-- +2.43.0 + diff --git a/SPECS/rsyslog/adding-new-functionality-for-omuxsock.patch b/SPECS/rsyslog/adding-new-functionality-for-omuxsock.patch new file mode 100644 index 00000000000..30282f1ca24 --- /dev/null +++ b/SPECS/rsyslog/adding-new-functionality-for-omuxsock.patch @@ -0,0 +1,1116 @@ +From ff3ec4c31dc3a214b67c9b368d81faba7b7a7ade Mon Sep 17 00:00:00 2001 +From: Billie Alsup +Date: Wed, 16 Apr 2025 14:00:29 -0700 +Subject: [PATCH 1/2] New functionality for omuxsock + +This builds on "PR#6121 net: Add NetworkNamespace APIS" +to add Network Namespace support for omuxsock. In +addition, new functionality is added to support +abstract socket names, as well as connected sockets. + +Changes are isolated to omuxsock to provide the new +functionality: + +a. Abstract unix socket names (including network namespaces) +b. Connected socket support (SOCK_STREAM, SOCK_SEQPACKET) +c. Support for load, and action configuration +d. Bug fix related to closing/reopening sockets + This bug would cause the socket to be closed and + reopened for every message sent. + +New tests are added as follows: + +a. uxsock_multiple.sh tests basic functionality with + multiple output instances. +b. uxsock_multiple_netns.sh is similar but using + multiple namespaces. +c. uxsock_simple_abstract.sh mirrors the existing + uxsock_simple.sh but tests with abstract socket + names. + +The uxsockrcvr test program was modified to support +these new tests. + +Signed-off-by: Billie Alsup + +Upstream Patch reference: https://patch-diff.githubusercontent.com/raw/rsyslog/rsyslog/pull/5630.patch +--- + plugins/omuxsock/omuxsock.c | 289 +++++++++++++++++++++++++---- + tests/Makefile.am | 16 +- + tests/uxsock_multiple-vg.sh | 3 + + tests/uxsock_multiple.sh | 95 ++++++++++ + tests/uxsock_multiple_netns-vg.sh | 3 + + tests/uxsock_multiple_netns.sh | 128 +++++++++++++ + tests/uxsock_simple.sh | 2 +- + tests/uxsock_simple_abstract-vg.sh | 3 + + tests/uxsock_simple_abstract.sh | 57 ++++++ + tests/uxsockrcvr.c | 131 +++++++++++-- + 10 files changed, 667 insertions(+), 60 deletions(-) + create mode 100755 tests/uxsock_multiple-vg.sh + create mode 100755 tests/uxsock_multiple.sh + create mode 100755 tests/uxsock_multiple_netns-vg.sh + create mode 100755 tests/uxsock_multiple_netns.sh + create mode 100755 tests/uxsock_simple_abstract-vg.sh + create mode 100755 tests/uxsock_simple_abstract.sh + +diff --git a/plugins/omuxsock/omuxsock.c b/plugins/omuxsock/omuxsock.c +index 99fbc22..54fe257 100644 +--- a/plugins/omuxsock/omuxsock.c ++++ b/plugins/omuxsock/omuxsock.c +@@ -26,6 +26,7 @@ + #include "rsyslog.h" + #include + #include ++#include + #include + #include + #include +@@ -43,6 +44,7 @@ + #include "glbl.h" + #include "errmsg.h" + #include "unicode-helper.h" ++#include "net.h" + + MODULE_TYPE_OUTPUT + MODULE_TYPE_NOKEEP +@@ -51,15 +53,21 @@ MODULE_CNFNAME("omuxsock") + /* internal structures + */ + DEF_OMOD_STATIC_DATA +-DEFobjCurrIf(glbl) ++DEFobjCurrIf(glbl) DEFobjCurrIf(net) + + #define INVLD_SOCK -1 + + typedef struct _instanceData { + permittedPeers_t *pPermPeers; +- uchar *sockName; +- int sock; +- struct sockaddr_un addr; ++ uchar *tplName; /**< Template name */ ++ uchar *sockName; /**< Socket name */ ++ char *namespace; /**< Network namespace */ ++ int sockType; /**< Socket type (DGRAM, STREAM, SEQPACKET) */ ++ int bAbstract; /**< True if an abstract socket address */ ++ int bConnected; /**< True if a connection oriented (STREAM, SEQPACKET) */ ++ int sock; /**< Socket descriptor */ ++ struct sockaddr_un addr; /**< Unix socket address */ ++ socklen_t addrLen; /**< The socket address length */ + } instanceData; + + +@@ -69,14 +77,20 @@ typedef struct wrkrInstanceData { + + /* config data */ + typedef struct configSettings_s { +- uchar *tplName; /* name of the default template to use */ +- uchar *sockName; /* name of the default template to use */ ++ uchar *tplName; /**< Name of the default template to use */ ++ uchar *sockName; /**< Name of the default socket to use */ ++ char *namespace; /**< Network namespace for abstract addresses */ ++ int sockType; /**< Socket type (DGRAM, STREAM, SEQPACKET) */ ++ int bAbstract; /**< True if default socket is abstract socket */ ++ int bConnected; /**< True if connection oriented (STREAM, SEQPACKET) */ + } configSettings_t; + static configSettings_t cs; + + /* module-global parameters */ + static struct cnfparamdescr modpdescr[] = { +- { "template", eCmdHdlrGetWord, 0 }, ++ {"template", eCmdHdlrGetWord, 0}, {"abstract", eCmdHdlrInt, 0}, ++ {"socketname", eCmdHdlrString, 0}, {"sockettype", eCmdHdlrString, 0}, ++ {"networknamespace", eCmdHdlrString, 0}, + }; + static struct cnfparamblk modpblk = + { CNFPARAMBLK_VERSION, +@@ -84,21 +98,91 @@ static struct cnfparamblk modpblk = + modpdescr + }; + ++/* tables for interfacing with the v6 config system */ ++/* action (instance) parameters */ ++static struct cnfparamdescr actpdescr[] = { ++ {"template", eCmdHdlrGetWord, 0}, {"abstract", eCmdHdlrInt, 0}, ++ {"socketname", eCmdHdlrString, 0}, {"sockettype", eCmdHdlrString, 0}, ++ {"networknamespace", eCmdHdlrString, 0}, ++}; ++static struct cnfparamblk actpblk = {CNFPARAMBLK_VERSION, sizeof(actpdescr) / sizeof(struct cnfparamdescr), actpdescr}; ++ + struct modConfData_s { +- rsconf_t *pConf; /* our overall config object */ +- uchar *tplName; /* default template */ ++ rsconf_t *pConf; /**< our overall config object */ ++ uchar *tplName; /**< default template */ ++ uchar *sockName; /**< Socket name */ ++ char *namespace; /**< Network namespace */ ++ int sockType; /**< Socket type (DGRAM, STREAM, SEQPACKET) */ ++ int bAbstract; /**< True if socket name is abstract */ ++ int bConnected; /**< True if socket is connection oriented (STREAM, SEQPACKET) */ + }; + + static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */ + static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current exec process */ + +- + static pthread_mutex_t mutDoAct = PTHREAD_MUTEX_INITIALIZER; + ++/** ++ * @brief A structure to map identifiers to socket information ++ * @details Socket information includes the socket type, and ++ * whether it is a connection-oriented socket type. ++ */ ++static struct { ++ const char *id; /**< The identifier used in configuration */ ++ int value; /**< The underlying socket type for this identifier */ ++ int connected; /**< True if the socket type is connection oriented */ ++} socketType_map[] = { ++ {"DGRAM", SOCK_DGRAM, 0}, ++ {"STREAM", SOCK_STREAM, 1}, ++#ifdef SOCK_SEQPACKET ++ {"SEQPACKET", SOCK_SEQPACKET, 1}, ++#endif /* def SOCK_SEQPACKET */ ++}; ++#define ARRAY_SIZE(n) (sizeof(n) / sizeof((n)[0])) ++ ++/** ++ * @brief Lookup an identifier to obtain socket information ++ * @param estr The identifer to lookup. The lookup is case ++ * insensitive. ++ * @param type The location to store the socket type ++ * @param connected The location to store an indicator of whether ++ * this socket type is connection oriented or not. ++ * @return RS_RET_OK on success, otherwise a failure code. ++ * @details This uses the above socketType_map structure to find ++ * information for a given socket type identifier. ++ */ ++static rsRetVal ATTR_NONNULL() _decodeSockType(es_str_t *estr, int *type, int *connected) { ++ DEFiRet; ++ char *cstr = es_str2cstr(estr, NULL); ++ size_t index; ++ ++ if (!cstr) { ++ ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); ++ } ++ ++ for (index = 0; index < ARRAY_SIZE(socketType_map); ++index) { ++ if (!strcasecmp(cstr, socketType_map[index].id)) { ++ *type = socketType_map[index].value; ++ *connected = socketType_map[index].connected; ++ FINALIZE ++ } ++ } ++ LogError(0, RS_RET_ERR, "omuxsock: bad socket type %s", cstr); ++ ABORT_FINALIZE(RS_RET_ERR); ++ ++finalize_it: ++ free(cstr); ++ RETiRet; ++} ++ + BEGINinitConfVars /* (re)set config variables to default values */ + CODESTARTinitConfVars + cs.tplName = NULL; + cs.sockName = NULL; ++ cs.namespace = NULL; ++ cs.sockType = SOCK_DGRAM; ++ cs.bAbstract = 0; ++ cs.bConnected = 0; + ENDinitConfVars + + +@@ -162,6 +246,11 @@ CODESTARTbeginCnfLoad + loadModConf = pModConf; + pModConf->pConf = pConf; + pModConf->tplName = NULL; ++ pModConf->sockName = NULL; ++ pModConf->namespace = NULL; ++ pModConf->sockType = SOCK_DGRAM; ++ pModConf->bAbstract = 0; ++ pModConf->bConnected = 0; + ENDbeginCnfLoad + + BEGINsetModCnf +@@ -190,6 +279,14 @@ CODESTARTsetModCnf + "was already set via legacy directive - may lead to inconsistent " + "results."); + } ++ } else if (!strcmp(modpblk.descr[i].name, "abstract")) { ++ loadModConf->bAbstract = !!(int)pvals[i].val.d.n; ++ } else if (!strcmp(modpblk.descr[i].name, "socketname")) { ++ loadModConf->sockName = (uchar *)es_str2cstr(pvals[i].val.d.estr, NULL); ++ } else if (!strcmp(modpblk.descr[i].name, "sockettype")) { ++ CHKiRet(_decodeSockType(pvals[i].val.d.estr, &loadModConf->sockType, &loadModConf->bConnected)); ++ } else if (!strcmp(modpblk.descr[i].name, "networknamespace")) { ++ loadModConf->namespace = es_str2cstr(pvals[i].val.d.estr, NULL); + } else { + dbgprintf("omuxsock: program error, non-handled " + "param '%s' in beginCnfLoad\n", modpblk.descr[i].name); +@@ -206,6 +303,10 @@ CODESTARTendCnfLoad + /* free legacy config vars */ + free(cs.tplName); + cs.tplName = NULL; ++ free(cs.sockName); ++ cs.sockName = NULL; ++ free(cs.namespace); ++ cs.namespace = NULL; + ENDendCnfLoad + + BEGINcheckCnf +@@ -220,11 +321,14 @@ ENDactivateCnf + BEGINfreeCnf + CODESTARTfreeCnf + free(pModConf->tplName); ++ free(pModConf->sockName); ++ free(pModConf->namespace); + ENDfreeCnf + + BEGINcreateInstance + CODESTARTcreateInstance + pData->sock = INVLD_SOCK; ++ pData->sockType = SOCK_DGRAM; + ENDcreateInstance + + BEGINcreateWrkrInstance +@@ -243,9 +347,85 @@ BEGINfreeInstance + CODESTARTfreeInstance + /* final cleanup */ + closeSocket(pData); ++ free(pData->tplName); + free(pData->sockName); ++ free(pData->namespace); + ENDfreeInstance + ++BEGINnewActInst ++ struct cnfparamvals *pvals = NULL; ++ int i; ++ int bHaveAbstract = 0; ++ int bHaveSocketType = 0; ++ uchar *tplToUse; ++ ++ CODESTARTnewActInst; ++ CODE_STD_STRING_REQUESTnewActInst(1); ++ if ((pvals = nvlstGetParams(lst, &actpblk, NULL)) == NULL) { ++ ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); ++ } ++ CHKiRet(createInstance(&pData)); ++ ++ for (i = 0; i < actpblk.nParams; ++i) { ++ if (!pvals[i].bUsed) continue; ++ 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, "abstract")) { ++ pData->bAbstract = !!(int)pvals[i].val.d.n; ++ bHaveAbstract = 1; ++ } else if (!strcmp(actpblk.descr[i].name, "socketname")) { ++ pData->sockName = (uchar *)es_str2cstr(pvals[i].val.d.estr, NULL); ++ } else if (!strcmp(actpblk.descr[i].name, "sockettype")) { ++ CHKiRet(_decodeSockType(pvals[i].val.d.estr, &pData->sockType, &pData->bConnected)); ++ bHaveSocketType = 1; ++ } else if (!strcmp(actpblk.descr[i].name, "networknamespace")) { ++ pData->namespace = es_str2cstr(pvals[i].val.d.estr, NULL); ++ } ++ } ++ ++ if (!pData->namespace && loadModConf->namespace) { ++ CHKmalloc(pData->namespace = strdup(loadModConf->namespace)); ++ } ++ if (!pData->sockName) { ++ if (bHaveAbstract) { ++ LogError(0, RS_RET_NO_SOCK_CONFIGURED, "omuxsock: abstract configured without socket name\n"); ++ ABORT_FINALIZE(RS_RET_NO_SOCK_CONFIGURED); ++ } ++ if (bHaveSocketType) { ++ LogError(0, RS_RET_NO_SOCK_CONFIGURED, "omuxsock: socket type configured without socket name\n"); ++ ABORT_FINALIZE(RS_RET_NO_SOCK_CONFIGURED); ++ } ++ /* ++ * Note that we explicitly want the semantics that these parameters be consumed ++ * as a group, and not be individually used as a default if the corresponding ++ * parameter is not provided in the instance. ++ */ ++ if (loadModConf != NULL && loadModConf->sockName != NULL) { ++ pData->sockName = ustrdup(loadModConf->sockName); ++ pData->bAbstract = loadModConf->bAbstract; ++ pData->sockType = loadModConf->sockType; ++ pData->bConnected = loadModConf->bConnected; ++ } else if (cs.sockName == NULL) { ++ LogError(0, RS_RET_NO_SOCK_CONFIGURED, "No output socket configured for omuxsock\n"); ++ ABORT_FINALIZE(RS_RET_NO_SOCK_CONFIGURED); ++ } else { ++ /* ++ * Ownership is transferred here. There is only one default through cs structure. ++ */ ++ pData->sockName = cs.sockName; ++ pData->bAbstract = cs.bAbstract; ++ pData->sockType = cs.sockType; ++ pData->bConnected = cs.bConnected; ++ cs.sockName = NULL; /* pData is now owner and will free it */ ++ } ++ } ++ ++ tplToUse = ustrdup((pData->tplName == NULL) ? getDfltTpl() : pData->tplName); ++ CHKiRet(OMSRsetEntry(*ppOMSR, 0, tplToUse, OMSR_NO_RQD_TPL_OPTS)); ++ ++ CODE_STD_FINALIZERnewActInst cnfparamvalsDestruct(pvals, &actpblk); ++ENDnewActInst ++ + BEGINfreeWrkrInstance + CODESTARTfreeWrkrInstance + ENDfreeWrkrInstance +@@ -257,7 +437,7 @@ CODESTARTdbgPrintInstInfo + ENDdbgPrintInstInfo + + +-/* Send a message via UDP ++/* Send a message + * rgehards, 2007-12-20 + */ + static rsRetVal sendMsg(instanceData *pData, char *msg, size_t len) +@@ -270,13 +450,27 @@ static rsRetVal sendMsg(instanceData *pData, char *msg, size_t len) + } + + if(pData->sock != INVLD_SOCK) { +- lenSent = sendto(pData->sock, msg, len, 0, (const struct sockaddr *)&pData->addr, +- sizeof(pData->addr)); ++ /* ++ * This style is perhaps easier to follow. However note that even for non-connection-oriented ++ * sockets, a simple send can be used, as long as connect was called earlier. The connect ++ * parameters are simply used as the default in subsequent send() invocations. ++ */ ++ if (pData->bConnected) { ++ lenSent = send(pData->sock, msg, len, 0); ++ } else { ++ lenSent = sendto(pData->sock, msg, len, 0, (const struct sockaddr *)&pData->addr, pData->addrLen); ++ } + if(lenSent != len) { + int eno = errno; + char errStr[1024]; +- DBGPRINTF("omuxsock suspending: sendto(), socket %d, error: %d = %s.\n", +- pData->sock, eno, rs_strerror_r(eno, errStr, sizeof(errStr))); ++ ++ /* ++ * XXX/rgerhards: how is this message correct, in that we are returning OK still? ++ * In reality, the remainder of the partially sent message (or never sent message) ++ * is simply dropped, and we do not change the state of the socket. ++ */ ++ DBGPRINTF("omuxsock suspending: send%s(), socket %d, error: %d = %s.\n", pData->bConnected ? "" : "to", ++ pData->sock, eno, rs_strerror_r(eno, errStr, sizeof(errStr))); + } + } + +@@ -291,23 +485,40 @@ static rsRetVal + openSocket(instanceData *pData) + { + DEFiRet; +- assert(pData->sock == INVLD_SOCK); ++ size_t nameLen; + +- if((pData->sock = socket(AF_UNIX, SOCK_DGRAM, 0)) < 0) { +- char errStr[1024]; +- int eno = errno; +- DBGPRINTF("error %d creating AF_UNIX/SOCK_DGRAM: %s.\n", +- eno, rs_strerror_r(eno, errStr, sizeof(errStr))); +- pData->sock = INVLD_SOCK; +- ABORT_FINALIZE(RS_RET_NO_SOCKET); ++ assert(pData->sock == INVLD_SOCK); + +- } ++ CHKiRet(net.netns_socket(&pData->sock, AF_UNIX, pData->sockType, 0, pData->namespace)); + + /* set up server address structure */ + memset(&pData->addr, 0, sizeof(pData->addr)); ++ /* ++ * For pathname addresses, the +1 is the terminating \0. ++ * For abstract addresses, the +1 is the leading \0 and note that there is NO terminating \0. ++ */ ++ nameLen = strlen((char *)pData->sockName); ++ if ((nameLen + 1) > sizeof(pData->addr.sun_path)) { ++ LogError(0, RS_TRUNCAT_TOO_LARGE, "Socket name '%s' is too long", pData->sockName); ++ ABORT_FINALIZE(RS_TRUNCAT_TOO_LARGE); ++ } ++ pData->addrLen = offsetof(struct sockaddr_un, sun_path) + 1 + nameLen; + pData->addr.sun_family = AF_UNIX; +- strncpy(pData->addr.sun_path, (char*)pData->sockName, sizeof(pData->addr.sun_path)); +- pData->addr.sun_path[sizeof(pData->addr.sun_path)-1] = '\0'; ++ /* ++ * Note destination is all \0 initially, so a non-abstract name is properly terminated, ++ * and an abstract name doesn't care what follows (and may consume the entire sun_path). ++ */ ++ strncpy(pData->addr.sun_path + pData->bAbstract, (char *)pData->sockName, nameLen); ++ ++ /* ++ * Note that connect is legal even for non-connected sockets, and the parameters so passed ++ * become defaults for the send function. ++ */ ++ if (pData->bConnected && connect(pData->sock, (struct sockaddr *)&pData->addr, pData->addrLen) == -1) { ++ LogError(errno, RS_RET_NO_SOCKET, "Error connecting to %ssocket %s", pData->bAbstract ? "abstract " : "", ++ pData->sockName); ++ ABORT_FINALIZE(RS_RET_NO_SOCKET); ++ } + + finalize_it: + if(iRet != RS_RET_OK) { +@@ -323,13 +534,13 @@ finalize_it: + static rsRetVal doTryResume(instanceData *pData) + { + DEFiRet; ++ if (pData->sock == INVLD_SOCK) { ++ DBGPRINTF("omuxsock trying to resume\n"); ++ iRet = openSocket(pData); + +- DBGPRINTF("omuxsock trying to resume\n"); +- closeSocket(pData); +- iRet = openSocket(pData); +- +- if(iRet != RS_RET_OK) { +- iRet = RS_RET_SUSPENDED; ++ if (iRet != RS_RET_OK) { ++ iRet = RS_RET_SUSPENDED; ++ } + } + + RETiRet; +@@ -375,12 +586,10 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) + } + + /* ok, if we reach this point, we have something for us */ +- p += sizeof(":omuxsock:") - 1; /* eat indicator sequence (-1 because of '\0'!) */ ++ p += sizeof(":omuxsock:") - 1; /* eat indicator sequence (-1 because of '\0'!) */ + CHKiRet(createInstance(&pData)); + + /* check if a non-standard template is to be applied */ +- if(*(p-1) == ';') +- --p; + CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, 0, getDfltTpl())); + + if(cs.sockName == NULL) { +@@ -389,7 +598,10 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) + } + + pData->sockName = cs.sockName; +- cs.sockName = NULL; /* pData is now owner and will fee it */ ++ pData->sockType = cs.sockType; ++ pData->bAbstract = cs.bAbstract; ++ pData->bConnected = cs.bConnected; ++ cs.sockName = NULL; /* pData is now owner and will free it */ + + CODE_STD_FINALIZERparseSelectorAct + ENDparseSelectorAct +@@ -423,6 +635,7 @@ CODEqueryEtryPt_STD_OMOD_QUERIES + CODEqueryEtryPt_STD_OMOD8_QUERIES + CODEqueryEtryPt_STD_CONF2_QUERIES + CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES ++CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES + ENDqueryEtryPt + + +@@ -440,9 +653,9 @@ BEGINmodInit() + CODESTARTmodInit + INITLegCnfVars + *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */ +-CODEmodInit_QueryRegCFSLineHdlr ++ CODEmodInit_QueryRegCFSLineHdlr; + CHKiRet(objUse(glbl, CORE_COMPONENT)); +- ++ CHKiRet(objUse(net, LM_NET_FILENAME)); + CHKiRet(regCfSysLineHdlr((uchar *)"omuxsockdefaulttemplate", 0, eCmdHdlrGetWord, setLegacyDfltTpl, + NULL, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"omuxsocksocket", 0, eCmdHdlrGetWord, NULL, &cs.sockName, NULL)); +diff --git a/tests/Makefile.am b/tests/Makefile.am +index a9713e0..2d63722 100644 +--- a/tests/Makefile.am ++++ b/tests/Makefile.am +@@ -1483,7 +1483,15 @@ endif + endif + + if ENABLE_OMUXSOCK +-TESTS += uxsock_simple.sh ++TESTS += uxsock_simple.sh \ ++ uxsock_simple_abstract.sh \ ++ uxsock_multiple.sh \ ++ uxsock_multiple_netns.sh ++if HAVE_VALGRIND ++TESTS += uxsock_simple_abstract-vg.sh \ ++ uxsock_multiple-vg.sh \ ++ uxsock_multiple_netns-vg.sh ++endif # HAVE_VALGRIND + endif + + if ENABLE_RELP +@@ -2622,6 +2630,12 @@ EXTRA_DIST= \ + testsuites/omprog-transactions-bin.sh \ + pipe_noreader.sh \ + uxsock_simple.sh \ ++ uxsock_simple_abstract.sh \ ++ uxsock_simple_abstract-vg.sh \ ++ uxsock_multiple.sh \ ++ uxsock_multiple-vg.sh \ ++ uxsock_multiple_netns.sh \ ++ uxsock_multiple_netns-vg.sh \ + asynwr_simple.sh \ + asynwr_simple_2.sh \ + asynwr_timeout.sh \ +diff --git a/tests/uxsock_multiple-vg.sh b/tests/uxsock_multiple-vg.sh +new file mode 100755 +index 0000000..5157947 +--- /dev/null ++++ b/tests/uxsock_multiple-vg.sh +@@ -0,0 +1,3 @@ ++#!/bin/bash ++export USE_VALGRIND="YES" ++source ${srcdir:-.}/uxsock_multiple.sh +diff --git a/tests/uxsock_multiple.sh b/tests/uxsock_multiple.sh +new file mode 100755 +index 0000000..93ea9e2 +--- /dev/null ++++ b/tests/uxsock_multiple.sh +@@ -0,0 +1,95 @@ ++#!/bin/bash ++# This tests basic omuxsock functionality. Multiple socket receivers are started ++# which sends all data to an output file, then a rsyslog instance is started which ++# generates messages and sends them to multiple unix sockets. Datagram sockets are ++# being used. ++# Based on uxsock_simple.sh added 2010-08-06 by Rgerhards ++# Updated 2025-04-16 for abstract sockets ++. ${srcdir:=.}/diag.sh init ++check_command_available timeout ++ ++uname ++if [ $(uname) != "Linux" ] ; then ++ echo "This test requires Linux (AF_UNIX abstract addresses)" ++ exit 77 ++fi ++ ++SOCKET_NAMES=( ++ "$RSYSLOG_DYNNAME-testbench-dgram-uxsock.0" ++ "$RSYSLOG_DYNNAME-testbench-dgram-uxsock.DGRAM" ++ "$RSYSLOG_DYNNAME-testbench-dgram-uxsock.STREAM" ++ "$RSYSLOG_DYNNAME-testbench-dgram-uxsock.SEQPACKET" ++) ++ ++# create the pipe and start a background process that copies data from ++# it to the "regular" work file ++generate_conf ++for name in "${SOCKET_NAMES[@]}"; do ++ ext=${name##*.} ++ if [ ${ext} == "0" ]; then ++ add_conf ' ++ ++ $template outfmt,"%msg:F,58:2%\n" ++ ++ module( ++ load = "../plugins/omuxsock/.libs/omuxsock" ++ template = "outfmt" ++ SocketName = "'${name}'" ++ abstract = "1" ++ ) ++ :msg, contains, "msgnum:" { ++ action( ++ type = "omuxsock" ++ ) ++ ' ++ else ++ add_conf ' ++ action( ++ type = "omuxsock" ++ SocketName = "'$name'" ++ abstract = "1" ++ SocketType = "'$ext'" ++ ) ++ ' ++ fi ++done ++add_conf ' ++} ++' ++BGPROCESS=() ++for name in "${SOCKET_NAMES[@]}"; do ++ ext=${name##*.} ++ if [ "${ext}" == "0" ]; then ++ type="" ++ else ++ type="-T${ext}" ++ fi ++ timeout 30s ./uxsockrcvr -a -s$name ${type} -o ${RSYSLOG_OUT_LOG}.${ext} -t 60 & ++ PID=($!) ++ BGPROCESS+=($PID) ++ echo background uxsockrcvr ${name} process id is $PID ++done ++ ++# now do the usual run ++startup ++# 10000 messages should be enough ++injectmsg 0 10000 ++shutdown_when_empty # shut down rsyslogd when done processing messages ++wait_shutdown ++ ++# wait for the cp process to finish, do pipe-specific cleanup ++echo shutting down uxsockrcvr... ++# TODO: we should do this more reliable in the long run! (message counter? timeout?) ++for pid in ${BGPROCESS[@]}; do ++ kill $pid ++ wait $pid ++done ++echo background processes have terminated, continue test... ++ ++# and continue the usual checks ++BASE=${RSYSLOG_OUT_LOG} ++for name in "${SOCKET_NAMES[@]}"; do ++ RSYSLOG_OUT_LOG=${BASE}.${name##*.} ++ seq_check 0 9999 ++done ++exit_test +diff --git a/tests/uxsock_multiple_netns-vg.sh b/tests/uxsock_multiple_netns-vg.sh +new file mode 100755 +index 0000000..4eec983 +--- /dev/null ++++ b/tests/uxsock_multiple_netns-vg.sh +@@ -0,0 +1,3 @@ ++#!/bin/bash ++export USE_VALGRIND="YES" ++source ${srcdir:-.}/uxsock_multiple_netns.sh +diff --git a/tests/uxsock_multiple_netns.sh b/tests/uxsock_multiple_netns.sh +new file mode 100755 +index 0000000..d233d30 +--- /dev/null ++++ b/tests/uxsock_multiple_netns.sh +@@ -0,0 +1,128 @@ ++#!/bin/bash ++# This tests basic omuxsock functionality with namespaces. ++# Multiple socket receivers are started which sends all data ++# to an output file, then a rsyslog instance is started which ++# generates messages and sends them to multiple unix sockets. ++# Multiple socket types are being used. ++# Based on uxsock_simple.sh added 2010-08-06 by Rgerhards ++# Updated 2025-04-16 for abstract sockets and multiple socket ++# types. ++echo =============================================================================== ++echo \[uxsock_multiple_netns.sh\]: test for transmitting to another namespace ++echo This test must be run with CAP_SYS_ADMIN capabilities [network namespace creation/change required] ++if [ "$EUID" -ne 0 ]; then ++ exit 77 # Not root, skip this test ++fi ++ ++. ${srcdir:=.}/diag.sh init ++check_command_available timeout ++ ++uname ++if [ $(uname) != "Linux" ] ; then ++ echo "This test requires Linux (AF_UNIX abstract addresses)" ++ exit 77 ++fi ++ ++NS_PREFIX=$(basename ${RSYSLOG_DYNNAME}) ++ ++SOCKET_NAMES=( ++ "$RSYSLOG_DYNNAME-testbench-dgram-uxsock.0" ++ "$RSYSLOG_DYNNAME-testbench-dgram-uxsock.DGRAM" ++ "$RSYSLOG_DYNNAME-testbench-dgram-uxsock.STREAM" ++ "$RSYSLOG_DYNNAME-testbench-dgram-uxsock.SEQPACKET" ++) ++ ++# create the pipe and start a background process that copies data from ++# it to the "regular" work file ++generate_conf ++for name in "${SOCKET_NAMES[@]}"; do ++ ext=${name##*.} ++ NS=${NS_PREFIX}.${ext} ++ if [ ${ext} == "0" ]; then ++ add_conf ' ++ ++ $template outfmt,"%msg:F,58:2%\n" ++ ++ module( ++ load = "../plugins/omuxsock/.libs/omuxsock" ++ template = "outfmt" ++ SocketName = "'${name}'" ++ abstract = "1" ++ NetworkNamespace="'${NS}'" ++ ) ++ :msg, contains, "msgnum:" { ++ action( ++ type = "omuxsock" ++ ) ++ ' ++ else ++ add_conf ' ++ action( ++ type = "omuxsock" ++ SocketName = "'$name'" ++ abstract = "1" ++ SocketType = "'$ext'" ++ NetworkNamespace="'${NS}'" ++ ) ++ action( ++ type = "omuxsock" ++ SocketName = "'$name'" ++ abstract = "1" ++ SocketType = "'$ext'" ++ NetworkNamespace="'${NS}.fail'" ++ ) ++ ' ++ fi ++done ++add_conf ' ++} ++' ++ ++BGPROCESS=() ++for name in "${SOCKET_NAMES[@]}"; do ++ ext=${name##*.} ++ NS=${NS_PREFIX}.${ext} ++ ip netns add "${NS}" ++ ip netns exec "${NS}" ip link set dev lo up ++ ip netns delete "${NS}.fail" > /dev/null 2>&1 ++ if [ "${ext}" == "0" ]; then ++ type="" ++ else ++ type="-T${ext}" ++ fi ++ timeout 30s ip netns exec "${NS}" ./uxsockrcvr -a -s$name ${type} -o ${RSYSLOG_OUT_LOG}.${ext} -t 60 & ++ PID=($!) ++ BGPROCESS+=($PID) ++ echo background uxsockrcvr ${name} process id is $PID ++done ++ ++# now do the usual run ++startup ++# 10000 messages should be enough ++injectmsg 0 10000 ++shutdown_when_empty # shut down rsyslogd when done processing messages ++wait_shutdown ++ ++# wait for the cp process to finish, do pipe-specific cleanup ++echo shutting down uxsockrcvr... ++# TODO: we should do this more reliable in the long run! (message counter? timeout?) ++for pid in ${BGPROCESS[@]}; do ++ kill $pid ++ wait $pid ++done ++echo background processes have terminated, continue test... ++ ++# Remove namespaces ++for name in "${SOCKET_NAMES[@]}"; do ++ ext=${name##*.} ++ NS=${NS_PREFIX}.${ext} ++ ip netns delete "${NS}" > /dev/null 2>&1 ++done ++ ++# and continue the usual checks ++BASE=${RSYSLOG_OUT_LOG} ++for name in "${SOCKET_NAMES[@]}"; do ++ RSYSLOG_OUT_LOG=${BASE}.${name##*.} ++ seq_check 0 9999 ++done ++exit_test +diff --git a/tests/uxsock_simple.sh b/tests/uxsock_simple.sh +index af97698..2cbd7ca 100755 +--- a/tests/uxsock_simple.sh ++++ b/tests/uxsock_simple.sh +@@ -12,7 +12,7 @@ if [ $(uname) = "FreeBSD" ] ; then + exit 77 + fi + +-# create the pipe and start a background process that copies data from ++# create the pipe and start a background process that copies data from + # it to the "regular" work file + generate_conf + add_conf ' +diff --git a/tests/uxsock_simple_abstract-vg.sh b/tests/uxsock_simple_abstract-vg.sh +new file mode 100755 +index 0000000..ff24ff3 +--- /dev/null ++++ b/tests/uxsock_simple_abstract-vg.sh +@@ -0,0 +1,3 @@ ++#!/bin/bash ++export USE_VALGRIND="YES" ++source ${srcdir:-.}/uxsock_simple_abstract.sh +diff --git a/tests/uxsock_simple_abstract.sh b/tests/uxsock_simple_abstract.sh +new file mode 100755 +index 0000000..1a0d09a +--- /dev/null ++++ b/tests/uxsock_simple_abstract.sh +@@ -0,0 +1,57 @@ ++#!/bin/bash ++# This tests basic omuxsock functionality. A socket receiver is started which sends ++# all data to an output file, then a rsyslog instance is started which generates ++# messages and sends them to the unix socket. Datagram sockets are being used. ++# added 2010-08-06 by Rgerhards ++# Updated 2025-04-16 for abstract sockets ++. ${srcdir:=.}/diag.sh init ++check_command_available timeout ++ ++uname ++if [ $(uname) != "Linux" ] ; then ++ echo "This test requires Linux (AF_UNIX abstract addresses)" ++ exit 77 ++fi ++ ++SOCKET_NAME="$RSYSLOG_DYNNAME-testbench-dgram-uxsock-abstract" ++ ++# create the pipe and start a background process that copies data from ++# it to the "regular" work file ++generate_conf ++add_conf ' ++$MainMsgQueueTimeoutShutdown 10000 ++ ++$template outfmt,"%msg:F,58:2%\n" ++ ++module( ++ load = "../plugins/omuxsock/.libs/omuxsock" ++ template = "outfmt" ++) ++ ++:msg, contains, "msgnum:" action( ++ type = "omuxsock" ++ SocketName = "'$SOCKET_NAME'" ++ abstract = "1" ++) ++' ++timeout 30s ./uxsockrcvr -a -s$SOCKET_NAME -o $RSYSLOG_OUT_LOG -t 60 & ++BGPROCESS=$! ++echo background uxsockrcvr process id is $BGPROCESS ++ ++# now do the usual run ++startup ++# 10000 messages should be enough ++injectmsg 0 10000 ++shutdown_when_empty # shut down rsyslogd when done processing messages ++wait_shutdown ++ ++# wait for the cp process to finish, do pipe-specific cleanup ++echo shutting down uxsockrcvr... ++# TODO: we should do this more reliable in the long run! (message counter? timeout?) ++kill $BGPROCESS ++wait $BGPROCESS ++echo background process has terminated, continue test... ++ ++# and continue the usual checks ++seq_check 0 9999 ++exit_test +diff --git a/tests/uxsockrcvr.c b/tests/uxsockrcvr.c +index 0d61ce9..05bbfc6 100644 +--- a/tests/uxsockrcvr.c ++++ b/tests/uxsockrcvr.c +@@ -31,6 +31,7 @@ + #include "config.h" + #include + #include ++#include + #include + #include + #include +@@ -53,8 +54,14 @@ + #define DFLT_TIMEOUT 60 + + char *sockName = NULL; +-int sock; ++int sockType = SOCK_DGRAM; ++int sockConnected = 0; + int addNL = 0; ++int abstract; ++#define MAX_FDS 4 ++int sockBacklog = MAX_FDS - 1; ++struct pollfd fds[MAX_FDS]; ++int nfds; + + + /* called to clean up on exit +@@ -62,8 +69,12 @@ int addNL = 0; + void + cleanup(void) + { +- unlink(sockName); +- close(sock); ++ int index; ++ ++ if (!abstract) unlink(sockName); ++ for (index = 0; index < nfds; ++index) { ++ close(fds[index].fd); ++ } + } + + +@@ -80,10 +91,46 @@ usage(void) + fprintf(stderr, "usage: uxsockrcvr -s /socket/name -o /output/file -l\n" + "-l adds newline after each message received\n" + "-s MUST be specified\n" ++ "-a Use abstract socket name\n" ++ "-T {DGRAM|STREAM" ++#ifdef SOCK_SEQPACKET ++ "|SEQPACKET" ++#endif /* def SOCK_SEQPACKET */ ++ "} Set socket type (default DGRAM)\n" + "if -o ist not specified, stdout is used\n"); + exit(1); + } + ++static struct { ++ const char *id; ++ int val; ++ int connected; ++} _sockType_map[] = { ++ {"DGRAM", SOCK_DGRAM, 0}, ++ {"STREAM", SOCK_STREAM, 1}, ++#ifdef SOCK_SEQPACKET ++ {"SEQPACKET", SOCK_SEQPACKET, 1}, ++#endif /* def SOCK_SEQPACKET */ ++ {NULL, 0, 0}, ++}; ++ ++static void _decode_sockType(const char *s) { ++ int index; ++ ++ for (index = 0; _sockType_map[index].id; ++index) { ++ if (!strcasecmp(s, _sockType_map[index].id)) { ++ sockType = _sockType_map[index].val; ++ sockConnected = _sockType_map[index].connected; ++ return; ++ } ++ } ++ fprintf(stderr, "?Illegal socket type '%s'. Valid socket types:\n", s); ++ ++ for (index = 0; _sockType_map[index].id; ++index) { ++ fprintf(stderr, " %s\n", _sockType_map[index].id); ++ } ++ exit(1); ++} + + int + main(int argc, char *argv[]) +@@ -96,15 +143,18 @@ main(int argc, char *argv[]) + struct sockaddr_un addr; /* address of server */ + struct sockaddr from; + socklen_t fromlen; +- struct pollfd fds[1]; ++ int index; + + if(argc < 2) { + fprintf(stderr, "error: too few arguments!\n"); + usage(); + } + +- while((opt = getopt(argc, argv, "s:o:lt:")) != EOF) { ++ while ((opt = getopt(argc, argv, "as:o:lt:T:")) != EOF) { + switch((char)opt) { ++ case 'a': ++ abstract = 1; ++ break; + case 'l': + addNL = 1; + break; +@@ -120,6 +170,9 @@ main(int argc, char *argv[]) + case 't': + timeout = atoi(optarg); + break; ++ case 'T': ++ _decode_sockType(optarg); ++ break; + default:usage(); + } + } +@@ -141,7 +194,7 @@ main(int argc, char *argv[]) + } + + /* Create a UNIX datagram socket for server */ +- if ((sock = socket(AF_UNIX, SOCK_DGRAM, 0)) < 0) { ++ if ((fds[0].fd = socket(AF_UNIX, sockType, 0)) < 0) { + perror("server: socket"); + exit(1); + } +@@ -151,40 +204,78 @@ main(int argc, char *argv[]) + /* Set up address structure for server socket */ + memset(&addr, 0, sizeof(addr)); + addr.sun_family = AF_UNIX; +- strcpy(addr.sun_path, sockName); ++ strncpy(addr.sun_path + abstract, sockName, sizeof(addr.sun_path) - abstract); ++ /* ++ * Abstract socket addresses do not require termination. ++ */ ++ if (!abstract && addr.sun_path[sizeof(addr.sun_path) - 1]) { ++ addr.sun_path[sizeof(addr.sun_path) - 1] = 0; ++ fprintf(stderr, "warning: socket path truncated: %s\n", addr.sun_path); ++ } + +- if (bind(sock, (struct sockaddr*) &addr, sizeof(addr)) < 0) { +- close(sock); ++ /* ++ * For pathname addresses, the +1 is the terminating \0 character. ++ * For abstract addresses, the +1 is the leading \0 character. ++ */ ++ if (bind(fds[0].fd, (struct sockaddr *)&addr, offsetof(struct sockaddr_un, sun_path) + strlen(sockName) + 1) < 0) { + perror("server: bind"); ++ close(fds[0].fd); ++ exit(1); ++ } ++ ++ if (sockConnected && listen(fds[0].fd, sockBacklog) == -1) { ++ perror("server: listen"); ++ close(fds[0].fd); + exit(1); + } + +- fds[0].fd = sock; + fds[0].events = POLLIN; ++ nfds = 1; + + /* we now run in an endless loop. We do not check who sends us + * data. This should be no problem for our testbench use. + */ + + while(1) { +- fromlen = sizeof(from); +- rlen = poll(fds, 1, timeout); ++ rlen = poll(fds, nfds, timeout); + if(rlen == -1) { + perror("uxsockrcvr : poll\n"); + exit(1); + } else if(rlen == 0) { + fprintf(stderr, "Socket timed out - nothing to receive\n"); + exit(1); +- } else { +- rlen = recvfrom(sock, data, 2000, 0, &from, &fromlen); +- if(rlen == -1) { +- perror("uxsockrcvr : recv\n"); +- exit(1); +- } else { +- fwrite(data, 1, rlen, fp); +- if(addNL) ++ } ++ if (sockConnected && fds[0].revents & POLLIN) { ++ fds[nfds].fd = accept(fds[0].fd, NULL, NULL); ++ fds[nfds].events = POLLIN; ++ fds[nfds].revents = 0; ++ if (fds[nfds].fd < 0) { ++ perror("accept"); ++ } else if (++nfds == MAX_FDS) { ++ fds[0].events = 0; ++ } ++ } ++ for (index = sockConnected; index < nfds; ++index) { ++ if (fds[index].revents & POLLIN) { ++ fromlen = sizeof(from); ++ rlen = recvfrom(fds[index].fd, data, 2000, 0, &from, &fromlen); ++ if (rlen == -1) { ++ perror("uxsockrcvr : recv\n"); ++ exit(1); ++ } else { ++ fwrite(data, 1, rlen, fp); ++ if(addNL) + fputc('\n', fp); ++ } + } ++ if (fds[index].revents & POLLHUP) { ++ close(fds[index].fd); ++ fds[index].fd = -1; ++ fds[index].events = 0; ++ } ++ } ++ while (nfds && fds[nfds - 1].fd < 0) { ++ --nfds; + } + } + +-- +2.43.0 + diff --git a/SPECS/rsyslog/fix-message-loss-bug-for-connected-Unix-domain-sockets.patch b/SPECS/rsyslog/fix-message-loss-bug-for-connected-Unix-domain-sockets.patch new file mode 100644 index 00000000000..069f0a0a6e1 --- /dev/null +++ b/SPECS/rsyslog/fix-message-loss-bug-for-connected-Unix-domain-sockets.patch @@ -0,0 +1,424 @@ +From ceea68eba615eacd694ee64ce2490ff27b1a9039 Mon Sep 17 00:00:00 2001 +From: Billie Alsup +Date: Wed, 22 Apr 2026 15:56:03 -0700 +Subject: [PATCH 1/6] omuxsock: fix message loss bug for connected Unix domain + sockets + +Resolves critical message loss issue where connected sockets (STREAM, +SEQPACKET) would silently drop all subsequent messages after any +send error, requiring process restart to recover functionality. +Discovered during commercial application integration. + +Impact: Prevents message loss and enables automatic recovery for +connected Unix domain socket configurations. + +Before: Send errors on connected sockets left socket in bad state +without suspension, causing all future messages to be lost silently. +After: Connected socket errors trigger proper cleanup and suspension, +allowing infrastructure to resume and reconnect automatically. + +Technical Overview: +- Fixed error handling to distinguish connected vs unconnected socket + behavior when send()/sendto() operations fail partially or completely +- Connected sockets now properly close and return RS_RET_SUSPENDED + to enable infrastructure-driven reconnection and recovery +- Unconnected sockets (DGRAM) maintain existing retry-on-next-message + behavior without suspension as they are inherently stateless +- Changed lenSent type from unsigned to ssize_t for proper error + detection and safe signed/unsigned comparison handling +- Enhanced error logging with transmission byte counts for better + debugging of partial write scenarios +- Added comprehensive tests that verify recovery behavior and fail + without this fix, confirming the bug reproduction and resolution + +Fixes message loss regression affecting STREAM and SEQPACKET Unix +domain socket configurations under error conditions. + +With the help of AI-Agents: GitHub Copilot + +Upstream Patch reference: https://patch-diff.githubusercontent.com/raw/rsyslog/rsyslog/pull/6750.patch +--- + plugins/omuxsock/omuxsock.c | 26 ++++++-- + tests/uxsock_multiple.sh | 24 ++++++- + tests/uxsock_multiple_netns.sh | 27 +++++++- + tests/uxsock_simple.sh | 6 +- + tests/uxsock_simple_abstract.sh | 14 +++- + tests/uxsockrcvr.c | 112 ++++++++++++++++++++++---------- + 6 files changed, 158 insertions(+), 51 deletions(-) + +diff --git a/plugins/omuxsock/omuxsock.c b/plugins/omuxsock/omuxsock.c +index 54fe257..3f62c55 100644 +--- a/plugins/omuxsock/omuxsock.c ++++ b/plugins/omuxsock/omuxsock.c +@@ -443,7 +443,7 @@ ENDdbgPrintInstInfo + static rsRetVal sendMsg(instanceData *pData, char *msg, size_t len) + { + DEFiRet; +- unsigned lenSent = 0; ++ ssize_t lenSent = 0; + + if(pData->sock == INVLD_SOCK) { + CHKiRet(doTryResume(pData)); +@@ -460,17 +460,29 @@ static rsRetVal sendMsg(instanceData *pData, char *msg, size_t len) + } else { + lenSent = sendto(pData->sock, msg, len, 0, (const struct sockaddr *)&pData->addr, pData->addrLen); + } +- if(lenSent != len) { ++ if ((size_t)lenSent != len) { + int eno = errno; + char errStr[1024]; + ++ if (lenSent != -1) { ++ eno = 0; ++ } ++ + /* +- * XXX/rgerhards: how is this message correct, in that we are returning OK still? +- * In reality, the remainder of the partially sent message (or never sent message) +- * is simply dropped, and we do not change the state of the socket. ++ * For connected sockets, we'll let the infrastructure resume us according to its own ++ * criteria. For unconnected sockets, we'll simply try again on the next message received. + */ +- DBGPRINTF("omuxsock suspending: send%s(), socket %d, error: %d = %s.\n", pData->bConnected ? "" : "to", +- pData->sock, eno, rs_strerror_r(eno, errStr, sizeof(errStr))); ++ if (pData->bConnected) { ++ LogError(eno, RS_RET_SUSPENDED, "omuxsock suspending: send(), socket %d, len %zu, sent %zd", ++ pData->sock, len, lenSent); ++ if (iRet == RS_RET_OK) { ++ iRet = RS_RET_SUSPENDED; ++ } ++ } else { ++ DBGPRINTF("omuxsock: sendto(), socket %d, len %zu, sent %zd, error: %d = %s.\n", pData->sock, len, ++ lenSent, eno, rs_strerror_r(eno, errStr, sizeof(errStr))); ++ } ++ closeSocket(pData); + } + } + +diff --git a/tests/uxsock_multiple.sh b/tests/uxsock_multiple.sh +index 93ea9e2..2504d85 100755 +--- a/tests/uxsock_multiple.sh ++++ b/tests/uxsock_multiple.sh +@@ -71,9 +71,16 @@ for name in "${SOCKET_NAMES[@]}"; do + done + + # now do the usual run ++RS_REDIR="> ${RSYSLOG_DYNNAME}.log 2>&1" + startup + # 10000 messages should be enough + injectmsg 0 10000 ++wait_queueempty ++echo resetting uxsockrcvr... ++for pid in ${BGPROCESS[@]}; do ++ kill -HUP $pid ++done ++injectmsg 10000 10000 + shutdown_when_empty # shut down rsyslogd when done processing messages + wait_shutdown + +@@ -86,10 +93,21 @@ for pid in ${BGPROCESS[@]}; do + done + echo background processes have terminated, continue test... + +-# and continue the usual checks ++# and continue the usual checks. + BASE=${RSYSLOG_OUT_LOG} + for name in "${SOCKET_NAMES[@]}"; do +- RSYSLOG_OUT_LOG=${BASE}.${name##*.} +- seq_check 0 9999 ++ SEQ_CHECK_FILE=${BASE}.${name##*.} ++ case ${name##*.} in ++ STREAM|SEQPACKET) ++ echo 10000 >> ${SEQ_CHECK_FILE} ++ ;; ++ esac ++ seq_check 0 19999 + done ++ ++# Verify that we get messages for when our receiver reset ++# We don't redirect with valgrind ++if [ "${USE_VALGRIND}" != "YES" ]; then ++ content_check "omuxsock suspending: send(), socket " ${RSYSLOG_DYNNAME}.log ++fi + exit_test +diff --git a/tests/uxsock_multiple_netns.sh b/tests/uxsock_multiple_netns.sh +index d233d30..6aa8700 100755 +--- a/tests/uxsock_multiple_netns.sh ++++ b/tests/uxsock_multiple_netns.sh +@@ -97,9 +97,16 @@ for name in "${SOCKET_NAMES[@]}"; do + done + + # now do the usual run ++RS_REDIR="> ${RSYSLOG_DYNNAME}.log 2>&1" + startup + # 10000 messages should be enough + injectmsg 0 10000 ++wait_queueempty ++echo resetting uxsockrcvr ++for pid in ${BGPROCESS[@]}; do ++ kill -HUP $pid ++done ++injectmsg 10000 10000 + shutdown_when_empty # shut down rsyslogd when done processing messages + wait_shutdown + +@@ -119,10 +126,24 @@ for name in "${SOCKET_NAMES[@]}"; do + ip netns delete "${NS}" > /dev/null 2>&1 + done + +-# and continue the usual checks ++# and continue the usual checks. ++# For connected sockets, we expect a single message loss ++# which causes the SUSPEND state. Add it back here to ++# simplify the seq_check. + BASE=${RSYSLOG_OUT_LOG} + for name in "${SOCKET_NAMES[@]}"; do +- RSYSLOG_OUT_LOG=${BASE}.${name##*.} +- seq_check 0 9999 ++ SEQ_CHECK_FILE=${BASE}.${name##*.} ++ case ${name##*.} in ++ STREAM|SEQPACKET) ++ echo 10000 >> ${SEQ_CHECK_FILE} ++ ;; ++ esac ++ seq_check 0 19999 + done ++ ++# Verify that we get messages for when our receiver reset ++# We don't redirect with valgrind ++if [ "${USE_VALGRIND}" != "YES" ]; then ++ content_check "omuxsock suspending: send(), socket " ${RSYSLOG_DYNNAME}.log ++fi + exit_test +diff --git a/tests/uxsock_simple.sh b/tests/uxsock_simple.sh +index 2cbd7ca..d33de7f 100755 +--- a/tests/uxsock_simple.sh ++++ b/tests/uxsock_simple.sh +@@ -31,6 +31,10 @@ echo background uxsockrcvr process id is $BGPROCESS + startup + # 10000 messages should be enough + injectmsg 0 10000 ++wait_queueempty ++echo resetting uxsockrcvr... ++kill -HUP $BGPROCESS ++injectmsg 10000 10000 + shutdown_when_empty # shut down rsyslogd when done processing messages + wait_shutdown + +@@ -42,5 +46,5 @@ wait $BGPROCESS + echo background process has terminated, continue test... + + # and continue the usual checks +-seq_check 0 9999 ++seq_check 0 19999 + exit_test +diff --git a/tests/uxsock_simple_abstract.sh b/tests/uxsock_simple_abstract.sh +index 1a0d09a..2185f15 100755 +--- a/tests/uxsock_simple_abstract.sh ++++ b/tests/uxsock_simple_abstract.sh +@@ -39,9 +39,14 @@ BGPROCESS=$! + echo background uxsockrcvr process id is $BGPROCESS + + # now do the usual run ++RS_REDIR="> ${RSYSLOG_DYNNAME}.log 2>&1" + startup + # 10000 messages should be enough + injectmsg 0 10000 ++wait_queueempty ++echo resetting uxsockrcvr... ++kill -HUP $BGPROCESS ++injectmsg 10000 10000 + shutdown_when_empty # shut down rsyslogd when done processing messages + wait_shutdown + +@@ -53,5 +58,12 @@ wait $BGPROCESS + echo background process has terminated, continue test... + + # and continue the usual checks +-seq_check 0 9999 ++seq_check 0 19999 ++ ++# Verify that we do NOT get messages for when our receiver reset. ++# We only expect these messages if we have a connected socket. ++# We don't redirect with valgrind ++if [ "${USE_VALGRIND}" != "YES" ]; then ++ check_not_present "omuxsock suspending: send(), socket " ${RSYSLOG_DYNNAME}.log ++fi + exit_test +diff --git a/tests/uxsockrcvr.c b/tests/uxsockrcvr.c +index 05bbfc6..0ac4f7a 100644 +--- a/tests/uxsockrcvr.c ++++ b/tests/uxsockrcvr.c +@@ -62,7 +62,7 @@ int abstract; + int sockBacklog = MAX_FDS - 1; + struct pollfd fds[MAX_FDS]; + int nfds; +- ++volatile int need_reset; + + /* called to clean up on exit + */ +@@ -71,10 +71,17 @@ cleanup(void) + { + int index; + ++ /* ++ * This function is called from a signal handler. Ensure any ++ * called functions are async-signal-safe as per ++ * signal-safety(7). close and unlink are safe. ++ */ + if (!abstract) unlink(sockName); + for (index = 0; index < nfds; ++index) { + close(fds[index].fd); ++ fds[index].fd = -1; + } ++ nfds = 0; + } + + +@@ -84,6 +91,10 @@ doTerm(int __attribute__((unused)) signum) + exit(1); + } + ++void doReset(int __attribute__((unused)) signum) { ++ need_reset = 1; ++} ++ + + void + usage(void) +@@ -132,6 +143,55 @@ static void _decode_sockType(const char *s) { + exit(1); + } + ++void initialize_socket(void) { ++ struct sockaddr_un addr; /* address of server */ ++ ++ /* Create a UNIX datagram socket for server */ ++ if ((fds[0].fd = socket(AF_UNIX, sockType, 0)) < 0) { ++ perror("server: socket"); ++ exit(1); ++ } ++ ++ /* Set up address structure for server socket */ ++ memset(&addr, 0, sizeof(addr)); ++ addr.sun_family = AF_UNIX; ++ const size_t sockname_len = strlen(sockName); ++ ++ /* Require non-abstract addresses to be terminated. */ ++ if (sockname_len >= (sizeof(addr.sun_path) - 1)) { ++ fprintf(stderr, "error: socket path would be truncated\n"); ++ exit(1); ++ } ++ memcpy(addr.sun_path + abstract, sockName, sockname_len); ++ /* ++ * Abstract socket addresses do not require termination. ++ */ ++ if (!abstract && addr.sun_path[sizeof(addr.sun_path) - 1]) { ++ addr.sun_path[sizeof(addr.sun_path) - 1] = 0; ++ } ++ ++ /* ++ * For pathname addresses, the +1 is the terminating \0 character. ++ * For abstract addresses, the +1 is the leading \0 character. ++ * Technically, Linux doesn't care about that terminating \0 for ++ * pathname addresses. ++ */ ++ if (bind(fds[0].fd, (struct sockaddr *)&addr, offsetof(struct sockaddr_un, sun_path) + sockname_len + 1) < 0) { ++ perror("server: bind"); ++ close(fds[0].fd); ++ exit(1); ++ } ++ ++ if (sockConnected && listen(fds[0].fd, sockBacklog) == -1) { ++ perror("server: listen"); ++ close(fds[0].fd); ++ exit(1); ++ } ++ ++ fds[0].events = POLLIN; ++ nfds = 1; ++} ++ + int + main(int argc, char *argv[]) + { +@@ -140,7 +200,6 @@ main(int argc, char *argv[]) + int timeout = DFLT_TIMEOUT; + FILE *fp = stdout; + unsigned char data[128*1024]; +- struct sockaddr_un addr; /* address of server */ + struct sockaddr from; + socklen_t fromlen; + int index; +@@ -193,52 +252,33 @@ main(int argc, char *argv[]) + exit(1); + } + +- /* Create a UNIX datagram socket for server */ +- if ((fds[0].fd = socket(AF_UNIX, sockType, 0)) < 0) { +- perror("server: socket"); ++ if (signal(SIGHUP, doReset) == SIG_ERR) { ++ perror("signal(SIGHUP, ...)"); + exit(1); + } + +- atexit(cleanup); +- +- /* Set up address structure for server socket */ +- memset(&addr, 0, sizeof(addr)); +- addr.sun_family = AF_UNIX; +- strncpy(addr.sun_path + abstract, sockName, sizeof(addr.sun_path) - abstract); +- /* +- * Abstract socket addresses do not require termination. +- */ +- if (!abstract && addr.sun_path[sizeof(addr.sun_path) - 1]) { +- addr.sun_path[sizeof(addr.sun_path) - 1] = 0; +- fprintf(stderr, "warning: socket path truncated: %s\n", addr.sun_path); +- } +- +- /* +- * For pathname addresses, the +1 is the terminating \0 character. +- * For abstract addresses, the +1 is the leading \0 character. +- */ +- if (bind(fds[0].fd, (struct sockaddr *)&addr, offsetof(struct sockaddr_un, sun_path) + strlen(sockName) + 1) < 0) { +- perror("server: bind"); +- close(fds[0].fd); +- exit(1); ++ for (index = 0; index < MAX_FDS; ++index) { ++ fds[index].fd = -1; + } +- +- if (sockConnected && listen(fds[0].fd, sockBacklog) == -1) { +- perror("server: listen"); +- close(fds[0].fd); +- exit(1); +- } +- +- fds[0].events = POLLIN; +- nfds = 1; ++ atexit(cleanup); + + /* we now run in an endless loop. We do not check who sends us + * data. This should be no problem for our testbench use. + */ + + while(1) { ++ if (need_reset) { ++ need_reset = 0; ++ cleanup(); ++ } ++ if (!nfds) { ++ initialize_socket(); ++ } + rlen = poll(fds, nfds, timeout); + if(rlen == -1) { ++ if (errno == EINTR) { ++ continue; ++ } + perror("uxsockrcvr : poll\n"); + exit(1); + } else if(rlen == 0) { +-- +2.43.0 + diff --git a/SPECS/rsyslog/rsyslog.spec b/SPECS/rsyslog/rsyslog.spec index 19114214a31..2b68cef0c2f 100644 --- a/SPECS/rsyslog/rsyslog.spec +++ b/SPECS/rsyslog/rsyslog.spec @@ -3,7 +3,7 @@ Summary: Rocket-fast system for log processing Name: rsyslog Version: 8.2308.0 -Release: 5%{?dist} +Release: 6%{?dist} License: GPLv3+ AND ASL 2.0 Vendor: Microsoft Corporation Distribution: Azure Linux @@ -17,6 +17,9 @@ Source3: rsyslog.conf Source4: https://www.rsyslog.com/files/download/rsyslog/%{name}-doc-%{base_version}.0.tar.gz Source5: rsyslog.logrotate Patch0: issue5158.patch +Patch1: add-network-namespace-APIs.patch +Patch2: adding-new-functionality-for-omuxsock.patch +Patch3: fix-message-loss-bug-for-connected-Unix-domain-sockets.patch BuildRequires: autogen BuildRequires: curl-devel BuildRequires: gnutls-devel @@ -82,6 +85,9 @@ BuildRequires: net-snmp-devel # Unpack the code source tarball %setup -q %patch 0 -p1 +%patch 1 -p1 +%patch 2 -p1 +%patch 3 -p1 # Unpack the documentation tarball in the folder created above %setup -q -a 4 -T -D # Remove documentation sources @@ -204,6 +210,11 @@ fi %{_libdir}/rsyslog/omsnmp.so %changelog +* Thu Jul 30 2026 BinduSri Adabala - 8.2308.0-6 +- Add new functionality for omuxsock. +- Add NetworkNamespace APIs. +- Fixes message loss regression affecting STREAM and SEQPACKET Unix domain sockets. + * Tue Jan 06 2026 Pawel Winogrodzki - 8.2308.0-5 - Bumping release to rebuild with new 'net-snmp' libs.