This topic describes how to manage alerts by using Simple Log Service SDK for Python and provides sample code.
Billing method
For the alerting feature, you are charged only when you receive alert notifications by using SMS messages and voice calls. For more information, see Simple Log Service pricing.
Notification method | Description |
SMS messages | Fees are incurred when you receive notifications by using SMS messages. You are charged based on the number of times that you receive alert notifications.
Note If an SMS message is longer than 70 characters, it may be sent in two messages. In this case, you are charged only once |
Voice calls | Fees are incurred when you receive notifications by using voice calls. You are charged based on the times that you receive alert notifications.
Note If a voice call is not answered, an SMS notification is sent. You are charged only once for the voice call whether the call is answered or not. The notification message does not incur fees.
|
Manage alert rules
The following sample code provides an example on how to manage alert rules. For more information about the parameters in the sample code, see Data structure of an alert 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',
'status': 'Enabled',
'schedule': {
'type': 'FixedRate',
'interval': '1m'
},
'configuration': {
'version': '2.0',
'type': 'default',
'dashboard': 'internal-alert-analysis',
'queryList': [{
# The type of alert data. Valid values: log, metric, and meta. The value log specifies a Logstore. The value metric specifies a Metricstore. The value meta specifies resource data.
'storeType': 'log',
# The supported regions.
'region': 'cn-huhehaote',
# The name of the project.
'project': 'demo-alert',
# The name of the Logstore or Metricstore.
'store': 'nginx-access-log',
# The query statement.
'query': 'status >= 400 | select count(*) as cnt',
# The type of the time span.
'timeSpanType': 'Truncated',
# The start time.
'start': '-1m',
# The end time.
'end': 'absolute',
# Specifies whether to enable Dedicated SQL.
'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 provides an example on how to manage alert resource data. 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()