[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 인스턴스 생성하면 자동으로 위 스크립트가 수행되도록 고민해 봐야겠다.💪