[python]CloudWatchアラーム自動登録スクリプト


Aurora MySQL Alarm Flow(アラーム設定のみ、モニタなし)



このドキュメントでは、AWS SNS/Lambdaの構成については説明しません.
SNS/Lambdaに設定されていると仮定すると、Cloudwatchの自動目覚まし時計登録内容のみが議論されます.🙏

Pythonスクリプト📝


CloudWatchには静的プローブ、異常プローブがあり、そのスクリプトは静的プローブです.
  • 静的検出:しきい値値値に基づいてアラーム
  • を発行する.
  • 異常検出:AWS内部で統計/機械学習アルゴリズムにより分析し警報を出す
    注意:Cloudwatchによる異常検出
  • import boto3
    
    '''
    RDS Metric을 추가할 경우 아래 rds_metric에 설정
    
        "RDS 지표명":['조건','통계','임계치']
    '''
    
    rds_metric = {
      "CPUUtilization": ['GreaterThanOrEqualToThreshold', 'Average', 75]
    , "DatabaseConnections": ['GreaterThanOrEqualToThreshold', 'Minimum', 500]
    
    }
    
    # AWS Client 함수
    def aws_client(aws_svc):
        aws_svc_client = boto3.client(
            aws_svc
            , aws_access_key_id='{AWS Access Key}'
            , aws_secret_access_key='{AWS Secret Key}'
            , region_name='{AWS Region}'
        )
        return aws_svc_client
    
    
    # Cloudwatch 알람 생성 함수
    def cloudwatch_create_alarm(inst_nm,metric_nm,metric_compare,metric_stat,metric_val):
        create_alarm = cloudwatch_client.put_metric_alarm (
            AlarmActions=["{구성한 AWS SNS의 ARN 입력}"],
            AlarmName="[TEST][%s][%s]" % (inst_nm, metric_nm),
            MetricName="%s" % metric_nm,
            Namespace="AWS/RDS",
            Statistic="%s" % metric_stat,
            Dimensions=[
                {
                    "Name": "DBInstanceIdentifier",
                    "Value": "%s" % inst_nm
                }
            ],
            Period=60,
            EvaluationPeriods=1,
            Threshold= metric_val,
            ComparisonOperator="%s" % metric_compare
        )
        return create_alarm
    
    
    # RDS 인스턴스 리스트 추출
    rds_client = aws_client('rds')
    get_instances = rds_client.get_paginator('describe_db_instances').paginate()
    
    rds_instance = []
    for instance_json in get_instances:
        for i in range(len(instance_json['DBInstances'])):
            rds_instance.append((instance_json['DBInstances'][i]['DBInstanceIdentifier']))
    
    cloudwatch_client = aws_client('cloudwatch')
    get_cloudwatch_alarm = cloudwatch_client.get_paginator('describe_alarms')
    
    
    # Cloudwatch 알람 설정
    for alarm_json in get_cloudwatch_alarm.paginate(AlarmNamePrefix='[TEST]'):
    
        alarm_check = alarm_json['MetricAlarms']
    
        # 알람 설정이 없을 경우
        if not alarm_check:
    
            for i in range(len(rds_instance)):
    
                for metric_name, metric_value in rds_metric.items():
    
                    # 신규 추가 : RDS 인스턴스 별 알람 생성
                    cloudwatch_create_alarm(rds_instance[i], metric_name, metric_value[0], metric_value[1], metric_value[2])
    
        else:
    
            for i in range(len(alarm_json['MetricAlarms'])):
    
    
                already_rds_instance = alarm_json['MetricAlarms'][i]['Dimensions'][0]['Value']
                already_rds_metric = alarm_json['MetricAlarms'][i]['MetricName']
    
                for metric_name, metric_value in rds_metric.items():
    
                    # 중복된 지표명이 있으면 Skip
                    if already_rds_metric == metric_name:
                        pass
    
                    # 인스턴스 별 추가 지표 등록 또는 지표 내용 변경 시 업데이트
                    else:
                        cloudwatch_create_alarm(already_rds_instance, metric_name, metric_value[0], metric_value[1],metric_value[2])
    

    スクリプト実行結果👀



    ここまでです。🙋‍♂️


    👉 主な業務は開発ではないので、欲しい結果が得られれば満足です.🙆‍♂️
    👉 RDSインスタンスを初めて作成する場合は、スクリプトの自動実行を考慮する必要があります.💪