이미지 로딩 중...
AI Generated
2025. 11. 18. · 2 Views
이상 패턴 탐지 및 알림 시스템 완벽 가이드
시스템 운영 중 발생하는 이상 패턴을 자동으로 감지하고, Slack이나 이메일로 즉시 알림을 받는 방법을 배워봅니다. 임계값 설정부터 급증 패턴 탐지, 알림 중복 방지까지 실무에 바로 적용 가능한 모니터링 시스템 구축 가이드입니다.
목차
1. 임계값 기반 알림
시작하며
여러분이 운영 중인 서버에서 갑자기 CPU 사용률이 90%를 넘어가는데, 여러분은 그 사실을 한참 뒤에야 알게 된 적 있나요? 혹은 디스크 공간이 거의 다 찼는데 사용자들이 먼저 문제를 제기해서 당황했던 경험은요?
이런 문제는 실제 개발 현장에서 매일같이 발생합니다. 시스템 리소스나 애플리케이션 메트릭이 위험한 수준에 도달했을 때 빠르게 대응하지 못하면, 서비스 장애로 이어질 수 있고, 이는 곧 사용자 경험 저하와 비즈니스 손실로 연결됩니다.
바로 이럴 때 필요한 것이 임계값 기반 알림 시스템입니다. 미리 설정한 기준값을 넘어서는 순간, 자동으로 담당자에게 알림을 보내서 문제가 커지기 전에 빠르게 대응할 수 있게 해줍니다.
개요
간단히 말해서, 임계값 기반 알림은 "이 값이 이 선을 넘으면 알려줘!"라는 규칙을 시스템에 설정해두는 것입니다. 마치 온도계에 경보 기능을 달아두는 것과 같죠.
왜 이 개념이 필요한지 실무 관점에서 생각해보면, 24시간 내내 모니터링 대시보드를 지켜볼 수는 없기 때문입니다. CPU 사용률이 80%를 넘거나, 메모리 사용량이 4GB를 초과하거나, 에러 로그가 분당 10건 이상 발생하는 경우 자동으로 알림을 받으면, 담당자는 다른 업무를 하다가도 즉시 대응할 수 있습니다.
기존에는 cron으로 정기적으로 스크립트를 돌려서 로그를 확인하고, 문제가 있으면 수동으로 팀원들에게 연락했다면, 이제는 임계값만 설정해두면 시스템이 알아서 판단하고 알림을 보낼 수 있습니다. 이 개념의 핵심 특징은 첫째, 설정이 간단하고 명확하다는 점입니다.
"CPU > 80%"처럼 누구나 이해할 수 있는 조건식으로 표현됩니다. 둘째, 즉시성입니다.
조건이 만족되는 순간 바로 알림이 발송되어 빠른 대응이 가능합니다. 셋째, 자동화입니다.
한 번 설정해두면 사람의 개입 없이 계속 작동합니다. 이러한 특징들이 시스템 안정성을 크게 향상시키는 이유입니다.
코드 예제
# threshold_alert.py
import psutil
import time
# 임계값 설정
CPU_THRESHOLD = 80.0 # CPU 사용률 80% 초과 시 알림
MEMORY_THRESHOLD = 85.0 # 메모리 사용률 85% 초과 시 알림
DISK_THRESHOLD = 90.0 # 디스크 사용률 90% 초과 시 알림
def check_system_metrics():
# 현재 시스템 메트릭 수집
cpu_percent = psutil.cpu_percent(interval=1)
memory_percent = psutil.virtual_memory().percent
disk_percent = psutil.disk_usage('/').percent
# 임계값 체크 및 알림
if cpu_percent > CPU_THRESHOLD:
send_alert(f"⚠️ CPU 사용률 위험: {cpu_percent}%")
if memory_percent > MEMORY_THRESHOLD:
send_alert(f"⚠️ 메모리 사용률 위험: {memory_percent}%")
if disk_percent > DISK_THRESHOLD:
send_alert(f"⚠️ 디스크 사용률 위험: {disk_percent}%")
def send_alert(message):
# 알림 전송 (여기서는 로그 출력)
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {message}")
# 실제로는 Slack, 이메일 등으로 전송
# 1분마다 체크
while True:
check_system_metrics()
time.sleep(60)
설명
이것이 하는 일: 이 코드는 시스템의 주요 리소스(CPU, 메모리, 디스크)를 1분마다 자동으로 체크하고, 각각의 사용률이 미리 정해둔 임계값을 넘으면 즉시 알림을 보냅니다. 마치 건강검진에서 혈압이나 혈당 수치가 정상 범위를 벗어나면 경고를 주는 것과 같습니다.
첫 번째 단계로, 코드 상단에서 각 리소스의 임계값을 상수로 정의합니다. CPU_THRESHOLD = 80.0은 "CPU 사용률이 80%를 넘으면 위험하다"는 기준을 설정하는 것입니다.
이렇게 상수로 분리해두면 나중에 임계값을 변경할 때 한 곳만 수정하면 되어 관리가 편리합니다. 실무에서는 서버 사양이나 서비스 특성에 맞게 이 값들을 조정합니다.
그 다음으로, check_system_metrics() 함수가 실행되면서 psutil 라이브러리를 통해 현재 시스템 상태를 실시간으로 수집합니다. psutil.cpu_percent()는 CPU 사용률을, virtual_memory().percent는 메모리 사용률을, disk_usage('/').percent는 디스크 사용률을 백분율로 반환합니다.
이 값들을 변수에 저장한 후, 각각을 임계값과 비교합니다. 세 번째 단계에서는 if 조건문을 통해 임계값 초과 여부를 검사합니다.
예를 들어 CPU 사용률이 85%라면 CPU_THRESHOLD(80%)를 초과했으므로 send_alert() 함수가 호출되어 "⚠️ CPU 사용률 위험: 85%"라는 메시지를 전송합니다. 마지막으로 while True 무한 루프 안에서 time.sleep(60)으로 60초마다 이 체크를 반복합니다.
여러분이 이 코드를 사용하면 서버에 문제가 생기기 전에 미리 경고를 받을 수 있어, 사전 대응이 가능합니다. 예를 들어 디스크 사용률이 90%에 도달했다는 알림을 받으면, 100%가 되어 서비스가 멈추기 전에 불필요한 파일을 삭제하거나 디스크를 증설할 수 있습니다.
또한 CPU나 메모리 문제를 조기에 발견하면 서버를 증설하거나 코드를 최적화할 수 있는 시간을 확보할 수 있습니다.
실전 팁
💡 임계값은 너무 낮게 설정하면 불필요한 알림이 너무 많이 와서 "알림 피로"가 생기고, 너무 높게 설정하면 이미 문제가 심각해진 후에야 알게 됩니다. 운영하면서 적절한 값을 찾아가세요 (보통 CPU/메모리는 80-85%, 디스크는 85-90%가 적당합니다).
💡 psutil 라이브러리는 pip install psutil 명령으로 설치할 수 있으며, Windows, Linux, macOS 모두에서 동작하는 크로스 플랫폼 라이브러리입니다.
💡 실제 운영 환경에서는 이 스크립트를 systemd 서비스나 supervisord로 등록해서 백그라운드에서 계속 실행되도록 하고, 서버 재부팅 시에도 자동으로 시작되게 설정하세요.
💡 send_alert() 함수는 현재 print만 하고 있지만, 실제로는 Slack webhook이나 SMTP 이메일 전송 코드로 교체해야 합니다. 다음 섹션에서 자세히 다룹니다.
💡 로그 파일에 알림 이력을 함께 저장해두면, 나중에 "언제 어떤 문제가 얼마나 자주 발생했는지" 분석할 수 있어 시스템 용량 계획에 도움이 됩니다.
2. 급증 패턴 탐지 알고리즘
시작하며
여러분의 서비스에서 트래픽이 평소 시간당 100건 정도인데, 갑자기 10분 만에 500건으로 급증했다면 어떻게 하시겠어요? 이게 정상적인 이벤트 때문인지, 아니면 봇 공격이나 시스템 오류인지 빠르게 판단해야 하는데, 사람이 일일이 그래프를 보고 있을 수는 없죠.
이런 문제는 실제로 서비스 운영 중 빈번하게 발생합니다. 단순히 "값이 높다"가 아니라 "평소보다 급격하게 증가했다"는 패턴을 감지하는 것이 중요합니다.
에러 로그가 갑자기 폭증하거나, API 호출이 평소의 10배로 늘어나거나, 데이터베이스 쿼리 시간이 급격히 증가하는 등의 이상 징후를 놓치면 대규모 장애로 이어질 수 있습니다. 바로 이럴 때 필요한 것이 급증 패턴 탐지 알고리즘입니다.
과거 데이터와 비교하여 현재 값이 비정상적으로 증가했는지 자동으로 판단하고, 즉시 알림을 보내 빠른 대응을 가능하게 합니다.
개요
간단히 말해서, 급증 패턴 탐지는 "지금 이 값이 평소와 비교했을 때 너무 갑자기 올라갔는지"를 자동으로 확인하는 알고리즘입니다. 마치 주식 시장에서 급등주를 찾아내는 것처럼, 메트릭의 급격한 변화를 감지합니다.
왜 이 개념이 필요한지 실무 관점에서 살펴보면, 절대값만으로는 문제를 감지하기 어려운 경우가 많기 때문입니다. 예를 들어 에러 로그가 분당 10건 발생한다고 해서 무조건 문제는 아닙니다.
평소에도 그 정도 나왔다면 정상이지만, 평소 1건이었는데 갑자기 10건으로 늘었다면 심각한 문제입니다. 급증 패턴 탐지는 이런 상대적 변화를 감지합니다.
기존에는 단순히 임계값만 체크했다면 (예: "에러 > 100건"), 이제는 변화율까지 고려할 수 있습니다 (예: "에러가 평균 대비 300% 증가"). 이를 통해 더 정확하고 맥락을 고려한 알림이 가능합니다.
이 알고리즘의 핵심 특징은 첫째, 이동 평균을 사용해서 과거 패턴을 파악한다는 점입니다. 최근 N개의 데이터 포인트 평균을 계산해서 "정상" 기준선을 만듭니다.
둘째, 표준편차나 백분율 변화를 활용해서 "얼마나 벗어났는지"를 수치화합니다. 셋째, 시간대별 패턴을 고려할 수 있습니다 (예: 낮과 밤의 트래픽 차이).
이러한 특징들이 오탐을 줄이고 진짜 이상 상황만 잡아내는 데 도움을 줍니다.
코드 예제
# spike_detection.py
from collections import deque
import statistics
class SpikeDetector:
def __init__(self, window_size=10, threshold_multiplier=2.0):
# 최근 데이터를 저장할 큐 (최대 window_size개)
self.history = deque(maxlen=window_size)
# 급증 판단 기준 (평균의 몇 배인가)
self.threshold_multiplier = threshold_multiplier
def add_data_point(self, value):
"""새로운 데이터 포인트 추가 및 급증 여부 반환"""
# 이전 데이터가 충분하면 급증 여부 체크
if len(self.history) >= 5: # 최소 5개 데이터 필요
avg = statistics.mean(self.history)
std_dev = statistics.stdev(self.history)
# 현재 값이 평균 + (표준편차 * 배수)를 초과하면 급증
threshold = avg + (std_dev * self.threshold_multiplier)
if value > threshold:
spike_percent = ((value - avg) / avg) * 100
print(f"🚨 급증 탐지! 현재값: {value:.1f}, "
f"평균: {avg:.1f}, 증가율: {spike_percent:.1f}%")
return True
# 히스토리에 현재 값 추가
self.history.append(value)
return False
# 사용 예시: API 요청 수 모니터링
detector = SpikeDetector(window_size=10, threshold_multiplier=2.0)
# 정상 트래픽
for count in [100, 105, 98, 102, 99, 103, 101, 100, 104, 98]:
detector.add_data_point(count)
# 갑작스런 급증!
detector.add_data_point(250) # 급증 알림 발생
설명
이것이 하는 일: 이 코드는 최근 일정 개수의 데이터 포인트를 계속 기억하고 있다가, 새로운 값이 들어올 때마다 "이 값이 과거 패턴에 비해 비정상적으로 높은가?"를 통계적으로 판단합니다. 마치 체온계가 평소 체온을 기억하고 있다가 갑자기 높은 열이 나면 알려주는 것과 같습니다.
첫 번째로, SpikeDetector 클래스의 init 메서드에서 deque(덱)를 사용해 최근 데이터를 저장합니다. deque는 maxlen을 설정하면 자동으로 오래된 데이터를 제거하고 새 데이터를 추가하는 "슬라이딩 윈도우" 역할을 합니다.
window_size=10이면 최근 10개 데이터만 유지하고, threshold_multiplier=2.0은 "평균보다 표준편차의 2배 이상 높으면 급증"이라는 기준을 설정합니다. 그 다음으로, add_data_point() 메서드가 호출되면 먼저 충분한 히스토리 데이터가 있는지 확인합니다(최소 5개).
데이터가 충분하면 statistics.mean()으로 평균을, stdev()로 표준편차를 계산합니다. 그리고 threshold = avg + (std_dev * threshold_multiplier) 공식으로 "급증"의 기준선을 동적으로 계산합니다.
예를 들어 평균이 100, 표준편차가 5라면, 기준선은 100 + (5 * 2) = 110이 됩니다. 세 번째 단계에서 현재 값(value)이 이 기준선을 초과하면 급증으로 판단하고, 증가율을 백분율로 계산해서 알림 메시지를 출력합니다.
그 후 현재 값을 히스토리에 추가하여 다음 비교를 위한 데이터로 활용합니다. 예시 코드에서는 평소 100 근처였던 값이 250으로 급증하자 즉시 탐지됩니다.
여러분이 이 알고리즘을 사용하면 단순 임계값으로는 잡아낼 수 없는 미묘한 이상 패턴을 감지할 수 있습니다. 예를 들어 새벽 시간대에는 트래픽이 10인데 갑자기 50으로 늘면 급증이지만, 낮 시간대에 트래픽이 500에서 550으로 늘어난 건 정상 범위일 수 있습니다.
이 알고리즘은 시간대별로 다른 기준을 자동으로 적용합니다. 또한 DDoS 공격, 메모리 누수로 인한 점진적 증가, 무한 루프로 인한 로그 폭증 등을 조기에 발견할 수 있습니다.
실전 팁
💡 window_size(윈도우 크기)는 모니터링 주기에 따라 조정하세요. 1분마다 체크한다면 window_size=10은 최근 10분, 10초마다 체크한다면 최근 100초를 의미합니다. 너무 작으면 민감하고, 너무 크면 둔감해집니다.
💡 threshold_multiplier 값이 작을수록 (예: 1.5) 민감하게 반응하고, 클수록 (예: 3.0) 확실한 급증만 잡아냅니다. 초기에는 2.0으로 시작해서 오탐/미탐 비율을 보고 조정하세요.
💡 시간대별로 다른 detector 인스턴스를 사용하면 더 정확합니다. 예를 들어 낮 시간대용, 밤 시간대용, 주말용 detector를 따로 만들어 각각의 "정상 패턴"을 학습시키세요.
💡 이 알고리즘은 "급증"만 탐지합니다. "급감"도 탐지하려면 if value < (avg - std_dev * threshold_multiplier) 조건을 추가하세요. 트래픽 급감도 서버 다운 등의 문제 신호일 수 있습니다.
💡 실제 운영에서는 Redis나 InfluxDB 같은 시계열 데이터베이스에 히스토리를 저장하면, 서버 재시작 후에도 과거 패턴을 유지할 수 있고, 여러 서버가 같은 히스토리를 공유할 수 있습니다.
3. Slack 이메일 알림 연동
시작하며
여러분이 만든 모니터링 시스템이 문제를 감지했는데, 그 알림이 서버 로그 파일에만 남아있다면 아무 소용이 없겠죠? 담당자가 실시간으로 로그를 확인하지 않는 한, 문제를 알아차릴 수 없으니까요.
이런 문제는 실제로 많은 스타트업과 중소기업에서 발생합니다. 훌륭한 모니터링 시스템을 구축해놓고도, 알림이 제대로 전달되지 않아서 장애 대응이 늦어지는 경우가 허다합니다.
특히 밤이나 주말에 발생한 문제는 출근해서야 알게 되는 경우도 있죠. 바로 이럴 때 필요한 것이 Slack이나 이메일 같은 실시간 커뮤니케이션 도구와의 연동입니다.
문제가 감지되는 순간, 팀원들의 Slack 채널에 알림이 울리거나, 담당자의 이메일로 긴급 알림이 전송되어 어디서든 즉시 확인하고 대응할 수 있게 됩니다.
개요
간단히 말해서, Slack/이메일 알림 연동은 모니터링 시스템이 감지한 이상 상황을 사람들이 실제로 사용하는 커뮤니케이션 채널로 자동으로 전송하는 것입니다. 마치 화재경보기가 울리면 소방서에 자동으로 신고되는 것처럼, 시스템 문제가 감지되면 담당자에게 자동으로 알려줍니다.
왜 이 개념이 필요한지 실무 관점에서 보면, 현대 개발팀은 대부분 Slack, Teams, Discord 같은 협업 도구를 사용하고, 중요한 알림은 이메일로 받기 때문입니다. 서버에 SSH로 접속해서 로그를 확인하는 건 이미 문제를 인지한 후의 행동이지, 문제를 발견하는 방법이 될 수 없습니다.
알림이 팀원들이 항상 확인하는 채널로 전송되어야 빠른 대응이 가능합니다. 기존에는 로그 파일을 주기적으로 확인하거나, 별도의 모니터링 대시보드를 열어두고 지켜봐야 했다면, 이제는 평소처럼 Slack을 사용하다가 문제가 생기면 즉시 알림을 받을 수 있습니다.
심지어 모바일 앱으로도 알림이 오기 때문에 외출 중에도 확인 가능합니다. 이 연동의 핵심 특징은 첫째, 즉시성입니다.
Slack webhook은 보통 1-2초 이내에 메시지가 도착하므로 실시간 대응이 가능합니다. 둘째, 가시성입니다.
팀 전체가 보는 채널에 알림이 오면 누군가는 반드시 확인하게 됩니다. 셋째, 풍부한 포맷입니다.
단순 텍스트가 아니라 색상, 이모지, 버튼, 링크 등을 활용해 정보를 명확하게 전달할 수 있습니다. 이러한 특징들이 알림의 효과를 극대화합니다.
코드 예제
# slack_email_alert.py
import requests
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
class AlertManager:
def __init__(self, slack_webhook_url, smtp_config):
self.slack_webhook = slack_webhook_url
self.smtp_config = smtp_config
def send_slack_alert(self, message, severity="warning"):
"""Slack으로 알림 전송"""
# 심각도에 따른 색상 설정
colors = {"info": "#36a64f", "warning": "#ff9900", "critical": "#ff0000"}
payload = {
"attachments": [{
"color": colors.get(severity, "#808080"),
"title": f"🚨 시스템 알림 [{severity.upper()}]",
"text": message,
"footer": "모니터링 시스템",
"ts": int(time.time())
}]
}
response = requests.post(self.slack_webhook, json=payload)
return response.status_code == 200
def send_email_alert(self, subject, body, recipients):
"""이메일로 알림 전송"""
msg = MIMEMultipart()
msg['From'] = self.smtp_config['from']
msg['To'] = ', '.join(recipients)
msg['Subject'] = f"[시스템 알림] {subject}"
msg.attach(MIMEText(body, 'plain'))
# SMTP 서버 연결 및 전송
with smtplib.SMTP(self.smtp_config['server'], self.smtp_config['port']) as server:
server.starttls()
server.login(self.smtp_config['user'], self.smtp_config['password'])
server.send_message(msg)
return True
# 사용 예시
import time
alert_mgr = AlertManager(
slack_webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
smtp_config={
'server': 'smtp.gmail.com',
'port': 587,
'user': 'your-email@gmail.com',
'password': 'your-app-password',
'from': 'alert@yourcompany.com'
}
)
# Slack 알림
alert_mgr.send_slack_alert("CPU 사용률 85% 초과", severity="warning")
# 이메일 알림
alert_mgr.send_email_alert(
subject="디스크 공간 부족",
body="서버 disk-01의 사용률이 92%에 도달했습니다.",
recipients=["admin@company.com", "devops@company.com"]
)
설명
이것이 하는 일: 이 코드는 모니터링 시스템이 감지한 문제를 Slack 채널에 예쁘게 포맷된 메시지로 보내거나, 담당자들의 이메일로 즉시 전송합니다. 마치 응급 상황에서 119에 자동으로 신고하는 시스템처럼, 시스템 이상을 사람들이 즉시 볼 수 있는 곳으로 보냅니다.
첫 번째로, AlertManager 클래스를 초기화할 때 Slack webhook URL과 SMTP 서버 설정을 받아 저장합니다. Slack webhook URL은 Slack 워크스페이스 설정에서 "Incoming Webhooks" 앱을 추가하면 얻을 수 있는 고유 URL입니다.
이 URL로 POST 요청을 보내면 지정된 채널에 메시지가 나타납니다. SMTP 설정은 Gmail, AWS SES, SendGrid 등 이메일 서버 정보를 담고 있습니다.
그 다음으로, send_slack_alert() 메서드는 메시지와 심각도(severity)를 받아서 Slack 메시지 포맷으로 변환합니다. severity에 따라 메시지 색상이 달라져서(info는 초록색, warning은 주황색, critical은 빨간색) 한눈에 심각도를 파악할 수 있습니다.
payload 딕셔너리를 JSON으로 변환하여 requests.post()로 webhook URL에 전송하면, 몇 초 안에 Slack 채널에 메시지가 나타납니다. 세 번째로, send_email_alert() 메서드는 제목, 본문, 수신자 리스트를 받아서 이메일을 구성합니다.
MIMEMultipart를 사용해 이메일 헤더(From, To, Subject)를 설정하고, MIMEText로 본문을 추가합니다. 그 후 smtplib를 통해 SMTP 서버에 연결하고, starttls()로 보안 연결을 수립한 뒤, login()으로 인증하고, send_message()로 실제 이메일을 전송합니다.
with 문을 사용하면 자동으로 연결이 종료됩니다. 여러분이 이 코드를 사용하면 서버실에 있지 않아도, 심지어 퇴근 후 집에 있어도 시스템 문제를 즉시 알 수 있습니다.
Slack 알림은 모바일 푸시로도 오기 때문에 이동 중에도 확인 가능하고, 이메일은 공식적인 기록으로 남아 나중에 문제 발생 시점과 대응 과정을 추적할 수 있습니다. 또한 심각도별로 색상을 다르게 하면, 팀원들이 "이건 즉시 봐야 하는 문제인지, 나중에 확인해도 되는 문제인지"를 빠르게 판단할 수 있습니다.
실전 팁
💡 Slack webhook URL은 절대 코드에 하드코딩하지 마세요. 환경변수나 설정 파일로 분리하고, .gitignore에 추가해서 GitHub에 노출되지 않도록 하세요. 노출되면 누구나 여러분의 Slack에 메시지를 보낼 수 있습니다.
💡 Gmail을 SMTP 서버로 사용할 때는 "앱 비밀번호"를 발급받아야 합니다. 일반 비밀번호로는 보안 정책상 로그인이 차단됩니다. Google 계정 설정 > 보안 > 2단계 인증 > 앱 비밀번호에서 생성할 수 있습니다.
💡 Slack 메시지에 버튼을 추가하면 클릭 한 번으로 서버 재시작, 로그 확인 등의 액션을 실행할 수 있습니다. "Interactive Components"를 활용하면 알림에서 바로 대응까지 가능합니다.
💡 이메일은 전송 속도가 느릴 수 있으므로(수 초~수십 초) 긴급 알림은 Slack을, 공식 기록용이나 관리자 보고용은 이메일을 사용하는 식으로 병행하세요.
💡 알림 전송 실패에 대비해 try-except로 예외 처리를 하고, 실패 시 로그 파일에 기록하거나 대체 채널로 전송하는 fallback 로직을 구현하세요. 알림 시스템 자체가 다운되면 안 됩니다.
4. 알림 중복 방지 로직
시작하며
여러분의 모니터링 시스템이 완벽하게 작동해서 CPU 과부하를 감지했습니다. 그런데 1분마다 체크하는 시스템이 10분 동안 같은 문제를 계속 감지하면서, Slack에 같은 알림을 10번 연속으로 보냈다면 어떨까요?
팀원들의 폰이 계속 울리면서 정작 중요한 새로운 알림을 놓칠 수도 있습니다. 이런 문제는 실제 운영 환경에서 매우 흔하게 발생합니다.
같은 이슈에 대해 수십, 수백 개의 중복 알림이 쏟아지면서 "알림 피로(alert fatigue)"가 생기고, 결국 팀원들이 알림 자체를 무시하게 되는 상황이 벌어집니다. 심지어 정말 중요한 새로운 문제가 발생해도 수많은 중복 알림에 묻혀서 놓치게 됩니다.
바로 이럴 때 필요한 것이 알림 중복 방지 로직입니다. 같은 문제에 대해서는 최초 1회만 알림을 보내고, 일정 시간이 지나거나 문제가 해결된 후 다시 발생했을 때만 새로 알림을 보내는 똑똑한 시스템을 만들어야 합니다.
개요
간단히 말해서, 알림 중복 방지는 "이 문제 이미 알렸으니까, 해결될 때까지는 또 알리지 말자"는 기억 기능을 시스템에 추가하는 것입니다. 마치 친구에게 "밥 먹었어?"라고 물었는데 대답을 못 들었다고 해서 1분마다 계속 묻지 않는 것처럼, 시스템도 한 번 알린 문제는 기억하고 있어야 합니다.
왜 이 개념이 필요한지 실무 관점에서 보면, 알림의 목적은 "문제를 알리는 것"이지 "팀원을 괴롭히는 것"이 아니기 때문입니다. CPU가 90%라는 알림을 받으면 담당자는 즉시 확인하고 조치를 취합니다.
하지만 조치하는 동안 같은 알림이 계속 오면 집중력만 떨어뜨립니다. 또한 Slack이나 이메일 채널이 같은 메시지로 도배되면 정작 다른 중요한 알림(예: 디스크 문제, 네트워크 문제)을 놓칠 수 있습니다.
기존에는 알림을 그냥 무조건 보냈다면, 이제는 "마지막으로 이 알림을 보낸 게 언제인지" 또는 "이 문제가 현재 진행 중인지"를 체크해서 스마트하게 알림을 관리합니다. 예를 들어 "같은 문제는 1시간에 최대 1회만 알림" 같은 정책을 적용할 수 있습니다.
이 로직의 핵심 특징은 첫째, 알림 이력을 저장한다는 점입니다. 어떤 문제를 언제 알렸는지 기록해둡니다.
둘째, 쿨다운(cooldown) 기간을 설정합니다. 같은 문제는 최소 N분 간격으로만 재알림합니다.
셋째, 문제 해결 감지입니다. 문제가 사라지면 "해결됨" 알림을 보내고, 그 후 다시 발생하면 새 문제로 간주합니다.
이러한 특징들이 알림의 효율성을 크게 높입니다.
코드 예제
# alert_deduplication.py
import time
from datetime import datetime, timedelta
class AlertDeduplicator:
def __init__(self, cooldown_minutes=60):
# 알림 이력: {alert_key: {'last_sent': timestamp, 'active': bool}}
self.alert_history = {}
# 재알림까지의 대기 시간 (분 단위)
self.cooldown = timedelta(minutes=cooldown_minutes)
def should_send_alert(self, alert_key, current_value, threshold):
"""알림을 보내야 하는지 판단"""
now = datetime.now()
is_problem = current_value > threshold
# 이 문제에 대한 이력이 있는지 확인
if alert_key in self.alert_history:
history = self.alert_history[alert_key]
last_sent = history['last_sent']
was_active = history['active']
if is_problem:
# 문제가 계속되는 중
if was_active:
# 쿨다운 시간이 지났는지 확인
if now - last_sent < self.cooldown:
return False # 아직 쿨다운 중이므로 알림 안 보냄
else:
# 쿨다운 지났으므로 재알림
self.alert_history[alert_key] = {'last_sent': now, 'active': True}
return True
else:
# 문제가 재발생 (이전에 해결됐었음)
self.alert_history[alert_key] = {'last_sent': now, 'active': True}
return True
else:
# 문제가 해결됨
if was_active:
# 해결 알림 전송
self.alert_history[alert_key] = {'last_sent': now, 'active': False}
print(f"✅ {alert_key} 문제 해결됨")
return False
else:
# 최초 발생
if is_problem:
self.alert_history[alert_key] = {'last_sent': now, 'active': True}
return True
return False
# 사용 예시
dedup = AlertDeduplicator(cooldown_minutes=30)
# 첫 번째 알림: 전송됨
if dedup.should_send_alert("cpu_high", 85, 80):
print("🚨 CPU 과부하 알림 전송")
# 5분 후 같은 문제: 전송 안 됨 (쿨다운 중)
time.sleep(1) # 실제로는 300초
if dedup.should_send_alert("cpu_high", 87, 80):
print("🚨 CPU 과부하 알림 전송")
else:
print("⏳ 쿨다운 중이므로 알림 생략")
# 문제 해결
dedup.should_send_alert("cpu_high", 70, 80) # ✅ 해결 메시지 출력
설명
이것이 하는 일: 이 코드는 각 알림에 대해 "언제 마지막으로 보냈는지"와 "현재 문제가 진행 중인지"를 기억하고 있다가, 새로운 알림 요청이 들어오면 "이미 알렸으니 또 보낼 필요 없다" 또는 "충분히 시간이 지났으니 다시 알려야겠다"를 판단합니다. 마치 스마트폰의 "방해 금지 모드"처럼, 같은 사람의 연속 전화를 일정 시간 차단하는 것과 유사합니다.
첫 번째로, AlertDeduplicator 클래스는 alert_history라는 딕셔너리를 통해 각 알림의 상태를 추적합니다. alert_key는 알림의 고유 식별자(예: "cpu_high", "memory_high", "disk_full")이고, 각 키에 대해 마지막 전송 시각(last_sent)과 문제 진행 여부(active)를 저장합니다.
cooldown_minutes는 같은 알림의 재전송까지 기다려야 하는 시간을 설정합니다. 그 다음으로, should_send_alert() 메서드는 현재 값과 임계값을 비교해서 문제 발생 여부(is_problem)를 판단합니다.
그리고 alert_history를 확인해서 이 문제가 처음인지, 계속되는 중인지, 재발한 건지를 구분합니다. 예를 들어 CPU가 85%로 처음 임계값(80%)을 넘으면 즉시 알림을 보내고, 이 정보를 history에 기록합니다.
세 번째 단계에서는 상황별 로직이 실행됩니다. (1) 문제가 계속되는 중이고 쿨다운 기간이 아직 안 지났으면 False를 반환해서 알림을 생략합니다.
(2) 쿨다운이 지났으면 True를 반환해서 "문제가 아직 계속되고 있다"는 재알림을 보냅니다. (3) 이전에 active=False였던 문제가 다시 발생하면 새 문제로 간주해서 즉시 알림을 보냅니다.
(4) 문제가 해결되면(current_value < threshold) active를 False로 바꾸고 "해결됨" 메시지를 출력합니다. 여러분이 이 로직을 사용하면 Slack 채널이 같은 알림으로 도배되는 것을 막을 수 있고, 팀원들은 정말 중요한 새로운 문제에만 집중할 수 있습니다.
예를 들어 CPU 문제가 30분 동안 계속될 때, 처음 1회만 알림이 오고 30분 후 재알림이 한 번 더 오는 식으로 적절한 빈도를 유지합니다. 또한 문제가 해결되면 자동으로 "해결됨" 상태를 기록해서, 다음에 같은 문제가 발생하면 "재발"로 명확히 알 수 있습니다.
실전 팁
💡 cooldown 시간은 문제의 심각도에 따라 다르게 설정하세요. Critical 문제는 10분, Warning은 30분, Info는 1시간처럼 차등 적용하면 중요한 문제는 더 자주 리마인드됩니다.
💡 alert_history는 메모리에만 있으므로 서버가 재시작되면 사라집니다. 영구 저장이 필요하면 Redis나 SQLite에 저장하세요. 특히 여러 서버가 있는 환경이라면 Redis를 공유 저장소로 사용해야 합니다.
💡 문제 해결 알림(✅)도 Slack으로 보내면 좋습니다. 담당자가 "아, 이 문제는 자동으로 해결됐구나" 또는 "내가 한 조치가 효과가 있었구나"를 확인할 수 있습니다.
💡 쿨다운 중에도 문제가 "악화"되면 즉시 알림을 보내도록 개선할 수 있습니다. 예를 들어 CPU 85%로 알림을 보냈는데, 쿨다운 중에 95%로 급증하면 "상황 악화" 알림을 추가로 보내는 식입니다.
💡 alert_key를 서버별로 분리하세요. "server1_cpu_high", "server2_cpu_high"처럼 하면 여러 서버를 모니터링할 때 각 서버의 상태를 독립적으로 추적할 수 있습니다.
5. 심각도별 알림 분류
시작하며
여러분의 모니터링 시스템이 하루에 수십 개의 알림을 보낸다고 생각해보세요. 디스크 사용률 70% 알림, CPU 85% 알림, 메모리 90% 알림, 에러 로그 발견 알림...
이 중에서 어떤 걸 먼저 봐야 할까요? 전부 똑같은 빨간색 알림으로 오면 우선순위를 정하기 어렵습니다.
이런 문제는 실제로 운영 규모가 커질수록 더 심각해집니다. 알림이 너무 많아지면 정작 서비스를 멈출 수 있는 치명적인 문제와, 나중에 봐도 되는 가벼운 경고를 구분하기 어려워집니다.
새벽 2시에 온콜 담당자의 폰이 울렸는데, 알고 보니 "디스크 60% 사용 중"이라는 여유 있는 경고였다면 불필요한 수면 방해가 됩니다. 바로 이럴 때 필요한 것이 심각도별 알림 분류 시스템입니다.
문제를 Info, Warning, Critical 같은 등급으로 나누고, 각 등급에 따라 다른 방식으로 알림을 보내서 담당자가 우선순위를 쉽게 파악하고 적절히 대응할 수 있게 합니다.
개요
간단히 말해서, 심각도별 알림 분류는 문제를 "얼마나 급한지"에 따라 등급을 매기고, 등급에 맞는 알림 방식을 사용하는 것입니다. 마치 병원 응급실에서 환자를 긴급도에 따라 트리아지(triage)하는 것처럼, 시스템 문제도 심각도를 분류합니다.
왜 이 개념이 필요한지 실무 관점에서 살펴보면, 모든 문제가 똑같이 긴급한 건 아니기 때문입니다. 디스크 사용률 60%는 "지켜봐야 할 상황"이지만, 95%는 "지금 당장 조치해야 할 위기"입니다.
CPU 순간적으로 80% 찍은 건 정상일 수 있지만, 10분 동안 95% 이상 유지되면 서비스 장애 직전입니다. 이런 차이를 반영하지 않으면 담당자는 매번 알림을 열어봐야 심각도를 파악할 수 있습니다.
기존에는 모든 알림을 동일하게 처리했다면, 이제는 Info는 Slack 메시지만, Warning은 Slack + 색상 강조, Critical은 Slack + 이메일 + SMS + 전화(PagerDuty) 같은 다단계 알림 전략을 사용합니다. 이를 통해 정말 중요한 문제는 놓치지 않으면서도, 사소한 알림으로 피로해지지 않습니다.
이 분류 시스템의 핵심 특징은 첫째, 명확한 기준입니다. "CPU 70-80%는 Warning, 80% 이상은 Critical"처럼 수치로 정의합니다.
둘째, 차등화된 응답입니다. 심각도에 따라 알림 채널, 빈도, 수신자를 다르게 설정합니다.
셋째, 시각적 구분입니다. 색상, 이모지, 제목 등으로 한눈에 심각도를 파악할 수 있게 합니다.
이러한 특징들이 효율적인 인시던트 대응을 가능하게 합니다.
코드 예제
# severity_classification.py
from enum import Enum
from dataclasses import dataclass
from typing import List
class Severity(Enum):
"""알림 심각도 레벨"""
INFO = 1 # 정보성 알림 (참고용)
WARNING = 2 # 경고 (주의 필요)
CRITICAL = 3 # 치명적 (즉시 조치 필요)
@dataclass
class AlertConfig:
"""심각도별 알림 설정"""
severity: Severity
slack: bool # Slack 알림 여부
email: bool # 이메일 알림 여부
sms: bool # SMS 알림 여부
color: str # Slack 메시지 색상
emoji: str # 이모지
recipients: List[str] # 수신자 목록
class SeverityClassifier:
def __init__(self):
# 심각도별 설정 정의
self.configs = {
Severity.INFO: AlertConfig(
severity=Severity.INFO,
slack=True, email=False, sms=False,
color="#36a64f", emoji="ℹ️",
recipients=["team-channel"]
),
Severity.WARNING: AlertConfig(
severity=Severity.WARNING,
slack=True, email=True, sms=False,
color="#ff9900", emoji="⚠️",
recipients=["team-channel", "devops@company.com"]
),
Severity.CRITICAL: AlertConfig(
severity=Severity.CRITICAL,
slack=True, email=True, sms=True,
color="#ff0000", emoji="🚨",
recipients=["emergency-channel", "oncall@company.com", "+82-10-1234-5678"]
)
}
def classify_and_alert(self, metric_name, value, thresholds):
"""메트릭 값에 따라 심각도 분류 및 알림"""
# 심각도 결정
if value >= thresholds['critical']:
severity = Severity.CRITICAL
elif value >= thresholds['warning']:
severity = Severity.WARNING
else:
severity = Severity.INFO
# 해당 심각도 설정 가져오기
config = self.configs[severity]
# 메시지 생성
message = f"{config.emoji} [{severity.name}] {metric_name}: {value}"
# 설정에 따라 알림 전송
if config.slack:
self.send_slack(message, config.color, config.recipients)
if config.email:
self.send_email(message, config.recipients)
if config.sms:
self.send_sms(message, config.recipients)
return severity
def send_slack(self, message, color, recipients):
print(f"📱 Slack 전송 ({color}): {message} -> {recipients}")
def send_email(self, message, recipients):
print(f"📧 Email 전송: {message} -> {recipients}")
def send_sms(self, message, recipients):
print(f"📞 SMS 전송: {message} -> {recipients}")
# 사용 예시
classifier = SeverityClassifier()
# CPU 사용률 체크
cpu_thresholds = {'warning': 70, 'critical': 85}
classifier.classify_and_alert("CPU 사용률", 65, cpu_thresholds) # INFO
classifier.classify_and_alert("CPU 사용률", 75, cpu_thresholds) # WARNING
classifier.classify_and_alert("CPU 사용률", 90, cpu_thresholds) # CRITICAL
설명
이것이 하는 일: 이 코드는 시스템 메트릭 값을 미리 정의된 임계값과 비교하여 자동으로 심각도를 판단하고, 심각도에 맞는 알림 채널(Slack, 이메일, SMS)과 수신자를 선택해서 적절한 방식으로 알림을 보냅니다. 마치 비행기 조종석의 경고등이 노란불과 빨간불로 나뉘어 긴급도를 표시하는 것처럼, 시스템 문제도 단계별로 구분합니다.
첫 번째로, Severity Enum으로 INFO, WARNING, CRITICAL 세 가지 등급을 정의합니다. Enum을 사용하면 오타 방지와 타입 안정성을 얻을 수 있습니다.
AlertConfig 데이터클래스는 각 심각도별로 어떤 채널을 사용할지, 어떤 색상과 이모지를 쓸지, 누구에게 보낼지를 구조화된 형태로 저장합니다. 예를 들어 CRITICAL은 모든 채널(Slack, 이메일, SMS)을 사용하지만, INFO는 Slack만 사용합니다.
그 다음으로, SeverityClassifier의 __init__에서 self.configs 딕셔너리에 각 심각도별 설정을 미리 정의해둡니다. INFO는 팀 전체 채널에만 조용히 알리고, WARNING은 DevOps 팀 이메일까지 보내고, CRITICAL은 긴급 채널 + 온콜 담당자 이메일 + SMS까지 모든 수단을 동원합니다.
이렇게 설정을 중앙화하면 나중에 정책 변경 시 한 곳만 수정하면 됩니다. 세 번째로, classify_and_alert() 메서드는 메트릭 이름(예: "CPU 사용률"), 현재 값(예: 90), 임계값들(예: warning=70, critical=85)을 받습니다.
먼저 if-elif 조건문으로 현재 값이 어느 구간에 속하는지 판단해서 severity를 결정합니다. 90은 critical 임계값(85)을 넘었으므로 Severity.CRITICAL이 됩니다.
그 후 해당 severity의 config를 가져와서, config에 설정된 채널들(slack=True, email=True, sms=True)을 통해 알림을 전송합니다. 여러분이 이 시스템을 사용하면 알림의 "노이즈"를 크게 줄일 수 있습니다.
디스크 60% 사용 같은 정보성 알림은 Slack에만 조용히 남고, 담당자는 출근해서 확인하면 됩니다. 반면 메모리 95% 같은 치명적 문제는 SMS로까지 오기 때문에 새벽이든 주말이든 즉시 대응할 수 있습니다.
또한 색상과 이모지로 시각적으로 구분되어, Slack 채널을 스크롤하다가도 "🚨 빨간색"은 눈에 즉시 띄지만 "ℹ️ 초록색"은 넘어갈 수 있습니다. 이렇게 중요도에 맞는 주의 집중이 가능해집니다.
실전 팁
💡 임계값은 절대적인 기준이 아니라 서비스 특성에 맞게 조정하세요. 배치 처리 서버는 CPU 95%도 정상일 수 있고, 실시간 API 서버는 80%도 위험할 수 있습니다. 운영 데이터를 보고 적절한 값을 찾아가세요.
💡 심각도를 4단계(INFO, WARNING, ERROR, CRITICAL)나 5단계로 더 세분화할 수도 있습니다. 하지만 너무 많으면 오히려 혼란스러우니, 3-4단계가 적당합니다.
💡 시간대를 고려하세요. 업무 시간(9-18시)과 업무 외 시간의 알림 정책을 다르게 가져갈 수 있습니다. 예를 들어 낮에는 WARNING도 이메일을 보내지만, 밤에는 CRITICAL만 SMS를 보내는 식입니다.
💡 Slack의 @here, @channel 멘션도 심각도에 따라 차등 적용하세요. CRITICAL은 @channel로 전체 알림, WARNING은 @here로 온라인 사용자만, INFO는 멘션 없이 조용히 보내면 적절합니다.
💡 PagerDuty, Opsgenie 같은 온콜 관리 시스템을 연동하면 CRITICAL 알림을 자동으로 당직자에게 에스컬레이션하고, 일정 시간 응답 없으면 다음 담당자에게 전달하는 고급 기능을 사용할 수 있습니다.
6. 알림 이력 관리
시작하며
여러분의 팀이 지난 한 달 동안 받은 알림이 몇 개나 되는지, 어떤 문제가 가장 자주 발생했는지, 평균 대응 시간은 얼마나 되는지 알고 계신가요? 알림을 받고 대응하는 것도 중요하지만, 이런 데이터를 모아서 분석하면 시스템을 근본적으로 개선할 수 있는 인사이트를 얻을 수 있습니다.
이런 문제는 실제로 많은 팀이 간과하는 부분입니다. 알림을 받고, Slack에서 대화하고, 문제를 해결하고 나면 그걸로 끝입니다.
하지만 "CPU 과부하 알림이 매주 금요일마다 온다"는 패턴을 발견하면 자동 스케일링을 설정할 수 있고, "디스크 알림이 한 달에 10번 왔다"는 데이터가 있으면 스토리지 증설 예산을 요청할 근거가 됩니다. 바로 이럴 때 필요한 것이 알림 이력 관리 시스템입니다.
모든 알림을 데이터베이스나 로그 파일에 체계적으로 저장하고, 나중에 조회, 검색, 분석할 수 있게 만들어서 지속적인 시스템 개선의 기반을 마련합니다.
개요
간단히 말해서, 알림 이력 관리는 "언제, 무슨 알림이 왔고, 얼마나 심각했고, 얼마 만에 해결됐는지"를 모두 기록으로 남기는 것입니다. 마치 병원에서 환자의 진료 기록을 남기는 것처럼, 시스템의 건강 기록을 체계적으로 관리합니다.
왜 이 개념이 필요한지 실무 관점에서 보면, 데이터 없이는 개선이 불가능하기 때문입니다. "서버가 자주 느려지는 것 같은데요"라는 주관적인 느낌이 아니라, "지난 한 달간 CPU 과부하 알림이 37건 발생했고, 그 중 82%가 금요일 오후에 집중되어 있습니다"라는 객관적인 데이터가 있어야 경영진을 설득하고 예산을 받을 수 있습니다.
또한 알림 정책의 효과를 측정할 수도 있습니다. 임계값을 조정한 후 오탐률이 줄었는지, 평균 대응 시간이 개선됐는지 등을 수치로 확인할 수 있습니다.
기존에는 알림이 Slack 메시지나 이메일로만 남고 시간이 지나면 묻혔다면, 이제는 구조화된 데이터베이스에 저장되어 언제든지 검색하고 통계를 낼 수 있습니다. "지난 분기 CRITICAL 알림 추이", "서버별 알림 발생 빈도", "주간 알림 패턴" 같은 리포트를 자동으로 생성할 수 있습니다.
이 시스템의 핵심 특징은 첫째, 구조화된 저장입니다. timestamp, severity, metric_name, value, threshold 등을 필드로 분리해서 저장합니다.
둘째, 검색 가능성입니다. 날짜 범위, 심각도, 메트릭 종류 등으로 필터링해서 조회할 수 있습니다.
셋째, 통계 분석입니다. 집계 쿼리를 통해 트렌드, 패턴, 이상치를 발견할 수 있습니다.
이러한 특징들이 데이터 기반 운영을 가능하게 합니다.
코드 예제
# alert_history.py
import sqlite3
from datetime import datetime
from typing import List, Dict
import json
class AlertHistoryManager:
def __init__(self, db_path="alert_history.db"):
self.db_path = db_path
self._init_database()
def _init_database(self):
"""데이터베이스 및 테이블 초기화"""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
severity TEXT NOT NULL,
metric_name TEXT NOT NULL,
current_value REAL NOT NULL,
threshold REAL NOT NULL,
message TEXT,
resolved_at TEXT,
resolution_time_seconds INTEGER
)
""")
# 검색 성능을 위한 인덱스
conn.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON alerts(timestamp)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_severity ON alerts(severity)")
def log_alert(self, severity, metric_name, current_value, threshold, message=""):
"""알림 발생 기록"""
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT INTO alerts
(timestamp, severity, metric_name, current_value, threshold, message)
VALUES (?, ?, ?, ?, ?, ?)
""", (
datetime.now().isoformat(),
severity,
metric_name,
current_value,
threshold,
message
))
print(f"✅ 알림 이력 저장됨: {metric_name} = {current_value}")
def resolve_alert(self, alert_id):
"""알림 해결 처리"""
with sqlite3.connect(self.db_path) as conn:
# 원래 발생 시각 조회
cursor = conn.execute("SELECT timestamp FROM alerts WHERE id = ?", (alert_id,))
row = cursor.fetchone()
if row:
created_at = datetime.fromisoformat(row[0])
resolved_at = datetime.now()
resolution_time = int((resolved_at - created_at).total_seconds())
conn.execute("""
UPDATE alerts
SET resolved_at = ?, resolution_time_seconds = ?
WHERE id = ?
""", (resolved_at.isoformat(), resolution_time, alert_id))
print(f"✅ 알림 해결됨 (ID: {alert_id}, 대응시간: {resolution_time}초)")
def get_statistics(self, days=30):
"""최근 N일간 통계 조회"""
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.execute("""
SELECT
severity,
COUNT(*) as count,
AVG(resolution_time_seconds) as avg_resolution_time
FROM alerts
WHERE timestamp >= datetime('now', '-' || ? || ' days')
GROUP BY severity
ORDER BY count DESC
""", (days,))
results = [dict(row) for row in cursor.fetchall()]
return results
# 사용 예시
history = AlertHistoryManager()
# 알림 발생 기록
history.log_alert("CRITICAL", "CPU 사용률", 92.5, 85.0, "서버1 CPU 과부하")
history.log_alert("WARNING", "메모리 사용률", 78.3, 75.0, "서버1 메모리 경고")
# 알림 해결 (ID 1번 알림)
history.resolve_alert(1)
# 통계 조회
stats = history.get_statistics(days=30)
for stat in stats:
print(f"{stat['severity']}: {stat['count']}건, "
f"평균 해결시간: {stat['avg_resolution_time']:.0f}초")
설명
이것이 하는 일: 이 코드는 발생한 모든 알림을 SQLite 데이터베이스에 구조화된 형태로 저장하고, 알림이 해결되면 해결 시각과 대응 시간을 기록하며, 나중에 기간별, 심각도별 통계를 조회할 수 있게 합니다. 마치 은행의 거래 내역처럼, 모든 알림의 생명주기(발생→해결)를 추적 가능한 형태로 남깁니다.
첫 번째로, AlertHistoryManager 클래스의 _init_database() 메서드는 SQLite 데이터베이스 파일을 생성하고 alerts 테이블을 만듭니다. 테이블은 id(고유 식별자), timestamp(발생 시각), severity(심각도), metric_name(메트릭 이름), current_value(현재 값), threshold(임계값), message(상세 메시지), resolved_at(해결 시각), resolution_time_seconds(대응 시간)을 컬럼으로 가집니다.
또한 timestamp와 severity에 인덱스를 걸어서 날짜별, 심각도별 조회 성능을 최적화합니다. 그 다음으로, log_alert() 메서드는 알림이 발생할 때마다 호출되어 정보를 INSERT합니다.
datetime.now().isoformat()로 현재 시각을 ISO 8601 형식("2025-11-18T15:30:45")으로 저장하면 나중에 날짜 계산과 정렬이 쉽습니다. 예를 들어 "CPU 사용률이 92.5%로 임계값 85%를 초과했다"는 정보가 모두 구조화되어 저장되므로, 나중에 "CPU 문제가 언제 얼마나 자주 발생했는지" 정확히 알 수 있습니다.
세 번째로, resolve_alert() 메서드는 알림이 해결되었을 때 호출됩니다. 먼저 해당 알림의 원래 발생 시각(timestamp)을 조회하고, 현재 시각(resolved_at)과의 차이를 계산해서 대응 시간(resolution_time_seconds)을 구합니다.
예를 들어 알림이 15:30:00에 발생하고 15:45:00에 해결되면 900초(15분)가 기록됩니다. 이 데이터가 쌓이면 "우리 팀의 평균 인시던트 대응 시간"을 측정할 수 있습니다.
마지막으로, get_statistics() 메서드는 최근 N일간의 데이터를 심각도별로 집계합니다. SQL의 GROUP BY와 집계 함수(COUNT, AVG)를 사용해서 "CRITICAL 알림이 37건 발생했고 평균 대응 시간은 23분이었다" 같은 통계를 쉽게 뽑아냅니다.
이런 통계를 주간/월간 리포트로 만들어 팀 회의에서 공유하면, "어떤 문제가 반복되는지", "대응 속도가 개선되고 있는지" 등을 객관적으로 논의할 수 있습니다. 여러분이 이 시스템을 사용하면 "감"이 아닌 "데이터"로 시스템을 관리할 수 있습니다.
예를 들어 3개월치 데이터를 분석해서 "매주 금요일 17시경 트래픽 급증으로 CPU 알림이 발생한다"는 패턴을 발견하면, 금요일 오후에 자동으로 서버를 증설하는 정책을 만들 수 있습니다. 또한 "임계값을 80%에서 85%로 올린 후 알림 건수가 50% 감소했지만 실제 장애는 증가하지 않았다"는 분석으로 최적의 임계값을 찾을 수 있습니다.
심지어 "평균 대응 시간이 한 달 만에 30분에서 10분으로 줄었다"는 개선 성과를 정량적으로 증명할 수도 있습니다.
실전 팁
💡 SQLite는 단일 서버 환경에 적합합니다. 여러 서버에서 알림을 기록해야 한다면 PostgreSQL, MySQL, MongoDB 같은 중앙 집중식 데이터베이스를 사용하세요.
💡 데이터가 계속 쌓이면 성능이 저하될 수 있으므로, 6개월~1년 이상 된 데이터는 별도의 아카이브 테이블로 이동하거나 삭제하는 정책을 만드세요. 통계만 남기고 상세 로그는 지우는 방법도 있습니다.
💡 Grafana, Metabase 같은 시각화 도구를 연동하면 SQL 쿼리 없이도 대시보드에서 그래프와 차트로 알림 트렌드를 한눈에 볼 수 있습니다.
💡 알림 이력에 "해결 방법"이나 "근본 원인" 필드를 추가하면 더욱 유용합니다. 예를 들어 "디스크 정리로 해결" 같은 노트를 남기면, 나중에 같은 문제 발생 시 바로 참고할 수 있는 runbook이 됩니다.
💡 resolved_at이 NULL인 레코드는 "아직 해결되지 않은 알림"입니다. 이 데이터를 조회하면 현재 진행 중인 모든 문제를 한눈에 볼 수 있는 "인시던트 대시보드"를 만들 수 있습니다.