카테고리 없음
[모니터링(monitoring)] slack 으로 에러 알림 받기
늘이
2024. 11. 14. 15:31
1. 워크스페이스 생성하기
1단계: 회사 또는 팀이름 지정
2단계: 이름 설정
3단계: 팀원 초대
4단계: 채널 목적 설정
채널 생성 확인
만들어진 채널에 오른쪽 마우스 클릭 -> 채널 세부정보 보기
앱 추가 클릭
webhook 검색
Slack에 추가 클릭
적용할 채널 선택 -> 수신 웹후크 통합 앱 추가 버튼 클릭
webhook url 및 사용 방법이 나와있고 아래를 보면 정보가 나와있다.
예시대로 cURL 요청 테스트
아래와 같이 슬랙 알람이 온 것을 확인할 수 있음
코드에 적용하기
import os
import logging
from logging.handlers import TimedRotatingFileHandler
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import Request
import requests
from fastapi.exception_handlers import http_exception_handler
from fastapi.exceptions import HTTPException
# 로그 디렉토리 생성 함수
def create_log_directory(path):
if not os.path.exists(path):
os.makedirs(path)
# 로그 디렉토리 생성
create_log_directory("logs/error")
create_log_directory("logs/request")
# 로깅 포맷 설정
log_formatter = logging.Formatter("%(asctime)s %(levelname)s [%(name)s] [%(filename)s:%(lineno)d] " \
"[trace_id=%(otelTraceID)s span_id=%(otelSpanID)s resource.service.name=%(otelServiceName)s] - %(message)s")
# 에러 로그 핸들러 설정
error_log_handler = TimedRotatingFileHandler("logs/error/error.log", when="midnight", interval=1, backupCount=90, encoding='utf-8')
error_log_handler.setFormatter(log_formatter)
error_log_handler.suffix = "%Y-%m-%d"
error_log_handler.setLevel(logging.ERROR)
# 요청/응답 로그 핸들러 설정
request_log_handler = TimedRotatingFileHandler("logs/request/request.log", when="midnight", interval=1, backupCount=90, encoding='utf-8')
request_log_handler.setFormatter(log_formatter)
request_log_handler.suffix = "%Y-%m-%d"
request_log_handler.setLevel(logging.INFO)
# 스트림 핸들러 설정
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(log_formatter)
# 로깅 기본 설정
logging.basicConfig(level=logging.INFO, handlers=[error_log_handler, request_log_handler, stream_handler])
logger = logging.getLogger(__name__)
# 슬랙 웹훅 URL 설정
SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/URL"
# Teams Webhook URL 설정
TEAMS_WEBHOOK_URL = "URL"
def send_slack_message(message: str):
payload = {
"channel": "#모니터링",
"username": "dews-monitoring",
"text": message,
"icon_emoji": ":ghost:"
}
requests.post(SLACK_WEBHOOK_URL, json=payload)
def send_teams_message(message: str):
payload = {
"@type": "MessageCard",
"@context": "http://schema.org/extensions",
"themeColor": "0072C6",
"summary": "Error Notification",
"sections": [{
"activityTitle": "📢 Server Error Notification",
"text": message
}]
}
headers = {"Content-Type": "application/json"}
requests.post(TEAMS_WEBHOOK_URL, json=payload, headers=headers)
async def global_exception_handler(request: Request, exc: Exception):
error_message = f"🖥 URL: [{request.method}] {request.url}\n\n🚨 Error: {exc}"
logger.error(error_message)
# send_slack_message(error_message)
send_teams_message(error_message)
return {"detail": "Internal server error"}
# 로깅 미들웨어 설정
class ExceptionLoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
try:
response = await call_next(request)
return response
except Exception as exc:
logger.error(f"Exception: {exc}", exc_info=True)
# 기존 예외 처리기를 호출
if isinstance(exc, HTTPException):
return await http_exception_handler(request, exc)
else:
# 글로벌 예외 처리기 호출
return await global_exception_handler(request, exc)
class RequestResponseLoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
client_ip = request.client.host
pid = os.getpid()
logger.info(f"Request from {client_ip}: {request.method} {request.url} - PID:{pid}")
response = await call_next(request)
logger.info(f"Response to {client_ip}: {response.status_code} - PID:{pid}")
return response