[amg] Azure managed grafana options for notification channels

src/amg/README.md

@@ -26,7 +26,7 @@ az grafana delete \
     -n MyGrafanaInstance
 ```
 
-### Configure folder, data sources and dashboard
+### Configure folders, data sources, notification channels and dashboards
 
 #### create a folder
 *Examples:*
@@ -44,6 +44,14 @@ az grafana data-source create \
     --definition ~/data-source-sql.json
 ```
 
+#### configure a notification channel 
+*Examples:*
+```
+az grafana notification-channel create \
+    -n MyGrafanaInstance \
+    --definition ~/notification-channel-teams.json
+```
+
 #### Create a dashboard
 *Examples:*
 ```

src/amg/azext_amg/_help.py

@@ -61,7 +61,6 @@
             }'
 """
 
-
 helps['grafana data-source update'] = """
     type: command
     short-summary: Update a data source.
@@ -87,6 +86,52 @@
     short-summary: query a data source having backend implementation
 """
 
+helps['grafana notification-channel'] = """
+    type: group
+    short-summary: Commands to manage notification channels of an instance.
+"""
+
+helps['grafana notification-channel list'] = """
+    type: command
+    short-summary: List all notification channels of an instance.
+"""
+
+helps['grafana notification-channel show'] = """
+    type: command
+    short-summary: get details of a notification channel
+"""
+
+helps['grafana notification-channel create'] = """
+    type: command
+    short-summary: Create a notification channel.
+    examples:
+        - name: create a notification channel for Teams
+          text: |
+            az grafana notification-channel create -n MyGrafana --definition '{
+              "name": "Teams",
+              "settings": {
+                "uploadImage": true,
+                "url": "https://webhook.office.com/IncomingWebhook/"
+               },
+              "type": "teams"
+            }'
+"""
+
+helps['grafana notification-channel update'] = """
+    type: command
+    short-summary: Update a notification channel.
+"""
+
+helps['grafana notification-channel delete'] = """
+    type: command
+    short-summary: delete a notification channel.
+"""
+
+helps['grafana notification-channel test'] = """
+    type: command
+    short-summary: tests a notification channels.
+"""
+
 helps['grafana dashboard'] = """
     type: group
     short-summary: Commands to manage dashboards of an instance.

src/amg/azext_amg/_params.py

@@ -9,7 +9,7 @@ def load_arguments(self, _):
 
     from knack.arguments import CLIArgumentType
     from azure.cli.core.commands.parameters import tags_type, get_three_state_flag, get_enum_type
-    from azure.cli.core.commands.validators import get_default_location_from_resource_group
+    from azure.cli.core.commands.validators import get_default_location_from_resource_group, validate_file_or_dict
     from ._validators import process_missing_resource_group_parameter
     from azext_amg.vendored_sdks.models import ZoneRedundancy
     grafana_name_type = CLIArgumentType(options_list="--grafana-name",
@@ -33,24 +33,34 @@ def load_arguments(self, _):
 
     with self.argument_context("grafana dashboard") as c:
         c.argument("uid", options_list=["--dashboard"], help="dashboard uid")
-        c.argument("definition", help="The complete dashboard model in json string, a path or url to a file with such content")
         c.argument("title", help="title of a dashboard")
         c.argument('overwrite', arg_type=get_three_state_flag(), help='Overwrite a dashboard with same uid')
 
+    with self.argument_context("grafana dashboard create") as c:
+        c.argument("definition", type=validate_file_or_dict, help="The complete dashboard model in json string, a path or url to a file with such content")
+
+    with self.argument_context("grafana dashboard update") as c:
+        c.argument("definition", type=validate_file_or_dict, help="The complete dashboard model in json string, a path or url to a file with such content")
+
     with self.argument_context("grafana dashboard import") as c:
         c.argument("definition", help="The complete dashboard model in json string, Grafana gallery id, a path or url to a file with such content")
 
     with self.argument_context("grafana data-source") as c:
         c.argument("data_source", help="name, id, uid which can identify a data source. CLI will search in the order of name, id, and uid, till finds a match")
-        c.argument("definition", help="json string with data source definition, or a path to a file with such content")
+        c.argument("definition", type=validate_file_or_dict, help="json string with data source definition, or a path to a file with such content")
+
+    with self.argument_context("grafana notification-channel") as c:
+        c.argument("notification_channel", help="id, uid which can identify a data source. CLI will search in the order of id, and uid, till finds a match")
+        c.argument("definition", type=validate_file_or_dict, help="json string with notification channel definition, or a path to a file with such content")
+        c.argument("short", action='store_true', help="list notification channels in short format.")
 
     with self.argument_context("grafana data-source query") as c:
         c.argument("conditions", nargs="+", help="space-separated condition in a format of `<name>=<value>`")
         c.argument("time_from", options_list=["--from"], help="start time in iso 8601, e.g. '2022-01-02T16:15:00'. Default: 1 hour early")
         c.argument("time_to", options_list=["--to"], help="end time in iso 8601, e.g. '2022-01-02T17:15:00'. Default: current time ")
-        c.argument("max_data_points", help="Maximum amount of data points that dashboard panel can render")
+        c.argument("max_data_points", type=int, help="Maximum amount of data points that dashboard panel can render. Default: 1000")
         c.argument("query_format", help="format of the resule, e.g. table, time_series")
-        c.argument("internal_ms", help="The time interval in milliseconds of time series")
+        c.argument("internal_ms", type=int, help="The time interval in milliseconds of time series. Default: 1000")
 
     with self.argument_context("grafana folder") as c:
         c.argument("title", help="title of the folder")

src/amg/azext_amg/commands.py

@@ -32,6 +32,14 @@ def load_command_table(self, _):
         g.custom_command('query', 'query_data_source')
         g.custom_command('update', 'update_data_source')
 
+    with self.command_group('grafana notification-channel') as g:
+        g.custom_command('list', 'list_notification_channels')
+        g.custom_show_command('show', 'show_notification_channel')
+        g.custom_command('create', 'create_notification_channel')
+        g.custom_command('update', 'update_notification_channel')
+        g.custom_command('delete', 'delete_notification_channel')
+        g.custom_command('test', 'test_notification_channel')
+
     with self.command_group('grafana folder') as g:
         g.custom_command('create', 'create_folder')
         g.custom_command('list', 'list_folders')

src/amg/azext_amg/custom.py

@@ -166,26 +166,25 @@ def list_dashboards(cmd, grafana_name, resource_group_name=None):
 
 
 def create_dashboard(cmd, grafana_name, definition, title=None, folder=None, resource_group_name=None, overwrite=None):
-    data = _try_load_dashboard_definition(cmd, resource_group_name, grafana_name, definition, for_import=False)
-    if "dashboard" in data:
-        payload = data
+    if "dashboard" in definition:
+        payload = definition
     else:
         logger.info("Adjust input by adding 'dashboard' field")
         payload = {}
-        payload["dashboard"] = data
+        payload['dashboard'] = definition
 
     if title:
         payload['dashboard']['title'] = title
 
     if folder:
         folder = _find_folder(cmd, resource_group_name, grafana_name, folder)
-        payload['folderId'] = folder["id"]
+        payload['folderId'] = folder['id']
 
-    payload["overwrite"] = overwrite or False
+    payload['overwrite'] = overwrite or False
 
-    if "id" in payload["dashboard"]:
+    if "id" in payload['dashboard']:
         logger.warning("Removing 'id' from dashboard to prevent the error of 'Not Found'")
-        del payload["dashboard"]["id"]
+        del payload['dashboard']['id']
 
     response = _send_request(cmd, resource_group_name, grafana_name, "post", "/api/dashboards/db",
                              payload)
@@ -200,31 +199,31 @@ def update_dashboard(cmd, grafana_name, definition, folder=None, resource_group_
 
 def import_dashboard(cmd, grafana_name, definition, folder=None, resource_group_name=None, overwrite=None):
     import copy
-    data = _try_load_dashboard_definition(cmd, resource_group_name, grafana_name, definition, for_import=True)
+    data = _try_load_dashboard_definition(cmd, resource_group_name, grafana_name, definition)
     if "dashboard" in data:
         payload = data
     else:
         logger.info("Adjust input by adding 'dashboard' field")
         payload = {}
-        payload["dashboard"] = data
+        payload['dashboard'] = data
 
     if folder:
         folder = _find_folder(cmd, resource_group_name, grafana_name, folder)
-        payload['folderId'] = folder["id"]
+        payload['folderId'] = folder['id']
 
-    payload["overwrite"] = overwrite or False
+    payload['overwrite'] = overwrite or False
 
-    payload["inputs"] = []
+    payload['inputs'] = []
 
     # provide parameter values for datasource
     data_sources = list_data_sources(cmd, grafana_name, resource_group_name)
-    for parameter in payload["dashboard"].get('__inputs', []):
+    for parameter in payload['dashboard'].get('__inputs', []):
         if parameter.get("type") == "datasource":
-            match = next((d for d in data_sources if d["type"] == parameter["pluginId"]), None)
+            match = next((d for d in data_sources if d['type'] == parameter['pluginId']), None)
             if match:
                 clone = copy.deepcopy(parameter)
-                clone["value"] = match["uid"]
-                payload["inputs"].append(clone)
+                clone['value'] = match['uid']
+                payload['inputs'].append(clone)
             else:
                 logger.warning("No data source was found matching the required parameter of %s", parameter['pluginId'])
 
@@ -233,24 +232,25 @@ def import_dashboard(cmd, grafana_name, definition, folder=None, resource_group_
     return json.loads(response.content)
 
 
-def _try_load_dashboard_definition(cmd, resource_group_name, grafana_name, definition, for_import):
+def _try_load_dashboard_definition(cmd, resource_group_name, grafana_name, definition):
     import re
 
-    if for_import:
-        try:  # see whether it is a gallery id
-            int(definition)
-            response = _send_request(cmd, resource_group_name, grafana_name, "get",
-                                     "/api/gnet/dashboards/" + definition)
-            return json.loads(response.content)["json"]
-        except ValueError:
-            pass
+    try:
+        int(definition)
+        response = _send_request(cmd, resource_group_name, grafana_name, "get",
+                                 "/api/gnet/dashboards/" + str(definition))
+        definition = json.loads(response.content)["json"]
+        return definition
+    except ValueError:
+        pass
 
     if re.match(r"^[a-z]+://", definition.lower()):
         response = requests.get(definition, verify=(not should_disable_connection_verify()))
         if response.status_code == 200:
             definition = json.loads(response.content.decode())
         else:
             raise ArgumentUsageError(f"Failed to dashboard definition from '{definition}'. Error: '{response}'.")
+
     else:
         definition = json.loads(_try_load_file_content(definition))
 
@@ -262,9 +262,7 @@ def delete_dashboard(cmd, grafana_name, uid, resource_group_name=None):
 
 
 def create_data_source(cmd, grafana_name, definition, resource_group_name=None):
-    definition = _try_load_file_content(definition)
-    payload = json.loads(definition)
-    response = _send_request(cmd, resource_group_name, grafana_name, "post", "/api/datasources", payload)
+    response = _send_request(cmd, resource_group_name, grafana_name, "post", "/api/datasources", definition)
     return json.loads(response.content)
 
 
@@ -283,10 +281,47 @@ def list_data_sources(cmd, grafana_name, resource_group_name=None):
 
 
 def update_data_source(cmd, grafana_name, data_source, definition, resource_group_name=None):
-    definition = _try_load_file_content(definition)
     data = _find_data_source(cmd, resource_group_name, grafana_name, data_source)
     response = _send_request(cmd, resource_group_name, grafana_name, "put", "/api/datasources/" + str(data['id']),
-                             json.loads(definition))
+                             definition)
+    return json.loads(response.content)
+
+
+def list_notification_channels(cmd, grafana_name, resource_group_name=None, short=False):
+    if short is False:
+        response = _send_request(cmd, resource_group_name, grafana_name, "get", "/api/alert-notifications")
+    elif short is True:
+        response = _send_request(cmd, resource_group_name, grafana_name, "get", "/api/alert-notifications/lookup")
+    return json.loads(response.content)
+
+
+def show_notification_channel(cmd, grafana_name, notification_channel, resource_group_name=None):
+    return _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel)
+
+
+def create_notification_channel(cmd, grafana_name, definition, resource_group_name=None):
+    response = _send_request(cmd, resource_group_name, grafana_name, "post", "/api/alert-notifications", definition)
+    return json.loads(response.content)
+
+
+def update_notification_channel(cmd, grafana_name, notification_channel, definition, resource_group_name=None):
+    data = _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel)
+    definition['id'] = data['id']
+    response = _send_request(cmd, resource_group_name, grafana_name, "put",
+                             "/api/alert-notifications/" + str(data['id']),
+                             definition)
+    return json.loads(response.content)
+
+
+def delete_notification_channel(cmd, grafana_name, notification_channel, resource_group_name=None):
+    data = _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel)
+    _send_request(cmd, resource_group_name, grafana_name, "delete", "/api/alert-notifications/" + str(data["id"]))
+
+
+def test_notification_channel(cmd, grafana_name, notification_channel, resource_group_name=None):
+    data = _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel)
+    response = _send_request(cmd, resource_group_name, grafana_name, "post", "/api/alert-notifications/test",
+                             data)
     return json.loads(response.content)
 
 
@@ -422,6 +457,20 @@ def _find_data_source(cmd, resource_group_name, grafana_name, data_source):
     return json.loads(response.content)
 
 
+def _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel):
+    response = _send_request(cmd, resource_group_name, grafana_name, "get",
+                             "/api/alert-notifications/" + notification_channel,
+                             raise_for_error_status=False)
+    if response.status_code >= 400:
+        response = _send_request(cmd, resource_group_name, grafana_name,
+                                 "get", "/api/alert-notifications/uid/" + notification_channel,
+                                 raise_for_error_status=False)
+    if response.status_code >= 400:
+        raise ArgumentUsageError(
+            f"Couldn't found notification channel {notification_channel}. Ex: {response.status_code}")
+    return json.loads(response.content)
+
+
 # For UX: we accept a file path for complex payload such as dashboard/data-source definition
 def _try_load_file_content(file_content):
     import os