prometheus

Prometheus module.

exception wmflib.prometheus.PrometheusError[source]

Bases: WmflibError

Custom exception class for errors of this module.

__doc__ = 'Custom exception class for errors of this module.'

__module__ = 'wmflib.prometheus'

class wmflib.prometheus.PrometheusBase[source]

Bases: object

Base class to interact with Prometheus-like APIs.

Initialize the instance.

__init__() → None[source]: Initialize the instance.

_query(url: str, params: Dict[str, str], timeout: float | Tuple[float, float]) → List[Dict][source]

Perform a generic query.

Parameters:

url (str) – the URL to query.
params (dict) – a dictionary of the GET parameters to pass to the URL.
timeout (wmflib.requests.TimeoutType) – 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

__dict__ = mappingproxy({'__module__': 'wmflib.prometheus', '__doc__': 'Base class to interact with Prometheus-like APIs.', '__init__': <function PrometheusBase.__init__>, '_query': <function PrometheusBase._query>, '__dict__': <attribute '__dict__' of 'PrometheusBase' objects>, '__weakref__': <attribute '__weakref__' of 'PrometheusBase' objects>, '__annotations__': {}})

__doc__ = 'Base class to interact with Prometheus-like APIs.'

__module__ = 'wmflib.prometheus'

__weakref__: list of weak references to the object (if defined)

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.

_prometheus_api: str = 'http://prometheus.svc.{site}.wmnet/{instance}/api/v1/query'

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_memory_MemTotal_bytes{instance=~"host1001:.*"}', "eqiad")
>>> results = prometheus.query(
...     'kube_deployment_created{deployment="mw-web.eqiad.main"}', "eqiad", instance="k8s")

The content of the first 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

__annotations__ = {'_prometheus_api': <class 'str'>}

__doc__ = 'Class to interact with a Prometheus API instance.\n\n Examples:\n ::\n\n >>> from wmflib.prometheus import Prometheus\n >>> prometheus = Prometheus()\n\n '

__module__ = 'wmflib.prometheus'

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.

_thanos_api: str = 'https://thanos-query.discovery.wmnet/api/v1/query'

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.

__annotations__ = {'_thanos_api': <class 'str'>}

__doc__ = 'Class to interact with a Thanos API endpoint.\n\n Examples:\n ::\n\n >>> from wmflib.prometheus import Thanos\n >>> thanos = Thanos()\n\n '

__module__ = 'wmflib.prometheus'