csit.infra.dash/app/cdash/utils/telemetry_data.py

   1 # Copyright (c) 2023 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 """A module implementing the parsing of OpenMetrics data and elementary
  15 operations with it.
  16 """
  17
  18
  19 import pandas as pd
  20
  21 from ..trending.graphs import select_trending_data
  22
  23
  24 class TelemetryData:
  25     """A class to store and manipulate the telemetry data.
  26     """
  27
  28     def __init__(self, tests: list=list()) -> None:
  29         """Initialize the object.
  30
  31         :param in_data: Input data.
  32         :param tests: List of selected tests.
  33         :type in_data: pandas.DataFrame
  34         :type tests: list
  35         """
  36
  37         self._tests = tests
  38         self._data = None
  39         self._unique_metrics = list()
  40         self._unique_metrics_labels = pd.DataFrame()
  41         self._selected_metrics_labels = pd.DataFrame()
  42
  43     def from_dataframe(self, in_data: pd.DataFrame=pd.DataFrame()) -> None:
  44         """Read the input from pandas DataFrame.
  45
  46         This method must be call at the begining to create all data structures.
  47         """
  48
  49         if in_data.empty:
  50             return
  51
  52         df = pd.DataFrame()
  53         metrics = set()  # A set of unique metrics
  54
  55         # Create a dataframe with metrics for selected tests:
  56         for itm in self._tests:
  57             sel_data = select_trending_data(in_data, itm)
  58             if sel_data is not None:
  59                 sel_data["test_name"] = itm["id"]
  60                 df = pd.concat([df, sel_data], ignore_index=True, copy=False)
  61         # Use only neccessary data:
  62         df = df[[
  63             "job",
  64             "build",
  65             "dut_type",
  66             "dut_version",
  67             "start_time",
  68             "passed",
  69             "test_name",
  70             "test_type",
  71             "result_receive_rate_rate_avg",
  72             "result_receive_rate_rate_stdev",
  73             "result_receive_rate_rate_unit",
  74             "result_pdr_lower_rate_value",
  75             "result_pdr_lower_rate_unit",
  76             "result_ndr_lower_rate_value",
  77             "result_ndr_lower_rate_unit",
  78             "telemetry"
  79         ]]
  80         # Transform metrics from strings to dataframes:
  81         lst_telemetry = list()
  82         for _, row in df.iterrows():
  83             d_telemetry = {
  84                 "metric": list(),
  85                 "labels": list(),  # list of tuple(label, value)
  86                 "value": list(),
  87                 "timestamp": list()
  88             }
  89             if row["telemetry"] is not None and \
  90                     not isinstance(row["telemetry"], float):
  91                 for itm in row["telemetry"]:
  92                     itm_lst = itm.replace("'", "").rsplit(" ", maxsplit=2)
  93                     metric, labels = itm_lst[0].split("{")
  94                     d_telemetry["metric"].append(metric)
  95                     d_telemetry["labels"].append(
  96                         [tuple(x.split("=")) for x in labels[:-1].split(",")]
  97                     )
  98                     d_telemetry["value"].append(itm_lst[1])
  99                     d_telemetry["timestamp"].append(itm_lst[2])
 100                 metrics.update(d_telemetry["metric"])
 101             lst_telemetry.append(pd.DataFrame(data=d_telemetry))
 102         df["telemetry"] = lst_telemetry
 103
 104         self._data = df
 105         self._unique_metrics = sorted(metrics)
 106
 107     def from_json(self, in_data: dict) -> None:
 108         """Read the input data from json.
 109         """
 110
 111         df = pd.read_json(in_data)
 112         lst_telemetry = list()
 113         metrics = set()  # A set of unique metrics
 114         for _, row in df.iterrows():
 115             telemetry = pd.DataFrame(row["telemetry"])
 116             lst_telemetry.append(telemetry)
 117             metrics.update(telemetry["metric"].to_list())
 118         df["telemetry"] = lst_telemetry
 119
 120         self._data = df
 121         self._unique_metrics = sorted(metrics)
 122
 123     def from_metrics(self, in_data: set) -> None:
 124         """Read only the metrics.
 125         """
 126         self._unique_metrics = in_data
 127
 128     def from_metrics_with_labels(self, in_data: dict) -> None:
 129         """Read only metrics with labels.
 130         """
 131         self._unique_metrics_labels = pd.DataFrame.from_dict(in_data)
 132
 133     def to_json(self) -> str:
 134         """Return the data transformed from dataframe to json.
 135
 136         :returns: Telemetry data transformed to a json structure.
 137         :rtype: dict
 138         """
 139         return self._data.to_json()
 140
 141     @property
 142     def unique_metrics(self) -> list:
 143         """Return a set of unique metrics.
 144
 145         :returns: A set of unique metrics.
 146         :rtype: set
 147         """
 148         return self._unique_metrics
 149
 150     @property
 151     def unique_metrics_with_labels(self) -> dict:
 152         """
 153         """
 154         return self._unique_metrics_labels.to_dict()
 155
 156     def get_selected_labels(self, metrics: list) -> dict:
 157         """Return a dictionary with labels (keys) and all their possible values
 158         (values) for all selected 'metrics'.
 159
 160         :param metrics: List of metrics we are interested in.
 161         :type metrics: list
 162         :returns: A dictionary with labels and all their possible values.
 163         :rtype: dict
 164         """
 165
 166         df_labels = pd.DataFrame()
 167         tmp_labels = dict()
 168         for _, row in self._data.iterrows():
 169             telemetry = row["telemetry"]
 170             for itm in metrics:
 171                 df = telemetry.loc[(telemetry["metric"] == itm)]
 172                 df_labels = pd.concat(
 173                     [df_labels, df],
 174                     ignore_index=True,
 175                     copy=False
 176                 )
 177                 for _, tm in df.iterrows():
 178                     for label in tm["labels"]:
 179                         if label[0] not in tmp_labels:
 180                             tmp_labels[label[0]] = set()
 181                         tmp_labels[label[0]].add(label[1])
 182
 183         selected_labels = dict()
 184         for key in sorted(tmp_labels):
 185             selected_labels[key] = sorted(tmp_labels[key])
 186
 187         self._unique_metrics_labels = df_labels[["metric", "labels"]].\
 188             loc[df_labels[["metric", "labels"]].astype(str).\
 189                 drop_duplicates().index]
 190
 191         return selected_labels
 192
 193     @property
 194     def str_metrics(self) -> str:
 195         """Returns all unique metrics as a string.
 196         """
 197         return TelemetryData.metrics_to_str(self._unique_metrics_labels)
 198
 199     @staticmethod
 200     def metrics_to_str(in_data: pd.DataFrame) -> str:
 201         """Convert metrics from pandas dataframe to string. Metrics in string
 202         are separated by '\n'.
 203
 204         :param in_data: Metrics to be converted to a string.
 205         :type in_data: pandas.DataFrame
 206         :returns: Metrics as a string.
 207         :rtype: str
 208         """
 209         metrics = str()
 210         for _, row in in_data.iterrows():
 211             labels = ','.join([f"{itm[0]}='{itm[1]}'" for itm in row["labels"]])
 212             metrics += f"{row['metric']}{{{labels}}}\n"
 213         return metrics[:-1]
 214
 215     def search_unique_metrics(self, string: str) -> list:
 216         """Return a list of metrics which name includes the given string.
 217
 218         :param string: A string which must be in the name of metric.
 219         :type string: str
 220         :returns: A list of metrics which name includes the given string.
 221         :rtype: list
 222         """
 223         return [itm for itm in self._unique_metrics if string in itm]
 224
 225     def filter_selected_metrics_by_labels(
 226             self,
 227             selection: dict
 228         ) -> pd.DataFrame:
 229         """Filter selected unique metrics by labels and their values.
 230
 231         :param selection: Labels and their values specified by the user.
 232         :type selection: dict
 233         :returns: Pandas dataframe with filtered metrics.
 234         :rtype: pandas.DataFrame
 235         """
 236
 237         def _is_selected(labels: list, sel: dict) -> bool:
 238             """Check if the provided 'labels' are selected by the user.
 239
 240             :param labels: List of labels and their values from a metric. The
 241                 items in this lists are two-item-lists whre the first item is
 242                 the label and the second one is its value.
 243             :param sel: User selection. The keys are the selected lables and the
 244                 values are lists with label values.
 245             :type labels: list
 246             :type sel: dict
 247             :returns: True if the 'labels' are selected by the user.
 248             :rtype: bool
 249             """
 250             passed = list()
 251             labels = dict(labels)
 252             for key in sel.keys():
 253                 if key in list(labels.keys()):
 254                     if sel[key]:
 255                         passed.append(labels[key] in sel[key])
 256                     else:
 257                         passed.append(True)
 258                 else:
 259                     passed.append(False)
 260             return bool(passed and all(passed))
 261
 262         self._selected_metrics_labels = pd.DataFrame()
 263         for _, row in self._unique_metrics_labels.iterrows():
 264             if _is_selected(row["labels"], selection):
 265                 self._selected_metrics_labels = pd.concat(
 266                     [self._selected_metrics_labels, row.to_frame().T],
 267                     ignore_index=True,
 268                     axis=0,
 269                     copy=False
 270                 )
 271         return self._selected_metrics_labels
 272
 273     def select_tm_trending_data(self, selection: dict) -> pd.DataFrame:
 274         """Select telemetry data for trending based on user's 'selection'.
 275
 276         The output dataframe includes these columns:
 277             - "job",
 278             - "build",
 279             - "dut_type",
 280             - "dut_version",
 281             - "start_time",
 282             - "passed",
 283             - "test_name",
 284             - "test_id",
 285             - "test_type",
 286             - "result_receive_rate_rate_avg",
 287             - "result_receive_rate_rate_stdev",
 288             - "result_receive_rate_rate_unit",
 289             - "result_pdr_lower_rate_value",
 290             - "result_pdr_lower_rate_unit",
 291             - "result_ndr_lower_rate_value",
 292             - "result_ndr_lower_rate_unit",
 293             - "tm_metric",
 294             - "tm_value".
 295
 296         :param selection: User's selection (metrics and labels).
 297         :type selection: dict
 298         :returns: Dataframe with selected data.
 299         :rtype: pandas.DataFrame
 300         """
 301
 302         df = pd.DataFrame()
 303
 304         if self._data is None:
 305             return df
 306         if self._data.empty:
 307             return df
 308         if not selection:
 309             return df
 310
 311         df_sel = pd.DataFrame.from_dict(selection)
 312         for _, row in self._data.iterrows():
 313             tm_row = row["telemetry"]
 314             for _, tm_sel in df_sel.iterrows():
 315                 df_tmp = tm_row.loc[tm_row["metric"] == tm_sel["metric"]]
 316                 for _, tm in df_tmp.iterrows():
 317                     if tm["labels"] == tm_sel["labels"]:
 318                         labels = ','.join(
 319                             [f"{itm[0]}='{itm[1]}'" for itm in tm["labels"]]
 320                         )
 321                         row["tm_metric"] = f"{tm['metric']}{{{labels}}}"
 322                         row["tm_value"] = tm["value"]
 323                         new_row = row.drop(labels=["telemetry", ])
 324                         df = pd.concat(
 325                             [df, new_row.to_frame().T],
 326                             ignore_index=True,
 327                             axis=0,
 328                             copy=False
 329                         )
 330         return df