안녕하세요 MANVSCLOUD 김수현입니다.
최근 운영하고 있는 Cloud DB for PostgreSQL에서 지속적인 백업 실패가 발생하는 문제가 있었습니다. Cloud DB for PostgreSQL은 관리형 데이터베이스로 백업 보관일과 백업 시간을 설정할 수 있는 편리한 기능을 제공합니다.
하지만 설정한 시간대에 백업이 제대로 이루어지지 않아 1일 1백업이 되지 않고 요구되는 백업 보관일만큼 데이터가 보관되지 않는 문제가 발생했습니다.
데이터는 시스템 운영에서 가장 중요한 자산 중 하나이며 백업은 이를 보호하는 최후의 보루입니다. 따라서 오늘은 관리형 데이터베이스 백업의 성공/실패 여부를 효과적으로 모니터링할 수 있는 알림 기능을 구현하는 방법을 공유드리고자 합니다.
관리형 데이터베이스도 백업에 실패할 수 있다
관리형 데이터베이스에서는 위 이미지와 같이 백업 보관일과 백업 시간을 지정해줄 수 있습니다. 이론적으로는 매일 지정한 시간대에 백업이 수행되며 설정된 보관일만큼 백업 데이터가 안전하게 보관되어야 합니다.
(선생님… 제 백업은 어디로 간 것입니까? 백업 시간은 무슨 일이 있었던 겁니까?…)
하지만 어느 날 Backup 리스트를 확인해보니 백업 날짜가 띄엄띄엄 누락되어 있는 현상을 발견했습니다.
데이터베이스 용량이 너무 커서 백업 소요 시간이 길어져 다음날로 미루어진 것도 아니었습니다. (5GB도 안쓰고 있다고…)
우선 하루에 한 번씩 백업이 제대로 수행되었는지 모니터링이 될 필요가 있었습니다. 그러나 매번 콘솔에 접속해서 백업 상태를 확인하는 것은 매우 번거롭고 시간 소모적입니다.
저는 매일 아침 출근과 동시에 필요한 항목들을 우선적으로 체크하는 편입니다. 따라서 관리형 데이터베이스 백업도 매일 아침마다 성공적으로 수행되었는지 Slack으로 알림을 받아볼 수 있도록 했습니다.
백업 여부 지켜보고 있다
자동화 시스템을 구현하기 위해 Cloud Functions을 활용하겠습니다. 이런 간단한 기능을 구현하고 자동화하는 부분은 역시 Serverless가 베스트 선택지입니다. 인프라 관리 부담 없이 필요한 기능만 집중적으로 구현할 수 있어서 제가 아주 좋아합니다.
긴 말하지 않고 코드부터 공유드려 보겠습니다.
먼저 구현 환경은 다음과 같습니다:
– Python 3.11
– Trigger: Cron (알림을 원하는 시간대로 설정)
import hashlib
import hmac
import base64
import requests
import time
import json
import re
from datetime import datetime, timezone, timedelta
def parse_datetime(date_str):
"""다양한 형식의 날짜 문자열을 파싱하여 datetime 객체로 변환"""
# 2025-02-20T12:08:21+0900 형식 파싱
if '+0900' in date_str:
pattern = r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})\+0900'
match = re.match(pattern, date_str)
if match:
base_dt = datetime.strptime(match.group(1), '%Y-%m-%dT%H:%M:%S')
return base_dt.replace(tzinfo=timezone(timedelta(hours=9)))
# Z로 끝나는 경우 (UTC)
if date_str.endswith('Z'):
base_dt = datetime.strptime(date_str[:-1], '%Y-%m-%dT%H:%M:%S')
return base_dt.replace(tzinfo=timezone.utc).astimezone(timezone(timedelta(hours=9)))
# 다른 ISO 형식 시도
try:
formats = [
'%Y-%m-%dT%H:%M:%S',
'%Y-%m-%d %H:%M:%S',
'%Y-%m-%d'
]
for fmt in formats:
try:
return datetime.strptime(date_str, fmt).replace(tzinfo=timezone(timedelta(hours=9)))
except ValueError:
continue
# 모든 형식 실패 시 마지막 시도
return datetime.strptime(date_str, '%Y-%m-%dT%H:%M:%S%z')
except ValueError:
raise ValueError(f"지원하지 않는 날짜 형식입니다: {date_str}")
def format_datetime(date_str):
"""ISO 형식의 날짜 문자열을 '년 월 일, 시 분 초' 형식으로 변환"""
dt = parse_datetime(date_str)
return dt.strftime('%Y년 %m월 %d일, %H시 %M분 %S초')
def check_backup_status(backup_info):
"""백업 상태를 확인하고 Slack 메시지 작성"""
today = datetime.now(timezone(timedelta(hours=9))).date()
today_str = today.strftime('%Y년 %m월 %d일')
messages = []
for db in backup_info['getCloudPostgresqlBackupListResponse']['cloudPostgresqlBackupList']:
service_name = db['cloudPostgresqlServiceName']
backup_time = db.get('backupTime', '랜덤')
last_backup_str = db['lastBackupDate']
# parse_datetime 함수를 사용하여 날짜 파싱
last_backup_dt = parse_datetime(last_backup_str)
last_backup_date = last_backup_dt.date()
# 오늘 백업 여부 확인
is_backup_today = (last_backup_date == today)
status = "성공" if is_backup_today else "실패"
# Slack 메시지 작성
message = {
"blocks": [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*데이터베이스 백업 상태 알림*"
}
},
{
"type": "divider"
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": f"*오늘 날짜:*\n{today_str}"
},
{
"type": "mrkdwn",
"text": f"*DB 서비스명:*\n{service_name}"
},
{
"type": "mrkdwn",
"text": f"*예약 백업 시간:*\n{backup_time}"
},
{
"type": "mrkdwn",
"text": f"*최종 백업 시간:*\n{format_datetime(last_backup_str)}"
},
{
"type": "mrkdwn",
"text": f"*일일 백업 여부:*\n{status}"
}
]
}
]
}
# 백업 실패 시 색상 표시
if not is_backup_today:
message["attachments"] = [
{
"color": "#FF0000",
"blocks": [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "⚠️ *경고: 오늘 백업이 수행되지 않았습니다!*"
}
}
]
}
]
messages.append(message)
return messages
def send_to_slack(messages, webhook_url):
"""Slack 웹훅 URL로 메시지 전송"""
results = []
for message in messages:
response = requests.post(
webhook_url,
data=json.dumps(message),
headers={'Content-Type': 'application/json'}
)
result = {
"status_code": response.status_code,
"success": response.status_code == 200,
"response": response.text
}
results.append(result)
return results
def get_cloud_postgresql_backup_list(access_key, secret_key):
"""네이버 클라우드 API를 통해 PostgreSQL 백업 목록을 가져옴"""
timestamp = str(int(time.time() * 1000))
secret_key = bytes(secret_key, 'UTF-8')
method = "GET"
api_server = "https://ncloud.apigw.ntruss.com"
uri = "/vpostgresql/v2/getCloudPostgresqlBackupList"
uri = uri + f"?responseFormatType=json"
# 서명 생성
message = method + " " + uri + "\n" + timestamp + "\n" + access_key
message = bytes(message, 'UTF-8')
signingKey = base64.b64encode(hmac.new(secret_key, message, digestmod=hashlib.sha256).digest())
# HTTP 헤더 설정
http_header = {
'x-ncp-apigw-signature-v2': signingKey,
'x-ncp-apigw-timestamp': timestamp,
'x-ncp-iam-access-key': access_key,
}
# API 요청 및 응답 처리
response = requests.get(api_server + uri, headers=http_header)
if response.status_code != 200:
raise Exception(f"API 요청 오류: {response.status_code}, {response.text}")
return response.json()
def main(args):
"""Cloud Functions 메인 함수"""
# 필수 파라미터 확인
required_params = ['DB_ACCESS_KEY', 'DB_SECRET_KEY', 'SLACK_WEBHOOK_URL']
for param in required_params:
if not args.get(param):
return {
"statusCode": 400,
"body": f"필수 파라미터가 누락되었습니다: {param}"
}
try:
# 백업 정보 가져오기
backup_info = get_cloud_postgresql_backup_list(
args.get('DB_ACCESS_KEY'),
args.get('DB_SECRET_KEY')
)
# 백업 상태 확인 및 Slack 메시지 생성
slack_messages = check_backup_status(backup_info)
# Slack으로 메시지 전송
slack_results = send_to_slack(slack_messages, args.get('SLACK_WEBHOOK_URL'))
return {
"statusCode": 200,
"body": {
"message": "백업 상태 확인 및 알림 전송 완료",
"backup_count": len(backup_info['getCloudPostgresqlBackupListResponse']['cloudPostgresqlBackupList']),
"slack_results": slack_results
}
}
except Exception as e:
return {
"statusCode": 500,
"body": f"오류 발생: {str(e)}"
}
위 코드는 Cloud DB For PostgreSQL의 백업 성공/실패 여부를 확인하고 웹훅으로 결과를 전달하는 기능을 수행합니다. 다른 관리형 데이터베이스를 사용하고 있을 경우 API 주소 및 파라미터를 적절히 변경하여 사용하실 수 있습니다.
- 디폴트 파라미터
{
"DB_ACCESS_KEY": "ACCESS_KEY",
"DB_SECRET_KEY": "SECRET_KEY",
"SLACK_WEBHOOK_URL": "WEBHOOK_URL"
}
- ACCESS_KEY: DB 백업 리스트를 조회할 수 있는 권한이 있는 ACCESS_KEY
- SECRET_KEY: DB 백업 리스트를 조회할 수 있는 권한이 있는 ACCESS_KEY의 SECRET_KEY
- WEBHOOK_URL: 꼭 SLACK WEBHOOK URL뿐만 아니라 원하는 WEBHOOK URL을 넣어주시면 됩니다. MS Teams나 Discord 등 다른 메신저 서비스의 웹훅 URL도 활용 가능합니다.
이 Cloud Function은 매일 지정한 시간에 실행되어 전날의 백업 상태를 확인하고 그 결과를 Slack 채널로 전송합니다.
저는 cron 트리거를 이용하여 출근 시간에 알림을 받는 방식을 선택했는데 원하는 트리거 방식이 있다면 변경해서 사용해도 좋습니다 🙂
Personal Comments
지금까지 관리형 데이터베이스 백업 여부를 효과적으로 모니터링하고 알림을 받을 수 있는 방법을 공유해보았습니다.
하지만 우리는 아직 근본적인 문제를 해결하지 못했습니다. 백업 여부에 대한 알림을 받았지만 백업이 실패했다는 사실 자체는 달라지지 않기 때문입니다. 알림은 문제를 감지하는 데는 도움이 되지만 문제를 해결하지는 못합니다.
다음 포스팅에서는 한 걸음 더 나아가 관리형 데이터베이스에서 제공하는 기본 백업 기능을 사용하지 않고 직접 백업을 자동화하고 원하는 기간만큼 안전하게 보관할 수 있는 방법을 공유드리겠습니다.
긴 글 읽어주셔서 감사합니다.
