generator.dashboards.operational_monitoring_dashboard

  1"""Class to describe Operational Monitoring Dashboard."""
  2
  3from __future__ import annotations
  4
  5from typing import Any, Dict, List
  6
  7from ..views import lookml_utils
  8from .dashboard import Dashboard
  9
 10
 11class OperationalMonitoringDashboard(Dashboard):
 12    """An Operational Monitoring dashboard."""
 13
 14    type: str = "operational_monitoring_dashboard"
 15
 16    def __init__(
 17        self,
 18        title: str,
 19        name: str,
 20        layout: str,
 21        namespace: str,
 22        defn: List[Dict[str, Any]],
 23    ):
 24        """Get an instance of a Operational Monitoring Dashboard."""
 25        self.dimensions = defn[0].get("dimensions", {})
 26        self.xaxis = defn[0]["xaxis"]
 27        self.compact_visualization = defn[0].get("compact_visualization", False)
 28        self.group_by_dimension = defn[0].get("group_by_dimension", None)
 29
 30        super().__init__(title, name, layout, namespace, defn)
 31
 32    @classmethod
 33    def from_dict(
 34        klass, namespace: str, name: str, defn: dict
 35    ) -> OperationalMonitoringDashboard:
 36        """Get a OperationalMonitoringDashboard from a dict representation."""
 37        title = defn["title"]
 38        return klass(title, name, "newspaper", namespace, defn["tables"])
 39
 40    def _map_series_to_colours(self, branches, explore):
 41        colours = [
 42            "#3FE1B0",
 43            "#0060E0",
 44            "#9059FF",
 45            "#B933E1",
 46            "#FF2A8A",
 47            "#FF505F",
 48            "#FF7139",
 49            "#FFA537",
 50            "#005E5D",
 51            "#073072",
 52            "#7F165B",
 53            "#A7341F",
 54        ]
 55        return {branch: color for branch, color in zip(branches, colours)}
 56
 57    def to_lookml(self):
 58        """Get this dashboard as LookML."""
 59        kwargs = {
 60            "name": self.name,
 61            "title": self.title,
 62            "layout": self.layout,
 63            "elements": [],
 64            "dimensions": [],
 65            "group_by_dimension": self.group_by_dimension,
 66            "alerts": None,
 67            "compact_visualization": self.compact_visualization,
 68        }
 69
 70        includes = []
 71        graph_index = 0
 72        for table_defn in self.tables:
 73            explore = table_defn["explore"]
 74            includes.append(
 75                f"/looker-hub/{self.namespace}/explores/{explore}.explore.lkml"
 76            )
 77
 78            if table_defn["table"].endswith("alerts"):
 79                kwargs["alerts"] = {
 80                    "explore": explore,
 81                    "col": 0,
 82                    "date": (
 83                        f"{self.xaxis}_date" if self.xaxis == "build_id" else self.xaxis
 84                    ),
 85                }
 86            else:
 87                if len(kwargs["dimensions"]) == 0:
 88                    kwargs["dimensions"] = [
 89                        {
 90                            "name": name,
 91                            "title": lookml_utils.slug_to_title(name),
 92                            "default": info["default"],
 93                            "options": info["options"],
 94                        }
 95                        for name, info in self.dimensions.items()
 96                    ]
 97
 98                series_colors = self._map_series_to_colours(
 99                    table_defn["branches"], explore
100                )
101                # determine metric groups
102                metric_groups = {}
103                for summary in table_defn.get("summaries", []):
104                    for metric_group in summary.get("metric_groups", []):
105                        if metric_group not in metric_groups:
106                            metric_groups[metric_group] = [summary["metric"]]
107                        elif summary["metric"] not in metric_groups[metric_group]:
108                            metric_groups[metric_group].append(summary["metric"])
109
110                seen_metric_groups = []
111                for summary in table_defn.get("summaries", []):
112                    summary_metric_groups = summary.get("metric_groups", [])
113                    if len(summary_metric_groups) == 0:
114                        # append a dummy entry if no metric group defined
115                        summary_metric_groups.append(None)
116
117                    for metric_group in summary_metric_groups:
118                        if (metric_group, summary["statistic"]) in seen_metric_groups:
119                            continue
120
121                        if self.compact_visualization:
122                            title = "Metric"
123                        else:
124                            if metric_group is None:
125                                title = lookml_utils.slug_to_title(summary["metric"])
126                            else:
127                                title = lookml_utils.slug_to_title(metric_group)
128
129                        if not self.group_by_dimension:
130                            kwargs["elements"].append(
131                                {
132                                    "title": title,
133                                    "metric": (
134                                        summary["metric"]
135                                        if metric_group is None
136                                        else ", ".join(
137                                            f'"{m}"'
138                                            for m in metric_groups[metric_group]
139                                        )
140                                    ),
141                                    "statistic": summary["statistic"],
142                                    "explore": explore,
143                                    "series_colors": series_colors,
144                                    "xaxis": self.xaxis,
145                                    "row": int(graph_index / 2) * 10,
146                                    "col": 0 if graph_index % 2 == 0 else 12,
147                                    "is_metric_group": metric_group is not None,
148                                }
149                            )
150                            if metric_group is not None:
151                                seen_metric_groups.append(
152                                    (metric_group, summary["statistic"])
153                                )
154                            graph_index += 1
155
156                        if self.group_by_dimension:
157                            kwargs["elements"].append(
158                                {
159                                    "title": f"{title} - By {self.group_by_dimension}",
160                                    "metric": (
161                                        summary["metric"]
162                                        if metric_group is None
163                                        else ", ".join(
164                                            f'"{m}"'
165                                            for m in metric_groups[metric_group]
166                                        )
167                                    ),
168                                    "statistic": summary["statistic"],
169                                    "explore": explore,
170                                    "series_colors": series_colors,
171                                    "xaxis": self.xaxis,
172                                    "row": int(graph_index / 2) * 10,
173                                    "col": 0 if graph_index % 2 == 0 else 12,
174                                    "is_metric_group": metric_group is not None,
175                                }
176                            )
177                            graph_index += 1
178
179                        if self.compact_visualization:
180                            # compact visualization only needs a single tile for all probes
181                            break
182
183                    if self.compact_visualization:
184                        # compact visualization only needs a single tile for all probes
185                        break
186
187        if "alerts" in kwargs and kwargs["alerts"] is not None:
188            kwargs["alerts"]["row"] = int(graph_index / 2) * 10
189
190        dash_lookml = lookml_utils.render_template(
191            "dashboard.lkml", "dashboards", **kwargs
192        )
193        return dash_lookml