ceph/monitoring/ceph-mixin/tests_dashboards/__init__.py

   1 import re
   2 import subprocess
   3 import sys
   4 import tempfile
   5 from dataclasses import asdict, dataclass, field
   6 from typing import Any, List
   7
   8 import yaml
   9
  10 from .util import replace_grafana_expr_variables
  11
  12
  13 @dataclass
  14 class InputSeries:
  15     series: str = ''
  16     values: str = ''
  17
  18 @dataclass
  19 class ExprSample:
  20     labels: str = ''
  21     value: float = -1
  22
  23 @dataclass
  24 class PromqlExprTest:
  25     expr: str = ''
  26     eval_time: str = '1m'
  27     exp_samples: List[ExprSample] = field(default_factory=list)
  28
  29 @dataclass
  30 class Test:
  31     interval: str = '1m'
  32     input_series: List[InputSeries] = field(default_factory=list)
  33     promql_expr_test: List[PromqlExprTest] = field(default_factory=list)
  34
  35
  36 @dataclass
  37 class TestFile:
  38     evaluation_interval: str = '1m'
  39     tests: List[Test] = field(default_factory=list)
  40
  41
  42 class PromqlTest:
  43     """
  44     Base class to provide prometheus query test capabilities. After setting up
  45     the query test with its input and expected output it's expected to run promtool.
  46
  47     https://prometheus.io/docs/prometheus/latest/configuration/unit_testing_rules/#test-yml
  48
  49     The workflow of testing would be something like:
  50
  51         # add prometheus query to test
  52         self.set_expression('bonding_slaves > 0')
  53
  54         # add some prometheus input series
  55         self.add_series('bonding_slaves{master="bond0"}', '2')
  56         self.add_series('bonding_slaves{master="bond1"}', '3')
  57         self.add_series('node_network_receive_bytes{instance="127.0.0.1",
  58             device="eth1"}', "10 100 230 22")
  59
  60         # expected output of the query
  61         self.add_exp_samples('bonding_slaves{master="bond0"}', 2)
  62         self.add_exp_samples('bonding_slaves{master="bond1"}', 3)
  63
  64         # at last, always call promtool with:
  65         self.assertTrue(self.run_promtool())
  66         # assertTrue means it expect promtool to succeed
  67     """
  68
  69     def __init__(self):
  70         self.test_output_file = tempfile.NamedTemporaryFile('w+')
  71
  72         self.test_file = TestFile()
  73         self.test = Test()
  74         self.promql_expr_test = PromqlExprTest()
  75         self.test.promql_expr_test.append(self.promql_expr_test)
  76         self.test_file.tests.append(self.test)
  77
  78         self.variables = {}
  79
  80     def __del__(self):
  81         self.test_output_file.close()
  82
  83
  84     def set_evaluation_interval(self, interval: int, unit: str = 'm') -> None:
  85         """
  86         Set the evaluation interval of the time series
  87
  88         Args:
  89             interval (int): number of units.
  90             unit (str): unit type: 'ms', 's', 'm', etc...
  91         """
  92         self.test_file.evaluation_interval = f'{interval}{unit}'
  93
  94     def set_interval(self, interval: int, unit: str = 'm') -> None:
  95         """
  96         Set the duration of the time series
  97
  98         Args:
  99             interval (int): number of units.
 100             unit (str): unit type: 'ms', 's', 'm', etc...
 101         """
 102         self.test.interval = f'{interval}{unit}'
 103
 104     def set_expression(self, expr: str) -> None:
 105         """
 106         Set the prometheus expression/query used to filter data.
 107
 108         Args:
 109              expr(str): expression/query.
 110         """
 111         self.promql_expr_test.expr = expr
 112
 113     def add_series(self, series: str, values: str) -> None:
 114         """
 115         Add a series to the input.
 116
 117         Args:
 118              series(str): Prometheus series.
 119              Notation: '<metric name>{<label name>=<label value>, ...}'
 120              values(str): Value of the series.
 121         """
 122         input_series = InputSeries(series=series, values=values)
 123         self.test.input_series.append(input_series)
 124
 125     def set_eval_time(self, eval_time: int, unit: str = 'm') -> None:
 126         """
 127         Set the time when the expression will be evaluated
 128
 129         Args:
 130              interval (int): number of units.
 131              unit (str): unit type: 'ms', 's', 'm', etc...
 132         """
 133         self.promql_expr_test.eval_time = f'{eval_time}{unit}'
 134
 135     def add_exp_samples(self, sample: str, values: Any) -> None:
 136         """
 137         Add an expected sample/output of the query given the series/input
 138
 139         Args:
 140              sample(str): Expected sample.
 141              Notation: '<metric name>{<label name>=<label value>, ...}'
 142              values(Any): Value of the sample.
 143         """
 144         expr_sample = ExprSample(labels=sample, value=values)
 145         self.promql_expr_test.exp_samples.append(expr_sample)
 146
 147     def set_variable(self, variable: str, value: str):
 148         """
 149         If a query makes use of grafonnet variables, for example
 150         '$osd_hosts', you should change this to a real value. Example:
 151
 152
 153         > self.set_expression('bonding_slaves{master="$osd_hosts"} > 0')
 154         > self.set_variable('osd_hosts', '127.0.0.1')
 155         > print(self.query)
 156         > bonding_slaves{master="127.0.0.1"} > 0
 157
 158         Args:
 159              variable(str): Variable name
 160              value(str): Value to replace variable with
 161
 162         """
 163         self.variables[variable] = value
 164
 165     def run_promtool(self):
 166         """
 167         Run promtool to test the query after setting up the input, output
 168         and extra parameters.
 169
 170         Returns:
 171              bool: True if successful, False otherwise.
 172         """
 173
 174         for variable, value in self.variables.items():
 175             expr = self.promql_expr_test.expr
 176             new_expr = replace_grafana_expr_variables(expr, variable, value)
 177             self.set_expression(new_expr)
 178
 179         test_as_dict = asdict(self.test_file)
 180         yaml.dump(test_as_dict, self.test_output_file)
 181
 182         args = f'promtool test rules {self.test_output_file.name}'.split()
 183         try:
 184             subprocess.run(args, check=True)
 185             return True
 186         except subprocess.CalledProcessError as process_error:
 187             print(yaml.dump(test_as_dict))
 188             print(process_error.stderr)
 189             return False