PAL: Process sh-run from telemetry
[csit.git] / resources / tools / presentation / input_data_parser.py
index 2cc2447..1e25138 100644 (file)
@@ -295,7 +295,7 @@ class ExecutionChecker(ResultVisitor):
 
     REGEX_TC_PAPI_CLI = re.compile(r'.*\((\d+.\d+.\d+.\d+.) - (.*)\)')
 
-    def __init__(self, metadata, mapping, ignore):
+    def __init__(self, metadata, mapping, ignore, for_output):
         """Initialisation.
 
         :param metadata: Key-value pairs to be included in "metadata" part of
@@ -303,9 +303,11 @@ class ExecutionChecker(ResultVisitor):
         :param mapping: Mapping of the old names of test cases to the new
             (actual) one.
         :param ignore: List of TCs to be ignored.
+        :param for_output: Output to be generated from downloaded data.
         :type metadata: dict
         :type mapping: dict
         :type ignore: list
+        :type for_output: str
         """
 
         # Type of message to parse out from the test messages
@@ -326,6 +328,8 @@ class ExecutionChecker(ResultVisitor):
         # Ignore list
         self._ignore = ignore
 
+        self._for_output = for_output
+
         # Number of PAPI History messages found:
         # 0 - no message
         # 1 - PAPI History of DUT1
@@ -333,6 +337,8 @@ class ExecutionChecker(ResultVisitor):
         self._conf_history_lookup_nr = 0
 
         self._sh_run_counter = 0
+        self._telemetry_kw_counter = 0
+        self._telemetry_msg_counter = 0
 
         # Test ID of currently processed test- the lowercase full path to the
         # test
@@ -357,7 +363,8 @@ class ExecutionChecker(ResultVisitor):
             u"dpdk-version": self._get_dpdk_version,
             u"teardown-papi-history": self._get_papi_history,
             u"test-show-runtime": self._get_show_run,
-            u"testbed": self._get_testbed
+            u"testbed": self._get_testbed,
+            u"test-telemetry": self._get_telemetry
         }
 
     @property
@@ -669,54 +676,80 @@ class ExecutionChecker(ResultVisitor):
         except (AttributeError, IndexError):
             sock = u""
 
-        runtime = loads(str(msg.message).replace(u' ', u'').replace(u'\n', u'').
-                        replace(u"'", u'"').replace(u'b"', u'"').
-                        replace(u'u"', u'"').split(u":", 1)[1])
-
-        try:
-            threads_nr = len(runtime[0][u"clocks"])
-        except (IndexError, KeyError):
-            return
-
         dut = u"dut{nr}".format(
             nr=len(self._data[u'tests'][self._test_id][u'show-run'].keys()) + 1)
 
-        oper = {
-            u"host": host,
-            u"socket": sock,
-            # Needed for json converter, enable when 'threads' is gone.
-            # u"runtime": runtime,
-            u"threads": OrderedDict({idx: list() for idx in range(threads_nr)})
-        }
+        self._data[u'tests'][self._test_id][u'show-run'][dut] = \
+            copy.copy(
+                {
+                    u"host": host,
+                    u"socket": sock,
+                    u"runtime": str(msg.message).replace(u' ', u'').
+                                replace(u'\n', u'').replace(u"'", u'"').
+                                replace(u'b"', u'"').replace(u'u"', u'"').
+                                split(u":", 1)[1]
+                }
+            )
 
-        for item in runtime:
-            for idx in range(threads_nr):
-                if item[u"vectors"][idx] > 0:
-                    clocks = item[u"clocks"][idx] / item[u"vectors"][idx]
-                elif item[u"calls"][idx] > 0:
-                    clocks = item[u"clocks"][idx] / item[u"calls"][idx]
-                elif item[u"suspends"][idx] > 0:
-                    clocks = item[u"clocks"][idx] / item[u"suspends"][idx]
-                else:
-                    clocks = 0.0
+    def _get_telemetry(self, msg):
+        """Called when extraction of VPP telemetry data is required.
 
-                if item[u"calls"][idx] > 0:
-                    vectors_call = item[u"vectors"][idx] / item[u"calls"][idx]
-                else:
-                    vectors_call = 0.0
+        :param msg: Message to process.
+        :type msg: Message
+        :returns: Nothing.
+        """
 
-                if int(item[u"calls"][idx]) + int(item[u"vectors"][idx]) + \
-                        int(item[u"suspends"][idx]):
-                    oper[u"threads"][idx].append([
-                        item[u"name"],
-                        item[u"calls"][idx],
-                        item[u"vectors"][idx],
-                        item[u"suspends"][idx],
-                        clocks,
-                        vectors_call
-                    ])
+        if self._telemetry_kw_counter > 1:
+            return
+        if not msg.message.count(u"vpp_runtime_calls"):
+            return
 
-        self._data[u'tests'][self._test_id][u'show-run'][dut] = copy.copy(oper)
+        if u"telemetry-show-run" not in \
+                self._data[u"tests"][self._test_id].keys():
+            self._data[u"tests"][self._test_id][u"telemetry-show-run"] = dict()
+
+        self._telemetry_msg_counter += 1
+        dut = f"dut{self._telemetry_msg_counter}"
+        runtime = {
+            u"source_type": u"node",
+            u"source_id": dut,
+            u"msg_type": u"metric",
+            u"log_level": u"INFO",
+            u"timestamp": msg.timestamp,
+            u"msg": u"show_runtime",
+            u"host": dut,  # No info, should be host IP
+            u"socket": u"",  # No info
+            u"data": list()
+        }
+        for line in msg.message.splitlines():
+            if not line.startswith(u"vpp_runtime_"):
+                continue
+            try:
+                params, value = line.rsplit(u" ", maxsplit=2)[:-1]
+                cut = params.index(u"{")
+                name = params[:cut].split(u"_", maxsplit=2)[-1]
+                labels = eval(
+                    u"dict" + params[cut:].replace('{', '(').replace('}', ')')
+                )
+                labels[u"graph_node"] = labels.pop(u"name")
+                runtime[u"data"].append(
+                    {
+                        u"name": name,
+                        u"value": value,
+                        u"labels": labels
+                    }
+                )
+            except (TypeError, ValueError, IndexError):
+                continue
+
+        self._data[u'tests'][self._test_id][u'telemetry-show-run'][dut] = \
+            copy.copy(
+                {
+                    u"host": dut,
+                    u"socket": u"",
+                    u"runtime": runtime
+                }
+            )
 
     def _get_ndrpdr_throughput(self, msg):
         """Get NDR_LOWER, NDR_UPPER, PDR_LOWER and PDR_UPPER from the test
@@ -1033,6 +1066,8 @@ class ExecutionChecker(ResultVisitor):
         """
 
         self._sh_run_counter = 0
+        self._telemetry_kw_counter = 0
+        self._telemetry_msg_counter = 0
 
         longname_orig = test.longname.lower()
 
@@ -1258,9 +1293,13 @@ class ExecutionChecker(ResultVisitor):
         :type test_kw: Keyword
         :returns: Nothing.
         """
-        if test_kw.name.count(u"Show Runtime On All Duts") or \
-                test_kw.name.count(u"Show Runtime Counters On All Duts") or \
-                test_kw.name.count(u"Vpp Show Runtime On All Duts"):
+        if self._for_output == u"trending":
+            return
+
+        if test_kw.name.count(u"Run Telemetry On All Duts"):
+            self._msg_type = u"test-telemetry"
+            self._telemetry_kw_counter += 1
+        elif test_kw.name.count(u"Show Runtime On All Duts"):
             self._msg_type = u"test-show-runtime"
             self._sh_run_counter += 1
         else:
@@ -1399,16 +1438,20 @@ class InputData:
           (as described in ExecutionChecker documentation)
     """
 
-    def __init__(self, spec):
+    def __init__(self, spec, for_output):
         """Initialization.
 
         :param spec: Specification.
+        :param for_output: Output to be generated from downloaded data.
         :type spec: Specification
+        :type for_output: str
         """
 
         # Specification:
         self._cfg = spec
 
+        self._for_output = for_output
+
         # Data store:
         self._input_data = pd.Series()
 
@@ -1483,7 +1526,7 @@ class InputData:
                 )
                 return None
         checker = ExecutionChecker(
-            metadata, self._cfg.mapping, self._cfg.ignore
+            metadata, self._cfg.mapping, self._cfg.ignore, self._for_output
         )
         result.visit(checker)
 
@@ -2022,15 +2065,6 @@ class InputData:
         """Print all operational data to console.
         """
 
-        tbl_hdr = (
-            u"Name",
-            u"Nr of Vectors",
-            u"Nr of Packets",
-            u"Suspends",
-            u"Cycles per Packet",
-            u"Average Vector Size"
-        )
-
         for job in self._input_data.values:
             for build in job.values:
                 for test_id, test_data in build[u"tests"].items():
@@ -2038,12 +2072,60 @@ class InputData:
                     if test_data.get(u"show-run", None) is None:
                         continue
                     for dut_name, data in test_data[u"show-run"].items():
-                        if data.get(u"threads", None) is None:
+                        if data.get(u"runtime", None) is None:
                             continue
+                        runtime = loads(data[u"runtime"])
+                        try:
+                            threads_nr = len(runtime[0][u"clocks"])
+                        except (IndexError, KeyError):
+                            continue
+                        threads = OrderedDict(
+                            {idx: list() for idx in range(threads_nr)})
+                        for item in runtime:
+                            for idx in range(threads_nr):
+                                if item[u"vectors"][idx] > 0:
+                                    clocks = item[u"clocks"][idx] / \
+                                             item[u"vectors"][idx]
+                                elif item[u"calls"][idx] > 0:
+                                    clocks = item[u"clocks"][idx] / \
+                                             item[u"calls"][idx]
+                                elif item[u"suspends"][idx] > 0:
+                                    clocks = item[u"clocks"][idx] / \
+                                             item[u"suspends"][idx]
+                                else:
+                                    clocks = 0.0
+
+                                if item[u"calls"][idx] > 0:
+                                    vectors_call = item[u"vectors"][idx] / \
+                                                   item[u"calls"][idx]
+                                else:
+                                    vectors_call = 0.0
+
+                                if int(item[u"calls"][idx]) + int(
+                                        item[u"vectors"][idx]) + \
+                                        int(item[u"suspends"][idx]):
+                                    threads[idx].append([
+                                        item[u"name"],
+                                        item[u"calls"][idx],
+                                        item[u"vectors"][idx],
+                                        item[u"suspends"][idx],
+                                        clocks,
+                                        vectors_call
+                                    ])
+
                         print(f"Host IP: {data.get(u'host', '')}, "
                               f"Socket: {data.get(u'socket', '')}")
-                        for thread_nr, thread in data[u"threads"].items():
-                            txt_table = prettytable.PrettyTable(tbl_hdr)
+                        for thread_nr, thread in threads.items():
+                            txt_table = prettytable.PrettyTable(
+                                (
+                                    u"Name",
+                                    u"Nr of Vectors",
+                                    u"Nr of Packets",
+                                    u"Suspends",
+                                    u"Cycles per Packet",
+                                    u"Average Vector Size"
+                                )
+                            )
                             avg = 0.0
                             for row in thread:
                                 txt_table.add_row(row)