# For those usages not covered by the Apache License, Version 2.0 please
# contact: bdiaz@whitestack.com or glavado@whitestack.com
##
+import base64
import logging
+from typing import Dict
import requests
def get_metric_value(self, metric_name: str, tags: dict):
query = self._build_query(metric_name, tags)
request_url = self._build_url(query)
+ request_headers = self._build_headers()
+ return self._query_metric(request_url, request_headers)
+
+ def _query_metric(self, request_url: str, headers: Dict[str, str]):
log.info("Querying Prometheus: %s", request_url)
r = requests.get(
- request_url, timeout=int(self.conf.get("global", "request_timeout"))
+ request_url,
+ timeout=int(self.conf.get("global", "request_timeout")),
+ headers=headers,
)
if r.status_code == 200:
json_response = r.json()
def _build_url(self, query: str):
return self.conf.get("prometheus", "url") + "/api/v1/query?" + query
+ def _build_headers(self) -> Dict[str, str]:
+ headers = {}
+ user = self.conf.get("prometheus", "user")
+ password = self.conf.get("prometheus", "password")
+ if user and password:
+ _phrase = f"{user}:{password}".encode("utf-8")
+ token = base64.b64encode(_phrase).decode("utf-8")
+ headers["Authorization"] = f"Basic {token}"
+ return headers
+
def _get_metric_value_from_response(self, json_response):
result = json_response["data"]["result"]
if len(result):