Support data source health check endpoint introduced with Grafana 9

CHANGELOG.md

@@ -6,6 +6,8 @@
 * Improve data source API by adding the `_by_uid` variants.
 * Improve data source API by adding universal `datasource.get()` method.
 * Improve data source API by adding a data source health-check probe.
+* Support data source health check endpoint introduced with Grafana 9.
+  Thanks, @jangaraj!
 
 
 ## 2.3.0 (2022-05-26)

examples/datasource-health.py

@@ -7,9 +7,11 @@
 import logging
 import os
 import sys
-from optparse import OptionParser
+import warnings
 
-from pkg_resources import parse_version
+warnings.filterwarnings("ignore", category=Warning, message="distutils Version classes are deprecated")
+from distutils.version import LooseVersion
+from optparse import OptionParser
 
 from grafana_client import GrafanaApi
 from grafana_client.client import GrafanaClientError
@@ -20,6 +22,11 @@
 logger = logging.getLogger(__name__)
 
 
+VERSION_7 = LooseVersion("7")
+VERSION_8 = LooseVersion("8")
+VERSION_9 = LooseVersion("9")
+
+
 def ensure_datasource(grafana: GrafanaApi, datasource: DatasourceModel):
     """
     Ensure data source exists and is configured like desired.
@@ -51,7 +58,7 @@ def ensure_datasource(grafana: GrafanaApi, datasource: DatasourceModel):
     return datasource
 
 
-def health_inquiry(grafana: GrafanaApi, datasource: DatasourceModel):
+def health_inquiry(grafana: GrafanaApi, datasource: DatasourceModel, grafana_version: LooseVersion = None):
     """
     Add a data source dynamically and run a data source health check on it.
     Be graceful if it exists already.
@@ -60,12 +67,20 @@ def health_inquiry(grafana: GrafanaApi, datasource: DatasourceModel):
     # Create data source.
     datasource = ensure_datasource(grafana, datasource)
     datasource_id = datasource["id"]
+    datasource_uid = datasource["uid"]
 
     # Resolve data source by UID.
     datasource = grafana.datasource.get(datasource_id=datasource_id)
 
     # Check data source health.
-    health = grafana.datasource.health_check(datasource=datasource)
+    if True or grafana_version and grafana_version >= VERSION_9:
+        try:
+            health = grafana.datasource.health(datasource_uid=datasource_uid)
+        except GrafanaClientError as ex:
+            logger.error(f"Data source health check for uid={datasource_uid} failed: {ex}. Response: {ex.response}")
+            raise
+    else:
+        health = grafana.datasource.health_check(datasource=datasource)
 
     # Delete data source again.
     grafana.datasource.delete_datasource_by_id(datasource_id)
@@ -81,11 +96,11 @@ def prometheus_demo(grafana: GrafanaApi):
     return health_info
 
 
-def run_healthcheck(grafana: GrafanaApi, grafana_version: str = None):
+def run_healthcheck(grafana: GrafanaApi, grafana_version: LooseVersion = None):
 
     # When called without options, invoke the Prometheus demo.
     if len(sys.argv) == 1:
-        if grafana_version < parse_version("8"):
+        if grafana_version < VERSION_8:
             raise NotImplementedError(
                 f"Data source health check subsystem on Grafana version {grafana_version} not supported for Prometheus"
             )
@@ -102,7 +117,7 @@ def run_healthcheck(grafana: GrafanaApi, grafana_version: str = None):
             parser.error("Options --type and --url required")
 
         # Sanity checks
-        if options.type == "prometheus" and grafana_version < parse_version("8"):
+        if options.type == "prometheus" and grafana_version < VERSION_8:
             raise NotImplementedError(
                 f"Data source health check subsystem on Grafana version {grafana_version} not supported for Prometheus"
             )
@@ -113,7 +128,7 @@ def run_healthcheck(grafana: GrafanaApi, grafana_version: str = None):
         datasource = datasource_factory(datasource)
 
         # Invoke the health check.
-        health_info = health_inquiry(grafana, datasource)
+        health_info = health_inquiry(grafana, datasource, grafana_version=grafana_version)
 
     # Display the outcome and terminate program based on success state.
     print(json.dumps(health_info, indent=2))
@@ -131,10 +146,10 @@ def run_healthcheck(grafana: GrafanaApi, grafana_version: str = None):
     grafana_client = grafana_client_factory(grafana_url=grafana_url, grafana_token=os.environ.get("GRAFANA_TOKEN"))
 
     grafana_info = grafana_client.health.check()
-    grafana_version = parse_version(grafana_info["version"])
+    grafana_version = LooseVersion(grafana_info["version"])
     logger.info(f"Connected to Grafana version {grafana_version} at {grafana_url}")
 
-    if grafana_version < parse_version("7"):
+    if grafana_version < VERSION_7:
         raise NotImplementedError(f"Data source health check subsystem not ready for Grafana version {grafana_version}")
 
     run_healthcheck(grafana_client, grafana_version=grafana_version)

examples/datasource-health.rst

@@ -43,6 +43,7 @@ Start Grafana::
     export GRAFANA_VERSION=7.5.16
     export GRAFANA_VERSION=8.5.6
     export GRAFANA_VERSION=9.0.0
+    export GRAFANA_VERSION=main
 
     docker run --rm -it \
         --publish=3000:3000 \

grafana_client/elements/datasource.py

@@ -18,6 +18,19 @@ def __init__(self, client):
         super(Datasource, self).__init__(client)
         self.client = client
 
+    def health(self, datasource_uid: str):
+        """
+        Makes a call to the health endpoint of a data source identified by the
+        given ``uid``.
+
+        Available in Grafana 9+.
+
+        https://grafana.com/docs/grafana/latest/developers/http_api/data_source/#check-data-source-health
+        """
+        path = f"/datasources/uid/{datasource_uid}/health"
+        r = self.client.GET(path)
+        return r
+
     def find_datasource(self, datasource_name):
         """