Merge state management command into values, adapt munin accordingly

4 files changed, 67 insertions, 90 deletions

diff --git a/doc/stats_dabmux_multi.py b/doc/stats_dabmux_multi.py
index 664f2f1..ff6f0a4 100755
--- a/doc/stats_dabmux_multi.py
+++ b/doc/stats_dabmux_multi.py
@@ -7,10 +7,13 @@ import sys
 import json
 import zmq
 import os
+import re
 
 config_top = """
 """
 
+#default data type is GAUGE
+
 config_template = """
 multigraph buffers_{ident}
 graph_title Contribution {ident} buffer
@@ -68,6 +71,19 @@ right.min -90
 right.max 0
 right.warning -40:0
 right.critical -80:0
+
+multigraph state_{ident}
+graph_title State of contribution {ident}
+graph_order state
+graph_args --base 1000 --lower-limit 0 --upper-limit 5
+graph_vlabel Current state of the input
+graph_category dabmux
+graph_info This graph shows the state for the {ident} ZMQ input
+
+state.info Input state
+state.label 0 Unknown, 1 NoData, 2 Unstable, 3 Silent, 4 Streaming
+state.warning 4:4
+state.critical 2:4
 """
 
 ctx = zmq.Context()
@@ -93,6 +109,8 @@ def connect():
 
     return sock
 
+re_state = re.compile(r"\w+ (\d+)")
+
 if len(sys.argv) == 1:
     sock = connect()
     sock.send("values")
@@ -111,6 +129,14 @@ if len(sys.argv) == 1:
         munin_values += "left.value {}\n".format(v['peak_left'])
         munin_values += "right.value {}\n".format(v['peak_right'])
 
+        if 'state' in v:
+            # If ODR-DabMux is v1.3.1-3 or older, it doesn't export state
+            match = re_state.match(v['state'])
+            if match:
+                munin_values += "state.value {}\n".format(match.group(1))
+            else:
+                print("Cannot parse state '{}'".format(v['state']))
+
     print(munin_values)
 
 elif len(sys.argv) == 2 and sys.argv[1] == "config":
diff --git a/doc/zmq_remote.py b/doc/zmq_remote.py
index 155390b..47da520 100755
--- a/doc/zmq_remote.py
+++ b/doc/zmq_remote.py
@@ -12,6 +12,9 @@ context = zmq.Context()
 
 sock = context.socket(zmq.REQ)
 
+poller = zmq.Poller()
+poller.register(sock, zmq.POLLIN)
+
 if len(sys.argv) < 2:
     print("Usage: program url cmd [args...]")
     sys.exit(1)
@@ -22,34 +25,37 @@ message_parts = sys.argv[2:]
 
 # first do a ping test
 
-print("Send ping")
-sock.send("ping".encode())
-data = sock.recv_multipart()
+print("ping")
+sock.send(b"ping")
 
-if len(data) != 1:
-    print("Received invalid number of parts: {}".format(len(data)))
-    for i,part in enumerate(data):
-        print("   {}".format(part))
-    sys.exit(1)
+socks = dict(poller.poll(1000))
+if socks:
+    if socks.get(sock) == zmq.POLLIN:
 
-if data[0] != b'ok':
-    print("Received invalid ping response: {}".format(data.decode()))
-    sys.exit(1)
+        data = sock.recv_multipart()
+        print("Received: {}".format(len(data)))
+        for i,part in enumerate(data):
+            print("   {}".format(part))
+
+        for i, part in enumerate(message_parts):
+            if i == len(message_parts) - 1:
+                f = 0
+            else:
+                f = zmq.SNDMORE
 
-print("Ping ok, sending request '{}'...".format(" ".join(message_parts)))
+            print("Send {}({}): '{}'".format(i, f, part))
 
-for i, part in enumerate(message_parts):
-    if i == len(message_parts) - 1:
-        f = 0
-    else:
-        f = zmq.SNDMORE
+            sock.send(part.encode(), flags=f)
 
-    sock.send(part.encode(), flags=f)
+        data = sock.recv_multipart()
 
-data = sock.recv_multipart()
+        print("Received: {}".format(len(data)))
+        for i,part in enumerate(data):
+            print(" RX {}: {}".format(i, part))
 
-print("Received {} entries:".format(len(data)))
-print("  " + "  ".join([d.decode() for d in data]))
+else:
+    print("ZMQ error: timeout")
+    context.destroy(linger=5)
 
 # This is free and unencumbered software released into the public domain.
 #
diff --git a/src/ManagementServer.cpp b/src/ManagementServer.cpp
index c44738a..3300c89 100644
--- a/src/ManagementServer.cpp
+++ b/src/ManagementServer.cpp
@@ -2,7 +2,7 @@
    Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications
    Research Center Canada)
 
-   Copyright (C) 2016
+   Copyright (C) 2017
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -146,35 +146,6 @@ std::string ManagementServer::getValuesJSON()
     return ss.str();
 }
 
-std::string ManagementServer::getStateJSON()
-{
-    boost::mutex::scoped_lock lock(m_statsmutex);
-
-    std::ostringstream ss;
-    ss << "{\n";
-
-    std::map<std::string,InputStat*>::iterator iter;
-    int i = 0;
-    for(iter = m_inputStats.begin(); iter != m_inputStats.end();
-            ++iter, i++)
-    {
-        const std::string& id = iter->first;
-        InputStat* stats = iter->second;
-
-        if (i > 0) {
-            ss << " ,\n";
-        }
-
-        ss << " \"" << id << "\" : ";
-        ss << stats->encodeStateJSON();
-        stats->reset();
-    }
-
-    ss << "}\n";
-
-    return ss.str();
-}
-
 void ManagementServer::restart()
 {
     m_restarter_thread = boost::thread(&ManagementServer::restart_thread,
@@ -244,9 +215,6 @@ void ManagementServer::handle_message(zmq::message_t& zmq_message)
         else if (data == "values") {
             answer << getValuesJSON();
         }
-        else if (data == "state") {
-            answer << getStateJSON();
-        }
         else if (data == "getptree") {
             boost::unique_lock<boost::mutex> lock(m_configmutex);
             boost::property_tree::json_parser::write_json(answer, m_pt);
@@ -305,44 +273,34 @@ std::string InputStat::encodeValuesJSON()
         "\"peak_left\": " << dB_l << ", "
         "\"peak_right\": " << dB_r << ", "
         "\"num_underruns\": " << num_underruns << ", "
-        "\"num_overruns\": " << num_overruns <<
-    " } }";
+        "\"num_overruns\": " << num_overruns << ", ";
 
-    return ss.str();
-}
-
-std::string InputStat::encodeStateJSON()
-{
-    std::ostringstream ss;
-
-    ss << "{ \"state\" : ";
+    ss << "\"state\": ";
 
     switch (determineState()) {
         case NoData:
-            ss << "\"NoData\"";
+            ss << "\"NoData (1)\"";
             break;
         case Unstable:
-            ss << "\"Unstable\"";
+            ss << "\"Unstable (2)\"";
             break;
         case Silence:
-            ss << "\"Silent\"";
+            ss << "\"Silent (3)\"";
             break;
         case Streaming:
-            ss << "\"Streaming\"";
+            ss << "\"Streaming (4)\"";
             break;
         default:
-            ss << "\"Unknown\"";
+            ss << "\"Unknown (0)\"";
     }
 
-    ss << " }";
+    ss << " } }";
 
     return ss.str();
 }
 
 input_state_t InputStat::determineState()
 {
-    boost::mutex::scoped_lock lock(m_mutex);
-
     time_t now = time(nullptr);
     input_state_t state;
 
diff --git a/src/ManagementServer.h b/src/ManagementServer.h
index aade3d0..e75aed3 100644
--- a/src/ManagementServer.h
+++ b/src/ManagementServer.h
@@ -2,7 +2,7 @@
    Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications
    Research Center Canada)
 
-   Copyright (C) 2016
+   Copyright (C) 2017
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -18,13 +18,11 @@
    The responds in JSON, and accepts the commands:
     - config
     - values
-      Inspired by the munin equivalent
-
-    - state
-      Returns the state of each input
+      Inspired by the munin equivalent, returns the configuration
+      and the statistics values for every exported stat.
 
     - getptree
-      Returns the internal boost property_tree that contains the 
+      Returns the internal boost property_tree that contains the
       multiplexer configuration DB.
 
    The server is using REQ/REP ZeroMQ sockets.
@@ -46,8 +44,7 @@
    along with ODR-DabMux.  If not, see <http://www.gnu.org/licenses/>.
    */
 
-#ifndef __MANAGEMENT_SERVER_H
-#define __MANAGEMENT_SERVER_H
+#pragma once
 
 #ifdef HAVE_CONFIG_H
 #   include "config.h"
@@ -278,8 +275,6 @@ class InputStat
 
         std::string encodeValuesJSON(void);
 
-        std::string encodeStateJSON(void);
-
         input_state_t determineState(void);
 
     private:
@@ -394,12 +389,6 @@ class ManagementServer
          */
         std::string getValuesJSON();
 
-        /* Return the state of each input
-         *
-         * returns: JSON encoded state
-         */
-        std::string getStateJSON();
-
         // mutex for accessing the map
         boost::mutex m_statsmutex;
 
@@ -412,5 +401,3 @@ class ManagementServer
 // a reference to it
 ManagementServer& get_mgmt_server();
 
-#endif
-