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:
6 # http://www.apache.org/licenses/LICENSE-2.0
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.
14 """A module implementing the parsing of OpenMetrics data and elementary
21 from ..trending.graphs import select_trending_data
25 """A class to store and manipulate the telemetry data.
28 def __init__(self, tests: list=list()) -> None:
29 """Initialize the object.
31 :param in_data: Input data.
32 :param tests: List of selected tests.
33 :type in_data: pandas.DataFrame
39 self._unique_metrics = list()
40 self._unique_metrics_labels = pd.DataFrame()
41 self._selected_metrics_labels = pd.DataFrame()
43 def from_dataframe(self, in_data: pd.DataFrame=pd.DataFrame()) -> None:
44 """Read the input from pandas DataFrame.
46 This method must be call at the begining to create all data structures.
53 metrics = set() # A set of unique metrics
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:
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",
80 # Transform metrics from strings to dataframes:
81 lst_telemetry = list()
82 for _, row in df.iterrows():
85 "labels": list(), # list of tuple(label, value)
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(",")]
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
105 self._unique_metrics = sorted(metrics)
107 def from_json(self, in_data: dict) -> None:
108 """Read the input data from json.
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
121 self._unique_metrics = sorted(metrics)
123 def from_metrics(self, in_data: set) -> None:
124 """Read only the metrics.
126 self._unique_metrics = in_data
128 def from_metrics_with_labels(self, in_data: dict) -> None:
129 """Read only metrics with labels.
131 self._unique_metrics_labels = pd.DataFrame.from_dict(in_data)
133 def to_json(self) -> str:
134 """Return the data transformed from dataframe to json.
136 :returns: Telemetry data transformed to a json structure.
139 return self._data.to_json()
142 def unique_metrics(self) -> list:
143 """Return a set of unique metrics.
145 :returns: A set of unique metrics.
148 return self._unique_metrics
151 def unique_metrics_with_labels(self) -> dict:
154 return self._unique_metrics_labels.to_dict()
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'.
160 :param metrics: List of metrics we are interested in.
162 :returns: A dictionary with labels and all their possible values.
166 df_labels = pd.DataFrame()
168 for _, row in self._data.iterrows():
169 telemetry = row["telemetry"]
171 df = telemetry.loc[(telemetry["metric"] == itm)]
172 df_labels = pd.concat(
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])
183 selected_labels = dict()
184 for key in sorted(tmp_labels):
185 selected_labels[key] = sorted(tmp_labels[key])
187 self._unique_metrics_labels = df_labels[["metric", "labels"]].\
188 loc[df_labels[["metric", "labels"]].astype(str).\
189 drop_duplicates().index]
191 return selected_labels
194 def str_metrics(self) -> str:
195 """Returns all unique metrics as a string.
197 return TelemetryData.metrics_to_str(self._unique_metrics_labels)
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'.
204 :param in_data: Metrics to be converted to a string.
205 :type in_data: pandas.DataFrame
206 :returns: Metrics as a string.
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"
215 def search_unique_metrics(self, string: str) -> list:
216 """Return a list of metrics which name includes the given string.
218 :param string: A string which must be in the name of metric.
220 :returns: A list of metrics which name includes the given string.
223 return [itm for itm in self._unique_metrics if string in itm]
225 def filter_selected_metrics_by_labels(
229 """Filter selected unique metrics by labels and their values.
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
237 def _is_selected(labels: list, sel: dict) -> bool:
238 """Check if the provided 'labels' are selected by the user.
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.
247 :returns: True if the 'labels' are selected by the user.
251 labels = dict(labels)
252 for key in sel.keys():
253 if key in list(labels.keys()):
255 passed.append(labels[key] in sel[key])
260 return bool(passed and all(passed))
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],
271 return self._selected_metrics_labels
273 def select_tm_trending_data(self, selection: dict) -> pd.DataFrame:
274 """Select telemetry data for trending based on user's 'selection'.
276 The output dataframe includes these columns:
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",
296 :param selection: User's selection (metrics and labels).
297 :type selection: dict
298 :returns: Dataframe with selected data.
299 :rtype: pandas.DataFrame
304 if self._data is None:
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"]:
319 [f"{itm[0]}='{itm[1]}'" for itm in tm["labels"]]
321 row["tm_metric"] = f"{tm['metric']}{{{labels}}}"
322 row["tm_value"] = tm["value"]
323 new_row = row.drop(labels=["telemetry", ])
325 [df, new_row.to_frame().T],