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