Python built-in logging으로 Slack Alert 보내기

이주환·2024년 3월 17일

파이썬과 떠나는 모험

목록 보기

6/6

안녕하세요. 저는 현재 데이터 관리 업무를 하고 있는 주니어 개발자 입니다.
최근 진행한 프로젝트에서 에러 메세지를 파일 로그, 슬랙 로그를 통해 내용을 받고 있었습니다. 그러기 위해선 파이썬에서 제공 해주는 로깅과 슬랙으로 메세지를 보내줄 사용자 정의 함수가 필요했기 때문에 에러 로그를 남기는 곳에는 항상 로그 메서드, 슬랙 얼럿 함수 이렇게 2개의 코드가 반복 되는 구조였는데요.

이러한 불필요한 반복을 줄이고자 참고한 자료를 토대로 사용된 방법을 공유 해보고자 합니다.

파이썬 내장 로거 알아보기

Python logging

Python에서 자체적으로 지원 해주는 로깅 모듈이며 콘솔, 파일 등 다양하게 핸들러를 구성하여 원하는 결과 값을 로그로 기록할 수 있습니다.

usage:

	import logging
    
    # NOTE: getLogger는 핸들러에 명시된 이름을 지정하지 않을 시 RootLogger의 핸들러를 지정하게 됩니다.
    # Logger = logging.getLogger()
    Logger = logging.getLogger(__file__)
    
    Logger.info("로그 테스트")

이렇게 가장 기본적인 파이썬에서 로그를 남길 수 있게 모듈을 제공 해주고 있는데요. 로그에 대해 간단하게 살펴보자면 로그라는 객체는 루트 로거, 파생 로거 구조로 생성되며 루트 로거는 파생되는 로거가 남긴 기록까지 저장 할 수 있습니다.

python3.12 docs 에서 명시된 내용입니다

이런 특성 덕분에 프로젝트에선 모듈별로 핸들러 명칭을 갖고, 모듈별로 로그 파일을 생성 하여 해당 모듈에 필요한 정보만 로그에 남기며 최종적으로 루트 로거 핸들러는 다른 모듈에서 기록하고 있는 내용을 모두 저장합니다.

하지만, 슬랙에도 메세지를 기록하면 데일리로 로그 파일을 분석 할 필요가 없어지게 되고 슬랙 알람만 확인하면 되겠죠. 그렇게 로그와 슬랙이 계속 따라다니는 상황이 발생 하게 되었습니다.

예시 코드는 대략 이렇습니다.

import logging
import slack_alert

Logger = getLogger(__file__)

def test(a: bool):
    if not a:
    	raise ValueError("Error!")
        
if __name__ == "__main__":
	a = False
	try:
    	test(a)
    except Exception as e:
    	logger.error(e)
        slack_alert(e)

예외적인 상황이 굉장히 많이 발생하는 프로젝트에서 이 2개가 서로 꼬리표를 달고 다니며 로그를 남기게 되는건 굉장히 피곤하게 될겁니다. 중요한 부분에 슬랙을 빼먹을 수도 있을 염려도 생기게 됩니다.

슬랙 로거 구성하기

참고한 깃 허브를 보면 2017년에 마지막으로 릴리즈 된 슬랙 로거 글이 있는데요. 이 패키지를 다운 받아서 모듈을 임포트 할 수 있지만 아무래도 로그는 커스텀 해야 할 일이 많을 것 같기 때문에 소스코드를 직접 들고와서 수정이 필요했습니다.

로거를 구성 하는건 다 똑같으니 필터링을 해서 직접 슬랙에 남길 로거만 채택하는 방법을 먼저 살펴보겠습니다.

위 깃허브 이슈 내용을 보면 extra 필드에 notify_slack이 True인 경우만 슬랙에 보내는게 어때 라며 외국 개발자가 PR을 보냈고 이 내용이 반영이 되었네요. 이 부분을 살펴보면 슬랙에 보낼 때 원하는 값을 수정할 수 있을 것 같습니다.

LogRecord 객체 추적

	class SlackLogFilter(logging.Filter):
    """
    Logging filter to decide when logging to Slack is requested, using
    the `extra` kwargs:

        `logger.info("...", extra={'notify_slack': True})`
    """

    def filter(self, record):
        return getattr(record, 'notify_slack', False)

위 사진에서 보여드린 코드인데요. 여기서 def filter(self, record): 에 해당하는 부분의 record 아규먼트를 확인 해보면 getattr에서 어떻게 notify_slack이 쓰이는지 알 수 있습니다.

Python3 logging 내부에서 해당 엑스트라 필드를 어떻게 가져오는지 살펴보겠습니다.

_logRecordFactory는 결국 LogRecord를 생성 해주는 객체입니다. 아래 코드를 보면 알 수 있듯이 전역에 선언된 LogRecord에 할당하는 것 뿐이었습니다.

_logRecordFactory

#
#   Determine which class to use when instantiating log records.
#
_logRecordFactory = LogRecord

def setLogRecordFactory(factory):
    """
    Set the factory to be used when instantiating a log record.

    :param factory: A callable which will be called to instantiate
    a log record.
    """
    global _logRecordFactory
    _logRecordFactory = factory

그리고 바로 extra가 있는지 없는지 검사를 하고 있는 것을 볼 수 있죠. 엑스트라에 있는 값은 딕셔너리 객체로 넘겨지는 것이 아니라 LogRecord에 Attribute로 넘겨지는 것을 볼 수 있습니다. 바로 이 부분이 되겠네요. rv.__dict__[key] = extra[key] 이제 이 내용을 알고 있다면 이 코드도 이해가 될 것 입니다.

LogRecord getattr

    def filter(self, record: LogRecord):
      return getattr(record, 'notify_slack', False)

이렇게 LogRecord가 notify_slack: True인 경우만 슬랙을 보내게 되겠죠.

Slack 핸들러 설정하기

slack.yaml

version: 1
handlers:
  slack_handler:
    webhook_url: 'web hook url'
    user_name: 'user name'
    channel: '#chnnael name'
    level: 'ERROR'
    option: {
      slack_filter: True,
    }

slacker:
    '':
      handlers: [slack_handler]

slack_loader.py

  class SlackLoader:  
      @staticmethod  
      def load_yaml_file(file_name):  
          with open(file_name) as f:  
              handlers = dict()  
              slack_list = yaml.safe_load(f)  

              for handle_name, handle_info in slack_list.get('handlers', {}).items():  
                  _url = handle_info['webhook_url']  
                  _user_name = handle_info['user_name']  
                  _channel = handle_info['channel']  
                  _level = handle_info['level']  
                  _option = handle_info.get('option', {})  

                  tmp_handlers = SlackHandler(_level, _url, _user_name, _channel, **_option)  

                  if tmp_handlers.slack_filter:  
                      tmp_filter = SlackLogFilter()  
                      tmp_handlers.addFilter(tmp_filter)  

                  handlers[handle_name] = tmp_handlers  

              for slacker_name, slacker_info in slack_list.get('slacker', {}).items():  
                  tmp_logger = logging.getLogger(slacker_name)  
                  for slack_handler in slacker_info['handlers']:  
                      if slack_handler in handlers.keys():  
                          tmp_logger.addHandler(handlers[slack_handler])

위 yaml파일이 slack 환경변수가 되며, 그 환경변수를 호출 하는 로더가 따로 있습니다. 그렇게 yaml파일의 경로를 넘겨 핸들러 설정을 추가 해주게 되고, 설정한 yaml파일에서 slack_filter: True 부분이 슬랙 핸들러를 활성화 할 것인지 여부를 결정 하는 곳 이기 때문에 이 곳에서 슬랙 핸들러 활성화를 시켜줍니다.

Slack env 활성화 하기

import logging
import traceback  
from pathlib import Path  
  
from utils.slack_logging import SlackLoader  
from utils.logger import logger_setting  
  
  
def init_env(args):  
    project_root_directory = Path(__file__).resolve().parent.parent  
    access_env_path = Path(project_root_directory, "envs", args.target)  
  
    try:  
        # NOTE: Logger 설정  
        logger_env_file = Path(access_env_path, "logger", 'logger.yml')  
        logger_name = f"{args.van}_{args.target}"  
        logger_setting(args, logger_name)  
  
        # NOTE: prefix brand name 
        # NOTE: used: slack_logging.py lineno: 72  
        setattr(logging.LogRecord, "brand_name", args.brand)  
  
        logger = logging.getLogger()  
        logger.info(f'*** logger_env_file: {logger_env_file}')  
  
        # NOTE: Slack setup
        slack_env_file = Path(access_env_path, "slack", 'slack.yml')  
        SlackLoader.load_yaml_file(slack_env_file)  
        logger.info(f'*** slack_env_file: {slack_env_file}')  
    except Exception as e:  
        print(f'****** env setting error: {e}'  
              f'{traceback.format_exc()}')  
        exit()

이렇게 루트 로거와 슬랙 전용 핸들러를 활성화 해줍니다. 그렇게 되면 슬랙 핸들러는 루트 로거의 핸들러에 포함 되게 됩니다. 아까 위에서 보았듯이 루트로거와 파생로거의 개념에 의해 그렇게 되는 듯 합니다.

로거를 통해 슬랙으로 보내기 전에 잠시 LogRecord에 직접적으로 속성을 추가 해야 할 필요를 느껴서 추가 해 보았습니다.
아래 속성 목록을 보게 되면 brand_name이 지정되어 있고, 앞으로 모든 루트 로거 포함한 로거에 대해 메세지를 생성할 땐 brand name 속성 값을 갖고 있게 됩니다.

LogRecord setattr

setattr(logging.LogRecord, "brand_name", snu)

{'__module__': 'logging',
'__doc__': '\n    A LogRecord instance represents an event being logged.\n\n    LogRecord instances are created every time something is logged. They\n    contain all the information pertinent to the event being logged. The\n    main information passed in is in msg and args, which are combined\n    using str(msg) % args to create the message field of the record. The\n    record also includes information such as when the record was created,\n    the source line where the logging call was made, and any exception\n    information to be logged.\n    ',
'__init__': <function LogRecord.__init__ at 0x103629430>,
'__repr__': <function LogRecord.__repr__ at 0x1036294c0>,
'getMessage': <function LogRecord.getMessage at 0x103629550>,
'__dict__': <attribute '__dict__' of 'LogRecord' objects>,
'__weakref__': <attribute '__weakref__' of 'LogRecord' objects>,
'brand_name': 'snu'}

슬랙 로그 테스트

로그를 보내는건 이제 간단한데요.

logger.info("TEST")
logger.error("basic test", extra={"notify_slack": True})
logger.error("test use extra field", extra={"notify_slack": True})

이렇게 두 가지 방법으로 로그를 남기게 되면, 첫번째 로그에 대해서는 루트 로거에 지정되어 있는 핸들러에 의해 남게 되게 됩니다.
두번째는 당연 슬랙으로 남겨지게 되죠.

위에 말씀 드린바와 같이 prefix값을 추가 하기 위해 LogRecord에 setattr을 하는 과정에서 SNU 값을 추가 해주면 extra field에 따로 명시할 필요 없이 기록을 남길 수 있게 됩니다. 만약, 모든 로그에 슬랙을 다 남기고 싶다면 이 방법으로 적용해도 되겠지만 그럼 슬랙이 매우 지저분해지는 불상사가 생기게 되기 때문에 그건 비추한다는 의견입니다.

부록

Slack formatting을 활용한 attachments 커스텀 하기

번외로 로그를 남길 때 터미널에서 보는 화면 처럼 흰 텍스트만 남는다면 어떤 내용인지 알아보기 위해 긴 텍스트를 읽어야 하는 불편 사항이 있을겁니다.

그래서 취향대로 슬랙 메세지를 꾸미는 방법을 이용 했는데요. 일단 공식문서에 따르면 어떻게 작성해야 하는지 방법이 자세하게 나와있습니다. 그 내용을 기반으로 작성을 해보겠습니다.

이미 존재하는 SlackFormatter 클래스를 손을 좀 볼건데요.

그 전에 슬랙 로거를 불러오는 객체에 포맷터를 활성화 시켜주어야합니다.

SlackLoader

class SlackLoader:
    @staticmethod
    def load_yaml_file(file_name):
        with open(file_name) as f:
            handlers = dict()
            slack_list = yaml.safe_load(f)
            slack_formatter = SlackFormatter()  # 추가

            for handle_name, handle_info in slack_list.get('handlers', {}).items():
                _url = handle_info['webhook_url']
                _user_name = handle_info['user_name']
                _channel = handle_info['channel']
                _level = handle_info['level']
                _option = handle_info.get('option', {})

                tmp_handlers = SlackHandler(_level, _url, _user_name, _channel, **_option)
                tmp_handlers.setFormatter(slack_formatter)  # 추가

소스코드 옆에 "# 추가"라고 붙은 문단만 수정을 하였는데요. 실제로 사용할 땐 이와 같은 클래스 구조가 아니어도 상관 없으니 편한대로 로그를 생성하는 코드를 작성한 후, 포맷터를 활성화 해주시면 됩니다.

이 내용도 python-slack-logger를 만들어주신 제작자의 깃허브에 작성 되어 있는 내용입니다.

SlackFormatter

class SlackFormatter(logging.Formatter):
    def format(self, record):
        ret = {}
        if record.levelname == 'INFO':
            ret['color'] = 'good'
        elif record.levelname == 'WARNING':
            ret['color'] = 'warning'
        elif record.levelname == 'ERROR':
            ret['color'] = '#E91E63'
        elif record.levelname == 'CRITICAL':
            ret['color'] = 'danger'

        ret["pretext"] = "pretext"  # 추가된 정보
        ret['author_name'] = record.levelname
        ret['title'] = record.name
        ret['ts'] = record.created
        ret['text'] = super(SlackFormatter, self).format(record)
        ret["footer"] = "footer"  # 추가된 정보
        ret["footer_icon"] = "https://platform.slack-edge.com/img/default_application_icon.png"  # 추가된 정보
        return ret

기존 소스 코드에서 제공 하고 있는 내용을 사용하고, 공식문서와 비교 하면서 필요한 컨텍스트를 추가 합니다.

저 같은 경우에 기존 코드에서 "pretext", "footer", "footer_icon" 이렇게 세개의 컨텍스트를 추가 하였습니다.

블로그 글에 의존 하여 수정하는 것 보다 공식문서를 보며 수정 하는 것이 더 올바른 범위에서 컨트롤이 가능하니 꼭 공식문서를 보며 추가 하시고, 제 코드는 참고만 해주시면 더 많은 사고를 확장 시키실 수 있습니다.

Slack Handler의 set attatchments 메서드 생성 및 활용

SlackHandler

class SlackHandler(HTTPHandler):
	def __init__(self, level, webhook_url, user_name, channel, **kwargs):
          o = urlparse(webhook_url)
          is_secure = o.scheme == 'https'
          HTTPHandler.__init__(self, o.netloc, o.path, method="POST", secure=is_secure)

          self.setLevel(level)
          self._user_name = user_name
          self._channel = channel

          self._icon_url = kwargs.get('icon_url')
          self._icon_emoji = kwargs.get('icon_emoji')
          self._level_emoji = kwargs.get('level_emoji')
          self._slack_filter = kwargs.get('slack_filter')
          self._mention = kwargs.get('mention') and kwargs.get('mention').lstrip('@')

          # NOTE: 커스텀 변수
          self._brand_name = ""
          self._emoji = ""

      @property
      def slack_filter(self):
          return self._slack_filter

      def get_emoji_text_by_level(self, level_name):
          return '' if not self._level_emoji \
              else ':bug:' if level_name == 'DEBUG' \
              else ':pencil2:' if level_name == 'INFO' \
              else ':warning:' if level_name == 'WARNING' \
              else ':no_entry:' if level_name == 'ERROR' \
              else ':rotating_light:' if level_name == 'CRITICAL' \
              else ''

      def mapLogRecord(self, record: logging.LogRecord) -> Dict[str, Any]:
          text_format: Union[str, Union] = self.format(record)
          self._emoji: str = self.get_emoji_text_by_level(record.levelname)
          self._brand_name: str = getattr(record, "brand_name")

          if dict == type(text_format):
              text = text_format.get("text")
          else:
              text = text_format

          if isinstance(self.formatter, SlackFormatter):
              text_format = self.set_attachments(text_format, text)
              payload = {
                  'attachments': [
                      text_format,
                  ],
              }
              if self._mention:
                  payload['text'] = '<@{0}>'.format(self._mention)
          else:
              if self._mention:
                  text = '<@{0}> {1}'.format(self._mention, text)
              payload = {
                  'text': text,
              }

          if self._user_name:
              payload['username'] = self._user_name

          if self._icon_url:
              payload['icon_url'] = self._icon_url

          if self._icon_emoji:
              payload['icon_emoji'] = self._icon_emoji

          if self._channel:
              payload['channel'] = self._channel

          ret = {
              'payload': json.dumps(payload),
          }
          return ret

      def set_attachments(self, record: Dict[str, Any], message: str) -> Dict[str, Any]:
          record["text"] = message
          record["author_name"] = f"{self._emoji} {record.get('author_name')}"
          record["title"] = self._brand_name.upper() if self._brand_name else "TMSS"

          return record

set_attachments 를 만들어서 SlackFormatter에서 수정했던 정보를 여기서 덮어씌우는 작업을 하게 됩니다.

그 값은 실제 사용자에게 받은 로그 객체(LogRecord)에서 사용할 어트리뷰트나 2차 가공되는 데이터를 활용하여 로그에 남길 메세지, 에러 카테고리 등 추가하는 메서드를 생성 합니다.

그 다음, mapLogRecord는 이전에 보셨듯이 실제 Json데이터를 말아서 던지면 SlackHttpRequest를 통해 POST 요청을 보내게 되는데요. 그 때 사용 할 페이로드 입니다.

테스트 로그 날리기

if __name__ == "__main__":
    logger = logging.getLogger(__name__)

    logger.error("font test", extra={'notify_slack': True})

테스트 목적으로 작성 했지만 불필요한 내용이 포함되어 그 부분만 제외 하였는데요.
글을 읽으신 후 직접 활용 해보시려고 하신다면 "저게 대체 어느 필드를 가린거지?" 라는 의문보다 한 번의 코드 실행이 더 이해 하시는데 도움이 되실 것 같다고 생각합니다.

끝맺음

불필요한 코드를 한 줄 한 줄 줄여나갈 때 마다 희열이 엄청난데 그로 인한 디버깅에 대한 부작용도 최근에 많이 겪는 것 같습니다. 재밌는 방법으로 재밌는 프로젝트를 많이 만들어내는 즐거운 개발 되세요!

이주환

안녕하새우

이전 포스트

Python built-in logging으로 Slack Alert 보내기

파이썬과 떠나는 모험

파이썬 내장 로거 알아보기

슬랙 로거 구성하기

슬랙 로그 테스트

부록

끝맺음

문자열로 클래스 객체 동적 임포트 하여 사용하기

0개의 댓글