+
+ 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)
+ )
+ return query_section
+
+ 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_data_from_response(self, json_response) -> List[Dict[str, str]]:
+ result = json_response["data"]["result"]
+ metrics_data = []
+ if len(result):
+ for metric in result:
+ metrics_labels = metric["metric"]
+ metric_value = float(metric["value"][1])
+ log.info("Metric value: %s", metric_value)
+ metrics_data.append({"labels": metrics_labels, "value": metric_value})
+ return metrics_data
+ else:
+ return None