| # -*- coding: utf-8 -*- | |
| # Copyright 2018 Whitestack, LLC | |
| # ************************************************************* | |
| # This file is part of OSM Monitoring module | |
| # All Rights Reserved to Whitestack, LLC | |
| # Licensed under the Apache License, Version 2.0 (the "License"); you may | |
| # not use this file except in compliance with the License. You may obtain | |
| # a copy of the License at | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT | |
| # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the | |
| # License for the specific language governing permissions and limitations | |
| # under the License. | |
| # For those usages not covered by the Apache License, Version 2.0 please | |
| # contact: glavado@whitestack.com or fbravo@whitestack.com | |
| ## | |
| import logging | |
| import requests | |
| import base64 | |
| import json | |
| from osm_mon.core.config import Config | |
| log = logging.getLogger(__name__) | |
| class GrafanaBackend: | |
| def __init__(self, config: Config): | |
| self.conf = config | |
| self.url = config.get('grafana', 'url') | |
| grafana_user = config.get("grafana", "user") | |
| grafana_password = config.get("grafana", "password") | |
| self.headers = { | |
| 'content-type': "application/json", | |
| 'authorization': "Basic %s" % base64.b64encode( | |
| (grafana_user + ":" + grafana_password).encode("utf-8")).decode() | |
| } | |
| def get_all_dashboard_uids(self): | |
| # Gets only dashboards that were automated by OSM (with tag 'osm_automated') | |
| response = requests.request("GET", self.url + "/api/search?tag=osm_automated", headers=self.headers) | |
| dashboards = response.json() | |
| dashboard_uids = [] | |
| for dashboard in dashboards: | |
| dashboard_uids.append(dashboard['uid']) | |
| log.debug("Searching for all dashboard uids: %s", dashboard_uids) | |
| return dashboard_uids | |
| def get_dashboard_status(self, uid): | |
| response = requests.request("GET", self.url + "/api/dashboards/uid/" + uid, headers=self.headers) | |
| log.debug("Searching for dashboard result: %s", response.text) | |
| return response | |
| def create_dashboard(self, uid, name, json_file, project_name=None): | |
| try: | |
| with open(json_file) as f: | |
| dashboard_data = f.read() | |
| dashboard_data = dashboard_data.replace('OSM_ID', uid).replace('OSM_NAME', name) | |
| response = requests.request( | |
| "POST", self.url + "/api/dashboards/db/", data=dashboard_data, headers=self.headers) | |
| # get team id | |
| if project_name is not None: | |
| name = project_name | |
| response_team = requests.request( | |
| "GET", self.url + "/api/teams/search?name={}".format(name), headers=self.headers) | |
| if len(json.loads(response_team.text)["teams"]) > 0: | |
| team_id = json.loads(response_team.text)["teams"][0]["id"] | |
| permission_data = {"items": [{"teamId": team_id, "permission": 2}, ]} | |
| # provide permission to dashboard | |
| dahboard_id = json.loads(response.text)["id"] | |
| response = requests.request( | |
| "POST", self.url + "/api/dashboards/id/{}/permissions".format(dahboard_id), json=permission_data, | |
| headers=self.headers) | |
| return response | |
| except Exception: | |
| log.exception("Exception processing message: ") | |
| def delete_dashboard(self, uid): | |
| response = requests.request("DELETE", self.url + "/api/dashboards/uid/" + uid, headers=self.headers) | |
| log.debug("Delete dashboard result: %s", response.text) | |
| return response | |
| def create_grafana_users(self, user): | |
| email = "{}@osm.etsi.org".format(user) | |
| user_payload = { | |
| "name": user, | |
| "email": email, | |
| "login": user, | |
| "password": user, | |
| } | |
| response_users = requests.request("POST", self.url + "/api/admin/users/", json=user_payload, | |
| headers=self.headers) | |
| json_data = json.loads(response_users.text) | |
| url = "/api/org/users/{}/".format(json_data["id"]) | |
| permission_payload = {"role": "Editor", } | |
| requests.request("PATCH", self.url + url, json=permission_payload, headers=self.headers) | |
| log.info("Grafana user created: %s", response_users.text) | |
| return response_users | |
| # create grafana team with member | |
| def create_grafana_teams_members(self, project_name, user_name, is_admin, proj_list): | |
| # check if user exist in grafana or not | |
| user_response = requests.request("GET", self.url + "/api/users/lookup?loginOrEmail={}".format(user_name), | |
| headers=self.headers) | |
| user_obj = json.loads(user_response.text) | |
| if user_response.status_code != 200: | |
| user_response = self.create_grafana_users(user_name) | |
| user_obj = json.loads(user_response.text) | |
| user_id = user_obj["id"] | |
| # Get Teams for user | |
| team_objs = requests.request("GET", self.url + "/api/users/{}/teams".format(user_id), headers=self.headers) | |
| team_obj = json.loads(team_objs.text) | |
| team_list = [] | |
| if len(team_obj): | |
| for team in team_obj: | |
| team_list.append(team["name"]) | |
| proj_unlink = set(team_list) - set(proj_list) | |
| for prj in proj_unlink: | |
| response_team = requests.request("GET", self.url + "/api/teams/search?name={}".format(prj), | |
| headers=self.headers) | |
| team_id = json.loads(response_team.text)["teams"][0]["id"] | |
| requests.request("DELETE", self.url + "/api/teams/{}/members/{}".format(team_id, user_id), | |
| headers=self.headers) | |
| # add member to team | |
| response_team = requests.request("GET", self.url + "/api/teams/search?name={}".format(project_name), | |
| headers=self.headers) | |
| team_id = json.loads(response_team.text)["teams"][0]["id"] | |
| if project_name not in team_list: | |
| member_payload = { | |
| "userId": user_id | |
| } | |
| requests.request("POST", self.url + "/api/teams/{}/members".format(team_id), json=member_payload, | |
| headers=self.headers) | |
| # if role is admin change permission to admin | |
| if is_admin: | |
| url = "/api/org/users/{}/".format(user_id) | |
| permission_payload = {"role": "Admin", } | |
| requests.request("PATCH", self.url + url, json=permission_payload, headers=self.headers) | |
| log.info("Member added to grafana team") | |
| return response_team | |
| # create grafana team | |
| def create_grafana_teams(self, team_name): | |
| team_payload = {"name": team_name, } | |
| response = requests.request("POST", self.url + "/api/teams", json=team_payload, headers=self.headers) | |
| log.info("Grafana Team is created: %s", response.text) | |
| def delete_grafana_users(self, user_name): | |
| # find user id | |
| response_id = requests.request("GET", self.url + "/api/users/lookup?loginOrEmail={}".format(user_name), | |
| headers=self.headers) | |
| try: | |
| user_id = json.loads(response_id.text)["id"] | |
| except Exception: | |
| log.exception("Exception processing message: ") | |
| # delete user | |
| response = requests.request("DELETE", self.url + "/api/admin/users/{}".format(user_id), headers=self.headers) | |
| log.info("Grafana user deleted: %s", response.text) | |
| return response | |
| def delete_grafana_team(self, project_name): | |
| team_obj = requests.request("GET", self.url + "/api/teams/search?name={}".format(project_name), | |
| headers=self.headers) | |
| team_id = json.loads(team_obj.text)["teams"][0]["id"] | |
| response = requests.request("DELETE", self.url + "/api/teams/{}".format(team_id), headers=self.headers) | |
| log.info("Grafana team deleated: %s", response.text) | |
| return response | |
| def update_grafana_teams(self, project_new_name, project_old_name): | |
| team_obj = requests.request("GET", self.url + "/api/teams/search?name={}".format(project_old_name), | |
| headers=self.headers) | |
| team_id = json.loads(team_obj.text)["teams"][0]["id"] | |
| data = {"name": project_new_name, } | |
| response = requests.request("PUT", self.url + "/api/teams/{}".format(team_id), json=data, headers=self.headers) | |
| log.info("Grafana team updated %s", response.text) | |
| return response |