log.debug("Searching for all dashboard uids: %s", dashboard_uids)
return dashboard_uids
+ def get_all_datasource_names(self, datasource_name_substr):
+ # Gets only dashboards that were created for prom-operator
+ response = requests.request(
+ "GET", self.url + "/api/datasources", headers=self.headers
+ )
+ datasources = response.json()
+ datasource_names = []
+ for datasource in datasources:
+ if datasource["name"].startswith(datasource_name_substr):
+ datasource_names.append(datasource["name"])
+ log.debug("Searching for all datasource names: %s", datasource_names)
+ return datasource_names
+
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):
+ def create_dashboard(
+ self, uid, name, json_file, project_name=None, datasource_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
)
+ if datasource_name:
+ dashboard_data = dashboard_data.replace(
+ "OSM_DATASOURCE_NAME", datasource_name
+ )
dashboard_json_data = json.loads(dashboard_data)
# Get folder id
if project_name:
except Exception:
log.exception("Exception processing message: ")
+ def create_datasource(self, datasource_name, datasource_type, datasource_url):
+ try:
+ datasource_data = {
+ "name": datasource_name,
+ "type": datasource_type,
+ "url": datasource_url,
+ "access": "proxy",
+ "readOnly": False,
+ "basicAuth": False,
+ }
+ response = requests.request(
+ "POST",
+ self.url + "/api/datasources",
+ data=json.dumps(datasource_data),
+ headers=self.headers,
+ )
+ log.info("Datasource %s is created in Grafana", datasource_name)
+ log.info("************* response: {}".format(response.__dict__))
+ return response
+ except Exception:
+ log.exception("Exception processing request for creating datasource: ")
+
def send_request_for_creating_dashboard(self, dashboard_data):
response = requests.request(
"POST",
log.debug("Dashboard %s deleted from Grafana", uid)
return response
+ def delete_datasource(self, datasource_name):
+ response = requests.request(
+ "DELETE",
+ self.url + "/api/datasources/name/" + datasource_name,
+ headers=self.headers,
+ )
+ log.debug("Datasource %s deleted from Grafana", datasource_name)
+ return response
+
def delete_admin_dashboard(self):
requests.request(
"DELETE",
log.info("New user %s created in Grafana", user)
return response_users
+ # Get Grafana users
+ def get_grafana_users(self):
+ response_users = requests.request(
+ "GET",
+ self.url + "/api/users",
+ headers=self.headers,
+ )
+ user_list = []
+ users = json.loads(response_users.text)
+ for user in users:
+ if user["name"] and user["name"] != "admin":
+ user_list.append(user["name"])
+ return user_list
+
# Create Grafana team with member
def create_grafana_teams_members(
self, project_name, user_name, is_admin, proj_list