prometheus.krkn_prometheus module

class prometheus.krkn_prometheus.KrknPrometheus(prometheus_url: str, prometheus_bearer_token: str | None = None)

Bases: object

__init__(prometheus_url: str, prometheus_bearer_token: str | None = None)

Instantiates a KrknPrometheus class with the Prometheus API Endpoint url and the bearer token to access it (optional if the endpoint doesn’t need authentication).

Parameters:

  • prometheus_url – the prometheus API endpoint

  • prometheus_bearer_token – the bearer token to authenticate the query (optional).

parse_metric(description: str, record: dict[slice(<class 'str'>, <built-in function any>, None)]) → str

Parses the expression contained in the Krkn alert description replacing them with the respective values contained in the record previously returned by a PromQL query.

Parameters:

  • description – the description containing the expression in the Krkn alert format

  • record – the PromQL record from where the data will be extracted

Returns:

the description with the expressions replaced by the correct values

process_alert(alert: dict[str, str], start_time: datetime, end_time: datetime)

Processes Krkn alarm in the format

expr: <promQL query> description: <the message that will be logged in the console> severity <the log level that will be used>

Description: The description may contain two kind of expressions that will be properly replaced and valorized; the supported expression are:

  • {{$value}} the scalar value returned by the query (if the query

is designed to return a scalar value

  • {{$label.<label_name>}} one of the metric properties

returned by the query

If a value is not found in the the description won’t be replaced.

Severity: The severity represents the log level that will be used to print the description in the console. The supported levels are: - info - debug - warning - error - critical

If a non existing value is set an error message will be print instead of the the description.

Params alert:

a dictionary containing the following keys : expr, description, severity

Parameters:

  • start_time – start time of the result set (if None no time filter is applied to the query)

  • end_time – end time of the result set (if None no time filter is applied to the query)

Returns:

returns the alert log line as a string to be uploaded as telemetry metadata, None if no alert has been selected

process_prom_query_in_range(query: str, start_time: datetime | None = None, end_time: datetime | None = None, granularity: int = 10) → list[dict[slice(<class 'str'>, <built-in function any>, None)]]

Executes a query to the Prometheus API in PromQL languag, between a start and end time

Parameters:

  • query – promQL query

  • start_time – start time of the result set (default now - 1 day)

  • end_time – end time of the result set (default min datetime)

Returns:

a list of records in dictionary format

process_query(query: str) → list[dict[slice(<class 'str'>, <built-in function any>, None)]]

Executes a query to the Prometheus API in PromQL language

Parameters:

query – promQL query

Returns:

a list of records in dictionary format

prom_cli: PrometheusConnect