AWS SNS/Lambda 구성은 이번 문서에서 다루지 않음.
SNS/Lambda까지 구성되었다는 가정 하에 Cloudwatch 자동 알람 등록 내용만 다룬다.🙏
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])
👉 위 스크립트를 직접 실행하는 것이 아닌, 처음 인스턴스 생성될 때 SNS을 통해 자동으로 실행되도록 구현해 봐야겠다.