installers/charm/osm-mon/src/grafana_datasource_handler.py

   1 #!/usr/bin/env python3
   2 # Copyright 2022 Canonical Ltd.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 # not use this file except in compliance with the License. You may obtain
   6 # a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 # License for the specific language governing permissions and limitations
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: legal@canonical.com
  18 #
  19 # To get in touch with the maintainers, please contact:
  20 # osm-charmers@lists.launchpad.net
  21 #
  22 #
  23 # Learn more at: https://juju.is/docs/sdk
  24
  25 """Module to handle Grafana datasource operations."""
  26
  27 import json
  28 import logging
  29 from dataclasses import dataclass
  30
  31 import requests
  32 from requests.auth import HTTPBasicAuth
  33
  34 logger = logging.getLogger(__name__)
  35
  36
  37 @dataclass
  38 class GrafanaConfig:
  39     """Values needed to make grafana API calls."""
  40
  41     user: str
  42     password: str
  43     url: str
  44
  45
  46 @dataclass
  47 class DatasourceConfig:
  48     """Information about the datasource to create."""
  49
  50     name: str
  51     url: str
  52
  53
  54 @dataclass
  55 class DatasourceResponse:
  56     """Return value used by GrafanaDataSourceHandler operations."""
  57
  58     is_success: bool
  59     message: str
  60     results: dict
  61
  62
  63 class GrafanaDataSourceHandler:
  64     """Handle grafana datasource pperations."""
  65
  66     @staticmethod
  67     def create_datasource(
  68         grafana_config: GrafanaConfig, datasource_config: DatasourceConfig
  69     ) -> DatasourceResponse:
  70         """Calls the Grafana API to create a new prometheus datasource."""
  71         try:
  72             auth = HTTPBasicAuth(grafana_config.user, grafana_config.password)
  73             url = grafana_config.url + "/api/datasources"
  74             datasource_data = {
  75                 "name": datasource_config.name,
  76                 "type": "prometheus",
  77                 "url": datasource_config.url,
  78                 "access": "proxy",
  79                 "readOnly": False,
  80                 "basicAuth": False,
  81             }
  82             response = requests.post(url, json=datasource_data, auth=auth)
  83             response_content = response.json()
  84             results = {"datasource-name": response_content.get("name")}
  85             return DatasourceResponse(
  86                 response.ok, response_content.get("message"), results=results
  87             )
  88         except Exception as e:
  89             logger.debug(f"Exception processing request for creating datasource: {e}")
  90             return DatasourceResponse(False, str(e), results={})
  91
  92     @staticmethod
  93     def list_datasources(grafana_config: GrafanaConfig) -> DatasourceResponse:
  94         """Calls the Grafana API to get a list of datasources."""
  95         try:
  96             auth = HTTPBasicAuth(grafana_config.user, grafana_config.password)
  97             url = grafana_config.url + "/api/datasources"
  98             response = requests.get(url, auth=auth)
  99             response_content = response.json()
 100             results = {"datasources": json.dumps(response_content)}
 101             message = response_content.get("message") if not response.ok else ""
 102             return DatasourceResponse(response.ok, message=message, results=results)
 103         except Exception as e:
 104             logger.debug(f"Exception processing request to list datasources: {e}")
 105             return DatasourceResponse(False, str(e), results={})
 106
 107     @staticmethod
 108     def delete_datasource(
 109         grafana_config: GrafanaConfig, datasource_name: str
 110     ) -> DatasourceResponse:
 111         """Calls the Grafana API to delete a given datasource by name."""
 112         try:
 113             auth = HTTPBasicAuth(grafana_config.user, grafana_config.password)
 114             url = grafana_config.url + f"/api/datasources/name/{datasource_name}"
 115             response = requests.delete(url, auth=auth)
 116             response_content = response.json()
 117             return DatasourceResponse(response.ok, response_content.get("message"), results={})
 118         except Exception as e:
 119             logger.debug(f"Exception processing request for deleting datasource: {e}")
 120             return DatasourceResponse(False, str(e), results={})