Replace the broker wrapper script with a C wrapper that calls execve() on qpidd. Implement parsing of the MATAHARI_PORT environment variable and the Broker Federation options in the C wrapper.
Broker federation is configured using the BROKER_FEDERATION environment variable set in /etc/sysconfig/matahari-broker. The contents of the variable should be a list of hosts or DNS SRV queries separated by any combinations of spaces, commas and/or semicolons.
Signed-off-by: Zane Bitter zbitter@redhat.com --- CMakeLists.txt | 12 +++-- matahari-broker.init.in | 9 +++- matahari-broker.service | 19 -------- matahari-broker.service.in | 22 +++++++++ matahari-broker.sysconf.in | 23 +++++----- matahari-brokerd.in | 8 --- src/CMakeLists.txt | 1 src/broker/CMakeLists.txt | 16 +++++++ src/broker/broker.c | 74 ++++++++++++++++++++++++++++++++ src/broker/broker.h | 12 +++++ src/broker/broker_federation.c | 93 ++++++++++++++++++++++++++++++++++++++++ src/broker/broker_federation.h | 6 +++ src/broker/broker_linux.c | 53 +++++++++++++++++++++++ src/broker/broker_os.h | 15 ++++++ src/include/config.h.in | 2 + 15 files changed, 319 insertions(+), 46 deletions(-) delete mode 100644 matahari-broker.service create mode 100644 matahari-broker.service.in delete mode 100644 matahari-brokerd.in create mode 100644 src/broker/CMakeLists.txt create mode 100644 src/broker/broker.c create mode 100644 src/broker/broker.h create mode 100644 src/broker/broker_federation.c create mode 100644 src/broker/broker_federation.h create mode 100644 src/broker/broker_linux.c create mode 100644 src/broker/broker_os.h
diff --git a/CMakeLists.txt b/CMakeLists.txt index f7555b4..4ec2d1e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -109,9 +109,6 @@ else(WIN32)
set (CPACK_GENERATOR "TGZ")
- configure_file(${CMAKE_CURRENT_SOURCE_DIR}/matahari-brokerd.in - ${CMAKE_CURRENT_BINARY_DIR}/matahari-brokerd) - configure_file(${CMAKE_CURRENT_SOURCE_DIR}/matahari.init.in ${CMAKE_CURRENT_BINARY_DIR}/matahari.init)
@@ -126,8 +123,6 @@ else(WIN32) GROUP_READ GROUP_WRITE GROUP_EXECUTE) install(CODE "execute_process(COMMAND chown root:qpidd ${localstatedir}/lib/matahari)")
- install(PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/matahari-brokerd DESTINATION sbin) - install(PROGRAMS ${CMAKE_CURRENT_BINARY_DIR}/matahari-broker DESTINATION ${initdir})
install(FILES ${CMAKE_CURRENT_BINARY_DIR}/matahari.sysconf DESTINATION ${sysconfdir}/sysconfig RENAME matahari) @@ -142,8 +137,13 @@ else(WIN32)
if(systemd_FOUND) set(systemdunitdir "/lib/systemd/system") - install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/matahari-broker.service DESTINATION ${systemdunitdir}) + + configure_file(${CMAKE_CURRENT_SOURCE_DIR}/matahari-broker.service.in + ${CMAKE_CURRENT_BINARY_DIR}/matahari-broker.service) + + install(FILES ${CMAKE_CURRENT_BINARY_DIR}/matahari-broker.service DESTINATION ${systemdunitdir}) endif(systemd_FOUND) + endif(WIN32)
INCLUDE(InstallRequiredSystemLibraries) diff --git a/matahari-broker.init.in b/matahari-broker.init.in index e74926c..d9ca4ee 100755 --- a/matahari-broker.init.in +++ b/matahari-broker.init.in @@ -46,12 +46,19 @@ pidfile=@localstatedir@/run/matahari-broker.pid # Source function library. . @sysconfdir@/rc.d/init.d/functions
+if [ -f @sysconfdir@/sysconfig/matahari ]; then + . @sysconfdir@/sysconfig/matahari +fi +if [ -f @sysconfdir@/sysconfig/matahari-broker ]; then + . @sysconfdir@/sysconfig/matahari-broker +fi + RETVAL=0
start() { echo -n $"Starting Matahari broker daemon: " install -o root -g qpidd -m 775 -d @localstatedir@/run/matahari - daemon --pidfile $pidfile --check $proc --user qpidd /usr/sbin/$prog --daemon + daemon --pidfile $pidfile --check $proc --user qpidd /usr/sbin/$prog $QPIDD_OPTIONS --daemon RETVAL=$? echo [ $RETVAL = 0 ] && touch $lockfile diff --git a/matahari-broker.service b/matahari-broker.service deleted file mode 100644 index 7e6e9f7..0000000 --- a/matahari-broker.service +++ /dev/null @@ -1,19 +0,0 @@ -[Unit] -Description=Matahari Broker is an AMQP broker. It receives, stores, routes and forwards messages using the AMQP protcol. -After=syslog.target network.target -Requires=syslog.target network.target - -[Install] -WantedBy=multi-user.target - -[Service] -Type=simple -NotifyAccess=all -SysVStartPriority=85 -User=qpidd -Group=qpidd - -ExecStart=/usr/sbin/matahari-brokerd - -# Restart options include: no, on-success, on-failure, on-abort or always -Restart=on-failure diff --git a/matahari-broker.service.in b/matahari-broker.service.in new file mode 100644 index 0000000..62898fb --- /dev/null +++ b/matahari-broker.service.in @@ -0,0 +1,22 @@ +[Unit] +Description=Matahari Broker is an AMQP broker. It receives, stores, routes and forwards messages using the AMQP protcol. +After=syslog.target network.target +Requires=syslog.target network.target + +[Install] +WantedBy=multi-user.target + +[Service] +Type=simple +NotifyAccess=all +SysVStartPriority=85 +User=qpidd +Group=qpidd + +EnvironmentFile=-@sysconfdir@/sysconfig/matahari +EnvironmentFile=-@sysconfdir@/sysconfig/matahari-broker + +ExecStart=/usr/sbin/matahari-brokerd $QPIDD_OPTIONS + +# Restart options include: no, on-success, on-failure, on-abort or always +Restart=on-failure diff --git a/matahari-broker.sysconf.in b/matahari-broker.sysconf.in index 9b1c2b5..367e00c 100644 --- a/matahari-broker.sysconf.in +++ b/matahari-broker.sysconf.in @@ -1,13 +1,12 @@ -# Try to use the centrally defined port (also used by the agents) -# Otherwise, use the default port specified at build-time - -if [ -f @sysconfdir@/sysconfig/matahari ] ; then - . @sysconfdir@/sysconfig/matahari -fi - -if [ -z $MATAHARI_PORT ]; then - MATAHARI_PORT=@MATAHARI_PORT@ -fi - -QPIDD_OPTIONS="--config @sysconfdir@/matahari-broker.conf --port $MATAHARI_PORT" +QPIDD_OPTIONS="--config @sysconfdir@/matahari-broker.conf" + +# Federated Broker configuration - list the brokers to federate with +# The following formats are allowed: +# Hostname and port +# <fqdn>:<port> +# DNS SRV record format ("_matahari._tcp" is REQUIRED) +# _matahari._tcp.<fqdn> +# +# Addresses need not all be in the same format +#FEDERATED_BROKERS="broker1.example.com:49000, _matahari._tcp.example.com"
diff --git a/matahari-brokerd.in b/matahari-brokerd.in deleted file mode 100644 index 0bab652..0000000 --- a/matahari-brokerd.in +++ /dev/null @@ -1,8 +0,0 @@ -#!/bin/sh - -if [ -f @sysconfdir@/sysconfig/matahari-broker ]; then - . @sysconfdir@/sysconfig/matahari-broker -fi - -/usr/sbin/qpidd --data-dir @localstatedir@/lib/matahari $QPIDD_OPTIONS $* - diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index cb47956..58ba8a2 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -161,6 +161,7 @@ endif(WITH-QMF)
### Subdirectories +add_subdirectory(broker) add_subdirectory(lib) add_subdirectory(host) add_subdirectory(network) diff --git a/src/broker/CMakeLists.txt b/src/broker/CMakeLists.txt new file mode 100644 index 0000000..7b48387 --- /dev/null +++ b/src/broker/CMakeLists.txt @@ -0,0 +1,16 @@ +SET(BROKER_DAEMON "matahari-brokerd") + +if(WITH-QMF) +if(NOT WIN32) + add_executable(${BROKER_DAEMON} broker.c broker_federation.c broker_${VARIANT}.c) + + if(HAVE_RESOLV_H) + target_link_libraries(${BROKER_DAEMON} mcommon) + target_link_libraries(${BROKER_DAEMON} resolv) + endif(HAVE_RESOLV_H) + + install(TARGETS ${BROKER_DAEMON} DESTINATION sbin) + +endif(NOT WIN32) +endif(WITH-QMF) + diff --git a/src/broker/broker.c b/src/broker/broker.c new file mode 100644 index 0000000..ffb2587 --- /dev/null +++ b/src/broker/broker.c @@ -0,0 +1,74 @@ +#include "config.h" + +#include <stdlib.h> +#include <stdint.h> +#include <string.h> +#include <stdio.h> + +#include "broker_os.h" +#include "broker.h" + + +uint16_t broker_get_port(void) +{ + static uint16_t port = 0; + + if (!port) { + char *end; + char *port_env = getenv("MATAHARI_PORT"); + if (port_env) { + unsigned long port_l = strtoul(port_env, &end, 10); + if (port_env != end && port_l < (1 << 16)) { + port = port_l; + } + } + } + + return port ? port : MATAHARI_PORT; +} + +static char **broker_args(int argc, char * const argv[]) +{ + int i = 0, o = 0; + char portarg[13]; + char **newargs = malloc(sizeof(char *) * (argc + 2 + 1)); + + if (argc) { + newargs[o++] = strdup(argv[i++]); + } + + snprintf(portarg, sizeof(portarg), "--port=%hu", broker_get_port()); + newargs[o++] = strdup(portarg); + newargs[o++] = strdup("--data-dir=" LOCAL_STATE_DIR "/lib/matahari"); + + while (i < argc) { + newargs[o++] = strdup(argv[i++]); + } + + newargs[o] = NULL; + + return newargs; +} + +static void broker_args_free(char **args) +{ + char **p = args; + + while (*args) { + free(*args++); + } + + free(p); +} + +int main(int argc, char *argv[]) +{ + char **arglist = broker_args(argc, argv); + + int ret = broker_os_start_broker(arglist); + + broker_args_free(arglist); + + return ret; +} + diff --git a/src/broker/broker.h b/src/broker/broker.h new file mode 100644 index 0000000..4c7340d --- /dev/null +++ b/src/broker/broker.h @@ -0,0 +1,12 @@ +#ifndef MH_BROKER_H +#define MH_BROKER_H + +#include <stdint.h> + + +#define MATAHARI_PORT 49000 + + +uint16_t broker_get_port(void); + +#endif diff --git a/src/broker/broker_federation.c b/src/broker/broker_federation.c new file mode 100644 index 0000000..ae8dccf --- /dev/null +++ b/src/broker/broker_federation.c @@ -0,0 +1,93 @@ +#include "config.h" + +#include <stdlib.h> +#include <string.h> +#include <stdio.h> + +#include <matahari/dnssrv.h> +#include <matahari/logging.h> + +#include "broker.h" +#include "broker_os.h" +#include "broker_federation.h" + + +#define DNS_SRV_PREFIX "_matahari._tcp." + +#define ROUTE_LIST(LOCAL, REMOTE) { \ + {REMOTE, LOCAL, "amq.direct"}, \ + {LOCAL, REMOTE, "amq.direct"}, \ + {REMOTE, LOCAL, "qmf.default.direct"}, \ + {LOCAL, REMOTE, "qmf.default.direct"}, \ + {REMOTE, LOCAL, "qmf.default.topic"}, \ + {LOCAL, REMOTE, "qmf.default.topic"} \ +} + + +static int broker_federate(const char *local, const char *remote) +{ + int i; + struct mh_qpid_route routes[] = ROUTE_LIST(local, remote); + + mh_trace("Adding routes for federated broker "%s"", remote); + + for (i = 0; i < (sizeof(routes) / sizeof(*routes)); i++) { + int ret = broker_os_add_qpid_route(&routes[i]); + if (ret) { + mh_err("Failed to add federated broker "%s"", remote); + return ret; + } + } + + return 0; +} + +static char *broker_lookup(const char *query) +{ +#ifdef HAVE_RESOLV_H + static char broker[1025 + 6]; + char *host; + mh_trace("Looking up DNS SRV record for "%s"", query); + + host = mh_dnssrv_lookup(query); + + if (host) { + snprintf(broker, sizeof(broker), "%s:%hu", host, MATAHARI_PORT); + return broker; + } + mh_warn("Failed to find DNS SRV record for "%s"", query); +#else + mh_warn("DNS SRV lookups are not supported"); +#endif + + return NULL; +} + +void broker_federation_configure(void) +{ + const char *brokers = getenv("FEDERATED_BROKERS"); + char local[16]; + char peers[1024]; + char *p = NULL; + char *peer; + + if (!brokers) { + return; + } + + snprintf(local, sizeof(local), "localhost:%hu", broker_get_port()); + + strncpy(peers, brokers, sizeof(peers) - 1); + peers[sizeof(peers) - 1] = '\0'; + + while ((peer = strtok_r(p ? NULL : peers, ",; ", &p)) != NULL) { + if (strncmp(peer, DNS_SRV_PREFIX, strlen(DNS_SRV_PREFIX)) == 0) { + peer = broker_lookup(peer); + } + + if (peer) { + broker_federate(local, peer); + } + } +} + diff --git a/src/broker/broker_federation.h b/src/broker/broker_federation.h new file mode 100644 index 0000000..12cfe61 --- /dev/null +++ b/src/broker/broker_federation.h @@ -0,0 +1,6 @@ +#ifndef MH_BROKER_FEDERATION_H +#define MH_BROKER_FEDERATION_H + +void broker_federation_configure(void); + +#endif diff --git a/src/broker/broker_linux.c b/src/broker/broker_linux.c new file mode 100644 index 0000000..54e3bff --- /dev/null +++ b/src/broker/broker_linux.c @@ -0,0 +1,53 @@ +#include <stdlib.h> +#include <string.h> +#include <stdio.h> +#include <errno.h> +#include <unistd.h> + +#include "broker_federation.h" +#include "broker_os.h" + + +int broker_os_start_broker(char * const args[]) +{ + pid_t pid = fork(); + + if (pid < 0) { + perror("fork()"); + return errno; + } + + if (pid) { + // exec() in parent process so that qpidd keeps the initial pid. + // Note that this means the child will become a zombie. That's just the + // way it goes. + + // execvp() will find the path to qpidd from $PATH. + execvp("qpidd", args); + + // It's an error if we get here + perror("exec()"); + return errno; + } else { + // Configure federated brokers in the child process + broker_federation_configure(); + return 0; + } +} + +int broker_os_add_qpid_route(const struct mh_qpid_route *route) +{ + char cmd[1024]; + int ret; + + snprintf(cmd, sizeof(cmd), "qpid-route --timeout=5 dynamic add %s %s %s", + route->dest, route->src, route->exchange); + + ret = system(cmd); + + if (ret < 0) { + perror("system()"); + } + return ret; +} + diff --git a/src/broker/broker_os.h b/src/broker/broker_os.h new file mode 100644 index 0000000..a961625 --- /dev/null +++ b/src/broker/broker_os.h @@ -0,0 +1,15 @@ +#ifndef MH_OS_BROKER_H +#define MH_OS_BROKER_H + +struct mh_qpid_route { + const char *dest; + const char *src; + const char *exchange; +}; + + +int broker_os_start_broker(char * const args[]); + +int broker_os_add_qpid_route(const struct mh_qpid_route *route); + +#endif diff --git a/src/include/config.h.in b/src/include/config.h.in index 6231970..64a64db 100644 --- a/src/include/config.h.in +++ b/src/include/config.h.in @@ -5,3 +5,5 @@ #cmakedefine HAVE_TIME 1 #cmakedefine HAVE_G_LIST_FREE_FULL 1 #cmakedefine HAVE_PK_GET_SYNC 1 + +#define LOCAL_STATE_DIR "@localstatedir@"