[모니터링(monitoring)] slack 으로 에러 알림 받기

 

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