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 | 88 +++++++++++++++++++++++++++++++ src/broker/broker.h | 35 +++++++++++++ src/broker/broker_federation.c | 112 ++++++++++++++++++++++++++++++++++++++++ src/broker/broker_federation.h | 33 ++++++++++++ src/broker/broker_linux.c | 71 +++++++++++++++++++++++++ src/broker/broker_os.h | 51 ++++++++++++++++++ src/include/config.h.in | 9 +++ 15 files changed, 463 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 f045887..e3a7e38 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -162,6 +162,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..f6d7d73 --- /dev/null +++ b/src/broker/broker.c @@ -0,0 +1,88 @@ +/* Copyright (C) 2011 Red Hat, Inc. + * Written by Zane Bitter zbitter@redhat.com + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +#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 *port_env = getenv("MATAHARI_PORT"); + if (port_env) { + sscanf(port_env, "%hu", &port); + } + } + + 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..3aff4d3 --- /dev/null +++ b/src/broker/broker.h @@ -0,0 +1,35 @@ +/* Copyright (C) 2011 Red Hat, Inc. + * Written by Zane Bitter zbitter@redhat.com + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +/** + * @file broker.h + * Interface to the broker configuration code. + */ + +#ifndef MH_BROKER_H +#define MH_BROKER_H + +#include <stdint.h> + +/** + * Get the port for the broker to listen on. + * @return the TCP port for the local broker + */ +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..5140657 --- /dev/null +++ b/src/broker/broker_federation.c @@ -0,0 +1,112 @@ +/* Copyright (C) 2011 Red Hat, Inc. + * Written by Zane Bitter zbitter@redhat.com + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +#include "config.h" + +#include <stdlib.h> +#include <string.h> +#include <stdio.h> + +#include <matahari/dnssrv.h> +#include <matahari/utilities.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 < DIMOF(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..0cacaef --- /dev/null +++ b/src/broker/broker_federation.h @@ -0,0 +1,33 @@ +/* Copyright (C) 2011 Red Hat, Inc. + * Written by Zane Bitter zbitter@redhat.com + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +/** + * @file broker_federation.h + * Interface to the broker federation configuration code. + */ + +#ifndef MH_BROKER_FEDERATION_H +#define MH_BROKER_FEDERATION_H + +/** + * Read the broker federation settings from the environment and set up broker + * federation. + */ +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..6dd98e8 --- /dev/null +++ b/src/broker/broker_linux.c @@ -0,0 +1,71 @@ +/* Copyright (C) 2011 Red Hat, Inc. + * Written by Zane Bitter zbitter@redhat.com + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +#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..788fe78 --- /dev/null +++ b/src/broker/broker_os.h @@ -0,0 +1,51 @@ +/* Copyright (C) 2011 Red Hat, Inc. + * Written by Zane Bitter zbitter@redhat.com + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + */ + +/** + * @file broker_os.h + * Interface to the OS-specific broker configuration code. + */ + +#ifndef MH_BROKER_OS_H +#define MH_BROKER_OS_H + +/** + * A route between two Qpid brokers. + */ +struct mh_qpid_route { + const char *dest; /**< The route destination */ + const char *src; /**< The route source */ + const char *exchange; /**< The exchange */ +}; + + +/** + * Start the broker with the supplied argument list. + * @param args a list of args to pass to the broker. The last item in the list + * must be NULL. + */ +int broker_os_start_broker(char * const args[]); + +/** + * Add a route between two Qpid brokers. + * @param route pointer to the route to add + * @return -1 on error, otherwise the result of the external route add command. + */ +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..fd248ad 100644 --- a/src/include/config.h.in +++ b/src/include/config.h.in @@ -1,3 +1,6 @@ +#ifndef MH_CONFIG_H +#define MH_CONFIG_H + #cmakedefine HAVE_STRING_H 1 #cmakedefine HAVE_SYS_IOCTL_H 1 #cmakedefine HAVE_ASPRINTF 1 @@ -5,3 +8,9 @@ #cmakedefine HAVE_TIME 1 #cmakedefine HAVE_G_LIST_FREE_FULL 1 #cmakedefine HAVE_PK_GET_SYNC 1 + +#define LOCAL_STATE_DIR "@localstatedir@" + +#define MATAHARI_PORT @MATAHARI_PORT@ + +#endif