84 lines
2.3 KiB
Python
Executable File
84 lines
2.3 KiB
Python
Executable File
from datetime import datetime, timedelta
|
|
from urllib.parse import urljoin
|
|
|
|
import requests
|
|
|
|
|
|
class PromAPI:
|
|
def __init__(self, endpoint='http://127.0.0.1:9090/'):
|
|
"""
|
|
:param endpoint: address of
|
|
"""
|
|
self.endpoint = endpoint
|
|
|
|
@staticmethod
|
|
def _to_timestamp(input_):
|
|
"""
|
|
Convert string input to UNIX timestamp for Prometheus
|
|
:param input_:
|
|
:return:
|
|
"""
|
|
if type(input_) == datetime:
|
|
return input_.timestamp()
|
|
if input_ == 'now':
|
|
return datetime.utcnow().isoformat('T')
|
|
if type(input_) is str:
|
|
input_ = float(input_)
|
|
if type(input_) in [int, float]:
|
|
if input_ > 0:
|
|
return input_
|
|
if input_ == 0: # return now
|
|
return datetime.utcnow().isoformat('T')
|
|
if input_ < 0:
|
|
return (datetime.utcnow() + timedelta(seconds=input_)).isoformat('T')
|
|
#assert type(input_) == float
|
|
|
|
def query(self, query='prometheus_build_info'):
|
|
return self._get(
|
|
uri='/api/v1/query',
|
|
params=dict(
|
|
query=query
|
|
)
|
|
)
|
|
|
|
def query_range(self, query='prometheus_build_info', start=-60, end='now', duration=60):
|
|
"""Get ser"""
|
|
params = {
|
|
'query': query
|
|
}
|
|
if end is not None:
|
|
params['end'] = self._to_timestamp(end) + 'Z'
|
|
if start:
|
|
params['start'] = self._to_timestamp(start) + 'Z'
|
|
if duration:
|
|
params['step'] = duration
|
|
print(params)
|
|
return self._get(
|
|
uri='/api/v1/query_range',
|
|
params=params
|
|
)
|
|
|
|
def series(self, match='prometheus_build_info', start=-86400, end='now'):
|
|
"""Get ser"""
|
|
params = {
|
|
'match[]': match
|
|
}
|
|
if end is not None:
|
|
params['end'] = self._to_timestamp(end) + 'Z'
|
|
if start:
|
|
params['start'] = self._to_timestamp(start) + 'Z'
|
|
print(params)
|
|
return self._get(
|
|
uri='/api/v1/series',
|
|
params=params
|
|
)
|
|
|
|
def _get(self, uri, params, method='GET'):
|
|
url = urljoin(self.endpoint, uri)
|
|
assert method == 'GET'
|
|
result = requests.get(
|
|
url=url,
|
|
params=params
|
|
)
|
|
return result.json()
|