C-Dash: Add telemetry panel

Signed-off-by: Tibor Frank <tifrank@cisco.com>
Change-Id: Idee88c1da9bebd433fa47f5d983d432c54b5fbae

1 files changed, 330 insertions, 0 deletions

diff --git a/csit.infra.dash/app/cdash/utils/telemetry_data.py b/csit.infra.dash/app/cdash/utils/telemetry_data.py
new file mode 100644
index 0000000000..e88b8eed06
--- /dev/null
+++ b/csit.infra.dash/app/cdash/utils/telemetry_data.py
@@ -0,0 +1,330 @@
+# Copyright (c) 2023 Cisco and/or its affiliates.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""A module implementing the parsing of OpenMetrics data and elementary
+operations with it.
+"""
+
+
+import pandas as pd
+
+from ..trending.graphs import select_trending_data
+
+
+class TelemetryData:
+    """A class to store and manipulate the telemetry data.
+    """
+
+    def __init__(self, tests: list=list()) -> None:
+        """Initialize the object.
+
+        :param in_data: Input data.
+        :param tests: List of selected tests.
+        :type in_data: pandas.DataFrame
+        :type tests: list
+        """
+
+        self._tests = tests
+        self._data = None
+        self._unique_metrics = list()
+        self._unique_metrics_labels = pd.DataFrame()
+        self._selected_metrics_labels = pd.DataFrame()
+
+    def from_dataframe(self, in_data: pd.DataFrame=pd.DataFrame()) -> None:
+        """Read the input from pandas DataFrame.
+
+        This method must be call at the begining to create all data structures.
+        """
+
+        if in_data.empty:
+            return
+
+        df = pd.DataFrame()
+        metrics = set()  # A set of unique metrics
+
+        # Create a dataframe with metrics for selected tests:
+        for itm in self._tests:
+            sel_data = select_trending_data(in_data, itm)
+            if sel_data is not None:
+                sel_data["test_name"] = itm["id"]
+                df = pd.concat([df, sel_data], ignore_index=True, copy=False)
+        # Use only neccessary data:
+        df = df[[
+            "job",
+            "build",
+            "dut_type",
+            "dut_version",
+            "start_time",
+            "passed",
+            "test_name",
+            "test_type",
+            "result_receive_rate_rate_avg",
+            "result_receive_rate_rate_stdev",
+            "result_receive_rate_rate_unit",
+            "result_pdr_lower_rate_value",
+            "result_pdr_lower_rate_unit",
+            "result_ndr_lower_rate_value",
+            "result_ndr_lower_rate_unit",
+            "telemetry"
+        ]]
+        # Transform metrics from strings to dataframes:
+        lst_telemetry = list()
+        for _, row in df.iterrows():
+            d_telemetry = {
+                "metric": list(),
+                "labels": list(),  # list of tuple(label, value)
+                "value": list(),
+                "timestamp": list()
+            }
+            if row["telemetry"] is not None and \
+                    not isinstance(row["telemetry"], float):
+                for itm in row["telemetry"]:
+                    itm_lst = itm.replace("'", "").rsplit(" ", maxsplit=2)
+                    metric, labels = itm_lst[0].split("{")
+                    d_telemetry["metric"].append(metric)
+                    d_telemetry["labels"].append(
+                        [tuple(x.split("=")) for x in labels[:-1].split(",")]
+                    )
+                    d_telemetry["value"].append(itm_lst[1])
+                    d_telemetry["timestamp"].append(itm_lst[2])
+                metrics.update(d_telemetry["metric"])
+            lst_telemetry.append(pd.DataFrame(data=d_telemetry))
+        df["telemetry"] = lst_telemetry
+
+        self._data = df
+        self._unique_metrics = sorted(metrics)
+
+    def from_json(self, in_data: dict) -> None:
+        """Read the input data from json.
+        """
+
+        df = pd.read_json(in_data)
+        lst_telemetry = list()
+        metrics = set()  # A set of unique metrics
+        for _, row in df.iterrows():
+            telemetry = pd.DataFrame(row["telemetry"])
+            lst_telemetry.append(telemetry)
+            metrics.update(telemetry["metric"].to_list())
+        df["telemetry"] = lst_telemetry
+
+        self._data = df
+        self._unique_metrics = sorted(metrics)
+
+    def from_metrics(self, in_data: set) -> None:
+        """Read only the metrics.
+        """
+        self._unique_metrics = in_data
+
+    def from_metrics_with_labels(self, in_data: dict) -> None:
+        """Read only metrics with labels.
+        """
+        self._unique_metrics_labels = pd.DataFrame.from_dict(in_data)
+
+    def to_json(self) -> str:
+        """Return the data transformed from dataframe to json.
+
+        :returns: Telemetry data transformed to a json structure.
+        :rtype: dict
+        """
+        return self._data.to_json()
+
+    @property
+    def unique_metrics(self) -> list:
+        """Return a set of unique metrics.
+
+        :returns: A set of unique metrics.
+        :rtype: set
+        """
+        return self._unique_metrics
+
+    @property
+    def unique_metrics_with_labels(self) -> dict:
+        """
+        """
+        return self._unique_metrics_labels.to_dict()
+
+    def get_selected_labels(self, metrics: list) -> dict:
+        """Return a dictionary with labels (keys) and all their possible values
+        (values) for all selected 'metrics'.
+
+        :param metrics: List of metrics we are interested in.
+        :type metrics: list
+        :returns: A dictionary with labels and all their possible values.
+        :rtype: dict
+        """
+
+        df_labels = pd.DataFrame()
+        tmp_labels = dict()
+        for _, row in self._data.iterrows():
+            telemetry = row["telemetry"]
+            for itm in metrics:
+                df = telemetry.loc[(telemetry["metric"] == itm)]
+                df_labels = pd.concat(
+                    [df_labels, df],
+                    ignore_index=True,
+                    copy=False
+                )
+                for _, tm in df.iterrows():
+                    for label in tm["labels"]:
+                        if label[0] not in tmp_labels:
+                            tmp_labels[label[0]] = set()
+                        tmp_labels[label[0]].add(label[1])
+
+        selected_labels = dict()
+        for key in sorted(tmp_labels):
+            selected_labels[key] = sorted(tmp_labels[key])
+
+        self._unique_metrics_labels = df_labels[["metric", "labels"]].\
+            loc[df_labels[["metric", "labels"]].astype(str).\
+                drop_duplicates().index]
+
+        return selected_labels
+
+    @property
+    def str_metrics(self) -> str:
+        """Returns all unique metrics as a string.
+        """
+        return TelemetryData.metrics_to_str(self._unique_metrics_labels)
+
+    @staticmethod
+    def metrics_to_str(in_data: pd.DataFrame) -> str:
+        """Convert metrics from pandas dataframe to string. Metrics in string
+        are separated by '\n'.
+
+        :param in_data: Metrics to be converted to a string.
+        :type in_data: pandas.DataFrame
+        :returns: Metrics as a string.
+        :rtype: str
+        """
+        metrics = str()
+        for _, row in in_data.iterrows():
+            labels = ','.join([f"{itm[0]}='{itm[1]}'" for itm in row["labels"]])
+            metrics += f"{row['metric']}{{{labels}}}\n"
+        return metrics[:-1]
+
+    def search_unique_metrics(self, string: str) -> list:
+        """Return a list of metrics which name includes the given string.
+
+        :param string: A string which must be in the name of metric.
+        :type string: str
+        :returns: A list of metrics which name includes the given string.
+        :rtype: list
+        """
+        return [itm for itm in self._unique_metrics if string in itm]
+
+    def filter_selected_metrics_by_labels(
+            self,
+            selection: dict
+        ) -> pd.DataFrame:
+        """Filter selected unique metrics by labels and their values.
+
+        :param selection: Labels and their values specified by the user.
+        :type selection: dict
+        :returns: Pandas dataframe with filtered metrics.
+        :rtype: pandas.DataFrame
+        """
+
+        def _is_selected(labels: list, sel: dict) -> bool:
+            """Check if the provided 'labels' are selected by the user.
+
+            :param labels: List of labels and their values from a metric. The
+                items in this lists are two-item-lists whre the first item is
+                the label and the second one is its value.
+            :param sel: User selection. The keys are the selected lables and the
+                values are lists with label values.
+            :type labels: list
+            :type sel: dict
+            :returns: True if the 'labels' are selected by the user.
+            :rtype: bool
+            """
+            passed = list()
+            labels = dict(labels)
+            for key in sel.keys():
+                if key in list(labels.keys()):
+                    if sel[key]:
+                        passed.append(labels[key] in sel[key])
+                    else:
+                        passed.append(True)
+                else:
+                    passed.append(False)
+            return bool(passed and all(passed))
+
+        self._selected_metrics_labels = pd.DataFrame()
+        for _, row in self._unique_metrics_labels.iterrows():
+            if _is_selected(row["labels"], selection):
+                self._selected_metrics_labels = pd.concat(
+                    [self._selected_metrics_labels, row.to_frame().T],
+                    ignore_index=True,
+                    axis=0,
+                    copy=False
+                )
+        return self._selected_metrics_labels
+
+    def select_tm_trending_data(self, selection: dict) -> pd.DataFrame:
+        """Select telemetry data for trending based on user's 'selection'.
+
+        The output dataframe includes these columns:
+            - "job",
+            - "build",
+            - "dut_type",
+            - "dut_version",
+            - "start_time",
+            - "passed",
+            - "test_name",
+            - "test_id",
+            - "test_type",
+            - "result_receive_rate_rate_avg",
+            - "result_receive_rate_rate_stdev",
+            - "result_receive_rate_rate_unit",
+            - "result_pdr_lower_rate_value",
+            - "result_pdr_lower_rate_unit",
+            - "result_ndr_lower_rate_value",
+            - "result_ndr_lower_rate_unit",
+            - "tm_metric",
+            - "tm_value".
+
+        :param selection: User's selection (metrics and labels).
+        :type selection: dict
+        :returns: Dataframe with selected data.
+        :rtype: pandas.DataFrame
+        """
+
+        df = pd.DataFrame()
+
+        if self._data is None:
+            return df
+        if self._data.empty:
+            return df
+        if not selection:
+            return df
+
+        df_sel = pd.DataFrame.from_dict(selection)
+        for _, row in self._data.iterrows():
+            tm_row = row["telemetry"]
+            for _, tm_sel in df_sel.iterrows():
+                df_tmp = tm_row.loc[tm_row["metric"] == tm_sel["metric"]]
+                for _, tm in df_tmp.iterrows():
+                    if tm["labels"] == tm_sel["labels"]:
+                        labels = ','.join(
+                            [f"{itm[0]}='{itm[1]}'" for itm in tm["labels"]]
+                        )
+                        row["tm_metric"] = f"{tm['metric']}{{{labels}}}"
+                        row["tm_value"] = tm["value"]
+                        new_row = row.drop(labels=["telemetry", ])
+                        df = pd.concat(
+                            [df, new_row.to_frame().T],
+                            ignore_index=True,
+                            axis=0,
+                            copy=False
+                        )
+        return df