通过Python SDK管理告警

import os
from aliyun.log import LogClient
from aliyun.log.resource_params import ResourceRecord
# 日志服务的服务入口。资源数据的写操作，必须使用河源地域，读操作可以使用其它地域。
endpoint = 'cn-heyuan.log.aliyuncs.com'
# 本示例从环境变量中获取AccessKey ID和AccessKey Secret。
accesskey_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
accesskey_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')
# 创建日志服务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()

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()

def create_webhook_integration():
    webhooks = [{
        'id': 'dingtalk',
        'name': 'Dingtalk Webhook',
        'type': 'dingtalk',
        'url': 'https://oapi.dingtalk.com/robot/send?access_token=**********',
        'method': 'POST',
        # 钉钉加签密钥。如果您在钉钉侧设置安全验证方式为签名校验，则需配置该字段。您可以在钉钉机器人管理界面获取加签密钥。
        # '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',
        # 飞书加签密钥。如果您在飞书侧设置安全验证方式为签名校验，则需配置该字段。您可以在飞书机器人管理界面获取加签密钥。
        # '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()

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()

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()

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()