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

6 files changed, 170 insertions, 21 deletions

diff --git a/configure.ac b/configure.ac
index 6e85db4..35b4ee0 100644
--- a/configure.ac
+++ b/configure.ac
@@ -47,7 +47,7 @@ dnl checks for libraries
 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 @@ struct osmo_pcap_conn {
 	/* 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 @@ struct osmo_pcap_conn *osmo_pcap_server_find(struct osmo_pcap_server *ser,
 					     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 @@ static void close_connection(struct osmo_pcap_conn *conn)
 		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 @@ static void restart_pcap(struct osmo_pcap_conn *conn)
 		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 @@ static void write_data(struct osmo_pcap_conn *conn, struct osmo_pcap_data *data)
 	struct tm *tm = localtime(&now);
 	int rc;
 
+	client_data(conn, data);
+
 	if (conn->no_store) {
 		conn->last_write = *tm;
 		return;
@@ -317,6 +400,7 @@ static int accept_cb(struct osmo_fd *fd, unsigned int when)
 		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 @@ DEFUN(cfg_server_client,
 
 	/* 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;