Add TAI bulletin expiry to RC and to munin

5 files changed, 177 insertions, 16 deletions

diff --git a/doc/stats_dabmux_multi.py b/doc/stats_dabmux_multi.py
index 679ae6b..605e717 100755
--- a/doc/stats_dabmux_multi.py
+++ b/doc/stats_dabmux_multi.py
@@ -1,7 +1,11 @@
 #!/usr/bin/env python2
 #
-# present statistics from dabmux Stats Server
-# to munin
+# present statistics from dabmux Stats Server and ZeroMQ RC
+# to munin. Expects Stats server on port 12720 and ZeroMQ RC
+# on port 12722.
+#
+# Copy this to /etc/munin/plugins/stats_dabmux_multi
+# and make it executable (chmod +x)
 
 import sys
 import json
@@ -10,7 +14,19 @@ import os
 import re
 
 config_top = """
-"""
+multigraph clocktai_expiry
+graph_title Time to expiry for TAI bulletin
+graph_order expiry
+graph_args --base 1000
+graph_vlabel Number of seconds until expiry
+graph_category dabmux
+graph_info This graph shows the number of remaining seconds this bulletin is valid
+
+expiry.info Seconds until expiry
+expiry.label Seconds until expiry
+expiry.min 0
+expiry.warning {onemonth}:
+""".format(onemonth=3600*24*30)
 
 #default data type is GAUGE
 
@@ -107,7 +123,45 @@ def do_transaction(command, sock):
     sys.stderr.write("Could not receive data for command '{}'\n".format(command))
     sys.exit(1)
 
-def connect():
+def do_multipart_transaction(message_parts, sock):
+    """To a send + receive transaction, quit whole program on timeout"""
+    if isinstance(message_parts, str):
+        sys.stderr.write("do_transaction expects a list!\n");
+        sys.exit(1)
+
+    for i, part in enumerate(message_parts):
+        if i == len(message_parts) - 1:
+            f = 0
+        else:
+            f = zmq.SNDMORE
+        sock.send(part, flags=f)
+
+    poller = zmq.Poller()
+    poller.register(sock, zmq.POLLIN)
+
+    socks = dict(poller.poll(1000))
+    if socks:
+        if socks.get(sock) == zmq.POLLIN:
+            rxpackets = sock.recv_multipart()
+            return rxpackets
+
+    raise RCException("Could not receive data for command '{}'\n".format(
+        message_parts))
+
+def get_rc_value(module, name, sock):
+    try:
+        parts = do_multipart_transaction([b"get", module.encode(), name.encode()],
+                sock)
+        if len(parts) != 1:
+            sys.stderr.write("Received unexpected multipart message {}\n".format(
+                parts))
+            sys.exit(1)
+        return parts[0].decode()
+    except RCException as e:
+        print("get {} {} fail: {}".format(module, name, e))
+        return ""
+
+def connect_to_stats():
     """Create a connection to the dabmux stats server
 
     returns: the socket"""
@@ -124,13 +178,46 @@ def connect():
 
     return sock
 
-re_state = re.compile(r"\w+ \((\d+)\)")
+def connect_to_rc():
+    """Create a connection to the dabmux RC
 
-if len(sys.argv) == 1:
-    sock = connect()
-    values = json.loads(do_transaction("values", sock))['values']
+    returns: the socket"""
+
+    sock = zmq.Socket(ctx, zmq.REQ)
+    sock.set(zmq.LINGER, 5)
+    sock.connect("tcp://localhost:12722")
+
+    try:
+        ping_answer = do_multipart_transaction([b"ping"], sock)
+
+        if not ping_answer == [b"ok"]:
+            sys.stderr.write("Wrong answer to ping\n")
+            sys.exit(1)
+    except RCException as e:
+        print("connect failed because: {}".format(e))
+        sys.exit(1)
 
+    return sock
+
+def handle_re(graph_name, re, rc_value, group_number=1):
+    match = re.search(rc_value)
+    if match:
+        return "{}.value {}\n".format(graph_name, match.group(group_number))
+    else:
+        return "{}.value U\n".format(graph_name)
+
+if len(sys.argv) == 1:
     munin_values = ""
+
+    sock_rc = connect_to_rc()
+    clocktai_expiry = get_rc_value("clocktai", "expiry", sock_rc)
+    re_clocktai_expiry = re.compile(r"(\d+)", re.X)
+    munin_values += "multigraph clocktai_expiry\n"
+    munin_values += handle_re("expiry", re_clocktai_expiry, clocktai_expiry)
+
+    sock_stats = connect_to_stats()
+    values = json.loads(do_transaction("values", sock_stats))['values']
+
     for ident in values:
         v = values[ident]['inputstat']
 
@@ -147,6 +234,7 @@ if len(sys.argv) == 1:
 
         if 'state' in v:
             # If ODR-DabMux is v1.3.1-3 or older, it doesn't export state
+            re_state = re.compile(r"\w+ \((\d+)\)")
             match = re_state.match(v['state'])
             if match:
                 munin_values += "multigraph state_{ident}\n".format(ident=ident_)
@@ -157,9 +245,9 @@ if len(sys.argv) == 1:
     print(munin_values)
 
 elif len(sys.argv) == 2 and sys.argv[1] == "config":
-    sock = connect()
+    sock_stats = connect_to_stats()
 
-    config = json.loads(do_transaction("config", sock))
+    config = json.loads(do_transaction("config", sock_stats))
 
     munin_config = config_top
 
diff --git a/src/ClockTAI.cpp b/src/ClockTAI.cpp
index 8c01127..c481841 100644
--- a/src/ClockTAI.cpp
+++ b/src/ClockTAI.cpp
@@ -76,6 +76,12 @@ static const char* tai_tz_url =
 
 static const char* tai_ietf_cache_file = "/tmp/odr-dabmux-leap-seconds.cache";
 
+ClockTAI::ClockTAI() :
+    RemoteControllable("clocktai")
+{
+    RC_ADD_PARAMETER(expiry, "Number of seconds until TAI Bulletin expires");
+}
+
 int ClockTAI::get_valid_offset()
 {
     int offset = 0;
@@ -152,6 +158,8 @@ int ClockTAI::get_offset()
     using namespace std::chrono;
     auto time_now = system_clock::now();
 
+    std::unique_lock<std::mutex> lock(m_data_mutex);
+
     if (not m_offset_valid) {
 #ifdef TEST
         m_offset = 37; // Valid in early 2017
@@ -353,6 +361,11 @@ void ClockTAI::update_cache(const char* cache_filename)
 
 bool ClockTAI::bulletin_is_valid()
 {
+    return bulletin_expiry_delay() > 0;
+}
+
+int64_t ClockTAI::bulletin_expiry_delay() const
+{
     // The bulletin contains one line that specifies an expiration date
     // in NTP time. If that point in time is in the future, we consider
     // the bulletin valid.
@@ -380,10 +393,12 @@ bool ClockTAI::bulletin_is_valid()
             const int64_t expiry_unix =
                 std::atol(expiry_data_str.c_str()) - ntp_unix_offset;
 
-            return expiry_unix > now;
+            if (expiry_unix > now) {
+                return expiry_unix - now;
+            }
         }
     }
-    return false;
+    return -1;
 }
 
 void ClockTAI::download_tai_utc_bulletin(const char* url)
@@ -418,6 +433,36 @@ void ClockTAI::download_tai_utc_bulletin(const char* url)
 #endif // HAVE_CURL
 }
 
+void ClockTAI::set_parameter(const string& parameter, const string& value)
+{
+    if (parameter == "expiry") {
+        throw ParameterError("Parameter '" + parameter +
+            "' is not read-only in controllable " + get_rc_name());
+    }
+    else {
+        throw ParameterError("Parameter '" + parameter +
+            "' is not exported by controllable " + get_rc_name());
+    }
+}
+
+const string ClockTAI::get_parameter(const string& parameter) const
+{
+    if (parameter == "expiry") {
+        std::unique_lock<std::mutex> lock(m_data_mutex);
+        int64_t expiry = bulletin_expiry_delay();
+        if (expiry > 0) {
+            return to_string(expiry);
+        }
+        else {
+            return "Bulletin expired or invalid!";
+        }
+    }
+    else {
+        throw ParameterError("Parameter '" + parameter +
+            "' is not exported by controllable " + get_rc_name());
+    }
+}
+
 #if 0
 // Example testing code
 void debug_tai_clk()
diff --git a/src/ClockTAI.h b/src/ClockTAI.h
index 4ee4072..ac4978c 100644
--- a/src/ClockTAI.h
+++ b/src/ClockTAI.h
@@ -39,14 +39,18 @@
 #include <sstream>
 #include <chrono>
 #include <future>
+#include <mutex>
+#include "RemoteControl.h"
 
 // EDI needs to know UTC-TAI, but doesn't need the CLOCK_TAI to be set.
 // We can keep this code, maybe for future use
 #define SUPPORT_SETTING_CLOCK_TAI 0
 
 /* Loads, parses and represents TAI-UTC offset information from the IETF bulletin */
-class ClockTAI {
+class ClockTAI : public RemoteControllable {
     public:
+        ClockTAI();
+
         // Fetch the bulletin from the IETF website and return the current
         // TAI-UTC offset.
         // Throws runtime_error on failure.
@@ -70,11 +74,14 @@ class ClockTAI {
         // Download of new bulletin is done asynchronously
         std::future<int> m_offset_future;
 
+        // Protect all data members, as RC functions are in another thread
+        mutable std::mutex m_data_mutex;
+
         // The currently used TAI-UTC offset
         int m_offset;
         int m_offset_valid = false;
 
-        std::stringstream m_bulletin;
+        mutable std::stringstream m_bulletin;
         std::chrono::system_clock::time_point m_bulletin_download_time;
 
         // Load bulletin into m_bulletin from the cache file
@@ -87,6 +94,10 @@ class ClockTAI {
         // true if the bulletin is valid.
         bool bulletin_is_valid(void);
 
+        // In how much time will the bulletin expire?
+        // returns a value in seconds, or -1 if it is expired or invalid
+        int64_t bulletin_expiry_delay(void) const;
+
         // Load bulletin into m_bulletin from the URL
         void download_tai_utc_bulletin(const char* url);
 
@@ -99,5 +110,12 @@ class ClockTAI {
         // static callback wrapper for cURL
         static size_t fill_bulletin_cb(
                 char *ptr, size_t size, size_t nmemb, void *ctx);
+
+        /* Remote control */
+        virtual void set_parameter(const std::string& parameter,
+               const std::string& value);
+
+        /* Getting a parameter always returns a string. */
+        virtual const std::string get_parameter(const std::string& parameter) const;
 };
 
diff --git a/src/DabMultiplexer.cpp b/src/DabMultiplexer.cpp
index 8265b34..f452685 100644
--- a/src/DabMultiplexer.cpp
+++ b/src/DabMultiplexer.cpp
@@ -81,6 +81,13 @@ DabMultiplexer::DabMultiplexer(
 {
     RC_ADD_PARAMETER(frames,
             "Show number of frames generated [read-only]");
+
+    rcs.enrol(&m_clock_tai);
+}
+
+DabMultiplexer::~DabMultiplexer()
+{
+    rcs.remove_controllable(&m_clock_tai);
 }
 
 void DabMultiplexer::set_edi_config(const edi_configuration_t& new_edi_conf)
diff --git a/src/DabMultiplexer.h b/src/DabMultiplexer.h
index 80b3ab9..499e023 100644
--- a/src/DabMultiplexer.h
+++ b/src/DabMultiplexer.h
@@ -54,8 +54,11 @@
 
 class DabMultiplexer : public RemoteControllable {
     public:
-        DabMultiplexer(
-                boost::property_tree::ptree pt);
+        DabMultiplexer(boost::property_tree::ptree pt);
+        DabMultiplexer(const DabMultiplexer& other) = delete;
+        DabMultiplexer& operator=(const DabMultiplexer& other) = delete;
+        ~DabMultiplexer();
+
         void prepare(bool require_tai_clock);
 
         unsigned long getCurrentFrame() { return currentFrame; }