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