Python Logging 설정 파일 매뉴얼

1. logging.config.fileConfig 개요

   - logging.config.fileConfig는 설정 파일을 사용하여 로깅 시스템을 구성하는 방법을 제공

import logging.config

logging.config.fileConfig('logging.conf', disable_existing_loggers=False)

 

2. logging.conf 파일 구조

   1) 기본 섹션

[loggers]
keys=root,main

[handlers]
keys=consoleHandler,fileHandler

[formatters]
keys=simpleFormatter,detailedFormatter

 

   2) 로거 섹션 설정

      1. 옵션 설명:

         - level: 로그 레벨 지정 (DEBUG, INFO, WARNING, ERROR, CRITICAL)
         - handlers: 사용할 핸들러 목록 (콤마로 구분)
         - qualname: 로거의 이름
         - propagate: 상위 로거로 로그 전파 여부 (0=비활성화, 1=활성화)

[logger_root]
level=DEBUG
handlers=consoleHandler
qualname=root

[logger_main]
level=INFO
handlers=fileHandler
qualname=main
propagate=1

 

   3) 핸들러 섹션 설정

      1. 옵션 설명:

         - class: 핸들러 클래스 지정
         - level: 핸들러 레벨
         - formatter: 사용할 포맷터
         - args: 핸들러 클래스 생성자 인자

[handler_consoleHandler]
class=StreamHandler
level=DEBUG
formatter=simpleFormatter
args=(sys.stdout,)

[handler_fileHandler]
class=handlers.RotatingFileHandler
level=INFO
formatter=detailedFormatter
args=('app.log', 'a', 1048576, 5)

 

     2. 주요 핸들러 클래스와 args 옵션:

         - StreamHandler

args=(sys.stdout,)

         - FileHandler

args=('app.log', 'a')  # 파일명, 모드

 

         - RotatingFileHandler

args=('app.log', 'a', 1048576, 5)  # 파일명, 모드, 최대크기(bytes), 백업수

 

         - TimedRotatingFileHandler

args=('app.log', 'midnight', 1, 30)  # 파일명, 시간단위, 간격, 백업유지수

 

   4) 포맷터 섹션 설정

      1. 포맷터 옵션:

         - format: 로그 메시지 형식
         - datefmt: 날짜/시간 형식

 

      2. 주요 포맷 문자열:

         - %(asctime)s: 시간
         - %(name)s: 로거 이름
         - %(levelname)s: 로그 레벨
         - %(message)s: 로그 메시지
         - %(pathname)s: 전체 경로
         - %(filename)s: 파일명
         - %(module)s: 모듈명
         - %(funcName)s: 함수명
         - %(lineno)d: 라인번호
         - %(process)d: 프로세스 ID
         - %(thread)d: 스레드 ID
         - %(threadName)s: 스레드 이름

[formatter_simpleFormatter]
format=%(levelname)s - %(message)s
datefmt=

[formatter_detailedFormatter]
format=%(asctime)s - %(name)s - %(levelname)s - %(message)s
datefmt=%Y-%m-%d %H:%M:%S

 

3. 전체 설정 파일 예시

[loggers]
keys=root,mainApp

[handlers]
keys=consoleHandler,fileHandler,rotatingFileHandler

[formatters]
keys=simpleFormatter,detailedFormatter

[logger_root]
level=DEBUG
handlers=consoleHandler
qualname=root

[logger_mainApp]
level=INFO
handlers=fileHandler,rotatingFileHandler
qualname=mainApp
propagate=0

[handler_consoleHandler]
class=StreamHandler
level=DEBUG
formatter=simpleFormatter
args=(sys.stdout,)

[handler_fileHandler]
class=FileHandler
level=INFO
formatter=detailedFormatter
args=('app.log', 'a')

[handler_rotatingFileHandler]
class=handlers.RotatingFileHandler
level=ERROR
formatter=detailedFormatter
args=('errors.log', 'a', 1048576, 5)

[formatter_simpleFormatter]
format=%(levelname)s - %(message)s
datefmt=

[formatter_detailedFormatter]
format=%(asctime)s - %(name)s - %(levelname)s - %(message)s
datefmt=%Y-%m-%d %H:%M:%S

 

4. 설정 파일 적용

import logging.config

# 기본 설정 파일 로드
logging.config.fileConfig('logging.conf', disable_existing_loggers=False)

# 로거 가져오기
logger = logging.getLogger('mainApp')

# 로그 출력
logger.debug('디버그 메시지')
logger.info('정보 메시지')
logger.error('에러 메시지')

 

5. 고급 설정 옵션

   1) defaults 섹션 사용

[defaults]
log_dir=logs
main_log=%(log_dir)s/main.log
error_log=%(log_dir)s/error.log

[handler_fileHandler]
args=('%(main_log)s', 'a')

 

2) 커스텀 필터 설정

[filters]
keys=customFilter

[filter_customFilter]
class=project.logging.CustomFilter
name=

 

3) 커스텀 핸들러 설정

[handler_customHandler]
class=project.logging.CustomHandler
formatter=detailedFormatter
args=()

 

6. 실전 설정 패턴

   1) 다중 로거와 핸들러 연결 패턴

      - 아래는 root 로거와 애플리케이션 전용 로거를 분리하여 각각 다른 파일에 로그를 기록하는 패턴이다.

 

      1. 이 설정의 주요 특징:

         - 루트 로거는 콘솔과 dev.log 파일에 모든 로그를 기록
         - 애플리케이션 로거는 app.log 파일에 별도로 로그를 기록
         - 모든 핸들러가 동일한 포맷터를 사용하여 일관된 로그 형식 유지
         - 각 로거와 핸들러의 이름을 명확하게 구분하여 관리 용이성 향상

[loggers]
keys=root, app

[handlers]
keys=consoleHandler, fileHandler, fileHandler_app

[formatters]
keys=simpleFormatter

[logger_root]
level=DEBUG
handlers=consoleHandler, fileHandler

[logger_app]
level=DEBUG
handlers=fileHandler_app
qualname=app

[handler_consoleHandler]
class=StreamHandler
level=DEBUG
formatter=simpleFormatter
args=(sys.stdout,)

[handler_fileHandler]
class=FileHandler
level=DEBUG
formatter=simpleFormatter
args=('dev.log',)

[handler_fileHandler_app]
class=FileHandler
level=DEBUG
formatter=simpleFormatter
args=('app.log',)

[formatter_simpleFormatter]
format=%(asctime)s - %(levelname)s - %(message)s

 

   2) 핸들러 명명 규칙

      - 위 예시에서 볼 수 있듯이, 핸들러 명명에는 다음과 같은 패턴 사용:
         - consoleHandler: 콘솔 출력용
         - fileHandler: 일반 파일 출력용
         - fileHandler_[용도]: 특정 용도의 파일 출력용 (예: fileHandler_app)

 

   3) 간단한 포맷터 설정

      - 로그 포맷을 간단하게 유지하면서도 필수 정보를 포함하는 방법:
         - 타임스탬프 (%(asctime)s)
         - 로그 레벨 (%(levelname)s)
         - 로그 메시지 (%(message)s)

[formatter_simpleFormatter]
format=%(asctime)s - %(levelname)s - %(message)s

 

   4) 다중 핸들러 설정 시 주의사항

      로거에 여러 핸들러를 설정할 때는 다음 사항을 고려해야 한다.

 

      1. 핸들러 나열 시 쉼표 뒤의 공백도 포함됨:

handlers=consoleHandler, fileHandler

 

      2. 각 핸들러의 레벨을 개별적으로 설정 가능:

[handler_consoleHandler]
level=DEBUG

[handler_fileHandler]
level=INFO

   5) 로거 상속 관리

      - app 로거와 같은 하위 로거를 설정할 때 propagate 속성을 명시적으로 지정하면 로그 중복을 제어할 수 있다.

[logger_app]
propagate=0  # 상위 로거로 전파하지 않음

 

7. 고급 로깅 구성 전략

   1) 환경별 로깅 설정

      - 개발, 테스트, 운영 환경에 따라 다른 로깅 설정을 사용하는 것이 일반적입니다. 다음은 환경별 설정 예시이다.

# development.conf
[logger_root]
level=DEBUG
handlers=consoleHandler, devFileHandler

# production.conf
[logger_root]
level=INFO
handlers=productionFileHandler, errorEmailHandler

 

      - 이를 코드에서 다음과 같이 활용할 수 있다:

import os
import logging.config

def setup_logging():
    """환경에 따른 로깅 설정을 로드합니다."""
    env = os.getenv('ENVIRONMENT', 'development')
    config_file = f'logging_{env}.conf'
    logging.config.fileConfig(config_file, disable_existing_loggers=False)

 

   2) 로거 이름 계층 구조 활용

      - 애플리케이션의 구조를 반영하는 로거 계층을 설정하는 방법이다.

[loggers]
keys=root,app,app.api,app.core,app.database

[logger_app]
level=INFO
handlers=appFileHandler
qualname=app

[logger_app.api]
level=DEBUG
handlers=apiFileHandler
qualname=app.api

[logger_app.core]
level=INFO
handlers=coreFileHandler
qualname=app.core

[logger_app.database]
level=WARNING
handlers=dbFileHandler,errorEmailHandler
qualname=app.database

 

      - 이러한 계층 구조는 다음과 같이 사용된다.

# api.py
logger = logging.getLogger('app.api')

# database.py
logger = logging.getLogger('app.database')

 

   3) 컨텍스트 필터 활용

      - 특정 컨텍스트 정보를 로그에 추가하는 필터 설정:

[filters]
keys=contextFilter

[filter_contextFilter]
class=myapp.logging.ContextFilter
name=

[handler_fileHandler]
class=FileHandler
level=DEBUG
formatter=detailedFormatter
filters=contextFilter
args=('app.log',)

 

      - 필터 구현 예시:

class ContextFilter(logging.Filter):
    def filter(self, record):
        # 현재 사용자 정보 추가
        record.user = getattr(thread_local, 'user', 'unknown')
        # 요청 ID 추가
        record.request_id = getattr(thread_local, 'request_id', '')
        return True

 

   4) 조건부 로깅 설정

      - 특정 조건에 따라 로그 레벨이나 핸들러를 조정하는 방법:

[handlers]
keys=conditionalFileHandler

[handler_conditionalFileHandler]
class=myapp.logging.ConditionalFileHandler
level=DEBUG
formatter=detailedFormatter
args=('app.log',)

 

      - 핸들러 구현 예시:

class ConditionalFileHandler(logging.FileHandler):
    def emit(self, record):
        if hasattr(record, 'important') and record.important:
            # 중요 로그는 다른 파일에도 기록
            with open('important.log', 'a') as f:
                f.write(self.format(record) + '\n')
        super().emit(record)

 

   5) 백업 정책 설정

      - 로그 파일 관리를 위한 상세 설정:

[handler_rotatingFileHandler]
class=handlers.RotatingFileHandler
level=DEBUG
formatter=detailedFormatter
# maxBytes=10MB, backupCount=5
args=('app.log', 'a', 10485760, 5)

[handler_timedRotatingFileHandler]
class=handlers.TimedRotatingFileHandler
level=DEBUG
formatter=detailedFormatter
# 매일 자정에 로그 롤오버, 30일간 보관
args=('app.log', 'midnight', 1, 30, 'utf-8')

 

   6) 로그 포맷 패턴

      - 다양한 용도별 로그 포맷 설정:

[formatters]
keys=simple,detailed,json

[formatter_simple]
format=%(asctime)s - %(levelname)s - %(message)s

[formatter_detailed]
format=%(asctime)s - %(name)s - %(levelname)s - %(pathname)s:%(lineno)d - %(message)s
datefmt=%Y-%m-%d %H:%M:%S

[formatter_json]
class=myapp.logging.JsonFormatter
format=%(asctime)s %(name)s %(levelname)s %(message)s

 

      - JSON 포맷터 구현 예시:

import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'line': record.lineno
        }
        
        if hasattr(record, 'request_id'):
            log_data['request_id'] = record.request_id
            
        if record.exc_info:
            log_data['exception'] = self.formatException(record.exc_info)
            
        return json.dumps(log_data)

 

9. 성능 최적화와 보안

   1) 로깅 성능 최적화

      - 로깅 시스템은 애플리케이션의 성능에 중요한 영향을 미칠 수 있다. 다음은 로깅 성능을 최적화하는 주요 방법들이다.

 

      1. 비동기 로깅 설정

         - 대량의 로그 처리 시 비동기 로깅을 사용하면 애플리케이션의 주 실행 흐름이 차단되지 않다. 다음은 QueueHandler를 사용한 비동기 로깅 설정이다.

[handlers]
keys=asyncFileHandler

[handler_asyncFileHandler]
class=handlers.QueueHandler
level=DEBUG
formatter=detailedFormatter
args=(queue,)

 

         - 이를 활용하는 Python 코드:

import queue
import logging.handlers
import threading

def setup_async_logging():
    """비동기 로깅 설정을 구성합니다."""
    # 로그 메시지를 담을 큐 생성
    log_queue = queue.Queue(-1)  # 무제한 큐 크기
    
    # 큐 핸들러 설정
    queue_handler = logging.handlers.QueueHandler(log_queue)
    
    # 리스너 설정
    file_handler = logging.FileHandler('app.log')
    file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
    
    listener = logging.handlers.QueueListener(
        log_queue, file_handler, respect_handler_level=True
    )
    
    # 리스너 시작
    listener.start()
    
    return queue_handler, listener

 

      2. 로그 버퍼링 설정

         - 파일 I/O 최적화를 위한 버퍼링 설정:

[handler_bufferedFileHandler]
class=handlers.MemoryHandler
level=DEBUG
formatter=detailedFormatter
# capacity=1000개 메시지, flushLevel=ERROR
args=(1000, 'ERROR', fileHandler)

 

   2) 보안 강화

      1. 민감 정보 필터링

         - 로그에서 민감한 정보를 자동으로 필터링하는 설정:

[filters]
keys=sensitiveFilter

[filter_sensitiveFilter]
class=myapp.logging.SensitiveDataFilter
name=

 

         - 필터 구현 예시:

import re

class SensitiveDataFilter(logging.Filter):
    """민감한 정보를 필터링하는 로그 필터"""
    
    def __init__(self):
        super().__init__()
        # 필터링할 패턴 정의
        self.patterns = {
            'credit_card': r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
            'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
            'password': r'password["\s:]+\S+',
            'api_key': r'api[-_]key["\s:]+\S+'
        }
        self.compiled_patterns = {
            k: re.compile(v) for k, v in self.patterns.items()
        }
    
    def filter(self, record):
        # 메시지에서 민감 정보 마스킹
        message = record.getMessage()
        for pattern_name, pattern in self.compiled_patterns.items():
            message = pattern.sub(f'[FILTERED_{pattern_name.upper()}]', message)
        
        # 마스킹된 메시지로 교체
        record.msg = message
        return True

 

      2. 로그 파일 보안

         - 로그 파일의 보안을 강화하는 커스텀 핸들러:

import os
import stat
from logging.handlers import RotatingFileHandler

class SecureRotatingFileHandler(RotatingFileHandler):
    """보안이 강화된 로그 파일 핸들러"""
    
    def _open(self):
        # 파일 생성 시 적절한 권한 설정
        if not os.path.exists(self.baseFilename):
            open(self.baseFilename, 'a').close()
            # 파일 소유자만 읽기/쓰기 가능하도록 설정
            os.chmod(self.baseFilename, stat.S_IRUSR | stat.S_IWUSR)
        
        return super()._open()
    
    def doRollover(self):
        # 로그 롤오버 시에도 보안 설정 유지
        super().doRollover()
        if os.path.exists(self.baseFilename):
            os.chmod(self.baseFilename, stat.S_IRUSR | stat.S_IWUSR)

 

         - 이 핸들러를 설정 파일에서 사용:

[handler_secureFileHandler]
class=myapp.logging.SecureRotatingFileHandler
level=DEBUG
formatter=detailedFormatter
args=('secure.log', 'a', 1048576, 5)

 

   3) 로그 분석과 모니터링

      1. 구조화된 로깅

         - 로그 분석을 용이하게 하는 JSON 형식의 로깅 설정:

[formatters]
keys=jsonFormatter

[formatter_jsonFormatter]
class=myapp.logging.StructuredJsonFormatter
format=%(message)s

 

         - 포맷터 구현:

import json
import datetime
import traceback

class StructuredJsonFormatter(logging.Formatter):
    """구조화된 JSON 형식의 로그 포맷터"""
    
    def format(self, record):
        # 기본 로그 데이터 구성
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno,
            'thread_id': record.thread,
            'thread_name': record.threadName,
            'process_id': record.process
        }
        
        # 예외 정보 추가
        if record.exc_info:
            log_data['exception'] = {
                'type': record.exc_info[0].__name__,
                'message': str(record.exc_info[1]),
                'stacktrace': traceback.format_exception(*record.exc_info)
            }
        
        # 추가 컨텍스트 정보
        if hasattr(record, 'extra_data'):
            log_data['extra'] = record.extra_data
        
        return json.dumps(log_data)

 

      2. 로그 집계를 위한 메타데이터 추가

         - 로그 집계 시스템과의 통합을 위한 메타데이터 설정:

[formatters]
keys=aggregationFormatter

[formatter_aggregationFormatter]
class=myapp.logging.AggregationFormatter
format=%(message)s

 

         - 포맷터 구현:

class AggregationFormatter(logging.Formatter):
    """로그 집계를 위한 메타데이터가 포함된 포맷터"""
    
    def format(self, record):
        # 기본 메시지 포맷
        message = super().format(record)
        
        # 환경 정보 추가
        env_data = {
            'environment': os.getenv('ENVIRONMENT', 'development'),
            'application': os.getenv('APPLICATION_NAME', 'unknown'),
            'version': os.getenv('APPLICATION_VERSION', '0.0.0'),
            'host': socket.gethostname(),
            'service': record.name.split('.')[0]
        }
        
        # 성능 메트릭 추가
        if hasattr(record, 'duration'):
            env_data['duration_ms'] = record.duration
        
        # JSON 형식으로 변환
        return json.dumps({
            'message': message,
            'metadata': env_data,
            'timestamp': self.formatTime(record)
        })

 

10. 엔터프라이즈 로깅 전략

   1) 고가용성 로깅 구성

      1. 장애 대응 로깅 핸들러

         - 로그 저장소에 문제가 발생했을 때도 로그를 안전하게 보존하는 폴백(Fallback) 핸들러를 구현할 수 있다.

class FallbackHandler(logging.Handler):
    """주 핸들러 실패 시 대체 핸들러로 전환하는 핸들러"""
    
    def __init__(self, primary_handler, fallback_handler):
        super().__init__()
        self.primary_handler = primary_handler
        self.fallback_handler = fallback_handler
        self.use_fallback = False
        
    def emit(self, record):
        try:
            if not self.use_fallback:
                try:
                    self.primary_handler.emit(record)
                    return
                except Exception as e:
                    self.use_fallback = True
                    logging.warning(f"주 핸들러 실패, 대체 핸들러로 전환: {e}")
            
            self.fallback_handler.emit(record)
        except Exception as e:
            self.handleError(record)

 

         - 설정 파일에서 이 핸들러를 사용하는 방법:

[handlers]
keys=primaryHandler,fallbackHandler,fallbackCompositeHandler

[handler_fallbackCompositeHandler]
class=myapp.logging.FallbackHandler
args=(primaryHandler, fallbackHandler)

 

      2. 로그 버퍼링과 재시도 메커니즘

         네트워크나 디스크 문제로 인한 일시적인 장애를 처리하는 재시도 메커니즘을 구현할 수 있다.

class RetryHandler(logging.Handler):
    """실패한 로그 작업을 재시도하는 핸들러"""
    
    def __init__(self, base_handler, max_retries=3, retry_interval=1):
        super().__init__()
        self.base_handler = base_handler
        self.max_retries = max_retries
        self.retry_interval = retry_interval
        self.buffer = []
        self.buffer_lock = threading.Lock()
        
    def emit(self, record):
        for attempt in range(self.max_retries):
            try:
                self.base_handler.emit(record)
                # 성공적으로 처리된 경우 버퍼된 로그도 처리 시도
                self._process_buffer()
                return
            except Exception as e:
                if attempt == self.max_retries - 1:
                    # 마지막 시도에서 실패하면 버퍼에 저장
                    with self.buffer_lock:
                        self.buffer.append(record)
                time.sleep(self.retry_interval)
    
    def _process_buffer(self):
        """버퍼에 있는 로그 레코드 처리 시도"""
        with self.buffer_lock:
            while self.buffer:
                record = self.buffer[0]
                try:
                    self.base_handler.emit(record)
                    self.buffer.pop(0)
                except Exception:
                    break

 

   2) 분산 시스템에서의 로깅

      1. 상관 관계 ID 추적

         - 마이크로서비스 아키텍처에서 요청 추적을 위한 로깅 설정:

class CorrelationIdFilter(logging.Filter):
    """요청 추적을 위한 상관 관계 ID 필터"""
    
    def filter(self, record):
        # 현재 실행 컨텍스트에서 상관 관계 ID 가져오기
        correlation_id = getattr(context, 'correlation_id', None)
        if correlation_id is None:
            correlation_id = str(uuid.uuid4())
            setattr(context, 'correlation_id', correlation_id)
        
        record.correlation_id = correlation_id
        return True

class DistributedFormatter(logging.Formatter):
    """분산 시스템을 위한 로그 포맷터"""
    
    def format(self, record):
        # 기본 로그 정보
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'message': record.getMessage(),
            'correlation_id': getattr(record, 'correlation_id', 'unknown'),
            'service': self.get_service_info(),
            'trace': self.get_trace_info(record)
        }
        
        return json.dumps(log_data)
    
    def get_service_info(self):
        """서비스 정보 수집"""
        return {
            'name': os.getenv('SERVICE_NAME', 'unknown'),
            'version': os.getenv('SERVICE_VERSION', 'unknown'),
            'instance': socket.gethostname()
        }
    
    def get_trace_info(self, record):
        """분산 추적 정보 수집"""
        return {
            'trace_id': getattr(record, 'trace_id', None),
            'span_id': getattr(record, 'span_id', None),
            'parent_id': getattr(record, 'parent_id', None)
        }

 

      2. 로그 집중화 설정

         - 여러 서비스의 로그를 중앙 집중화하는 설정:

[handlers]
keys=syslogHandler,centralizedHandler

[handler_syslogHandler]
class=logging.handlers.SysLogHandler
level=INFO
formatter=jsonFormatter
args=('/dev/log', handlers.SysLogHandler.LOG_USER)

[handler_centralizedHandler]
class=myapp.logging.CentralizedLogHandler
level=INFO
formatter=jsonFormatter
args=('logs.example.com', 514)

 

         - 중앙 집중식 로그 핸들러 구현:

class CentralizedLogHandler(logging.Handler):
    """중앙 집중식 로깅을 위한 핸들러"""
    
    def __init__(self, host, port):
        super().__init__()
        self.host = host
        self.port = port
        self.socket = None
        self.connect()
        
    def connect(self):
        """로그 서버에 연결"""
        if self.socket is not None:
            self.socket.close()
        
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.connect((self.host, self.port))
    
    def emit(self, record):
        try:
            message = self.format(record)
            self.socket.sendall(message.encode() + b'\n')
        except Exception:
            self.handleError(record)
            # 연결 재시도
            self.connect()

 

   3) 로그 데이터 라이프사이클 관리

      1. 로그 보존 정책 구현

         - 로그 데이터의 수명 주기를 관리하는 핸들러:

class RetentionHandler(logging.handlers.RotatingFileHandler):
    """로그 보존 정책을 구현한 핸들러"""
    
    def __init__(self, filename, mode='a', maxBytes=0, backupCount=0,
                 encoding=None, delay=False, retention_days=30):
        super().__init__(filename, mode, maxBytes, backupCount, encoding, delay)
        self.retention_days = retention_days
        self.cleanup_old_logs()
    
    def cleanup_old_logs(self):
        """오래된 로그 파일 정리"""
        base_dir = os.path.dirname(self.baseFilename)
        base_name = os.path.basename(self.baseFilename)
        
        current_time = time.time()
        retention_seconds = self.retention_days * 24 * 60 * 60
        
        for filename in os.listdir(base_dir):
            if filename.startswith(base_name):
                filepath = os.path.join(base_dir, filename)
                file_time = os.path.getmtime(filepath)
                
                if current_time - file_time > retention_seconds:
                    try:
                        os.remove(filepath)
                        logging.info(f"오래된 로그 파일 삭제: {filepath}")
                    except OSError as e:
                        logging.error(f"로그 파일 삭제 실패: {e}")

 

      2. 로그 압축 및 아카이브

         - 오래된 로그를 압축하고 아카이브하는 핸들러:

class ArchiveHandler(logging.handlers.RotatingFileHandler):
    """로그 파일을 자동으로 압축하고 아카이브하는 핸들러"""
    
    def __init__(self, filename, mode='a', maxBytes=0, backupCount=0,
                 encoding=None, delay=False, archive_dir=None):
        super().__init__(filename, mode, maxBytes, backupCount, encoding, delay)
        self.archive_dir = archive_dir or os.path.join(
            os.path.dirname(filename), 'archive'
        )
        os.makedirs(self.archive_dir, exist_ok=True)
    
    def doRollover(self):
        """로그 롤오버 시 이전 로그 파일을 압축하여 아카이브"""
        if self.stream:
            self.stream.close()
            self.stream = None
            
        if self.backupCount > 0:
            # 가장 오래된 백업 파일 압축 및 이동
            for i in range(self.backupCount - 1, 0, -1):
                sfn = self.rotation_filename("%s.%d" % (self.baseFilename, i))
                dfn = self.rotation_filename("%s.%d" % (self.baseFilename, i + 1))
                if os.path.exists(sfn):
                    if os.path.exists(dfn):
                        os.remove(dfn)
                    os.rename(sfn, dfn)
                    
            dfn = self.rotation_filename(self.baseFilename + ".1")
            if os.path.exists(dfn):
                os.remove(dfn)
            self.rotate(self.baseFilename, dfn)
            
            # 아카이브로 이동
            archive_path = os.path.join(
                self.archive_dir,
                f"{os.path.basename(dfn)}_{time.strftime('%Y%m%d_%H%M%S')}.gz"
            )
            with open(dfn, 'rb') as f_in:
                with gzip.open(archive_path, 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
            os.remove(dfn)
            
        if not self.delay:
            self.stream = self._open()

 

11. 로깅 시스템 테스트와 모니터링

   1) 로깅 시스템 테스트

      1. 메모리 기반 로그 핸들러

테스트 시에는 실제 파일 시스템이나 네트워크를 사용하지 않고 메모리에서 로그를 검증할 수 있다.

class MemoryHandler(logging.Handler):
    """테스트를 위한 메모리 기반 로그 핸들러"""
    
    def __init__(self):
        super().__init__()
        self.messages = []
        self.lock = threading.Lock()
    
    def emit(self, record):
        """로그 레코드를 메모리에 저장"""
        with self.lock:
            self.messages.append({
                'level': record.levelname,
                'message': self.format(record),
                'timestamp': self.formatTime(record),
                'logger_name': record.name
            })
    
    def get_messages(self, level=None):
        """저장된 메시지 검색"""
        with self.lock:
            if level:
                return [msg for msg in self.messages 
                       if msg['level'] == level]
            return self.messages.copy()
    
    def clear(self):
        """저장된 메시지 초기화"""
        with self.lock:
            self.messages.clear()

 

         - 이 핸들러를 사용한 테스트 예시:

import unittest

class TestLogging(unittest.TestCase):
    def setUp(self):
        # 테스트용 로거 설정
        self.memory_handler = MemoryHandler()
        self.logger = logging.getLogger('test_logger')
        self.logger.addHandler(self.memory_handler)
        self.logger.setLevel(logging.DEBUG)
    
    def tearDown(self):
        # 테스트 후 정리
        self.logger.removeHandler(self.memory_handler)
        self.memory_handler.clear()
    
    def test_log_levels(self):
        """로그 레벨별 메시지 테스트"""
        self.logger.debug("디버그 메시지")
        self.logger.info("정보 메시지")
        self.logger.warning("경고 메시지")
        
        messages = self.memory_handler.get_messages()
        self.assertEqual(len(messages), 3)
        
        debug_messages = self.memory_handler.get_messages('DEBUG')
        self.assertEqual(len(debug_messages), 1)
        self.assertEqual(debug_messages[0]['message'], "디버그 메시지")

 

      2. 로그 포맷 검증

         - 로그 메시지의 형식이 예상대로인지 검증하는 테스트:

class TestLogFormatting(unittest.TestCase):
    def setUp(self):
        self.memory_handler = MemoryHandler()
        self.logger = logging.getLogger('format_test')
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        self.memory_handler.setFormatter(formatter)
        self.logger.addHandler(self.memory_handler)
        self.logger.setLevel(logging.DEBUG)
    
    def test_log_format(self):
        """로그 포맷 검증"""
        test_message = "테스트 메시지"
        self.logger.info(test_message)
        
        messages = self.memory_handler.get_messages()
        self.assertEqual(len(messages), 1)
        
        log_message = messages[0]['message']
        # 타임스탬프 형식 검증
        self.assertRegex(
            log_message,
            r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3} - INFO - 테스트 메시지'
        )

 

   2) 로깅 시스템 모니터링

      1. 로깅 메트릭 수집

         - 로깅 시스템의 성능과 상태를 모니터링하기 위한 메트릭 수집기:

class LoggingMetricsCollector:
    """로깅 시스템 메트릭 수집기"""
    
    def __init__(self):
        self.metrics = {
            'log_counts': defaultdict(int),
            'error_counts': defaultdict(int),
            'processing_times': [],
            'queue_sizes': [],
            'last_flush_time': None
        }
        self.lock = threading.Lock()
    
    def record_log(self, level, processing_time):
        """로그 발생 기록"""
        with self.lock:
            self.metrics['log_counts'][level] += 1
            self.metrics['processing_times'].append(processing_time)
    
    def record_error(self, error_type):
        """로깅 오류 기록"""
        with self.lock:
            self.metrics['error_counts'][error_type] += 1
    
    def record_queue_size(self, size):
        """로그 큐 크기 기록"""
        with self.lock:
            self.metrics['queue_sizes'].append(size)
    
    def get_statistics(self):
        """수집된 메트릭 통계 계산"""
        with self.lock:
            stats = {
                'total_logs': sum(self.metrics['log_counts'].values()),
                'logs_by_level': dict(self.metrics['log_counts']),
                'total_errors': sum(self.metrics['error_counts'].values()),
                'errors_by_type': dict(self.metrics['error_counts']),
                'avg_processing_time': (
                    sum(self.metrics['processing_times']) /
                    len(self.metrics['processing_times'])
                    if self.metrics['processing_times'] else 0
                ),
                'max_queue_size': max(self.metrics['queue_sizes'])
                if self.metrics['queue_sizes'] else 0
            }
            return stats

 

      2. 로깅 시스템 헬스체크

         - 로깅 시스템의 상태를 주기적으로 확인하는 헬스체크 구현:

class LoggingHealthChecker:
    """로깅 시스템 헬스체크"""
    
    def __init__(self, handlers):
        self.handlers = handlers
        self.last_check_time = None
        self.health_status = {'overall': 'unknown'}
    
    def check_handler_health(self, handler):
        """개별 핸들러의 상태 확인"""
        try:
            # 핸들러 타입별 검사
            if isinstance(handler, logging.FileHandler):
                return self._check_file_handler(handler)
            elif isinstance(handler, logging.handlers.SocketHandler):
                return self._check_socket_handler(handler)
            elif isinstance(handler, logging.handlers.QueueHandler):
                return self._check_queue_handler(handler)
            return {'status': 'unknown'}
        except Exception as e:
            return {
                'status': 'error',
                'error': str(e)
            }
    
    def _check_file_handler(self, handler):
        """파일 핸들러 상태 검사"""
        if not os.path.exists(handler.baseFilename):
            return {
                'status': 'error',
                'error': 'Log file does not exist'
            }
        
        try:
            # 파일 쓰기 권한 확인
            with open(handler.baseFilename, 'a'):
                pass
            
            # 디스크 공간 확인
            stat = os.statvfs(handler.baseFilename)
            free_space = stat.f_bavail * stat.f_frsize
            
            return {
                'status': 'healthy' if free_space > 1024*1024*100 else 'warning',
                'free_space_mb': free_space // (1024*1024)
            }
        except Exception as e:
            return {
                'status': 'error',
                'error': str(e)
            }
    
    def _check_socket_handler(self, handler):
        """소켓 핸들러 연결 상태 검사"""
        try:
            # 연결 테스트
            test_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            test_socket.settimeout(2)
            test_socket.connect((handler.host, handler.port))
            test_socket.close()
            
            return {
                'status': 'healthy',
                'connection': 'established'
            }
        except Exception as e:
            return {
                'status': 'error',
                'error': str(e)
            }
    
    def _check_queue_handler(self, handler):
        """큐 핸들러 상태 검사"""
        queue_size = handler.queue.qsize()
        return {
            'status': 'healthy' if queue_size < 1000 else 'warning',
            'queue_size': queue_size
        }
    
    def perform_health_check(self):
        """전체 로깅 시스템 상태 검사"""
        self.last_check_time = datetime.datetime.now()
        handler_status = {}
        
        for handler in self.handlers:
            handler_name = handler.__class__.__name__
            handler_status[handler_name] = self.check_handler_health(handler)
        
        # 전체 상태 판단
        has_error = any(
            status['status'] == 'error'
            for status in handler_status.values()
        )
        has_warning = any(
            status['status'] == 'warning'
            for status in handler_status.values()
        )
        
        self.health_status = {
            'overall': 'error' if has_error else
                      'warning' if has_warning else 'healthy',
            'handlers': handler_status,
            'last_check_time': self.last_check_time.isoformat()
        }
        
        return self.health_status

 

 - reference : 

https://docs.python.org/3/library/logging.config.html

logging.config — Logging configuration