status is back online + ZMQ async stats

4 files changed, 335 insertions, 155 deletions

diff --git a/scripts/automation/trex_control_plane/client/trex_async_client.py b/scripts/automation/trex_control_plane/client/trex_async_client.py
new file mode 100644
index 00000000..ea716057
--- /dev/null
+++ b/scripts/automation/trex_control_plane/client/trex_async_client.py
@@ -0,0 +1,172 @@
+#!/router/bin/python
+
+try:
+    # support import for Python 2
+    import outer_packages
+except ImportError:
+    # support import for Python 3
+    import client.outer_packages
+from client_utils.jsonrpc_client import JsonRpcClient, BatchMessage
+
+import json
+import threading
+import time
+import zmq
+import re
+
+from common.trex_stats import *
+from common.trex_streams import *
+
+# basic async stats class
+class TrexAsyncStats(object):
+    def __init__ (self):
+        self.ref_point = None
+
+    def update (self, snapshot):
+
+        self.current = snapshot
+
+        if self.ref_point == None:
+            self.ref_point = self.current
+        
+
+    def get (self, field):
+
+        if not field in self.current:
+            return None
+
+        return self.current[field]
+
+    def get_rel (self, field):
+        if not field in self.current:
+            return None
+
+        return self.current[field] - self.ref_point[field]
+
+
+# describes the general stats provided by TRex
+class TrexAsyncStatsGeneral(TrexAsyncStats):
+    def __init__ (self):
+        super(TrexAsyncStatsGeneral, self).__init__()
+
+
+# per port stats
+class TrexAsyncStatsPort(TrexAsyncStats):
+    def __init__ (self):
+        super(TrexAsyncStatsPort, self).__init__()
+
+
+# stats manager
+class TrexAsyncStatsManager():
+    def __init__ (self, port_count):
+        self.port_count = port_count
+
+        self.general_stats = TrexAsyncStatsGeneral()
+        self.port_stats = {}
+
+    def get_general_stats (self):
+        return self.general_stats
+
+    def get_port_stats (self, port_id):
+
+        if not port_id in self.port_stats:
+            return None
+
+        return self.port_stats[port_id]
+
+
+    def update (self, snapshot):
+
+        if snapshot['name'] == 'trex-global':
+            self.__handle_snapshot(snapshot['data'])
+        else:
+            # for now ignore the rest
+            return
+
+    def __handle_snapshot (self, snapshot):
+
+        general_stats = {}
+        port_stats = {}
+
+        # filter the values per port and general
+        for key, value in snapshot.iteritems():
+            
+            # match a pattern of ports
+            m = re.search('.*\-([0-8])', key)
+            if m:
+                port_id = m.group(1)
+
+                if not port_id in port_stats:
+                    port_stats[port_id] = {}
+
+                port_stats[port_id][key] = value
+
+            else:
+                # no port match - general stats
+                general_stats[key] = value
+
+        # update the general object with the snapshot
+        self.general_stats.update(general_stats)
+
+        # update all ports
+        for port_id, data in port_stats.iteritems():
+
+            if not port_id in self.port_stats:
+                self.port_stats[port_id] = TrexAsyncStatsPort()
+
+            self.port_stats[port_id].update(data)
+
+
+
+
+
+class TrexAsyncClient():
+    def __init__ (self, port):
+
+        self.port = port
+
+        self.raw_snapshot = {}
+
+        self.stats = TrexAsyncStatsManager(1)
+
+        self.active = True
+        self.t = threading.Thread(target = self._run)
+
+        # kill this thread on exit and don't add it to the join list
+        self.t.setDaemon(True)
+        self.t.start()
+
+
+    def _run (self):
+
+        #  Socket to talk to server
+        self.context = zmq.Context()
+        self.socket = self.context.socket(zmq.SUB)
+
+        self.c = "tcp://localhost:{0}".format(self.port)
+        print "Connecting To ZMQ Publisher At {0}".format(self.c)
+
+        self.socket.connect(self.c)
+        self.socket.setsockopt(zmq.SUBSCRIBE, '')
+
+        while self.active:
+            msg = json.loads(self.socket.recv_string())
+
+            key = msg['name']
+            self.raw_snapshot[key] = msg['data']
+
+            self.stats.update(msg)
+
+
+    def get_stats (self):
+        return self.stats
+
+    def get_raw_snapshot (self):
+        #return str(self.stats.global_stats.get('m_total_tx_bytes')) + " / " + str(self.stats.global_stats.get_rel('m_total_tx_bytes'))
+        return self.raw_snapshot
+
+
+    def stop (self):
+        self.active = False
+        self.t.join()
+
diff --git a/scripts/automation/trex_control_plane/client/trex_stateless_client.py b/scripts/automation/trex_control_plane/client/trex_stateless_client.py
index 8231fe33..c180e0d1 100755
--- a/scripts/automation/trex_control_plane/client/trex_stateless_client.py
+++ b/scripts/automation/trex_control_plane/client/trex_stateless_client.py
@@ -13,6 +13,8 @@ from common.trex_stats import *
 from common.trex_streams import *
 from collections import namedtuple
 
+from trex_async_client import TrexAsyncClient
+
 RpcCmdData = namedtuple('RpcCmdData', ['method', 'params'])
 
 class RpcResponseStatus(namedtuple('RpcResponseStatus', ['success', 'id', 'msg'])):
@@ -27,10 +29,10 @@ class RpcResponseStatus(namedtuple('RpcResponseStatus', ['success', 'id', 'msg']
 class CTRexStatelessClient(object):
     """docstring for CTRexStatelessClient"""
 
-    def __init__(self, username, server="localhost", port=5050, virtual=False):
+    def __init__(self, username, server="localhost", sync_port=5050, async_port = 4500, virtual=False):
         super(CTRexStatelessClient, self).__init__()
         self.user = username
-        self.comm_link = CTRexStatelessClient.CCommLink(server, port, virtual)
+        self.comm_link = CTRexStatelessClient.CCommLink(server, sync_port, virtual)
         self.verbose = False
         self._conn_handler = {}
         self._active_ports = set()
@@ -39,6 +41,9 @@ class CTRexStatelessClient(object):
         self._server_version = None
         self.__err_log = None
 
+        self._async_client = TrexAsyncClient(async_port)
+
+
     # ----- decorator methods ----- #
     def force_status(owned=True, active_and_owned=False):
         def wrapper(func):
@@ -100,6 +105,12 @@ class CTRexStatelessClient(object):
             return rc, err
         return self._init_sync()
 
+    def get_stats_async (self):
+        return self._async_client.get_stats()
+
+    def get_connection_port (self):
+        return self.comm_link.port
+
     def disconnect(self):
         return self.comm_link.disconnect()
 
@@ -300,10 +311,10 @@ class CTRexStatelessClient(object):
                                                        self.transmit(command.method, command.params),
                                                        self.ack_success_test)
 
-    def get_global_stats(self):
-        command = RpcCmdData("get_global_stats", {})
-        return self._handle_get_global_stats_response(command, self.transmit(command.method, command.params))
-        # return self.transmit("get_global_stats")
+#    def get_global_stats(self):
+#        command = RpcCmdData("get_global_stats", {})
+#        return self._handle_get_global_stats_response(command, self.transmit(command.method, command.params))
+#        # return self.transmit("get_global_stats")
 
     @force_status(owned=True, active_and_owned=True)
     def get_port_stats(self, port_id=None):
diff --git a/scripts/automation/trex_control_plane/console/trex_console.py b/scripts/automation/trex_control_plane/console/trex_console.py
index ec23eb0c..bd79cb42 100755
--- a/scripts/automation/trex_control_plane/console/trex_console.py
+++ b/scripts/automation/trex_control_plane/console/trex_console.py
@@ -34,7 +34,6 @@ from common.text_opts import *
 from client_utils.general_utils import user_input
 
 
-from client_utils.jsonrpc_client import TrexStatelessClient
 import trex_status
 from collections import namedtuple
 
@@ -177,6 +176,7 @@ class TRexConsole(cmd.Cmd):
         dotext = 'do_'+text
         return [a[3:]+' ' for a in self.get_names() if a.startswith(dotext)]
 
+
     # set verbose on / off
     def do_verbose (self, line):
         '''Shows or set verbose mode\n'''
@@ -770,6 +770,10 @@ def setParserOptions():
                         default = 5050,
                         type = int)
 
+    parser.add_argument("-z", "--pub", help = "TRex Async Publisher Port  [default is 4500]\n",
+                        default = 4500,
+                        type = int)
+
     parser.add_argument("-u", "--user", help = "User Name  [default is random generated]\n",
                         default = 'user_' + ''.join(random.choice(string.digits) for _ in range(5)),
                         type = str)
@@ -782,11 +786,10 @@ def setParserOptions():
 
 def main():
     parser = setParserOptions()
-    options = parser.parse_args()#sys.argv[1:])
+    options = parser.parse_args()
 
     # Stateless client connection
-    # stateless_client = TrexStatelessClient(options.server, options.port, options.user)
-    stateless_client = CTRexStatelessClient(options.user, options.server, options.port)
+    stateless_client = CTRexStatelessClient(options.user, options.server, options.port, options.pub)
 
     # console
     try:
diff --git a/scripts/automation/trex_control_plane/console/trex_status.py b/scripts/automation/trex_control_plane/console/trex_status.py
index 2c5a648f..4cd07358 100644
--- a/scripts/automation/trex_control_plane/console/trex_status.py
+++ b/scripts/automation/trex_control_plane/console/trex_status.py
@@ -11,6 +11,8 @@ import datetime
 
 g_curses_active = False
 
+################### utils #################
+
 # simple percetange show
 def percentage (a, total):
     x = int ((float(a) / total) * 100)
@@ -18,15 +20,25 @@ def percentage (a, total):
 
 # simple float to human readable
 def float_to_human_readable (size, suffix = "bps"):
-    for unit in ['','K','M','G']:
-        if abs(size) < 1024.0:
-            return "%3.1f %s%s" % (size, unit, suffix)
-        size /= 1024.0
+    for unit in ['','K','M','G','T']:
+        if abs(size) < 1000.0:
+            return "%3.2f %s%s" % (size, unit, suffix)
+        size /= 1000.0
     return "NaN"
 
+
+################### panels #################
+
 # panel object
 class TrexStatusPanel(object):
-    def __init__ (self, h, l, y, x, headline):
+    def __init__ (self, h, l, y, x, headline, status_obj):
+
+        self.status_obj = status_obj
+
+        self.log = status_obj.log
+        self.stateless_client = status_obj.stateless_client
+        self.general_stats = status_obj.general_stats
+
         self.h = h
         self.l = l
         self.y = y
@@ -53,64 +65,26 @@ class TrexStatusPanel(object):
         return self.win
 
 
-# total stats (ports + global)
-class Stats():
-    def __init__ (self, rpc_client, port_list, interval = 100):
-
-        self.rpc_client = rpc_client
-
-        self.port_list = port_list
-        self.port_stats = {}
-
-        self.interval = interval
-        self.delay_count = 0
-
-    def get_port_stats (self, port_id):
-        if self.port_stats.get(port_id):
-            return self.port_stats[port_id]
-        else:
-            return None
-
-    def query_sync (self):
-        self.delay_count += 1
-        if self.delay_count < self.interval:
-            return
-
-        self.delay_count = 0
-
-        # query global stats
-
-        # query port stats
-
-        rc, resp_list = self.rpc_client.get_port_stats(self.port_list)
-        if not rc:
-            return
-
-        for i, rc in enumerate(resp_list):
-            if rc[0]:
-                self.port_stats[self.port_list[i]] = rc[1]
-
-
 # various kinds of panels
 
 # Server Info Panel
 class ServerInfoPanel(TrexStatusPanel):
     def __init__ (self, h, l, y, x, status_obj):
 
-        super(ServerInfoPanel, self).__init__(h, l, y ,x ,"Server Info:")
-
-        self.status_obj = status_obj
+        super(ServerInfoPanel, self).__init__(h, l, y ,x ,"Server Info:", status_obj)
 
     def draw (self):
 
-        if self.status_obj.server_version == None:
+        if not self.status_obj.server_version :
             return
 
-        self.clear()
+        if not self.status_obj.server_sys_info:
+            return
 
-        connection_details = self.status_obj.rpc_client.get_connection_details()
 
-        self.getwin().addstr(3, 2, "{:<30} {:30}".format("Server:",self.status_obj.server_sys_info["hostname"] + ":" + str(connection_details['port'])))
+        self.clear()
+
+        self.getwin().addstr(3, 2, "{:<30} {:30}".format("Server:",self.status_obj.server_sys_info["hostname"] + ":" + str(self.stateless_client.get_connection_port())))
         self.getwin().addstr(4, 2, "{:<30} {:30}".format("Version:", self.status_obj.server_version["version"]))
         self.getwin().addstr(5, 2, "{:<30} {:30}".format("Build:", 
                                                                     self.status_obj.server_version["build_date"] + " @ " + 
@@ -123,7 +97,7 @@ class ServerInfoPanel(TrexStatusPanel):
 
         self.getwin().addstr(9, 2, "{:<30} {:<30}".format("Ports Count:", self.status_obj.server_sys_info["port_count"]))
 
-        ports_owned = " ".join(str(x) for x in self.status_obj.rpc_client.get_owned_ports())
+        ports_owned = " ".join(str(x) for x in self.status_obj.owned_ports)
 
         if not ports_owned:
             ports_owned = "None"
@@ -134,26 +108,39 @@ class ServerInfoPanel(TrexStatusPanel):
 class GeneralInfoPanel(TrexStatusPanel):
     def __init__ (self, h, l, y, x, status_obj):
 
-        super(GeneralInfoPanel, self).__init__(h, l, y ,x ,"General Info:")
-
-        self.status_obj = status_obj
+        super(GeneralInfoPanel, self).__init__(h, l, y ,x ,"General Info:", status_obj)
 
     def draw (self):
-        pass
+        self.clear()
+
+        self.getwin().addstr(3, 2, "{:<30} {:0.2f} %".format("CPU util.:", self.general_stats.get("m_cpu_util")))
+
+        self.getwin().addstr(5, 2, "{:<30} {:} / {:}".format("Total Tx. rate:",
+                                                               float_to_human_readable(self.general_stats.get("m_tx_bps")),
+                                                               float_to_human_readable(self.general_stats.get("m_tx_pps"), suffix = "pps")))
+
+        # missing RX field
+        #self.getwin().addstr(5, 2, "{:<30} {:} / {:}".format("Total Rx. rate:",
+        #                                                       float_to_human_readable(self.general_stats.get("m_rx_bps")),
+        #                                                       float_to_human_readable(self.general_stats.get("m_rx_pps"), suffix = "pps")))
+
+        self.getwin().addstr(7, 2, "{:<30} {:} / {:}".format("Total Tx:",
+                                                             float_to_human_readable(self.general_stats.get_rel("m_total_tx_bytes"), suffix = "B"),
+                                                             float_to_human_readable(self.general_stats.get_rel("m_total_tx_pkts"), suffix = "pkts")))
 
 # all ports stats
 class PortsStatsPanel(TrexStatusPanel):
     def __init__ (self, h, l, y, x, status_obj):
 
-        super(PortsStatsPanel, self).__init__(h, l, y ,x ,"Trex Ports:")
+        super(PortsStatsPanel, self).__init__(h, l, y ,x ,"Trex Ports:", status_obj)
 
-        self.status_obj = status_obj
 
     def draw (self):
 
         self.clear()
+        return
 
-        owned_ports = self.status_obj.rpc_client.get_owned_ports()
+        owned_ports = self.status_obj.owned_ports
         if not owned_ports:
             self.getwin().addstr(3, 2, "No Owned Ports - Please Acquire One Or More Ports")
             return
@@ -193,33 +180,23 @@ class PortsStatsPanel(TrexStatusPanel):
 class ControlPanel(TrexStatusPanel):
     def __init__ (self, h, l, y, x, status_obj):
 
-        super(ControlPanel, self).__init__(h, l, y, x, "")
+        super(ControlPanel, self).__init__(h, l, y, x, "", status_obj)
 
-        self.status_obj = status_obj
 
     def draw (self):
         self.clear()
 
         self.getwin().addstr(1, 2, "'g' - general, '0-{0}' - specific port, 'f' - freeze, 'c' - clear stats, 'p' - ping server, 'q' - quit"
-                             .format(self.status_obj.rpc_client.get_port_count() - 1))
+                             .format(self.status_obj.stateless_client.get_port_count() - 1))
 
-        index = 3
-
-        cut = len(self.status_obj.log) - 4
-        if cut < 0:
-            cut = 0
-
-        for l in self.status_obj.log[cut:]:
-            self.getwin().addstr(index, 2, l)
-            index += 1
+        self.log.draw(self.getwin(), 2, 3)
 
 # specific ports panels
 class SinglePortPanel(TrexStatusPanel):
     def __init__ (self, h, l, y, x, status_obj, port_id):
 
-        super(SinglePortPanel, self).__init__(h, l, y, x, "Port {0}".format(port_id))
+        super(SinglePortPanel, self).__init__(h, l, y, x, "Port {0}".format(port_id), status_obj)
 
-        self.status_obj = status_obj
         self.port_id = port_id
 
     def draw (self):
@@ -227,7 +204,7 @@ class SinglePortPanel(TrexStatusPanel):
 
         self.clear()
 
-        if not self.port_id in self.status_obj.rpc_client.get_owned_ports():
+        if not self.port_id in self.status_obj.stateless_client.get_owned_ports():
              self.getwin().addstr(y, 2, "Port {0} is not owned by you, please acquire the port for more info".format(self.port_id))
              return
 
@@ -292,96 +269,119 @@ class SinglePortPanel(TrexStatusPanel):
 
         y += 2
 
-# status object
-class TrexStatus():
-    def __init__ (self, stdscr, rpc_client):
-        self.stdscr = stdscr
+################### main objects #################
+
+# status log
+class TrexStatusLog():
+    def __init__ (self):
         self.log = []
-        self.rpc_client = rpc_client
 
-        self.snapshot = self.rpc_client.snapshot()
+    def add_event (self, msg):
+        self.log.append("[{0}] {1}".format(str(datetime.datetime.now().time()), msg))
 
-        # fetch server info
-        self.get_server_info()
+    def draw (self, window, x, y, max_lines = 4):
+        index = y
+
+        cut = len(self.log) - max_lines
+        if cut < 0:
+            cut = 0
+
+        for msg in self.log[cut:]:
+            window.addstr(index, x, msg)
+            index += 1
+
+# status commands
+class TrexStatusCommands():
+    def __init__ (self, status_object):
+
+        self.status_object = status_object
 
-        # create stats objects
-        self.stats = Stats(rpc_client, self.rpc_client.get_owned_ports())
+        self.stateless_client = status_object.stateless_client
+        self.log = self.status_object.log
 
-        # register actions
         self.actions = {}
-        self.actions[ord('q')] = self.action_quit
-        self.actions[ord('p')] = self.action_ping
-        self.actions[ord('f')] = self.action_freeze 
+        self.actions[ord('q')] = self._quit
+        self.actions[ord('p')] = self._ping
+        self.actions[ord('f')] = self._freeze 
 
-        self.actions[ord('g')] = self.action_show_ports_stats
+        self.actions[ord('g')] = self._show_ports_stats
 
-        for port_id in xrange(0, self.rpc_client.get_port_count()):
-            self.actions[ord('0') + port_id] = self.action_show_port_generator(port_id)
+        # register all the available ports shortcuts
+        for port_id in xrange(0, self.stateless_client.get_port_count()):
+            self.actions[ord('0') + port_id] = self._show_port_generator(port_id)
+
+
+    # handle a key pressed
+    def handle (self, ch):
+        if ch in self.actions:
+            return self.actions[ch]()
+        else:
+            self.log.add_event("Unknown key pressed, please see legend")
+            return True
+
+    # show all ports
+    def _show_ports_stats (self):
+        self.log.add_event("Switching to all ports view")
+        self.status_object.stats_panel = self.status_object.ports_stats_panel
 
-        
-    # all ports stats
-    def action_show_ports_stats (self):
-        self.add_log_event("Switching to all ports view")
-        self.stats_panel = self.ports_stats_panel
-        
         return True
 
-    # function generator for different ports requests
-    def action_show_port_generator (self, port_id):
-        def action_show_port():
-            self.add_log_event("Switching panel to port {0}".format(port_id))
-            self.stats_panel = self.ports_panels[port_id]
+
+     # function generator for different ports requests
+    def _show_port_generator (self, port_id):
+        def _show_port():
+            self.log.add_event("Switching panel to port {0}".format(port_id))
+            self.status_object.stats_panel = self.status_object.ports_panels[port_id]
 
             return True
 
-        return action_show_port
+        return _show_port
 
-    def action_freeze (self):
-        self.update_active = not self.update_active
-        self.add_log_event("Update continued" if self.update_active else "Update stopped")
+    def _freeze (self):
+        self.status_object.update_active = not self.status_object.update_active
+        self.log.add_event("Update continued" if self.status_object.update_active else "Update stopped")
 
         return True
 
-    def action_quit(self):
+    def _quit(self):
         return False
 
-    def action_ping (self):
-        self.add_log_event("Pinging RPC server")
+    def _ping (self):
+        self.log.add_event("Pinging RPC server")
 
-        rc, msg = self.rpc_client.ping_rpc_server()
+        rc, msg = self.stateless_client.ping()
         if rc:
-            self.add_log_event("Server replied: '{0}'".format(msg))
+            self.log.add_event("Server replied: '{0}'".format(msg))
         else:
-            self.add_log_event("Failed to get reply")
+            self.log.add_event("Failed to get reply")
 
         return True
 
-    def get_server_info (self):
-
-        self.server_version = self.rpc_client.get_rpc_server_version()
-        self.server_sys_info = self.rpc_client.get_system_info()
-
-
-    def add_log_event (self, msg):
-        self.log.append("[{0}] {1}".format(str(datetime.datetime.now().time()), msg))
-
-    # control panel
-    def update_control (self):
-        self.control_panel.clear()
+# status object
+# 
+#
+#
+class TrexStatus():
+    def __init__ (self, stdscr, stateless_client):
+        self.stdscr = stdscr
 
-        self.control_panel.getwin().addstr(1, 2, "'g' - general, '0-{0}' - specific port, 'f' - freeze, 'c' - clear stats, 'p' - ping server, 'q' - quit"
-                                           .format(self.rpc_client.get_port_count() - 1))
+        self.stateless_client = stateless_client
+        self.general_stats = stateless_client.get_stats_async().get_general_stats()
 
-        index = 3
+        # fetch server info
+        rc, self.server_sys_info = self.stateless_client.get_system_info()
+        if not rc:
+            return
 
-        cut = len(self.log) - 4
-        if cut < 0:
-            cut = 0
+        rc, self.server_version = self.stateless_client.get_version()
+        if not rc:
+            return
 
-        for l in self.log[cut:]:
-            self.control_panel.getwin().addstr(index, 2, l)
-            index += 1
+        self.owned_ports = self.stateless_client.get_acquired_ports()
 
+        self.log = TrexStatusLog()
+        self.cmds = TrexStatusCommands(self)
+  
     def generate_layout (self):
         self.max_y = self.stdscr.getmaxyx()[0]
         self.max_x = self.stdscr.getmaxyx()[1]
@@ -394,7 +394,7 @@ class TrexStatus():
         self.ports_stats_panel    = PortsStatsPanel(int(self.max_y * 0.8), self.max_x / 2, 0, 0, self)
 
         self.ports_panels = {}
-        for i in xrange(0, self.rpc_client.get_port_count()):
+        for i in xrange(0, self.stateless_client.get_port_count()):
             self.ports_panels[i] = SinglePortPanel(int(self.max_y * 0.8), self.max_x / 2, 0, 0, self, i)
 
         # at start time we point to the main one 
@@ -411,14 +411,8 @@ class TrexStatus():
         # no key , continue
         if ch == curses.ERR:
             return True
-        
-        # check for registered function        
-        if ch in self.actions:
-            return self.actions[ch]()
-        else:
-            self.add_log_event("Unknown key pressed, please see legend")
-
-        return True
+    
+        return self.cmds.handle(ch)
 
     # main run entry point
     def run (self):
@@ -454,14 +448,14 @@ class TrexStatus():
             sleep(0.01)
 
 
-def show_trex_status_internal (stdscr, rpc_client):
-    trex_status = TrexStatus(stdscr, rpc_client)
+def show_trex_status_internal (stdscr, stateless_client):
+    trex_status = TrexStatus(stdscr, stateless_client)
     trex_status.run()
 
-def show_trex_status (rpc_client):
+def show_trex_status (stateless_client):
 
     try:
-        curses.wrapper(show_trex_status_internal, rpc_client)
+        curses.wrapper(show_trex_status_internal, stateless_client)
     except KeyboardInterrupt:
         curses.endwin()