aboutsummaryrefslogtreecommitdiffstats
path: root/csit.infra.dash/app/cdash/utils/telemetry_data.py
blob: e88b8eed06346d64ec0c198c71c65651dfba32ea (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
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