prometheus

Prometheus module.

exception wmflib.prometheus.PrometheusError[source]

Bases: WmflibError

Custom exception class for errors of this module.

class wmflib.prometheus.PrometheusBase[source]

Bases: object

Base class to interact with Prometheus-like APIs.

Initialize the instance.

class wmflib.prometheus.Prometheus[source]

Bases: PrometheusBase

Class to interact with a Prometheus API instance.

Examples

>>> from wmflib.prometheus import Prometheus
>>> prometheus = Prometheus()

Initialize the instance.

query(query: str, site: str, *, instance: str = 'ops', timeout: float | Tuple[float, float] = 10.0) → List[Dict][source]

Perform a generic query.

Examples

>>> results = prometheus.query('node_uname_info{instance=~"host1001:.*"}', 'eqiad', instance='global')
>>> results = prometheus.query('node_memory_MemTotal_bytes{instance=~"host1001:.*"}', 'eqiad')

The content of the last results will be something like:

[
    {
        'metric': {
            '__name__': 'node_memory_MemTotal_bytes',
            'cluster': 'management',
            'instance': 'host1001:9100',
            'job': 'node',
            'site': 'eqiad'
        },
        'value': [1636569623.988, '67225329664']
    }
]

Parameters:

query (str) – a prometheus query string.
site (str) – The site to use for queries. Must be one of wmflib.constants.ALL_DATACENTERS
instance (str, optional) – The prometheus instance to query on the given site, see https://wikitech.wikimedia.org/wiki/Prometheus#Instances for the full list of available instances.
timeout (wmflib.requests.TimeoutType, optional) – How many seconds to wait for prometheus to reply before giving up. This is passed directly to the requests library.

Returns:

returns an empty list if there are no results otherwise return a list of results of the form: {'metric': {}, 'value': [$timestamp, $value]}.

Return type:

list

Raises:

wmflib.prometheus.PrometheusError – on error

class wmflib.prometheus.Thanos[source]

Bases: PrometheusBase

Class to interact with a Thanos API endpoint.

Examples

>>> from wmflib.prometheus import Thanos
>>> thanos = Thanos()

Initialize the instance.

query(query: str, *, timeout: float | Tuple[float, float] = 10.0) → List[Dict][source]

Perform a generic query.

Examples

>>> results = thanos.query('node_memory_MemTotal_bytes{instance=~"host1001:.*"}')
>>> results = thanos.query('node_uname_info{instance=~"host1001:.*"}')

The content of the last results will be something like:

[
    {
        'metric': {
            '__name__': 'node_uname_info',
            'cluster': 'management',
            'domainname': '(none)',
            'instance': 'host1001:9100',
            'job': 'node',
            'machine': 'x86_64',
            'nodename': 'host1001',
            'prometheus': 'ops',
            'release': '5.10.0-11-amd64',
            'site': 'eqiad',
            'sysname': 'Linux',
            'version': '#1 SMP Debian 5.10.92-2 (2022-02-28)'
        },
        'value': [1648898872.82, '1']
    }
]

Parameters:

query (str) – a prometheus query string.
timeout (wmflib.requests.TimeoutType, optional) – How many seconds to wait for prometheus to reply before giving up. This is passed directly to the requests library.

Returns:

returns an empty list if there are no results otherwise return a list of results of the form: {'metric': {}, 'value': [$timestamp, $value]}.

Return type:

list

Raises:

wmflib.prometheus.PrometheusError – on error.