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",