Make inputZMQ use the statistics, add munin script

5 files changed, 22 insertions, 5 deletions

diff --git a/src/DabMux.cpp b/src/DabMux.cpp
index bd8e685..b735cf7 100644
--- a/src/DabMux.cpp
+++ b/src/DabMux.cpp
@@ -117,11 +117,14 @@ typedef DWORD32 uint32_t;
 #include "utils.h"
 #include "ParserCmdline.h"
 #include "ParserConfigfile.h"
-
+#include "StatsServer.h"
 #include "Log.h"
 
 using namespace std;
 
+/* Global stats server */
+StatsServer global_stats(12720); //TODO define port
+
 static unsigned char Padding_FIB[] = {
     0xff, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
     0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
diff --git a/src/Makefile.am b/src/Makefile.am
index 6dea558..910d5bc 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -38,7 +38,7 @@ ZMQ_LIBS    =
 endif
 
 crc_dabmux_CPPFLAGS =-I$(FARSYNC_DIR) $(GITVERSION_FLAGS)
-crc_dabmux_LDADD    =$(FEC_LIBS) $(ZMQ_LIBS) -lpthread
+crc_dabmux_LDADD    =$(FEC_LIBS) $(ZMQ_LIBS) -lpthread -lboost_thread -lboost_system
 crc_dabmux_SOURCES  =DabMux.cpp \
                      dabInput.h dabInput.cpp \
                      dabInputBridgeUdp.h dabInputBridgeUdp.cpp \
@@ -85,6 +85,7 @@ crc_dabmux_SOURCES  =DabMux.cpp \
                      Interleaver.h Interleaver.cpp \
                      ReedSolomon.h ReedSolomon.cpp \
                      mpeg.h mpeg.c \
+                     StatsServer.h StatsServer.cpp \
                      TcpServer.h TcpServer.cpp \
                      TcpSocket.h TcpSocket.cpp
 
diff --git a/src/TestStatsServer.cpp b/src/TestStatsServer.cpp
index 7ec0342..0d86195 100644
--- a/src/TestStatsServer.cpp
+++ b/src/TestStatsServer.cpp
@@ -18,6 +18,10 @@ int main(int argc, char **argv)
             if (stats_example[i] == 0) {
                 serv.notifyUnderrun("foo");
             }
+
+            if (stats_example[i] == 56) {
+                serv.notifyOverrun("foo");
+            }
         }
     }
 
diff --git a/src/dabInputZmq.cpp b/src/dabInputZmq.cpp
index fc975bc..09b4688 100644
--- a/src/dabInputZmq.cpp
+++ b/src/dabInputZmq.cpp
@@ -35,6 +35,7 @@
 #include "dabInput.h"
 #include "dabInputZmq.h"
 #include "dabInputFifo.h"
+#include "StatsServer.h"
 
 #include <stdio.h>
 #include <zmq.h>
@@ -49,6 +50,7 @@
 
 #ifdef HAVE_INPUT_ZEROMQ
 
+extern StatsServer global_stats;
 
 struct dabInputOperations dabInputZmqOperations = {
     dabInputZmqInit,
@@ -68,8 +70,6 @@ struct dabInputOperations dabInputZmqOperations = {
 int dabInputZmqInit(void** args)
 {
     dabInputZmqData* input = new dabInputZmqData;
-    memset(&input->stats, 0, sizeof(input->stats));
-    input->stats.id = dabInputFifoData::nb++;
     input->zmq_context = zmq_ctx_new();
     if (input->zmq_context == NULL) {
         etiLog.log(error, "Failed to initialise ZeroMQ context: %s\n", zmq_strerror(errno));
@@ -108,11 +108,14 @@ int dabInputZmqOpen(void* args, const char* inputUri)
         return 1;
     }
 
+    global_stats.registerInput(uri);
+
     input->uri = uri;
     return 0;
 }
 
 
+// size corresponds to a frame size. It is constant for a given bitrate
 int dabInputZmqReadFrame(dabInputOperations* ops, void* args, void* buffer, int size)
 {
     int rc;
@@ -123,6 +126,7 @@ int dabInputZmqReadFrame(dabInputOperations* ops, void* args, void* buffer, int
         rc = dabInputZmqReadFromSocket(input, size);
     }
     else {
+        global_stats.notifyOverrun(input->uri);
         rc = 0;
     }
 
@@ -132,15 +136,20 @@ int dabInputZmqReadFrame(dabInputOperations* ops, void* args, void* buffer, int
         if (input->prebuffering == 0)
             etiLog.log(info, "inputZMQ %s input pre-buffering complete\n",
                 input->uri.c_str());
+        global_stats.notifyUnderrun(input->uri);
         memset(buffer, 0, size);
         return size;
     }
 
+    // Save stats data in bytes, not in frames
+    global_stats.notifyBuffer(input->uri, input->frame_buffer.size() * size);
+
     if (input->frame_buffer.empty()) {
         etiLog.log(warn, "inputZMQ %s input empty, re-enabling pre-buffering\n",
                 input->uri.c_str());
         // reset prebuffering
         input->prebuffering = INPUT_ZMQ_PREBUFFERING;
+        global_stats.notifyUnderrun(input->uri);
         memset(buffer, 0, size);
         return size;
     }
diff --git a/src/dabInputZmq.h b/src/dabInputZmq.h
index 17a4954..237209f 100644
--- a/src/dabInputZmq.h
+++ b/src/dabInputZmq.h
@@ -47,6 +47,7 @@
 #include <string>
 #include "dabInput.h"
 #include "dabInputFifo.h"
+#include "StatsServer.h"
 
 // Number of frames to prebuffer
 #define INPUT_ZMQ_PREBUFFERING 10
@@ -61,7 +62,6 @@ struct dabInputZmqData {
     void* zmq_context;
     void* zmq_sock;
     std::list<char*> frame_buffer; //stores elements of type char[<framesize>]
-    dabInputFifoStats stats;
     int prebuffering;
     std::string uri;
 };