This topic describes how to manage alerts by using Simple Log Service SDK for Python and provides sample code.
Prerequisites
A Resource Access Management (RAM) user is created, and the required permissions are granted to the RAM user. For more information, see Create a RAM user and grant permissions to the RAM user.
The ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables are configured. For more information, see Configure environment variables.
Important The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. We recommend that you use the AccessKey pair of a RAM user to call API operations or perform routine O&M.
We recommend that you do not save the AccessKey ID or AccessKey secret in your project code. Otherwise, the AccessKey pair may be leaked, and the security of all resources within your account may be compromised.
Simple Log Service SDK for Python V0.7.9 or later is installed. For more information, see Install Simple Log Service SDK for Python.
Manage alert monitoring rules
The following sample code is provided for your reference. For more information about the parameters in the sample code, see Data structure of an alert monitoring rule.
import os
from aliyun.log import LogClient
# The Simple Log Service endpoint.
endpoint = 'cn-huhehaote.log.aliyuncs.com'
# Configure environment variables. In this example, the AccessKey ID and AccessKey secret are obtained from environment variables.
accesskey_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
accesskey_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')
# Create a Simple Log Service client.
client = LogClient(endpoint, accesskey_id, accesskey_secret)
project = 'demo-alert'
alert_id = 'nginx-status-error'
def create_alert():
alert = {
'name': alert_id,
'displayName': 'Nginx Status Error',
'type': 'Alert',
'state': 'Enabled',
'schedule': {
'type': 'FixedRate',
'interval': '1m'
},
'configuration': {
'version': '2.0',
'type': 'default',
'dashboard': 'internal-alert-analysis',
'queryList': [{
'storeType': 'log',
'region': 'cn-huhehaote',
'project': 'demo-alert',
'store': 'nginx-access-log',
'query': 'status >= 400 | select count(*) as cnt',
'timeSpanType': 'Truncated',
'start': '-1m',
'end': 'absolute',
'powerSqlMode': 'auto'
}],
'groupConfiguration': {
'type': 'no_group',
'fields': []
},
'joinConfigurations': [],
'severityConfigurations': [{
'severity': 6,
'evalCondition': {
'condition': 'cnt > 0',
'countCondition': ''
}
}],
'labels': [{
'key': 'service',
'value': 'nginx'
}],
'annotations': [{
'key': 'title',
'value': 'Nginx Status Error'
}, {
'key': 'desc',
'value': 'Nginx Status Error, count: ${cnt}'
}],
'autoAnnotation': True,
'sendResolved': False,
'threshold': 1,
'noDataFire': False,
'noDataSeverity': 6,
'policyConfiguration': {
'alertPolicyId': 'sls.builtin.dynamic',
'actionPolicyId': 'test-action-policy',
'repeatInterval': '1m',
'useDefault': False
}
}
}
res = client.create_alert(project, alert)
res.log_print()
def get_and_update_alert():
res = client.get_alert(project, alert_id)
res.log_print()
alert = res.get_body()
alert['configuration']['queryList'][0]['query'] = 'status >= 400 | select count(*) as cnt'
res = client.update_alert(project, alert)
res.log_print()
def enable_and_disable_alert():
res = client.disable_alert(project, alert_id)
res.log_print()
res = client.enable_alert(project, alert_id)
res.log_print()
def list_alerts():
res = client.list_alert(project, offset=0, size=100)
res.log_print()
def delete_alert():
res = client.delete_alert(project, alert_id)
res.log_print()
if __name__ == '__main__':
create_alert()
get_and_update_alert()
enable_and_disable_alert()
list_alerts()
delete_alert()
Manage alert resource data
The following sample code is provided for your reference. For more information about the parameters in the sample code, see Data structure of alert resource data.
Manage users
import os
from aliyun.log import LogClient
from aliyun.log.resource_params import ResourceRecord
# The Simple Log Service endpoint. For the resource data, the write operations support only the Simple Log Service endpoint for the China (Heyuan) region, and the read operations support the Simple Log Service endpoints for other regions.
endpoint = 'cn-heyuan.log.aliyuncs.com'
# Configure environment variables. In this example, the AccessKey ID and AccessKey secret are obtained from environment variables.
accesskey_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
accesskey_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')
# Create a Simple Log Service client.
client = LogClient(endpoint, accesskey_id, accesskey_secret)
user_resource_name = 'sls.common.user'
def create_user():
user = {
'user_id': 'alex',
'user_name': 'Alex',
'email': [
'****@example.com'
],
'country_code': '86',
'phone': '133****3333',
'enabled': True,
'sms_enabled': True,
'voice_enabled': True
}
record = ResourceRecord(user['user_id'], user['user_name'], user)
res = client.create_resource_record(user_resource_name, record)
print('[create user]')
res.log_print()
def get_user():
res = client.get_resource_record(user_resource_name, 'alex')
print('[get user]')
print(res.get_record().to_dict())
def update_user():
user = {
'user_id': 'alex',
'user_name': 'Alex',
'email': [
'****@example.com'
],
'country_code': '86',
'phone': '133****3333',
'enabled': False,
'sms_enabled': True,
'voice_enabled': True
}
record = ResourceRecord(user['user_id'], user['user_name'], user)
res = client.update_resource_record(user_resource_name, record)
print('[update user]')
res.log_print()
def list_users():
res = client.list_resource_records(user_resource_name, offset=0, size=100)
print('[list users]')
print([r.to_dict() for r in res.get_records()])
def delete_user():
res = client.delete_resource_record(user_resource_name, ['alex'])
print('[delete user]')
res.log_print()
if __name__ == '__main__':
create_user()
get_user()
update_user()
list_users()
delete_user()
Manage user groups
def create_user_group():
user_group = {
'user_group_id': 'devops',
'user_group_name': 'DevOps Team',
'enabled': True,
'members': ['alex']
}
record = ResourceRecord(user_group['user_group_id'], user_group['user_group_name'], user_group)
res = client.create_resource_record('sls.common.user_group', record)
print('[create user group]')
res.log_print()
Manage webhook integration
def create_webhook_integration():
webhooks = [{
'id': 'dingtalk',
'name': 'Dingtalk Webhook',
'type': 'dingtalk',
'url': 'https://oapi.dingtalk.com/robot/send?access_token=**********',
'method': 'POST',
# The value of Additional Signature for your DingTalk chatbot. If you select Additional Signature for Security Settings when you create the DingTalk chatbot, you must configure this field. You can obtain the value of Additional Signature on the chatbot management page of DingTalk.
# 'secret': 'SEC**********',
'headers': []
}, {
'id': 'wechat',
'name': 'Wechat Webhook',
'type': 'wechat',
'url': 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=**********',
'method': 'POST',
'headers': []
}, {
'id': 'feishu',
'name': 'Feishu Webhook',
'type': 'lark',
'url': 'https://open.feishu.cn/open-apis/bot/v2/hook/**********',
'method': 'POST',
# The value of Set signature verification for your Lark bot. If you select Set signature verification for Security settings when you create the Lark bot, you must configure this field. You can obtain the value of Set signature verification on the bot management page of Lark.
# 'secret': '**********',
'headers': []
}, {
'id': 'slack',
'name': 'Slack Webhook',
'type': 'slack',
'url': 'https://hooks.slack.com/services/**********',
'method': 'POST',
'headers': []
}, {
'id': 'webhook',
'name': 'Common Webhook',
'type': 'custom',
'url': 'https://example.com/***********',
'method': 'POST',
'headers': [{
'key': 'Authorization',
'value': 'Basic YWRtaW46Zm9vYmFy'
}]
}]
for webhook in webhooks:
record = ResourceRecord(webhook['id'], webhook['name'], webhook)
res = client.create_resource_record('sls.alert.action_webhook', record)
print('[create webhook integration] ' + webhook['id'])
res.log_print()
Manage action policies
def create_action_policy():
action_policy = {
'action_policy_id': 'test-action-policy',
'action_policy_name': 'Test Action Policy',
'primary_policy_script': 'fire(type="sms", users=["alex"], groups=[], oncall_groups=[], receiver_type="static", external_url="", external_headers={}, template_id="sls.builtin.cn", period="any")',
'secondary_policy_script': 'fire(type="voice", users=["alex"], groups=[], oncall_groups=[], receiver_type="static", external_url="", external_headers={}, template_id="sls.builtin.cn", period="any")',
'escalation_start_enabled': False,
'escalation_start_timeout': '10m',
'escalation_inprogress_enabled': False,
'escalation_inprogress_timeout': '30m',
'escalation_enabled': True,
'escalation_timeout': '1h'
}
record = ResourceRecord(
action_policy['action_policy_id'], action_policy['action_policy_name'], action_policy)
res = client.create_resource_record('sls.alert.action_policy', record)
print('[create action policy]')
res.log_print()
Manage alert policies
def create_alert_policy():
alert_policy = {
'policy_id': 'test-alert-policy',
'policy_name': 'Test Alert Policy',
'parent_id': '',
'group_script': 'fire(action_policy="test-action-policy", group={"alert.alert_id": alert.alert_id}, group_by_all_labels=true, group_wait="15s", group_interval="5m", repeat_interval="1h")',
'inhibit_script': '',
'silence_script': ''
}
record = ResourceRecord(alert_policy['policy_id'], alert_policy['policy_name'], alert_policy)
res = client.create_resource_record('sls.alert.alert_policy', record)
print('[create alert policy]')
res.log_print()
Manage alert templates
def create_content_template():
template = {
'template_id': 'test-template',
'template_name': 'Test Template',
'templates': {
'sms': {
'locale': 'zh-CN',
'content': ''
},
'voice': {
'locale': 'zh-CN',
'content': ''
},
'email': {
'locale': 'zh-CN',
'subject': 'SLS Alert',
'content': ''
},
'message_center': {
'locale': 'zh-CN',
'content': ''
},
'dingtalk': {
'locale': 'zh-CN',
'title': 'SLS Alert',
'content': ''
},
'wechat': {
'locale': 'zh-CN',
'title': 'SLS Alert',
'content': ''
},
'lark': {
'locale': 'zh-CN',
'title': 'SLS Alert',
'content': ''
},
'slack': {
'locale': 'zh-CN',
'title': 'SLS Alert',
'content': ''
},
'webhook': {
'locale': 'zh-CN',
'send_type': 'batch',
'limit': 0,
'content': ''
},
'fc': {
'locale': 'zh-CN',
'limit': 0,
'send_type': 'batch',
'content': ''
},
'event_bridge': {
'locale': 'zh-CN',
'subject': 'SLS Alert',
'content': ''
},
}
}
record = ResourceRecord(template['template_id'], template['template_name'], template)
res = client.create_resource_record('sls.alert.content_template', record)
print('[create content template]')
res.log_print()