enable ZeroMQ interface
teamd -Z "tcp:\\127.0.0.1:5555"
teamdctl -Z "tcp:\\127.0.0.1:5555" config dump
Take care when you use this type of connection.
Security must be ensured by firewall or other tools.
There could be added verification by certificate with SSL layer.
It could be added in next version.
changes:
fix problem with autotools and wrong version of libzmq
Signed-off-by: Jiří Župka <jzupka(a)redhat.com>
---
configure.ac | 12 ++
include/private/misc.h | 1 +
include/teamdctl.h | 2 +-
libteamdctl/Makefile.am | 4 +-
libteamdctl/cli_zmq.c | 271 +++++++++++++++++++++++++++++++++++++
libteamdctl/libteamdctl.c | 7 +-
libteamdctl/teamdctl_private.h | 2 +
teamd/Makefile.am | 10 +-
teamd/teamd.c | 30 ++++-
teamd/teamd.h | 12 ++
teamd/teamd_state.c | 17 +++
teamd/teamd_zmq.c | 298 +++++++++++++++++++++++++++++++++++++++++
teamd/teamd_zmq.h | 41 ++++++
teamd/teamd_zmq_common.h | 58 ++++++++
utils/teamdctl.c | 67 ++++++++-
15 files changed, 811 insertions(+), 21 deletions(-)
create mode 100644 libteamdctl/cli_zmq.c
create mode 100644 teamd/teamd_zmq.c
create mode 100644 teamd/teamd_zmq.h
create mode 100644 teamd/teamd_zmq_common.h
diff --git a/configure.ac b/configure.ac
index b72a282..64da77f 100644
--- a/configure.ac
+++ b/configure.ac
@@ -76,6 +76,18 @@ AC_ARG_ENABLE([dbus],
fi
fi
+have_zmq=no
+AC_ARG_ENABLE([zmq],
+ AS_HELP_STRING([--disable-zmq], [disable ZeroMQ API @<:@default=enabled@:>@]))
+ if test "x$enable_zmq" != "xno"; then
+ PKG_CHECK_MODULES([ZMQ], [libzmq >= 3.2.0],
+ [AC_DEFINE(ENABLE_ZMQ, [1], [ZMQ API.]) have_zmq=yes],
+ have_zmq=no)
+ if test "x$have_zmq$enable_zmq" = xnoyes; then
+ AC_MSG_ERROR([*** ZeroMQ support requested but libraries not found])
+ fi
+fi
+
AC_CONFIG_FILES([Makefile
include/Makefile \
libteam/Makefile \
diff --git a/include/private/misc.h b/include/private/misc.h
index 0b56eca..ccd035d 100644
--- a/include/private/misc.h
+++ b/include/private/misc.h
@@ -24,6 +24,7 @@
#include <stdlib.h>
#include <errno.h>
#include <string.h>
+#include <stdarg.h>
#include <unistd.h>
#include <sys/socket.h>
#include <inttypes.h>
diff --git a/include/teamdctl.h b/include/teamdctl.h
index 9026411..b773bc4 100644
--- a/include/teamdctl.h
+++ b/include/teamdctl.h
@@ -43,7 +43,7 @@ void teamdctl_set_log_fn(struct teamdctl *tdc,
int teamdctl_get_log_priority(struct teamdctl *tdc);
void teamdctl_set_log_priority(struct teamdctl *tdc, int priority);
int teamdctl_connect(struct teamdctl *tdc, const char *team_name,
- const char *cli_type);
+ const char *addr, const char *cli_type);
void teamdctl_disconnect(struct teamdctl *tdc);
int teamdctl_refresh(struct teamdctl *tdc);
int teamdctl_port_add(struct teamdctl *tdc, const char *port_devname);
diff --git a/libteamdctl/Makefile.am b/libteamdctl/Makefile.am
index d3854fc..edec72b 100644
--- a/libteamdctl/Makefile.am
+++ b/libteamdctl/Makefile.am
@@ -6,9 +6,9 @@ AM_CFLAGS = -fvisibility=hidden -ffunction-sections -fdata-sections
AM_LDFLAGS = -Wl,--gc-sections -Wl,--as-needed
lib_LTLIBRARIES = libteamdctl.la
-libteamdctl_la_SOURCES = libteamdctl.c cli_usock.c cli_dbus.c
+libteamdctl_la_SOURCES = libteamdctl.c cli_usock.c cli_dbus.c cli_zmq.c
libteamdctl_la_CFLAGS= $(AM_CFLAGS) $(JANSSON_CFLAGS) $(DBUS_CFLAGS) -I${top_srcdir}/include -D_GNU_SOURCE
-libteamdctl_la_LIBADD= $(JANSSON_LIBS) $(DBUS_LIBS)
+libteamdctl_la_LIBADD= $(JANSSON_LIBS) $(DBUS_LIBS) $(ZMQ_LIBS) $(LIBDAEMON_LIBS)
libteamdctl_la_LDFLAGS = $(AM_LDFLAGS) -version-info @LIBTEAMDCTL_CURRENT@:@LIBTEAMDCTL_REVISION@:@LIBTEAMDCTL_AGE@
pkgconfigdir = $(libdir)/pkgconfig
diff --git a/libteamdctl/cli_zmq.c b/libteamdctl/cli_zmq.c
new file mode 100644
index 0000000..5037019
--- /dev/null
+++ b/libteamdctl/cli_zmq.c
@@ -0,0 +1,271 @@
+/* cli_zmq.c - Teamd daemon control library teamd ZeroMQ library client
+
+ * Copyright (C) 2013 Jiri Zupka <jzupka(a)redhat.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser 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 library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include "config.h"
+
+#ifdef ENABLE_ZMQ
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdarg.h>
+#include <sys/types.h>
+#include <zmq.h>
+#include <errno.h>
+#include <unistd.h>
+#include <teamdctl.h>
+#include "teamdctl_private.h"
+#include "../teamd/teamd_zmq_common.h"
+
+struct cli_zmq_priv {
+ void *context;
+ void *sock;
+};
+
+static int cli_zmq_process_msg(struct teamdctl *tdc, char *msg,
+ char **p_replystr)
+{
+ char *str;
+ char *rest = msg;
+
+ str = teamd_zmq_msg_getline(&rest);
+ if (!str) {
+ err(tdc, "zmq: Incomplete message.\n");
+ return -EINVAL;;
+ }
+
+ if (!strcmp(TEAMD_ZMQ_REPLY_SUCC_PREFIX, str)) {
+ *p_replystr = rest;
+ } else if (!strcmp(TEAMD_ZMQ_REPLY_ERR_PREFIX, str)) {
+ str = teamd_zmq_msg_getline(&rest);
+ if (!str) {
+ err(tdc, "zmq: Incomplete message.\n");
+ return -EINVAL;;
+ }
+ err(tdc, "zmq: Error message received: \"%s\"", str);
+ str = teamd_zmq_msg_getline(&rest);
+ if (!str) {
+ err(tdc, "zmq: Incomplete message.\n");
+ return -EINVAL;;
+ }
+ err(tdc, "zmq: Error message content: \"%s\"", str);
+ return -EINVAL;;
+ } else {
+ err(tdc, "zmq: Unsupported message type.\n");
+ return -EINVAL;
+ }
+ return 0;
+}
+
+void my_free_msg(void *data, void *hind) {
+ free(data);
+}
+
+static int cli_zmq_send(void *sock, char *buf)
+{
+ int ret;
+ int buflen = strlen(buf);
+
+ ret = zmq_send(sock, buf, buflen, 0);
+
+ if (ret == -1) {
+ teamd_log_warn("zmq: send failed: %s", strerror(errno));
+ return -errno;
+ }
+ free(buf);
+ return 0;
+}
+
+static int cli_zmq_recv(void *sock, char **p_str)
+{
+ int ret;
+ zmq_msg_t msg;
+ char *buf;
+
+ if (zmq_msg_init(&msg) == -1) {
+ teamd_log_dbg("zmq: Unable initiate message for receive.");
+ return -errno;
+ }
+
+ ret = zmq_msg_recv(&msg, sock, 0);
+
+ if (ret == -1) {
+ teamd_log_warn("zmq: send failed: %s", strerror(errno));
+ return -errno;
+ }
+
+ buf = malloc(ret + 1);
+ memcpy(buf, zmq_msg_data(&msg), ret);
+ buf[ret] = '\0';
+
+ if (zmq_msg_close(&msg) == -1) {
+ free(buf);
+ teamd_log_dbg("zmq: Unable close message.");
+ return -errno;
+ }
+
+ *p_str = buf;
+ return 0;
+}
+
+#define WAIT_MSEC TEAMDCTL_REPLY_TIMEOUT
+
+static int myasprintf(char **p_str, const char *fmt, ...)
+{
+ char *newstr;
+ va_list ap;
+ int ret;
+
+ va_start(ap, fmt);
+ ret = vasprintf(&newstr, fmt, ap);
+ va_end(ap);
+ if (ret == -1)
+ return -ENOMEM;
+ free(*p_str);
+ *p_str = newstr;
+ return 0;
+}
+
+static int cli_zmq_method_call(struct teamdctl *tdc, const char *method_name,
+ char **p_reply, void *priv,
+ const char *fmt, va_list ap)
+{
+ struct cli_zmq_priv *cli_zmq = priv;
+ char *str;
+ char *msg = NULL;
+ char *recvmsg = recvmsg;
+ char *replystr;
+ int err;
+
+ dbg(tdc, "zmq: Calling method \"%s\"", method_name);
+ err = myasprintf(&msg, "%s\n%s\n", TEAMD_ZMQ_REQUEST_PREFIX,
+ method_name);
+ if (err)
+ return err;
+ while (*fmt) {
+ switch (*fmt++) {
+ case 's': /* string */
+ str = va_arg(ap, char *);
+ err = myasprintf(&msg, "%s%s\n", msg, str);
+ if (err)
+ goto free_msg;
+ break;
+ default:
+ err(tdc, "zmq: Unknown argument type requested.");
+ err = -EINVAL;
+ goto free_msg;
+ }
+ }
+
+ err = cli_zmq_send(cli_zmq->sock, msg);
+ if (err)
+ goto send_err;
+
+ err = cli_zmq_recv(cli_zmq->sock, &recvmsg);
+ if (err)
+ goto send_err;
+
+ err = cli_zmq_process_msg(tdc, recvmsg, &replystr);
+ if (err)
+ goto free_recvmsg;
+
+ if (p_reply) {
+ replystr = strdup(replystr);
+ if (!replystr) {
+ err = -ENOMEM;
+ goto free_recvmsg;
+ }
+ *p_reply = replystr;
+ }
+
+free_recvmsg:
+ free(recvmsg);
+ goto send_err;
+free_msg:
+ free(msg);
+send_err:
+ return err;
+}
+
+static int cli_zmq_init(struct teamdctl *tdc, const char *team_name,
+ void *priv)
+{
+ int err;
+ struct cli_zmq_priv *cli_zmq = priv;
+ void *context, *sock;
+ int recv_timeo;
+
+ context = zmq_ctx_new();
+ if (!context) {
+ teamd_log_err("zmq: Failed to create context.");
+ return -errno;
+ }
+
+ sock = zmq_socket(context, ZMQ_REQ);
+ if (!sock) {
+ teamd_log_err("zmq: Failed to create socket.");
+ return -errno;
+ }
+
+ err = zmq_connect(sock, tdc->addr);
+ if (err == -1) {
+ err(tdc, "zmq: Failed to connect socket (%s).", tdc->addr);
+ zmq_close(sock);
+ zmq_ctx_destroy(context);
+ return -errno;
+ }
+
+ recv_timeo = WAIT_MSEC;
+ err = zmq_setsockopt(sock, ZMQ_RCVTIMEO, &recv_timeo,
+ sizeof(recv_timeo));
+ if (err == -1) {
+ err(tdc, "zmq: Failed set socket timeout.");
+ zmq_close(sock);
+ zmq_ctx_destroy(context);
+ return -errno;
+ }
+
+ cli_zmq->sock = sock;
+ cli_zmq->context = context;
+
+ return 0;
+}
+
+void cli_zmq_fini(struct teamdctl *tdc, void *priv)
+{
+ struct cli_zmq_priv *cli_zmq = priv;
+
+ zmq_close(cli_zmq->sock);
+ zmq_ctx_destroy(cli_zmq->context);
+}
+
+static const struct teamdctl_cli cli_zmq = {
+ .name = "zmq",
+ .init = cli_zmq_init,
+ .fini = cli_zmq_fini,
+ .method_call = cli_zmq_method_call,
+ .priv_size = sizeof(struct cli_zmq_priv),
+};
+
+const struct teamdctl_cli *teamdctl_cli_zmq_get(void)
+{
+ return &cli_zmq;
+}
+
+#endif /* ENABLE_ZMQ */
diff --git a/libteamdctl/libteamdctl.c b/libteamdctl/libteamdctl.c
index 33cbb3f..02daed0 100644
--- a/libteamdctl/libteamdctl.c
+++ b/libteamdctl/libteamdctl.c
@@ -226,7 +226,7 @@ static void cli_fini(struct teamdctl *tdc)
**/
TEAMDCTL_EXPORT
int teamdctl_connect(struct teamdctl *tdc, const char *team_name,
- const char *cli_type)
+ const char *addr, const char *cli_type)
{
int err;
int i;
@@ -235,6 +235,9 @@ int teamdctl_connect(struct teamdctl *tdc, const char *team_name,
#ifdef ENABLE_DBUS
teamdctl_cli_dbus_get(),
#endif
+#ifdef ENABLE_ZMQ
+ teamdctl_cli_zmq_get(),
+#endif
};
#define TEAMDCTL_CLI_LIST_SIZE ARRAY_SIZE(teamdctl_cli_list)
@@ -258,6 +261,8 @@ int teamdctl_connect(struct teamdctl *tdc, const char *team_name,
if (!cli_type && orig_log_prio < LOG_DEBUG)
teamdctl_set_log_priority(tdc, LOG_EMERG);
+ tdc->addr = (char *) addr;
+
err = cli_init(tdc, team_name);
/* restore original log priority */
diff --git a/libteamdctl/teamdctl_private.h b/libteamdctl/teamdctl_private.h
index 0e36212..026832a 100644
--- a/libteamdctl/teamdctl_private.h
+++ b/libteamdctl/teamdctl_private.h
@@ -40,6 +40,7 @@ struct teamdctl {
const char *file, int line, const char *fn,
const char *format, va_list args);
int log_priority;
+ char *addr;
const struct teamdctl_cli *cli;
void *cli_priv;
struct {
@@ -97,6 +98,7 @@ struct teamdctl_cli {
const struct teamdctl_cli *teamdctl_cli_usock_get(void);
const struct teamdctl_cli *teamdctl_cli_dbus_get(void);
+const struct teamdctl_cli *teamdctl_cli_zmq_get(void);
#define TEAMDCTL_REPLY_TIMEOUT 5000 /* ms */
diff --git a/teamd/Makefile.am b/teamd/Makefile.am
index c5f3401..e2d3bd9 100644
--- a/teamd/Makefile.am
+++ b/teamd/Makefile.am
@@ -8,13 +8,13 @@ AM_CPPFLAGS='-DLOCALSTATEDIR="$(localstatedir)"'
teamd_CFLAGS= $(LIBDAEMON_CFLAGS) $(JANSSON_CFLAGS) $(DBUS_CFLAGS) -I${top_srcdir}/include -D_GNU_SOURCE
-teamd_LDADD = $(top_builddir)/libteam/libteam.la $(LIBDAEMON_LIBS) $(JANSSON_LIBS) $(DBUS_LIBS)
+teamd_LDADD = $(top_builddir)/libteam/libteam.la $(LIBDAEMON_LIBS) $(JANSSON_LIBS) $(DBUS_LIBS) $(ZMQ_LIBS)
bin_PROGRAMS=teamd
teamd_SOURCES=teamd.c teamd_common.c teamd_json.c teamd_config.c teamd_state.c \
teamd_workq.c teamd_events.c teamd_per_port.c \
teamd_option_watch.c teamd_ifinfo_watch.c teamd_link_watch.c \
- teamd_ctl.c teamd_dbus.c teamd_usock.c teamd_sriov.c \
+ teamd_ctl.c teamd_dbus.c teamd_zmq.c teamd_usock.c teamd_sriov.c \
teamd_bpf_chef.c teamd_hash_func.c teamd_balancer.c \
teamd_runner_basic_ones.c teamd_runner_activebackup.c \
teamd_runner_loadbalance.c teamd_runner_lacp.c
@@ -22,6 +22,6 @@ teamd_SOURCES=teamd.c teamd_common.c teamd_json.c teamd_config.c teamd_state.c \
EXTRA_DIST = example_configs dbus redhat
noinst_HEADERS = teamd.h teamd_workq.h teamd_bpf_chef.h teamd_ctl.h \
- teamd_json.h teamd_dbus.h teamd_usock.h teamd_dbus_common.h \
- teamd_usock_common.h teamd_config.h teamd_state.h \
- teamd_sriov.h
+ teamd_json.h teamd_dbus.h teamd_zmq.h teamd_usock.h \
+ teamd_dbus_common.h teamd_usock_common.h teamd_config.h \
+ teamd_state.h teamd_sriov.h
diff --git a/teamd/teamd.c b/teamd/teamd.c
index c4783f6..509e175 100644
--- a/teamd/teamd.c
+++ b/teamd/teamd.c
@@ -49,6 +49,7 @@
#include "teamd_state.h"
#include "teamd_usock.h"
#include "teamd_dbus.h"
+#include "teamd_zmq.h"
#include "teamd_sriov.h"
static const struct teamd_runner *teamd_runner_list[] = {
@@ -97,7 +98,7 @@ static void print_help(const struct teamd_context *ctx) {
" -v --version Show version\n"
" -f --config-file=FILE Load the specified configuration file\n"
" -c --config=TEXT Use given config string (This causes configuration\n"
- " file will be ignored)\n"
+ " file will be ignored)\n"
" -p --pid-file=FILE Use the specified PID file\n"
" -g --debug Increase verbosity\n"
" -r --force-recreate Force team device recreation in case it\n"
@@ -106,6 +107,7 @@ static void print_help(const struct teamd_context *ctx) {
" -t --team-dev=DEVNAME Use the specified team device\n"
" -n --no-ports Start without ports\n"
" -D --dbus-enable Enable D-Bus interface\n"
+ " -Z --zmq-enable=ADDRESS Enable ZeroMQ interface\n"
" -U --usock-enable Enable UNIX domain socket interface\n"
" -u --usock-disable Disable UNIX domain socket interface\n",
ctx->argv0);
@@ -136,12 +138,13 @@ static int parse_command_line(struct teamd_context *ctx,
{ "team-dev", required_argument, NULL, 't' },
{ "no-ports", no_argument, NULL, 'n' },
{ "dbus-enable", no_argument, NULL, 'D' },
+ { "zmq-enable", required_argument, NULL, 'Z' },
{ "usock-enable", no_argument, NULL, 'U' },
{ "usock-disable", no_argument, NULL, 'u' },
{ NULL, 0, NULL, 0 }
};
- while ((opt = getopt_long(argc, argv, "hdkevf:c:p:grot:nDUu",
+ while ((opt = getopt_long(argc, argv, "hdkevf:c:p:grot:nDZ:Uu",
long_options, NULL)) >= 0) {
switch(opt) {
@@ -201,6 +204,15 @@ static int parse_command_line(struct teamd_context *ctx,
ctx->dbus.enabled = true;
#endif
break;
+ case 'Z':
+#ifndef ENABLE_ZMQ
+ fprintf(stderr, "ZeroMQ support is not compiled-in\n");
+ return -1;
+#else
+ ctx->zmq.enabled = true;
+ ctx->zmq.addr = optarg;
+#endif
+ break;
case 'U':
ctx->usock.enabled = true;
break;
@@ -1299,10 +1311,16 @@ skip_create:
goto usock_fini;
}
+ err = teamd_zmq_init(ctx);
+ if (err) {
+ teamd_log_err("Failed to init zmq.");
+ goto dbus_fini;
+ }
+
err = teamd_add_ports(ctx);
if (err) {
teamd_log_err("Failed to add ports.");
- goto dbus_fini;
+ goto zmq_fini;
}
/*
@@ -1312,11 +1330,12 @@ skip_create:
err = teamd_dbus_expose_name(ctx);
if (err) {
teamd_log_err("Failed to expose dbus name.");
- goto dbus_fini;
+ goto zmq_fini;
}
return 0;
-
+zmq_fini:
+ teamd_zmq_fini(ctx);
dbus_fini:
teamd_dbus_fini(ctx);
usock_fini:
@@ -1357,6 +1376,7 @@ team_free:
static void teamd_fini(struct teamd_context *ctx)
{
+ teamd_zmq_fini(ctx);
teamd_dbus_fini(ctx);
teamd_usock_fini(ctx);
teamd_sriov_fini(ctx);
diff --git a/teamd/teamd.h b/teamd/teamd.h
index dbfd6b2..88c28f1 100644
--- a/teamd/teamd.h
+++ b/teamd/teamd.h
@@ -42,6 +42,10 @@
#include <dbus/dbus.h>
#endif
+#ifdef ENABLE_ZMQ
+#include <zmq.h>
+#endif
+
#define teamd_log_err(args...) daemon_log(LOG_ERR, ##args)
#define teamd_log_warn(args...) daemon_log(LOG_WARNING, ##args)
#define teamd_log_info(args...) daemon_log(LOG_INFO, ##args)
@@ -128,6 +132,14 @@ struct teamd_context {
DBusConnection * con;
} dbus;
#endif
+#ifdef ENABLE_ZMQ
+ struct {
+ bool enabled;
+ void * context;
+ void * sock;
+ char * addr;
+ } zmq;
+#endif
struct {
bool enabled;
int sock;
diff --git a/teamd/teamd_state.c b/teamd/teamd_state.c
index f4aeb4d..6191ea1 100644
--- a/teamd/teamd_state.c
+++ b/teamd/teamd_state.c
@@ -617,6 +617,18 @@ static int setup_state_dbus_enabled_get(struct teamd_context *ctx,
return 0;
}
+static int setup_state_zmq_enabled_get(struct teamd_context *ctx,
+ struct team_state_gsc *gsc,
+ void *priv)
+{
+#ifdef ENABLE_ZMQ
+ gsc->data.bool_val = ctx->zmq.enabled;
+#else
+ gsc->data.bool_val = false;
+#endif
+ return 0;
+}
+
static int setup_state_debug_level_get(struct teamd_context *ctx,
struct team_state_gsc *gsc,
void *priv)
@@ -666,6 +678,11 @@ static const struct teamd_state_val setup_state_vals[] = {
.getter = setup_state_dbus_enabled_get,
},
{
+ .subpath = "zmq_enabled",
+ .type = TEAMD_STATE_ITEM_TYPE_BOOL,
+ .getter = setup_state_zmq_enabled_get,
+ },
+ {
.subpath = "debug_level",
.type = TEAMD_STATE_ITEM_TYPE_INT,
.getter = setup_state_debug_level_get,
diff --git a/teamd/teamd_zmq.c b/teamd/teamd_zmq.c
new file mode 100644
index 0000000..8a69b5d
--- /dev/null
+++ b/teamd/teamd_zmq.c
@@ -0,0 +1,298 @@
+/*
+ * teamd_zmq.c - Teamd ZeroMQ socket api
+ * Copyright (C) 2013 Jiri Zupka <jzupka(a)redhat.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser 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 library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include "config.h"
+
+#ifdef ENABLE_ZMQ
+
+#include <stdbool.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <errno.h>
+#include <ctype.h>
+#include <private/misc.h>
+#include <private/list.h>
+#include <team.h>
+
+#include "teamd.h"
+#include "teamd_zmq.h"
+#include "teamd_zmq_common.h"
+#include "teamd_ctl.h"
+#include "teamd_config.h"
+
+struct zmq_ops_priv {
+ char *rcv_msg_args;
+ void *sock;
+};
+
+struct zmq_acc_conn {
+ struct list_item list;
+ int sock;
+};
+
+static int zmq_op_get_args(void *ops_priv, const char *fmt, ...)
+{
+ va_list ap;
+ struct zmq_ops_priv *zmq_ops_priv = ops_priv;
+ char **pstr;
+ char *str;
+ char *rest = zmq_ops_priv->rcv_msg_args;
+ int err = 0;
+
+ va_start(ap, fmt);
+ while (*fmt) {
+ switch (*fmt++) {
+ case 's': /* string */
+ pstr = va_arg(ap, char **);
+ str = teamd_zmq_msg_getline(&rest);
+ if (!str) {
+ teamd_log_err("Insufficient number of arguments in message.");
+ err = -EINVAL;
+ goto out;
+ }
+ *pstr = str;
+ break;
+ default:
+ teamd_log_err("Unknown argument type requested");
+ err = -EINVAL;
+ goto out;
+ }
+ }
+out:
+ va_end(ap);
+ return err;
+}
+
+void my_free_msg(void *data, void *hind) {
+ free(data);
+}
+
+static void zmq_custom_send(struct zmq_ops_priv *zmq_ops_priv,
+ char *buf, size_t buflen)
+{
+ int ret;
+
+ ret = zmq_send(zmq_ops_priv->sock, buf, buflen, 0);
+ if (ret == -1)
+ teamd_log_warn("zmq: send failed: %s", strerror(errno));
+}
+
+static int zmq_op_reply_err(void *ops_priv, const char *err_code,
+ const char *err_msg)
+{
+ struct zmq_ops_priv *zmq_ops_priv = ops_priv;
+ char *strbuf;
+ int err;
+
+ err = asprintf(&strbuf, "%s\n%s\n%s\n", TEAMD_ZMQ_REPLY_ERR_PREFIX,
+ err_code, err_msg);
+ if (err == -1)
+ return -ENOMEM;
+ zmq_custom_send(zmq_ops_priv, strbuf, strlen(strbuf));
+ free(strbuf);
+ return 0;
+}
+
+static int zmq_op_reply_succ(void *ops_priv, const char *msg)
+{
+ struct zmq_ops_priv *zmq_ops_priv = ops_priv;
+ char *strbuf;
+ int err;
+
+ err = asprintf(&strbuf, "%s\n%s", TEAMD_ZMQ_REPLY_SUCC_PREFIX,
+ msg ? msg : "");
+ if (err == -1)
+ return -ENOMEM;
+ zmq_custom_send(zmq_ops_priv, strbuf, strlen(strbuf));
+ free(strbuf);
+ return 0;
+}
+
+static const struct teamd_ctl_method_ops teamd_zmq_ctl_method_ops = {
+ .get_args = zmq_op_get_args,
+ .reply_err = zmq_op_reply_err,
+ .reply_succ = zmq_op_reply_succ,
+};
+
+static int process_rcv_msg(struct teamd_context *ctx, char *rcv_msg)
+{
+ struct zmq_ops_priv zmq_ops_priv;
+ char *str;
+ char *rest = rcv_msg;
+
+ str = teamd_zmq_msg_getline(&rest);
+ if (!str) {
+ teamd_log_dbg("zmq: Incomplete message.");
+ return 0;
+ }
+ if (strcmp(TEAMD_ZMQ_REQUEST_PREFIX, str)) {
+ teamd_log_dbg("zmq: Unsupported message type.");
+ return 0;
+ }
+
+ str = teamd_zmq_msg_getline(&rest);
+ if (!str) {
+ teamd_log_dbg("zmq: Incomplete message.");
+ return 0;
+ }
+ if (!teamd_ctl_method_exists(str)) {
+ teamd_log_dbg("zmq: Unknown method \"%s\".", str);
+ return 0;
+ }
+
+ zmq_ops_priv.sock = ctx->zmq.sock;
+ zmq_ops_priv.rcv_msg_args = rest;
+
+ teamd_log_dbg("zmq: calling method \"%s\"", str);
+
+ return teamd_ctl_method_call(ctx, str, &teamd_zmq_ctl_method_ops,
+ &zmq_ops_priv);
+}
+
+static int callback_zmq(struct teamd_context *ctx, int events, void *priv)
+{
+ int err = 0;
+ int poolmask;
+ size_t poolmask_size = sizeof(poolmask);
+
+ err = zmq_getsockopt(ctx->zmq.sock, ZMQ_EVENTS, &poolmask,
+ &poolmask_size);
+ if (err == -1)
+ return -errno;
+
+ while (poolmask & ZMQ_POLLIN) {
+ zmq_msg_t msg;
+
+ zmq_msg_init(&msg);
+ if (zmq_msg_recv(&msg, ctx->zmq.sock, 0) == -1) {
+ zmq_msg_close(&msg);
+ return -errno;
+ }
+
+ err = process_rcv_msg(ctx, zmq_msg_data(&msg));
+
+ zmq_msg_close(&msg);
+
+ if (err == -1)
+ break;
+
+ err = zmq_getsockopt(ctx->zmq.sock, ZMQ_EVENTS, &poolmask,
+ &poolmask_size);
+ if (err == -1)
+ return -errno;
+ }
+ return err;
+}
+
+#define ZMQ_MAX_CLIENT_COUNT 10
+
+static int teamd_zmq_sock_open(struct teamd_context *ctx)
+{
+ int err;
+ void *context, *sock;
+ int rc;
+ const char *addr;
+
+ context = zmq_ctx_new();
+ if (!context) {
+ teamd_log_err("zmq: Failed to create context.");
+ return -errno;
+ }
+
+ sock = zmq_socket(context, ZMQ_REP);
+ if (!sock) {
+ teamd_log_err("zmq: Failed to create socket.");
+ return -errno;
+ }
+
+ if (ctx->zmq.addr) {
+ addr = ctx->zmq.addr;
+ } else {
+ err = teamd_config_string_get(ctx, &addr, "$.runner.addr");
+ if (err) {
+ teamd_log_err("zmq: Failed to get address from config.");
+ return err;
+ }
+ }
+
+ rc = zmq_bind(sock, addr);
+ if (rc != 0) {
+ teamd_log_err("zmq: Failed to bind socket.");
+ err = -errno;
+ goto close_sock;
+ }
+
+ ctx->zmq.context = context;
+ ctx->zmq.sock = sock;
+ return 0;
+
+close_sock:
+ zmq_close(sock);
+ zmq_ctx_destroy(context);
+ return err;
+}
+
+static void teamd_zmq_sock_close(struct teamd_context *ctx)
+{
+ zmq_close(ctx->zmq.sock);
+ zmq_ctx_destroy(ctx->zmq.context);
+}
+
+#define ZMQ_CB_NAME "zmq"
+
+int teamd_zmq_init(struct teamd_context *ctx)
+{
+ int err;
+ int fd;
+ size_t fd_size;
+
+ if (!ctx->zmq.enabled)
+ return 0;
+ err = teamd_zmq_sock_open(ctx);
+ if (err)
+ return err;
+
+
+ fd_size = sizeof(fd);
+ zmq_getsockopt(ctx->zmq.sock, ZMQ_FD, &fd, &fd_size);
+
+ err = teamd_loop_callback_fd_add(ctx, ZMQ_CB_NAME, ctx, callback_zmq,
+ fd, TEAMD_LOOP_FD_EVENT_READ);
+ if (err)
+ goto sock_close;
+ teamd_loop_callback_enable(ctx, ZMQ_CB_NAME, ctx);
+ return 0;
+sock_close:
+ teamd_zmq_sock_close(ctx);
+ return err;
+}
+
+void teamd_zmq_fini(struct teamd_context *ctx)
+{
+ if (!ctx->zmq.enabled)
+ return;
+ teamd_loop_callback_del(ctx, ZMQ_CB_NAME, ctx);
+ teamd_zmq_sock_close(ctx);
+}
+
+#endif /* ENABLE_ZMQ */
diff --git a/teamd/teamd_zmq.h b/teamd/teamd_zmq.h
new file mode 100644
index 0000000..cde9e55
--- /dev/null
+++ b/teamd/teamd_zmq.h
@@ -0,0 +1,41 @@
+/*
+ * teamd_zmq.h - Teamd ZeroMQ api
+ * Copyright (C) 2013 Jiri Zupka <jzupka(a)redhat.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser 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 library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#ifndef _TEAMD_ZMQ_H_
+#define _TEAMD_ZMQ_H_
+
+#ifdef ENABLE_ZMQ
+
+int teamd_zmq_init(struct teamd_context *ctx);
+void teamd_zmq_fini(struct teamd_context *ctx);
+
+#else
+
+static inline int teamd_zmq_init(struct teamd_context *ctx)
+{
+ return 0;
+}
+
+static inline void teamd_zmq_fini(struct teamd_context *ctx)
+{
+}
+
+#endif /* ENABLE_ZMQ */
+
+#endif /* _TEAMD_ZMQ_H_ */
diff --git a/teamd/teamd_zmq_common.h b/teamd/teamd_zmq_common.h
new file mode 100644
index 0000000..254c21c
--- /dev/null
+++ b/teamd/teamd_zmq_common.h
@@ -0,0 +1,58 @@
+/*
+ * teamd_zmq_common.h - Teamd unix socket api common things
+ * Copyright (C) 2013 Jiri Zupka <jzupka(a)redhat.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser 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 library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#ifndef _TEAMD_ZMQ_COMMON_H_
+#define _TEAMD_ZMQ_COMMON_H_
+
+#include <stdio.h>
+#include <errno.h>
+#include <sys/ioctl.h>
+#include <linux/sockios.h>
+
+#include "teamd.h"
+
+#define TEAMD_ZMQ_REQUEST_PREFIX "REQUEST"
+#define TEAMD_ZMQ_REPLY_ERR_PREFIX "REPLY_ERROR"
+#define TEAMD_ZMQ_REPLY_SUCC_PREFIX "REPLY_SUCCESS"
+
+static inline char *teamd_zmq_msg_getline(char **p_rest)
+{
+ char *start = NULL;
+ char *rest = NULL;
+ char *str = *p_rest;
+
+ if (!str)
+ return NULL;
+ while (1) {
+ if (*str == '\0')
+ break;
+ if ((*str != '\n') && !start)
+ start = str;
+ if ((*str == '\n') && start) {
+ *str = '\0';
+ rest = str + 1;
+ break;
+ }
+ str++;
+ }
+ *p_rest = rest;
+ return start;
+}
+
+#endif /* _TEAMD_ZMQ_COMMON_H_ */
diff --git a/utils/teamdctl.c b/utils/teamdctl.c
index 69a00d1..009922d 100644
--- a/utils/teamdctl.c
+++ b/utils/teamdctl.c
@@ -148,17 +148,19 @@ static int stateview_json_setup_process(char **prunner_name, json_t *dump_json)
char *runner_name;
char *kernel_team_mode_name;
int dbus_enabled;
+ int zmq_enabled;
int debug_level;
int daemonized;
int pid;
char *pid_file;
pr_out("setup:\n");
- err = json_unpack(dump_json, "{s:{s:s, s:s, s:b, s:i, s:b, s:i, s:s}}",
+ err = json_unpack(dump_json, "{s:{s:s, s:s, s:b, s:b, s:i, s:b, s:i, s:s}}",
"setup",
"runner_name", &runner_name,
"kernel_team_mode_name", &kernel_team_mode_name,
"dbus_enabled", &dbus_enabled,
+ "zmq_enabled", &zmq_enabled,
"debug_level", &debug_level,
"daemonized", &daemonized,
"pid", &pid,
@@ -171,6 +173,7 @@ static int stateview_json_setup_process(char **prunner_name, json_t *dump_json)
pr_out("runner: %s\n", runner_name);
pr_out2("kernel team mode: %s\n", kernel_team_mode_name);
pr_out2("D-BUS enabled: %s\n", boolyesno(dbus_enabled));
+ pr_out2("ZeroMQ enabled: %s\n", boolyesno(zmq_enabled));
pr_out2("debug level: %d\n", debug_level);
pr_out2("daemonized: %s\n", boolyesno(daemonized));
pr_out2("PID: %d\n", pid);
@@ -291,6 +294,7 @@ static int stateview_json_port_link_watches_list_process(json_t *port_link_watch
}
return 0;
}
+
static int stateview_json_port_link_watches_process(json_t *port_link_watches_json)
{
int err;
@@ -775,6 +779,7 @@ static struct command_type command_types[] = {
.params = {"PORTDEV"},
},
};
+
#define COMMAND_TYPE_COUNT ARRAY_SIZE(command_types)
static bool __cmd_executable(struct command_type *command_type)
@@ -876,6 +881,31 @@ static int check_team_devname(char *team_devname)
return 0;
}
+static int check_teamd_team_devname(struct teamdctl *tdc,
+ const char *team_devname)
+{
+ int ret = 0;
+ json_t* root;
+ json_error_t error;
+ json_t* j_device_name;
+ const char* teamd_device_name;
+
+ root = json_loads(teamdctl_config_get_raw(tdc), 0, &error);
+ j_device_name = json_object_get(root, "device");
+
+ teamd_device_name = json_string_value(j_device_name);
+
+ if (strcmp(team_devname, teamd_device_name) != 0) {
+ pr_err("Unable to access to %s through connected teamd daemon because daemon controls %s.\n",
+ team_devname, teamd_device_name);
+ ret = -1;
+ }
+
+ json_decref(j_device_name);
+ json_decref(root);
+ return ret;
+}
+
static int call_command(struct teamdctl *tdc, int argc, char **argv,
struct command_type *command_type)
{
@@ -900,8 +930,9 @@ static void print_help(const char *argv0) {
" -v --verbose Increase output verbosity\n"
" -o --oneline Force output to one line if possible\n"
" -D --force-dbus Force to use D-Bus interface\n"
+ " -Z --force-zmq=ADDRESS Force to use ZeroMQ interface [-Z[Address]]\n"
" -U --force-usock Force to use UNIX domain socket interface\n",
- argv0);
+ argv0);
pr_out("Commands:\n");
for (i = 0; i < COMMAND_TYPE_COUNT; i++) {
command_type = &command_types[i];
@@ -924,6 +955,7 @@ int main(int argc, char **argv)
{ "verbose", no_argument, NULL, 'v' },
{ "oneline", no_argument, NULL, 'o' },
{ "force-dbus", no_argument, NULL, 'D' },
+ { "force-zmq", required_argument, NULL, 'Z' },
{ "force-usock", no_argument, NULL, 'U' },
{ NULL, 0, NULL, 0 }
};
@@ -932,10 +964,12 @@ int main(int argc, char **argv)
struct command_type *command_type;
struct teamdctl *tdc;
int ret;
+ char *addr = NULL;
bool force_dbus = false;
+ bool force_zmq = false;
bool force_usock = false;
- while ((opt = getopt_long(argc, argv, "hvoDU",
+ while ((opt = getopt_long(argc, argv, "hvoDZ:U",
long_options, NULL)) >= 0) {
switch(opt) {
@@ -956,6 +990,15 @@ int main(int argc, char **argv)
force_dbus = true;
#endif
break;
+ case 'Z':
+#ifndef ENABLE_ZMQ
+ fprintf(stderr, "ZeroMQ support is not compiled-in\n");
+ return EXIT_FAILURE;
+#else
+ force_zmq = true;
+ addr = optarg;
+#endif
+ break;
case 'U':
force_usock = true;
break;
@@ -970,8 +1013,10 @@ int main(int argc, char **argv)
}
}
- if (force_usock && force_dbus) {
- pr_err("Either UNIX domain socket interface or D-Bus interface can be forced at a time.\n");
+ if ((force_usock && force_dbus) ||
+ (force_usock && force_zmq) ||
+ (force_dbus && force_zmq)) {
+ pr_err("Only one interface could be forced at a time (UNIX domain socket, D-Bus, ZMQ).\n");
print_help(argv0);
return EXIT_FAILURE;
}
@@ -1005,13 +1050,21 @@ int main(int argc, char **argv)
pr_err("teamdctl_alloc failed\n");
return EXIT_FAILURE;
}
- err = teamdctl_connect(tdc, team_devname,
- force_usock ? "usock" : (force_dbus ? "dbus" : NULL));
+
+ err = teamdctl_connect(tdc, team_devname, addr,
+ (force_usock ? "usock" : (force_dbus ?
+ "dbus": (force_zmq ? "zmq" :NULL))));
if (err) {
pr_err("teamdctl_connect failed (%s)\n", strerror(-err));
ret = EXIT_FAILURE;
goto teamdctl_free;
}
+
+ if (check_teamd_team_devname(tdc, team_devname)) {
+ ret = EXIT_FAILURE;
+ goto teamdctl_disconnect;
+ }
+
err = call_command(tdc, argc, argv, command_type);
if (err) {
pr_err("command call failed (%s)\n", strerror(-err));
--
1.8.3.1
