log = logging.getLogger(__name__)
-OSM_METRIC_PREFIX = 'osm_'
+OSM_METRIC_PREFIX = "osm_"
class PrometheusBackend(BaseBackend):
-
def __init__(self, config: Config):
super().__init__(config)
self.conf = config
query = self._build_query(metric_name, tags)
request_url = self._build_url(query)
log.info("Querying Prometheus: %s", request_url)
- r = requests.get(request_url, timeout=int(self.conf.get('global', 'request_timeout')))
+ r = requests.get(
+ request_url, timeout=int(self.conf.get("global", "request_timeout"))
+ )
if r.status_code == 200:
json_response = r.json()
- if json_response['status'] == 'success':
+ if json_response["status"] == "success":
return self._get_metric_value_from_response(json_response)
else:
- log.warning("Prometheus response is not success. Got status %s", json_response['status'])
+ log.warning(
+ "Prometheus response is not success. Got status %s",
+ json_response["status"],
+ )
else:
- log.warning("Error contacting Prometheus. Got status code %s: %s", r.status_code, r.text)
+ log.warning(
+ "Error contacting Prometheus. Got status code %s: %s",
+ r.status_code,
+ r.text,
+ )
return None
def _build_query(self, metric_name: str, tags: dict) -> str:
query_section_tags = []
for k, v in tags.items():
- query_section_tags.append(k + '=\"' + v + '\"')
- query_section = "query={0}{{{1}}}".format(OSM_METRIC_PREFIX + metric_name, ','.join(query_section_tags))
+ query_section_tags.append(k + '="' + v + '"')
+ query_section = "query={0}{{{1}}}".format(
+ OSM_METRIC_PREFIX + metric_name, ",".join(query_section_tags)
+ )
return query_section
def _build_url(self, query: str):
- return self.conf.get('prometheus', 'url') + "/api/v1/query?" + query
+ return self.conf.get("prometheus", "url") + "/api/v1/query?" + query
def _get_metric_value_from_response(self, json_response):
- result = json_response['data']['result']
+ result = json_response["data"]["result"]
if len(result):
- metric_value = float(result[0]['value'][1])
+ metric_value = float(result[0]["value"][1])
log.info("Metric value: %s", metric_value)
return metric_value
else: