Skip to content
Snippets Groups Projects
Commit 3bf4bfff authored by Christoph Alt's avatar Christoph Alt
Browse files

added own class for annotations

parent efdfc5bd
Branches
No related merge requests found
Pipeline #43383 passed with stages
in 34 seconds
from dataclasses import dataclass
@dataclass
class Annotation:
# type: str = "influx"
# uid: str
dataSource: str
iconColor: str
name: str
query: str
def to_json_data(self):
return {
"dataSource": self.dataSource,
"enable": True,
"iconColor": self.iconColor,
"name": self.name,
"query": self.query,
"target": {
"limit": 100,
"matchAny": False,
"tags": [],
"type": "dashboard"
}
}
......@@ -2,10 +2,12 @@ from dataclasses import asdict, dataclass
from typing import List, Union
from grafanalib.core import (Dashboard, GridPos, Panel, Repeat, RowPanel, Stat,
Template, Templating, Time, TimeSeries)
Template, Templating, Time, TimeSeries,
Annotations)
from grafanalib.influxdb import InfluxDBTarget
from dashboards.influx_queries import Query
from dashboards.annotations import Annotation
def get_influx_target(target_query: str, **kwargs) -> InfluxDBTarget:
......@@ -15,16 +17,19 @@ def get_influx_target(target_query: str, **kwargs) -> InfluxDBTarget:
)
def get_dashboard_variable_query(name: str, template_query: str, dataSource: str, **kwargs) -> Template:
return Template(
name,
template_query,
dataSource=dataSource,
type="query",
includeAll=True,
multi=True,
**kwargs
)
def get_annotation(*args, **kwargs):
return Annotations([Annotation(*args, **kwargs).to_json_data()])
def get_dashboard_variable_query(name: str, template_query: str,
dataSource: str, **kwargs) -> Template:
return Template(name,
template_query,
dataSource=dataSource,
type="query",
includeAll=True,
multi=True,
**kwargs)
@dataclass
......@@ -35,15 +40,16 @@ class DashboardOptions:
timezone: str = "browser"
def build_row_repeat_dashboard(options: DashboardOptions,
row_repeat_var: Template,
panel_repeat_var: Template,
dataSource: str,
measurment_name: str,
panel_query: Query,
unit: str,
other_vars: List[Template] = [],
) -> Dashboard:
def build_row_repeat_dashboard(
options: DashboardOptions,
row_repeat_var: Template,
panel_repeat_var: Template,
dataSource: str,
measurment_name: str,
panel_query: Query,
unit: str,
other_vars: List[Template] = [],
) -> Dashboard:
"""Build a Dashboard that takes one query and repeats that with 2 variables."""
dashboard = Dashboard(
**asdict(options),
......@@ -52,11 +58,10 @@ def build_row_repeat_dashboard(options: DashboardOptions,
title=f'{row_repeat_var.name}: ${row_repeat_var.name}',
panels=[
TimeSeries(
title=f'{panel_repeat_var.name}: ${panel_repeat_var.name}',
title=
f'{panel_repeat_var.name}: ${panel_repeat_var.name}',
dataSource=dataSource,
targets=[
get_influx_target(str(panel_query))
],
targets=[get_influx_target(str(panel_query))],
repeat=Repeat('h', panel_repeat_var.name),
unit=unit,
pointSize=9,
......@@ -72,7 +77,8 @@ def build_row_repeat_dashboard(options: DashboardOptions,
def build_dashboard(options: DashboardOptions,
*, templating=[],
*,
templating=[],
**kwargs) -> Dashboard:
dashboard = Dashboard(
**asdict(options),
......@@ -88,8 +94,7 @@ def get_grid_pos(h: int, w: int, x: int, y: int):
return GridPos(h, w, x, y)
def pack_in_row(title: str,
panels: Union[Panel, List[Panel]],
def pack_in_row(title: str, panels: Union[Panel, List[Panel]],
**kwargs) -> RowPanel:
if isinstance(panels, Panel):
......@@ -123,8 +128,6 @@ def get_stat_panel(title: str,
return Stat(
title=title,
dataSource=dataSource,
targets=[
get_influx_target(str(stat_query), alias=alias)
],
targets=[get_influx_target(str(stat_query), alias=alias)],
**new_kwargs,
)
......@@ -5,7 +5,7 @@ from dashboards.dashboard_base import (DashboardOptions, build_dashboard,
build_row_repeat_dashboard,
get_dashboard_variable_query,
get_grid_pos, get_stat_panel,
pack_in_row)
pack_in_row, get_annotation)
from dashboards.influx_queries import (Query, get_variable_condition,
join_conditions, join_variable_and,
show_tag_values)
......@@ -150,13 +150,15 @@ def dashboard_phasefieldallenchan():
power_query,
repeat=panel_repeat_var,
format='W')
dashboard = build_row_repeat_dashboard(options, row_repeat_var,
panel_repeat_var, dataSource,
measurment_name, query, unit)
dashboard.rows = [
pack_in_row("Power Consumption", panel_power), *dashboard.rows
]
annotation_query = "SELECT \"commit\" from( SELECT first(\"mlupsPerProcess\") FROM \"PhaseFieldAllenCahn\" WHERE host =~ /^$host$/ AND $timeFilter GROUP BY \"commit\" )"
dashboard.annotations = get_annotation("InfluxDB-1", "red", "commits",
annotation_query)
return dashboard.auto_panel_ids()
......
from dashboards.annotations import Annotation
import json
expected_str = r"""
{
"dataSource": "InfluxDB-1",
"enable": true,
"iconColor": "red",
"name": "commit",
"query": "SELECT \"commit\" from( SELECT first(\"mlupsPerProcess\") FROM \"PhaseFieldAllenCahn\" WHERE host =~ /^$host$/ AND $timeFilter GROUP BY \"commit\" )",
"target": {
"limit": 100,
"matchAny": false,
"tags": [],
"type": "dashboard"
}
}
"""
def test_annotation():
query = "SELECT \"commit\" from( SELECT first(\"mlupsPerProcess\") FROM \"PhaseFieldAllenCahn\" WHERE host =~ /^$host$/ AND $timeFilter GROUP BY \"commit\" )"
anno = Annotation(
"InfluxDB-1",
"red",
"commit",
query,
)
expected = json.loads(expected_str)
assert expected == anno.to_json_data()
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment