server: Add zmq based event and data interface to the server
To allow easily extracting or streaming the data to an external
analysis system, zeromq can be configured (and reconfigured). The
system works as fire and forget and no loss detection is present.
A simple go based client application is provided to subscribe to
the publisher.
Change-Id: I4f3e6d675023a81b7d2ee19bf1f44a2be0ca003c
diff --git a/configure.ac b/configure.ac
index 6e85db4..35b4ee0 100644
--- a/configure.ac
+++ b/configure.ac
@@ -47,7 +47,7 @@
PKG_CHECK_MODULES(LIBOSMOCORE, libosmocore >= 0.3.2)
PKG_CHECK_MODULES(LIBOSMOGSM, libosmogsm >= 0.3.2)
PKG_CHECK_MODULES(LIBOSMOVTY, libosmovty >= 0.3.0)
-PKG_CHECK_MODULES(LIBZMQ, libzmq >= 4.0.0)
+PKG_CHECK_MODULES(LIBZMQ, libzmq >= 3.2.2)
# Coverage build taken from WebKit's configure.in
diff --git a/contrib/zmq_recv.go b/contrib/zmq_recv.go
new file mode 100644
index 0000000..cd47fec
--- /dev/null
+++ b/contrib/zmq_recv.go
@@ -0,0 +1,25 @@
+package main
+
+import (
+ "fmt";
+ "strings";
+ zmq "github.com/pebbe/zmq4"
+)
+
+func main() {
+ subscriber, _ := zmq.NewSocket(zmq.SUB)
+ defer subscriber.Close()
+ subscriber.Connect("tcp://localhost:6666")
+
+ subscriber.SetSubscribe("")
+
+ for {
+ msg, _ := subscriber.RecvMessage(0)
+ if (strings.HasPrefix(msg[0], "event.v1")) {
+ fmt.Println("Got event message.. %d", len(msg), msg)
+ } else if (strings.HasPrefix(msg[0], "data.v1")) {
+ fmt.Println("Got data message.. %d", len(msg), msg)
+ }
+ }
+
+}
diff --git a/doc/zmq_data.txt b/doc/zmq_data.txt
new file mode 100644
index 0000000..b74dd62
--- /dev/null
+++ b/doc/zmq_data.txt
@@ -0,0 +1,41 @@
+ZeroMQ data protocol v1
+=======================
+
+The osmo-pcap-server can be configured to publish PCAP data to
+zero to many subscribers. The following document describes the
+data format used.
+
+Multiple clients might be connected to the osmo-pcap-server and
+use different link headers depending on the underlying device
+data is being captured from.
+
+The messages published are in two categories. These are client
+events and client data. Client events are generated on connect,
+disconnect, link type change and client data is sent for each
+frame.
+
+Client Events
+^^^^^^^^^^^^^
+
+A multi-part message with event.v1.<EVENT_NAME>.<CLIENT_NAME>
+as the first part followed by textual data will be generated.
+<CLIENT_NAME> is the configured name and <EVENT_NAME> can be
+any of:
+
+ * connect
+ * disconnect
+ * closingtracefile
+
+It might contain more information, such as the filename of the
+tracefile that was closed. There is no guarantee for the order
+and amount of connect/disconnect messages.
+
+
+Client Data
+^^^^^^^^^^^
+
+A multi-part message with data.v1.<CLIENT_NAME> to allow to
+filter for data and a specific client if wanted.
+
+It is followed by the pcap_file_header structure as the second
+part and then the data as third part.
diff --git a/include/osmo-pcap/osmo_pcap_server.h b/include/osmo-pcap/osmo_pcap_server.h
index 3af9ded..5af0838 100644
--- a/include/osmo-pcap/osmo_pcap_server.h
+++ b/include/osmo-pcap/osmo_pcap_server.h
@@ -58,6 +58,7 @@
/* Remote connection */
struct osmo_fd rem_fd;
int local_fd;
+ char *curr_filename;
/* pcap stuff */
struct pcap_file_header file_hdr;
@@ -98,5 +99,6 @@
const char *name);
void osmo_pcap_server_delete(struct osmo_pcap_conn *conn);
void vty_server_init(struct osmo_pcap_server *server);
+void osmo_pcap_server_close_trace(struct osmo_pcap_conn *conn);
#endif
diff --git a/src/osmo_server_network.c b/src/osmo_server_network.c
index d530ef7..6812f35 100644
--- a/src/osmo_server_network.c
+++ b/src/osmo_server_network.c
@@ -31,11 +31,99 @@
#include <sys/socket.h>
#include <sys/types.h>
+#include <zmq.h>
+
#include <fcntl.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
+static void pcap_zmq_send(void *publ, const void *data, size_t len, int flags)
+{
+ int rc;
+ zmq_msg_t msg;
+
+ rc = zmq_msg_init_size(&msg, len);
+ if (rc != 0) {
+ /* sigh.. we said SNDMORE but can't... */
+ LOGP(DSERVER, LOGL_ERROR, "Failed to init rc=%d errno=%d/%s\n",
+ rc, errno, strerror(errno));
+ return;
+ }
+ memcpy(zmq_msg_data(&msg), data, len);
+ rc = zmq_msg_send(&msg, publ, flags);
+ if (rc == -1) {
+ /* is the zmq_msg now owned? leak??? */
+ LOGP(DSERVER, LOGL_ERROR, "Failed to send data rc=%d errno=%d/%s\n",
+ rc, errno, strerror(errno));
+ return;
+ }
+}
+
+static void client_event(struct osmo_pcap_conn *conn,
+ const char *event, const char *data)
+{
+ char *event_name;
+
+ if (!conn->server->zmq_publ)
+ return;
+
+ /*
+ * This multi-part support is insane... so if we lose the first
+ * or the last part of the multipart message stuff is going out
+ * of sync. *great* As we can't do anything about it right now
+ * just close the eyese and send it.
+ */
+ event_name = talloc_asprintf(conn, "event.v1.%s.%s",
+ event, conn->name);
+ pcap_zmq_send(conn->server->zmq_publ,
+ event_name, strlen(event_name),
+ data ? ZMQ_SNDMORE : 0);
+ talloc_free(event_name);
+ if (data)
+ pcap_zmq_send(conn->server->zmq_publ, data, strlen(data), 0);
+}
+
+static void client_data(struct osmo_pcap_conn *conn,
+ struct osmo_pcap_data *data)
+{
+ char *event_name;
+
+ if (!conn->server->zmq_publ)
+ return;
+
+ /*
+ * This multi-part support is insane... so if we lose the first
+ * or the last part of the multipart message stuff is going out
+ * of sync. *great* As we can't do anything about it right now
+ * just close the eyese and send it.
+ */
+ event_name = talloc_asprintf(conn, "data.v1.%s", conn->name);
+ pcap_zmq_send(conn->server->zmq_publ, event_name, strlen(event_name), ZMQ_SNDMORE);
+ talloc_free(event_name);
+
+ pcap_zmq_send(conn->server->zmq_publ,
+ &conn->file_hdr, sizeof(conn->file_hdr),
+ ZMQ_SNDMORE);
+ pcap_zmq_send(conn->server->zmq_publ,
+ &data->data[0], data->len,
+ 0);
+}
+
+void osmo_pcap_server_close_trace(struct osmo_pcap_conn *conn)
+{
+ if (conn->local_fd >= 0) {
+ close(conn->local_fd);
+ conn->local_fd = -1;
+ }
+
+ if (conn->curr_filename) {
+ client_event(conn, "closingtracefile", conn->curr_filename);
+ talloc_free(conn->curr_filename);
+ conn->curr_filename = NULL;
+ }
+}
+
static void close_connection(struct osmo_pcap_conn *conn)
{
if (conn->rem_fd.fd >= 0) {
@@ -44,44 +132,39 @@
osmo_fd_unregister(&conn->rem_fd);
}
- if (conn->local_fd >= 0) {
- close(conn->local_fd);
- conn->local_fd = -1;
- }
+ osmo_pcap_server_close_trace(conn);
+ client_event(conn, "disconnect", NULL);
}
static void restart_pcap(struct osmo_pcap_conn *conn)
{
time_t now = time(NULL);
struct tm *tm = localtime(&now);
- char *filename;
int rc;
- if (conn->local_fd >= 0) {
- close(conn->local_fd);
- conn->local_fd = -1;
- }
+ osmo_pcap_server_close_trace(conn);
/* omit any storing/creation of the file */
if (conn->no_store) {
conn->last_write = *tm;
+ talloc_free(conn->curr_filename);
+ conn->curr_filename = NULL;
return;
}
- filename = talloc_asprintf(conn, "%s/trace-%s-%d%.2d%.2d_%.2d%.2d%.2d.pcap",
+ conn->curr_filename = talloc_asprintf(conn, "%s/trace-%s-%d%.2d%.2d_%.2d%.2d%.2d.pcap",
conn->server->base_path, conn->name,
tm->tm_year + 1900, tm->tm_mon + 1, tm->tm_mday,
tm->tm_hour, tm->tm_min, tm->tm_sec);
- if (!filename) {
+ if (!conn->curr_filename) {
LOGP(DSERVER, LOGL_ERROR, "Failed to assemble filename for %s.\n", conn->name);
return;
}
- conn->local_fd = creat(filename, 0440);
+ conn->local_fd = creat(conn->curr_filename, 0440);
if (conn->local_fd < 0) {
- LOGP(DSERVER, LOGL_ERROR, "Failed to file: '%s'\n", filename);
- talloc_free(filename);
+ LOGP(DSERVER, LOGL_ERROR, "Failed to file: '%s'\n", conn->curr_filename);
return;
}
@@ -90,12 +173,10 @@
LOGP(DSERVER, LOGL_ERROR, "Failed to write the header: %d\n", errno);
close(conn->local_fd);
conn->local_fd = -1;
- talloc_free(filename);
return;
}
conn->last_write = *tm;
- talloc_free(filename);
}
static void link_data(struct osmo_pcap_conn *conn, struct osmo_pcap_data *data)
@@ -127,6 +208,8 @@
struct tm *tm = localtime(&now);
int rc;
+ client_data(conn, data);
+
if (conn->no_store) {
conn->last_write = *tm;
return;
@@ -317,6 +400,7 @@
if (conn->remote_addr.s_addr == addr.sin_addr.s_addr) {
LOGP(DSERVER, LOGL_NOTICE,
"New connection from %s\n", conn->name);
+ client_event(conn, "connect", NULL);
new_connection(server, conn, new_fd);
return 0;
}
diff --git a/src/osmo_server_vty.c b/src/osmo_server_vty.c
index 5428445..d13ea6f 100644
--- a/src/osmo_server_vty.c
+++ b/src/osmo_server_vty.c
@@ -134,10 +134,7 @@
/* Checking no-store and maybe closing a pcap file */
if (argc >= 3) {
- if (conn->local_fd >= 0) {
- close(conn->local_fd);
- conn->local_fd = -1;
- }
+ osmo_pcap_server_close_trace(conn);
conn->no_store = 1;
} else
conn->no_store = 0;