In my Python 3.11 program I manage to fetch all alert_rules in Azure like so:
def alert_rules(self, tag:str = None):
processed = list()
try:
rules = list()
# List all Metric Alert Rules
for rule in self.management_client.metric_alerts.list_by_subscription():
rules.append(rule)
# List all Scheduled Query (Log) Alert Rules
for rule in self.management_client.scheduled_query_rules.list_by_subscription():
rules.append(rule)
if tag:
alert_rules_with_tag = []
for rule in rules:
if rule.tags and tag in rule.tags:
alert_rules_with_tag.append(rule)
rules = alert_rules_with_tag
for rule in rules:
processed.append(self.process_rule(rule))
except Exception as e:
logger.exception(f"Exception when listing alert rules: {e}")
return None
finally:
return processed
And I process each alert_rule like so:
def process_rule(self, rule):
processed = dict()
try:
alerts_for_rule = self.alerts_for_rule(rule_id = rule.id)
logger.info(f"alerts for rule {rule.id}: {devtools.pformat(alerts_for_rule)}")
processed_alerts = []
for alert in alerts_for_rule:
processed_alerts.append(self.process_alert(alert))
processed = {
'name': rule.name
, 'id': rule.id
, 'type': rule.type
, 'alerts': processed_alerts
, 'tags': rule.tags if hasattr(rule, 'tags') else {}
}
except Exception as e:
logger.exception(f"Exception in rule_to_dict: {e}")
finally:
return processed
This all works great, except when I try to fetch all the alerts for the rule, it fails:
def alerts_for_rule(self, rule_id):
alerts = []
try:
raw = self.management_client.alerts.get_all(alert_rule = rule_id)
logger.warning(f"*** Alerts for rule '{rule_id}' {devtools.pformat(raw)}")
out = list()
for alert in raw:
alerts.append(self.process_alert(alert))
except SomeAzureException as ex:
logger.error(f"Azure exception occurred while fetching alerts for rule {rule_id}: {ex}")
except Exception as e:
logger.exception(f"Exception occurred while fetching alerts for rule {rule_id}: {e}")
finally:
return alerts
Basically, I get an empty list of alerts for every rule, even rules I know to have alerts associated with them (looking in the portal).
So my question is; what is the value I should pass as rule_id to management_client.alerts.get_all(alert_rule = rule_id) to make this work? I tried the intuitive choice rule.id which obviously does not work. The documentation does not specify any details except that is it of type str.
4
So my question is; what is the value I should pass as
rule_idtomanagement_client.alerts.get_all(alert_rule = rule_id)to make this work?
The code you are using is good to execute without any conflicts.
Need to check below:
Basically the rule_id is nothing but the resource_Id of an alert rule which should be in the below format.
/subscriptions/{subscription}/resourceGroups/{resourceGroup}/providers/Microsoft.Insights/alertrules/{alertRule}
You can check the format of the rule_Id and make changes accordingly in the script.
Check whether that the alert rule has any active alerts triggered. If the alerts are not triggered, it could lead to an empty result.
Best case would be an example code showing how it is used.
As you have asked a sample code, below is the code which can be used to meet your requirements.
import logging
from azure.identity import DefaultAzureCredential
from azure.mgmt.alertsmanagement import AlertsManagementClient
import os
os.environ["AZURE_CLIENT_ID"]="CLIENTID"
os.environ["AZURE_CLIENT_SECRET"] = "CLIENTSECRET"
os.environ["AZURE_SUBSCRIPTION_ID"] = "SUBSCRIPTIONID"
os.environ["AZURE_TENANT_ID"] = "TENANTID"
credential = DefaultAzureCredential()
client = AlertsManagementClient(
credential=credential,
subscription_id="SUBSCRIPTIONID"
)
def alerts_for_rule(self, rule_id):
alerts = []
try:
logging.debug(f"retrieve alerts for rule ID: {rule_id}")
raw = self.management_client.alerts.get_all()
filtered = [alert for alert in raw if alert.alert_rule == rule_id]
if not filtered:
logging.warning(f"No alerts found")
for alert in filtered:
alerts.append(self.process_alert(alert))
except Exception as e:
logging.error(f"Exception occurred")
finally:
return alerts