Get specific alert from Azure Monitor using Python

Question

I want to get a specific alert from Azure Monitor using python (through an Azure function). The Azure Monitor will triggered the az function for each Event.

Currently I'm using get_all from azure.mgmt.alertsmanagement.operations module, this allows me to get all alerts. Also already tested get_by_id but I was obliged to specify the alert_id while I'm looking to get it automatically.

import logging
import urllib3
import os
import json
import requests
from azure.identity import ClientSecretCredential
from azure.mgmt.alertsmanagement import AlertsManagementClient


subscription_id =""
client_id =""
client_secret =""
tenant_id = ""

credential = ClientSecretCredential(
      tenant_id=tenant_id, 
      client_id=client_id, 
      client_secret=client_secret
)

print("===Auth Azure Monitor===")
client = AlertsManagementClient(
    credential,
    subscription_id
)

print("=== Get alert event from Az Monitor & Post it to monitoring platform === ")
headers = {'Authorization': f'authtoken {token}'}

for alert in client.alerts.get_all():
    if alert.name == "alert_rule_name" :
        attributes = {'CLASS': 'EVENT',
                      'severity': 'CRITICAL',
                      'msg': alert.name,
                      'lastModifiedDateTime': json.dumps(alert.properties.essentials.last_modified_date_time, indent=4, sort_keys=True, default=str)
                      }
        payload = [{'eventSourceHostName': alert.properties.essentials.target_resource_name, 'attributes': attributes}]
        print("JSON_PAYLOAD :", payload)
## Some code here to push the Alert to a monitoring platform ..

Please, find below the json sent by Azure Monitor with get_all :

{'value': [{'properties': {'essentials': {
'severity': 'Sev2', 
'signalType': 'Metric', 
'alertState': 'New', 
'monitorCondition': 'Fired', 
'monitorService': 'Platform', 
'targetResource': '/subscriptions/sub_id/resourcegroups/rsg_name/providers/microsoft.compute/virtualmachines/vm_name', 
'targetResourceName': 'vm_name', 
'targetResourceGroup': 'rsg_name', 
'targetResourceType': 'virtualmachines', 
'sourceCreatedId': '5f33r_rsg_name_microsoft.insights_metricAlerts_alert_rule_name-1899618006', 
'alertRule': '/subscriptions/sub_id/resourceGroups/rsg_name/providers/microsoft.insights/metricAlerts/alert_rule_name', 
'startDateTime': '2023-05-09T13:32:28.1880147Z', 
'lastModifiedDateTime': '2023-05-09T13:32:28.1880147Z', 
'lastModifiedUserName': 'System', 
'actionStatus': {'isSuppressed': False}, 'description': ''}
}, 
'id': '/subscriptions/sub_id/providers/Microsoft.AlertsManagement/alerts/2222-5555-88888', 
'type': 'Microsoft.AlertsManagement/alerts', 
'name': 'alert_rule_name'}, 

As you see, I'm filtering by [if alert.name == "alert_rule_name"] and this is not what I'm looking for (I got a list of Events).

Is there a way to get the alert ID from the payload when Azure Monitor call my function ? This is to use this ID to get a specific alert (event).

Thanks in advance

Answer 1

Azure Monitor trigger the below Azure Function which parse only one Event and forward it to another destination in order to notify the support team :

import azure.functions as func
import os
import json
import requests
import urllib3
import logging

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("===== Auth Monitoring Platform =====")
    urllib3.disable_warnings()
    creds = {'username': "user", 'password': "****", 'tenantName': '*'}
    logging.debug(f'Retrieving authorization token')
    resp = requests.post(f'https://{api_url}', json=creds, verify=False)
    if resp.status_code != 200:
            logging.error(f'{resp.headers}')
            logging.error(f'{resp.text}')
            exit(1)
            
    token = resp.json()["response"]["authToken"]

    logging.info("===== Get Alert JSON & Prepare Post to Monitor Platform =====")

    req_body = req.get_json()
    
    msg_detail = {'firedDateTime' : req_body['data']['essentials']['firedDateTime'], 
                   'operator':       req_body['data']['alertContext']['condition']['allOf'][0]['operator'],
                   'threshold':      req_body['data']['alertContext']['condition']['allOf'][0]['threshold'],
                   'metricValue':    req_body['data']['alertContext']['condition']['allOf'][0]['metricValue']
                   }
    headers = {'Authorization': f'authtoken {token}'}
    attributes = {  'CLASS':           'EVENT',
                    'severity':        'CRITICAL',
                    'msg':             req_body['data']['essentials']['alertRule'],
                    'msg_detail':     mc_long_msg,
                    'object':       req_body['data']['essentials']['configurationItems'][0],
                    'object_class': req_body['data']['alertContext']['condition']['allOf'][0]['metricNamespace'],
                    'mc_parameter':    req_body['data']['alertContext']['condition']['allOf'][0]['metricName']
                 }
    
    payload = [{'eventSource': req_body['data']['essentials']['configurationItems'][0], 'attributes': attributes}]
    params = {'param1':"value1" , 'param2':"value2"}
    logging.info(f'Event header: {headers}')
    logging.info(f'Event payload: {payload}')     
    
    response = requests.post(f'https://{api_url}',
                       headers=headers,
                       params=params,
                       json=payload,
                       verify=False
                )
    if response.status_code != 200:
        logging.error(f'{response.headers}')
        logging.error(f'{response.text}')
        exit(1)
    
    logging.debug(f'{response.headers}')
    logging.debug(f'{response.text}')
    logging.info(f'Payload: {payload}')
    
    return func.HttpResponse(
        "This HTTP triggered function executed successfully.",
        status_code=200
    )