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
23 from ..trending.graphs import select_trending_data
27 """A class to store and manipulate the telemetry data.
30 def __init__(self, tests: list=list()) -> None:
31 """Initialize the object.
33 :param in_data: Input data.
34 :param tests: List of selected tests.
35 :type in_data: pandas.DataFrame
41 self._unique_metrics = list()
42 self._unique_metrics_labels = pd.DataFrame()
43 self._selected_metrics_labels = pd.DataFrame()
45 def from_dataframe(self, in_data: pd.DataFrame=pd.DataFrame()) -> None:
46 """Read the input from pandas DataFrame.
48 This method must be called at the beginning to create all data
55 metrics = set() # A set of unique metrics
57 # Create a dataframe with metrics for selected tests:
59 for itm in self._tests:
60 sel_data = select_trending_data(in_data, itm)
61 if sel_data is not None:
62 sel_data["test_name"] = itm["id"]
63 lst_items.append(sel_data)
64 df = pd.concat(lst_items, ignore_index=True, copy=False)
66 # Use only neccessary data:
76 "result_receive_rate_rate_avg",
77 "result_receive_rate_rate_stdev",
78 "result_receive_rate_rate_unit",
79 "result_pdr_lower_rate_value",
80 "result_pdr_lower_rate_unit",
81 "result_ndr_lower_rate_value",
82 "result_ndr_lower_rate_unit",
85 # Transform metrics from strings to dataframes:
86 lst_telemetry = list()
87 for _, row in df.iterrows():
90 "labels": list(), # list of tuple(label, value)
95 # If there is no telemetry data, use empty dictionary
96 if row["telemetry"] is None or isinstance(row["telemetry"], float):
97 lst_telemetry.append(pd.DataFrame(data=d_telemetry))
100 # Read telemetry data
101 # - list of uncompressed strings List[str, ...], or
102 # - list with only one compressed string List[str]
104 tm_data = zlib.decompress(
105 binascii.a2b_base64(row["telemetry"][0].encode())
106 ).decode().split("\n")
107 except (binascii.Error, zlib.error, AttributeError, IndexError):
108 tm_data = row["telemetry"]
110 # Pre-process telemetry data
112 itm_lst = itm.replace("'", "").rsplit(" ", maxsplit=2)
113 metric, labels = itm_lst[0].split("{")
114 d_telemetry["metric"].append(metric)
115 d_telemetry["labels"].append(
116 [tuple(x.split("=")) for x in labels[:-1].split(",")]
118 d_telemetry["value"].append(itm_lst[1])
119 d_telemetry["timestamp"].append(itm_lst[2])
121 metrics.update(d_telemetry["metric"])
122 lst_telemetry.append(pd.DataFrame(data=d_telemetry))
123 df["telemetry"] = lst_telemetry
126 self._unique_metrics = sorted(metrics)
128 def from_json(self, in_data: dict) -> None:
129 """Read the input data from json.
132 df = pd.read_json(in_data)
133 lst_telemetry = list()
134 metrics = set() # A set of unique metrics
135 for _, row in df.iterrows():
136 telemetry = pd.DataFrame(row["telemetry"])
137 lst_telemetry.append(telemetry)
138 metrics.update(telemetry["metric"].to_list())
139 df["telemetry"] = lst_telemetry
142 self._unique_metrics = sorted(metrics)
144 def from_metrics(self, in_data: set) -> None:
145 """Read only the metrics.
147 self._unique_metrics = in_data
149 def from_metrics_with_labels(self, in_data: dict) -> None:
150 """Read only metrics with labels.
152 self._unique_metrics_labels = pd.DataFrame.from_dict(in_data)
154 def to_json(self) -> str:
155 """Return the data transformed from dataframe to json.
157 :returns: Telemetry data transformed to a json structure.
160 return self._data.to_json()
163 def unique_metrics(self) -> list:
164 """Return a set of unique metrics.
166 :returns: A set of unique metrics.
169 return self._unique_metrics
172 def unique_metrics_with_labels(self) -> dict:
175 return self._unique_metrics_labels.to_dict()
177 def get_selected_labels(self, metrics: list) -> dict:
178 """Return a dictionary with labels (keys) and all their possible values
179 (values) for all selected 'metrics'.
181 :param metrics: List of metrics we are interested in.
183 :returns: A dictionary with labels and all their possible values.
189 for _, row in self._data.iterrows():
190 telemetry = row["telemetry"]
192 df = telemetry.loc[(telemetry["metric"] == itm)]
193 lst_labels.append(df)
194 for _, tm in df.iterrows():
195 for label in tm["labels"]:
196 if label[0] not in tmp_labels:
197 tmp_labels[label[0]] = set()
198 tmp_labels[label[0]].add(label[1])
200 df_labels = pd.concat(lst_labels, ignore_index=True, copy=False)
201 selected_labels = dict()
202 for key in sorted(tmp_labels):
203 selected_labels[key] = sorted(tmp_labels[key])
205 self._unique_metrics_labels = df_labels[["metric", "labels"]].\
206 loc[df_labels[["metric", "labels"]].astype(str).\
207 drop_duplicates().index]
209 return selected_labels
212 def str_metrics(self) -> str:
213 """Returns all unique metrics as a string.
215 return TelemetryData.metrics_to_str(self._unique_metrics_labels)
218 def metrics_to_str(in_data: pd.DataFrame) -> str:
219 """Convert metrics from pandas dataframe to string. Metrics in string
220 are separated by '\n'.
222 :param in_data: Metrics to be converted to a string.
223 :type in_data: pandas.DataFrame
224 :returns: Metrics as a string.
228 for _, row in in_data.iterrows():
229 labels = ','.join([f"{itm[0]}='{itm[1]}'" for itm in row["labels"]])
230 metrics += f"{row['metric']}{{{labels}}}\n"
233 def search_unique_metrics(self, string: str) -> list:
234 """Return a list of metrics which name includes the given string.
236 :param string: A string which must be in the name of metric.
238 :returns: A list of metrics which name includes the given string.
241 return [itm for itm in self._unique_metrics if string in itm]
243 def filter_selected_metrics_by_labels(
247 """Filter selected unique metrics by labels and their values.
249 :param selection: Labels and their values specified by the user.
250 :type selection: dict
251 :returns: Pandas dataframe with filtered metrics.
252 :rtype: pandas.DataFrame
255 def _is_selected(labels: list, sel: dict) -> bool:
256 """Check if the provided 'labels' are selected by the user.
258 :param labels: List of labels and their values from a metric. The
259 items in this lists are two-item-lists whre the first item is
260 the label and the second one is its value.
261 :param sel: User selection. The keys are the selected lables and the
262 values are lists with label values.
265 :returns: True if the 'labels' are selected by the user.
269 labels = dict(labels)
270 for key in sel.keys():
271 if key in list(labels.keys()):
273 passed.append(labels[key] in sel[key])
278 return bool(passed and all(passed))
280 self._selected_metrics_labels = pd.DataFrame()
282 for _, row in self._unique_metrics_labels.iterrows():
283 if _is_selected(row["labels"], selection):
284 lst_items.append(row.to_frame().T)
285 self._selected_metrics_labels = \
286 pd.concat(lst_items, ignore_index=True, axis=0, copy=False)
287 return self._selected_metrics_labels
289 def select_tm_trending_data(
292 ignore_host: bool = False
294 """Select telemetry data for trending based on user's 'selection'.
296 The output dataframe includes these columns:
306 - "result_receive_rate_rate_avg",
307 - "result_receive_rate_rate_stdev",
308 - "result_receive_rate_rate_unit",
309 - "result_pdr_lower_rate_value",
310 - "result_pdr_lower_rate_unit",
311 - "result_ndr_lower_rate_value",
312 - "result_ndr_lower_rate_unit",
316 :param selection: User's selection (metrics and labels).
317 :param ignore_host: Ignore 'hostname' and 'hook' labels in metrics.
318 :type selection: dict
319 :type ignore_host: bool
320 :returns: Dataframe with selected data.
321 :rtype: pandas.DataFrame
324 if self._data is None:
325 return pd.DataFrame()
327 return pd.DataFrame()
329 return pd.DataFrame()
331 df_sel = pd.DataFrame.from_dict(selection)
333 for _, row in self._data.iterrows():
334 tm_row = row["telemetry"]
335 for _, tm_sel in df_sel.iterrows():
336 df_tmp = tm_row.loc[tm_row["metric"] == tm_sel["metric"]]
337 for _, tm in df_tmp.iterrows():
340 if tm["labels"][2:] == tm_sel["labels"][2:]:
342 [f"{i[0]}='{i[1]}'" for i in tm["labels"][2:]]
346 if tm["labels"] == tm_sel["labels"]:
348 [f"{i[0]}='{i[1]}'" for i in tm["labels"]]
352 row["tm_metric"] = f"{tm['metric']}{{{labels}}}"
353 row["tm_value"] = tm["value"]
355 row.drop(labels=["telemetry", ]).to_frame().T
359 lst_rows, ignore_index=True, axis=0, copy=False
362 return pd.DataFrame()