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:
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'