Python實現(xiàn)結(jié)構(gòu)化日志系統(tǒng)的完整方案和最佳實踐
引言
在現(xiàn)代軟件系統(tǒng)中,日志不僅是調(diào)試和問題排查的工具,更是系統(tǒng)可觀測性的核心組成部分。隨著微服務(wù)、分布式系統(tǒng)和云原生架構(gòu)的普及,傳統(tǒng)文本日志已無法滿足復(fù)雜系統(tǒng)的監(jiān)控、分析和調(diào)試需求。結(jié)構(gòu)化日志應(yīng)運而生,成為現(xiàn)代日志系統(tǒng)的標(biāo)準(zhǔn)實踐。
根據(jù)2023年DevOps現(xiàn)狀報告顯示,采用結(jié)構(gòu)化日志的團隊部署頻率提高2.6倍,故障恢復(fù)時間縮短3.2倍。本文將深入探討結(jié)構(gòu)化日志系統(tǒng)的設(shè)計原理、實現(xiàn)方法和最佳實踐,提供完整的Python實現(xiàn)方案。
1. 日志系統(tǒng)基礎(chǔ)概念
1.1 日志的重要性與價值
日志系統(tǒng)為軟件系統(tǒng)提供了以下關(guān)鍵價值:
- 故障排查:快速定位和解決生產(chǎn)環(huán)境問題
- 性能監(jiān)控:跟蹤系統(tǒng)性能和資源使用情況
- 安全審計:記錄用戶操作和安全事件
- 業(yè)務(wù)分析:分析用戶行為和應(yīng)用使用模式
- 合規(guī)要求:滿足法律和行業(yè)規(guī)定的日志保留要求
1.2 日志系統(tǒng)的演進歷程

1.3 日志質(zhì)量的金字塔模型

2. 結(jié)構(gòu)化日志基礎(chǔ)
2.1 什么是結(jié)構(gòu)化日志
結(jié)構(gòu)化日志是將日志數(shù)據(jù)以機器可讀的格式(通常是JSON)進行組織,而不是傳統(tǒng)的純文本格式。結(jié)構(gòu)化日志包含:
- 固定字段:時間戳、級別、消息、來源等
- 上下文字段:請求ID、用戶ID、會話ID等
- 業(yè)務(wù)字段:操作類型、資源ID、結(jié)果狀態(tài)等
- 性能字段:耗時、內(nèi)存使用、請求大小等
2.2 結(jié)構(gòu)化日志 vs 非結(jié)構(gòu)化日志
| 維度 | 結(jié)構(gòu)化日志 | 非結(jié)構(gòu)化日志 |
|---|---|---|
| 格式 | JSON、鍵值對 | 純文本 |
| 可讀性 | 機器友好 | 人類友好 |
| 查詢能力 | 強大(支持字段篩選) | 有限(文本搜索) |
| 存儲效率 | 較高 | 較低 |
| 解析復(fù)雜度 | 簡單 | 復(fù)雜 |
| 擴展性 | 容易添加新字段 | 需要修改格式 |
2.3 結(jié)構(gòu)化日志的數(shù)學(xué)表示
設(shè)日志事件為一個元組: L=(t,l,m,C)
其中:
- t :時間戳
- l:日志級別
- m:消息模板
- C:上下文鍵值對集合,C={k1?:v1?,k2?:v2?,...,kn?:vn?}
結(jié)構(gòu)化日志可以表示為:
Lstruct?=JSON({timestamp:t,level:l,message:m}∪C)
日志查詢可以形式化為:
Query(Lstruct?,Φ)={L∣∀(k,v)∈Φ,L.C[k]=v}
其中 Φ 是查詢條件的鍵值對集合。
3. 日志系統(tǒng)架構(gòu)設(shè)計
3.1 現(xiàn)代日志系統(tǒng)架構(gòu)

3.2 日志處理流水線
典型的日志處理流水線包含以下階段:
- 收集:從應(yīng)用收集原始日志
- 解析:提取結(jié)構(gòu)化字段
- 豐富:添加元數(shù)據(jù)(主機名、環(huán)境等)
- 過濾:移除敏感信息或無用數(shù)據(jù)
- 轉(zhuǎn)換:格式轉(zhuǎn)換和標(biāo)準(zhǔn)化
- 路由:根據(jù)規(guī)則分發(fā)到不同目的地
- 存儲:持久化存儲
- 索引:建立快速檢索索引
3.3 分布式日志追蹤
在微服務(wù)架構(gòu)中,分布式追蹤是結(jié)構(gòu)化日志的關(guān)鍵組成部分。使用以下字段實現(xiàn)追蹤:
trace_id:整個請求鏈路的唯一標(biāo)識span_id:單個操作段的標(biāo)識parent_span_id:父操作的標(biāo)識service_name:服務(wù)名稱operation_name:操作名稱
追蹤系統(tǒng)的數(shù)學(xué)表示:
設(shè)請求 R經(jīng)過 n 個服務(wù),則:
T(R)={S1?,S2?,...,Sn?}
每個服務(wù)操作 Si? 包含:Si?=(tstart?,tend?,trace_id,span_idi?,parent_span_idi?,metadatai?)
請求總耗時:Δt=max(tend?)−min(tstart?)
4. Python結(jié)構(gòu)化日志實現(xiàn)
4.1 基礎(chǔ)結(jié)構(gòu)化日志框架
"""
結(jié)構(gòu)化日志系統(tǒng)實現(xiàn)
設(shè)計原則:
1. 結(jié)構(gòu)化優(yōu)先:所有日志輸出為結(jié)構(gòu)化格式
2. 上下文感知:自動捕獲和傳遞上下文
3. 性能友好:異步處理,最小化性能影響
4. 可擴展性:支持自定義處理器和格式器
5. 安全性:內(nèi)置敏感信息過濾
"""
import json
import logging
import sys
import time
import uuid
import inspect
import threading
from typing import Dict, Any, Optional, List, Union, Callable
from datetime import datetime
from enum import Enum
from dataclasses import dataclass, field, asdict
from abc import ABC, abstractmethod
from queue import Queue, Empty
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import traceback
import hashlib
import zlib
from collections import defaultdict
# 類型別名
LogData = Dict[str, Any]
ContextDict = Dict[str, Any]
class LogLevel(Enum):
"""日志級別枚舉"""
TRACE = 0 # 最詳細的跟蹤信息
DEBUG = 1 # 調(diào)試信息
INFO = 2 # 常規(guī)信息
WARN = 3 # 警告信息
ERROR = 4 # 錯誤信息
FATAL = 5 # 嚴(yán)重錯誤
@classmethod
def from_string(cls, level_str: str) -> 'LogLevel':
"""從字符串轉(zhuǎn)換日志級別"""
level_map = {
'trace': cls.TRACE,
'debug': cls.DEBUG,
'info': cls.INFO,
'warn': cls.WARN,
'warning': cls.WARN,
'error': cls.ERROR,
'fatal': cls.FATAL,
'critical': cls.FATAL
}
return level_map.get(level_str.lower(), cls.INFO)
@classmethod
def to_standard_level(cls, level: 'LogLevel') -> int:
"""轉(zhuǎn)換為標(biāo)準(zhǔn)logging級別"""
mapping = {
cls.TRACE: 5, # 低于DEBUG
cls.DEBUG: logging.DEBUG,
cls.INFO: logging.INFO,
cls.WARN: logging.WARNING,
cls.ERROR: logging.ERROR,
cls.FATAL: logging.CRITICAL
}
return mapping[level]
@dataclass
class LogRecord:
"""結(jié)構(gòu)化日志記錄"""
# 基礎(chǔ)字段
timestamp: str
level: str
message: str
logger_name: str
# 上下文字段
trace_id: Optional[str] = None
span_id: Optional[str] = None
request_id: Optional[str] = None
user_id: Optional[str] = None
session_id: Optional[str] = None
correlation_id: Optional[str] = None
# 執(zhí)行上下文
filename: Optional[str] = None
function: Optional[str] = None
line_no: Optional[int] = None
thread_id: Optional[int] = None
thread_name: Optional[str] = None
process_id: Optional[int] = None
# 應(yīng)用程序上下文
app_name: Optional[str] = None
app_version: Optional[str] = None
environment: Optional[str] = None
hostname: Optional[str] = None
service_name: Optional[str] = None
# 性能指標(biāo)
duration_ms: Optional[float] = None
memory_mb: Optional[float] = None
cpu_percent: Optional[float] = None
# 自定義字段
extra: Dict[str, Any] = field(default_factory=dict)
# 錯誤信息
error_type: Optional[str] = None
error_message: Optional[str] = None
stack_trace: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""轉(zhuǎn)換為字典"""
result = asdict(self)
# 移除None值以減小體積
return {k: v for k, v in result.items() if v is not None}
def to_json(self, indent: Optional[int] = None) -> str:
"""轉(zhuǎn)換為JSON字符串"""
return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False)
def get_field_hash(self) -> str:
"""獲取字段內(nèi)容的哈希值(用于去重)"""
# 排除一些動態(tài)字段
excluded_fields = {'timestamp', 'duration_ms', 'memory_mb', 'cpu_percent'}
data = {k: v for k, v in self.to_dict().items()
if k not in excluded_fields and v is not None}
content = json.dumps(data, sort_keys=True, ensure_ascii=False)
return hashlib.md5(content.encode()).hexdigest()
def is_similar_to(self, other: 'LogRecord', threshold: float = 0.9) -> bool:
"""判斷兩個日志記錄是否相似(用于去重)"""
if self.level != other.level:
return False
# 計算消息相似度(簡化的編輯距離)
from difflib import SequenceMatcher
message_similarity = SequenceMatcher(
None, self.message, other.message
).ratio()
return message_similarity >= threshold
class LogContext:
"""日志上下文管理器"""
def __init__(self):
# 線程本地存儲
self._local = threading.local()
self._global_context = {}
self._context_stack = []
@property
def current(self) -> Dict[str, Any]:
"""獲取當(dāng)前上下文"""
if not hasattr(self._local, 'context'):
self._local.context = {}
return self._local.context
@current.setter
def current(self, context: Dict[str, Any]):
"""設(shè)置當(dāng)前上下文"""
self._local.context = context
def get(self, key: str, default: Any = None) -> Any:
"""獲取上下文值"""
return self.current.get(key, self._global_context.get(key, default))
def set(self, key: str, value: Any, global_scope: bool = False):
"""設(shè)置上下文值"""
if global_scope:
self._global_context[key] = value
else:
self.current[key] = value
def update(self, data: Dict[str, Any], global_scope: bool = False):
"""批量更新上下文"""
if global_scope:
self._global_context.update(data)
else:
self.current.update(data)
def clear(self):
"""清除當(dāng)前線程上下文"""
if hasattr(self._local, 'context'):
self._local.context.clear()
def push_context(self, context: Dict[str, Any]):
"""壓入新的上下文層"""
if not hasattr(self._local, 'context_stack'):
self._local.context_stack = []
# 保存當(dāng)前上下文
current_copy = self.current.copy()
self._local.context_stack.append(current_copy)
# 更新為新上下文(合并)
new_context = current_copy.copy()
new_context.update(context)
self.current = new_context
def pop_context(self) -> Dict[str, Any]:
"""彈出上下文層"""
if not hasattr(self._local, 'context_stack') or not self._local.context_stack:
old_context = self.current.copy()
self.clear()
return old_context
old_context = self.current
self.current = self._local.context_stack.pop()
return old_context
def context_manager(self, **kwargs):
"""上下文管理器"""
return LogContextManager(self, kwargs)
def get_all_context(self) -> Dict[str, Any]:
"""獲取所有上下文(包括全局)"""
result = self._global_context.copy()
result.update(self.current)
return result
class LogContextManager:
"""上下文管理器"""
def __init__(self, log_context: LogContext, context_data: Dict[str, Any]):
self.log_context = log_context
self.context_data = context_data
def __enter__(self):
self.log_context.push_context(self.context_data)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.log_context.pop_context()
class StructuredFormatter(ABC):
"""結(jié)構(gòu)化日志格式化器抽象基類"""
@abstractmethod
def format(self, record: LogRecord) -> str:
"""格式化日志記錄"""
pass
class JSONFormatter(StructuredFormatter):
"""JSON格式化器"""
def __init__(
self,
indent: Optional[int] = None,
ensure_ascii: bool = False,
sort_keys: bool = False,
include_metadata: bool = True
):
self.indent = indent
self.ensure_ascii = ensure_ascii
self.sort_keys = sort_keys
self.include_metadata = include_metadata
def format(self, record: LogRecord) -> str:
"""格式化為JSON"""
data = record.to_dict()
# 添加格式化元數(shù)據(jù)
if self.include_metadata:
data['_metadata'] = {
'format_version': '1.0',
'formatter': 'json',
'timestamp_ns': time.time_ns()
}
return json.dumps(
data,
indent=self.indent,
ensure_ascii=self.ensure_ascii,
sort_keys=self.sort_keys
)
class NDJSONFormatter(StructuredFormatter):
"""NDJSON格式化器(每行一個JSON)"""
def __init__(self, **kwargs):
self.json_formatter = JSONFormatter(**kwargs)
def format(self, record: LogRecord) -> str:
"""格式化為NDJSON"""
return self.json_formatter.format(record)
class LogFilter(ABC):
"""日志過濾器抽象基類"""
@abstractmethod
def filter(self, record: LogRecord) -> bool:
"""過濾日志記錄,返回True表示保留"""
pass
class LevelFilter(LogFilter):
"""級別過濾器"""
def __init__(self, min_level: LogLevel):
self.min_level = min_level
def filter(self, record: LogRecord) -> bool:
"""根據(jù)級別過濾"""
record_level = LogLevel.from_string(record.level)
return record_level.value >= self.min_level.value
class RateLimitFilter(LogFilter):
"""速率限制過濾器"""
def __init__(self, max_per_second: int = 10, window_seconds: int = 1):
self.max_per_second = max_per_second
self.window_seconds = window_seconds
self.log_counts = defaultdict(int)
self.window_start = time.time()
def filter(self, record: LogRecord) -> bool:
"""速率限制"""
current_time = time.time()
# 檢查是否需要重置窗口
if current_time - self.window_start >= self.window_seconds:
self.log_counts.clear()
self.window_start = current_time
# 獲取日志哈希作為鍵
log_key = record.get_field_hash()
current_count = self.log_counts[log_key]
if current_count < self.max_per_second:
self.log_counts[log_key] = current_count + 1
return True
return False
class SensitiveDataFilter(LogFilter):
"""敏感數(shù)據(jù)過濾器"""
def __init__(self):
# 敏感數(shù)據(jù)模式(可以擴展)
self.sensitive_patterns = [
r'(?i)(password|passwd|pwd)[=:]\s*["\']?([^"\'\s]+)["\']?',
r'(?i)(api[_-]?key|secret[_-]?key)[=:]\s*["\']?([^"\'\s]+)["\']?',
r'(?i)(token)[=:]\s*["\']?([^"\'\s]+)["\']?',
r'(?i)(credit[_-]?card|cc)[=:]\s*["\']?(\d[ -]*?){13,16}["\']?',
r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', # 電話號碼
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', # 郵箱
]
self.compiled_patterns = [re.compile(pattern) for pattern in self.sensitive_patterns]
def filter(self, record: LogRecord) -> bool:
"""過濾敏感信息"""
# 對消息進行脫敏
record.message = self._mask_sensitive_data(record.message)
# 對extra字段進行脫敏
for key, value in record.extra.items():
if isinstance(value, str):
record.extra[key] = self._mask_sensitive_data(value)
return True
def _mask_sensitive_data(self, text: str) -> str:
"""脫敏文本中的敏感信息"""
if not isinstance(text, str):
return text
masked_text = text
for pattern in self.compiled_patterns:
masked_text = pattern.sub(self._mask_replacer, masked_text)
return masked_text
def _mask_replacer(self, match) -> str:
"""替換匹配的敏感信息"""
full_match = match.group(0)
# 根據(jù)匹配內(nèi)容決定脫敏策略
if '@' in full_match: # 郵箱
parts = full_match.split('@')
if len(parts[0]) > 2:
return parts[0][:2] + '***@' + parts[1]
else:
return '***@' + parts[1]
elif any(keyword in full_match.lower() for keyword in ['password', 'passwd', 'pwd']):
return 'password=***'
elif any(keyword in full_match.lower() for keyword in ['key', 'token', 'secret']):
return match.group(1) + '=***'
elif re.match(r'\d', full_match.replace('-', '').replace(' ', '')):
# 數(shù)字類型(信用卡、電話等)
digits = re.sub(r'[^\d]', '', full_match)
if 10 <= len(digits) <= 16:
return digits[:4] + '*' * (len(digits) - 8) + digits[-4:]
return '***'
4.2 高級日志處理器
class LogHandler(ABC):
"""日志處理器抽象基類"""
def __init__(
self,
level: LogLevel = LogLevel.INFO,
formatter: Optional[StructuredFormatter] = None,
filters: Optional[List[LogFilter]] = None
):
self.level = level
self.formatter = formatter or JSONFormatter()
self.filters = filters or []
# 性能統(tǒng)計
self.processed_count = 0
self.dropped_count = 0
self.start_time = time.time()
@abstractmethod
def emit(self, record: LogRecord):
"""輸出日志記錄"""
pass
def handle(self, record: LogRecord) -> bool:
"""處理日志記錄"""
# 檢查級別
record_level = LogLevel.from_string(record.level)
if record_level.value < self.level.value:
self.dropped_count += 1
return False
# 應(yīng)用過濾器
for filter_obj in self.filters:
if not filter_obj.filter(record):
self.dropped_count += 1
return False
# 格式化
formatted = self.formatter.format(record)
# 輸出
try:
self.emit(record)
self.processed_count += 1
return True
except Exception as e:
# 處理器錯誤處理
print(f"日志處理器錯誤: {e}")
self.dropped_count += 1
return False
def get_stats(self) -> Dict[str, Any]:
"""獲取處理器統(tǒng)計信息"""
uptime = time.time() - self.start_time
return {
'processed': self.processed_count,
'dropped': self.dropped_count,
'uptime_seconds': uptime,
'rate_per_second': self.processed_count / max(uptime, 0.001),
'handler_type': self.__class__.__name__
}
class ConsoleHandler(LogHandler):
"""控制臺處理器"""
def __init__(
self,
level: LogLevel = LogLevel.INFO,
formatter: Optional[StructuredFormatter] = None,
output_stream: Any = sys.stdout,
use_colors: bool = True
):
super().__init__(level, formatter)
self.output_stream = output_stream
self.use_colors = use_colors
# 顏色映射
self.color_map = {
'TRACE': '\033[90m', # 灰色
'DEBUG': '\033[36m', # 青色
'INFO': '\033[32m', # 綠色
'WARN': '\033[33m', # 黃色
'ERROR': '\033[31m', # 紅色
'FATAL': '\033[41m\033[37m', # 紅底白字
'RESET': '\033[0m' # 重置
}
def emit(self, record: LogRecord):
"""輸出到控制臺"""
formatted = self.formatter.format(record)
if self.use_colors:
color = self.color_map.get(record.level.upper(), '')
reset = self.color_map['RESET']
output = f"{color}{formatted}{reset}"
else:
output = formatted
print(output, file=self.output_stream)
class FileHandler(LogHandler):
"""文件處理器"""
def __init__(
self,
filename: Union[str, Path],
level: LogLevel = LogLevel.INFO,
formatter: Optional[StructuredFormatter] = None,
mode: str = 'a',
encoding: str = 'utf-8',
buffering: int = 1 # 行緩沖
):
super().__init__(level, formatter)
self.filename = Path(filename)
self.mode = mode
self.encoding = encoding
self.buffering = buffering
# 確保目錄存在
self.filename.parent.mkdir(parents=True, exist_ok=True)
# 打開文件
self._open_file()
def _open_file(self):
"""打開文件"""
self.file = open(
self.filename,
mode=self.mode,
encoding=self.encoding,
buffering=self.buffering
)
def emit(self, record: LogRecord):
"""輸出到文件"""
formatted = self.formatter.format(record)
self.file.write(formatted + '\n')
self.file.flush()
def close(self):
"""關(guān)閉文件"""
if hasattr(self, 'file') and self.file:
self.file.close()
def rotate(self, max_size_mb: float = 100, backup_count: int = 5):
"""日志輪轉(zhuǎn)"""
if not self.filename.exists():
return
file_size_mb = self.filename.stat().st_size / (1024 * 1024)
if file_size_mb < max_size_mb:
return
# 關(guān)閉當(dāng)前文件
self.close()
# 重命名舊文件
for i in range(backup_count - 1, 0, -1):
old_file = self.filename.with_suffix(f".{i}.log")
new_file = self.filename.with_suffix(f".{i+1}.log")
if old_file.exists():
old_file.rename(new_file)
# 重命名當(dāng)前文件
current_backup = self.filename.with_suffix(".1.log")
self.filename.rename(current_backup)
# 重新打開文件
self._open_file()
class RotatingFileHandler(FileHandler):
"""自動輪轉(zhuǎn)的文件處理器"""
def __init__(
self,
filename: Union[str, Path],
level: LogLevel = LogLevel.INFO,
formatter: Optional[StructuredFormatter] = None,
max_size_mb: float = 100,
backup_count: int = 5,
check_interval: int = 10 # 檢查間隔(處理的日志條數(shù))
):
super().__init__(filename, level, formatter)
self.max_size_mb = max_size_mb
self.backup_count = backup_count
self.check_interval = check_interval
self.processed_since_check = 0
def handle(self, record: LogRecord) -> bool:
"""處理日志記錄(添加輪轉(zhuǎn)檢查)"""
self.processed_since_check += 1
if self.processed_since_check >= self.check_interval:
self.rotate(self.max_size_mb, self.backup_count)
self.processed_since_check = 0
return super().handle(record)
class AsyncHandler(LogHandler):
"""異步處理器"""
def __init__(
self,
base_handler: LogHandler,
max_queue_size: int = 10000,
worker_count: int = 1,
drop_when_full: bool = False
):
super().__init__(base_handler.level, base_handler.formatter, base_handler.filters)
self.base_handler = base_handler
# 隊列設(shè)置
self.max_queue_size = max_queue_size
self.queue = Queue(maxsize=max_queue_size)
self.drop_when_full = drop_when_full
# 工作線程
self.worker_count = worker_count
self.executor = ThreadPoolExecutor(
max_workers=worker_count,
thread_name_prefix="AsyncLogger"
)
# 啟動消費者
self.running = True
for i in range(worker_count):
self.executor.submit(self._worker_loop)
def emit(self, record: LogRecord):
"""異步處理日志記錄"""
try:
if self.drop_when_full and self.queue.full():
self.dropped_count += 1
return
self.queue.put_nowait(record)
except Exception as e:
# 隊列滿或其他錯誤
self.dropped_count += 1
print(f"異步日志隊列錯誤: {e}")
def _worker_loop(self):
"""工作線程循環(huán)"""
while self.running:
try:
# 阻塞獲取日志記錄(帶超時)
try:
record = self.queue.get(timeout=1.0)
except Empty:
continue
# 使用基礎(chǔ)處理器處理
self.base_handler.handle(record)
# 標(biāo)記任務(wù)完成
self.queue.task_done()
except Exception as e:
print(f"異步日志工作線程錯誤: {e}")
def shutdown(self, timeout: float = 5.0):
"""關(guān)閉異步處理器"""
self.running = False
# 等待隊列清空
self.queue.join()
# 關(guān)閉執(zhí)行器
self.executor.shutdown(wait=True, timeout=timeout)
# 關(guān)閉基礎(chǔ)處理器
if hasattr(self.base_handler, 'close'):
self.base_handler.close()
def get_stats(self) -> Dict[str, Any]:
"""獲取統(tǒng)計信息(包括隊列信息)"""
base_stats = super().get_stats()
base_stats.update({
'queue_size': self.queue.qsize(),
'queue_max_size': self.max_queue_size,
'queue_full': self.queue.full(),
'worker_count': self.worker_count,
'is_running': self.running,
'base_handler_stats': self.base_handler.get_stats()
})
return base_stats
class BatchHandler(LogHandler):
"""批量處理器"""
def __init__(
self,
base_handler: LogHandler,
batch_size: int = 100,
flush_interval: float = 1.0, # 秒
compression: bool = False
):
super().__init__(base_handler.level, base_handler.formatter, base_handler.filters)
self.base_handler = base_handler
self.batch_size = batch_size
self.flush_interval = flush_interval
self.compression = compression
# 批處理緩沖區(qū)
self.buffer: List[LogRecord] = []
self.last_flush_time = time.time()
# 啟動定時刷新線程
self.flush_thread = threading.Thread(target=self._flush_loop, daemon=True)
self.running = True
self.flush_thread.start()
def emit(self, record: LogRecord):
"""添加到批處理緩沖區(qū)"""
self.buffer.append(record)
# 檢查是否需要刷新
if (len(self.buffer) >= self.batch_size or
(time.time() - self.last_flush_time) >= self.flush_interval):
self._flush_buffer()
def _flush_buffer(self):
"""刷新緩沖區(qū)"""
if not self.buffer:
return
# 準(zhǔn)備批量數(shù)據(jù)
batch_records = self.buffer.copy()
self.buffer.clear()
try:
# 批量處理
if self.compression:
# 壓縮批量數(shù)據(jù)
batch_data = self._compress_batch(batch_records)
# 這里需要基礎(chǔ)處理器支持批量數(shù)據(jù)
# 簡化實現(xiàn):逐個處理
for record in batch_records:
self.base_handler.handle(record)
else:
for record in batch_records:
self.base_handler.handle(record)
self.last_flush_time = time.time()
except Exception as e:
print(f"批量日志處理錯誤: {e}")
# 錯誤處理:將記錄放回緩沖區(qū)(避免丟失)
self.buffer.extend(batch_records)
def _compress_batch(self, records: List[LogRecord]) -> bytes:
"""壓縮批量數(shù)據(jù)"""
batch_json = json.dumps([r.to_dict() for r in records])
return zlib.compress(batch_json.encode())
def _flush_loop(self):
"""定時刷新循環(huán)"""
while self.running:
time.sleep(self.flush_interval)
self._flush_buffer()
def shutdown(self):
"""關(guān)閉批量處理器"""
self.running = False
self._flush_buffer() # 最后一次刷新
if self.flush_thread.is_alive():
self.flush_thread.join(timeout=2.0)
if hasattr(self.base_handler, 'shutdown'):
self.base_handler.shutdown()
def get_stats(self) -> Dict[str, Any]:
"""獲取統(tǒng)計信息"""
base_stats = super().get_stats()
base_stats.update({
'buffer_size': len(self.buffer),
'batch_size': self.batch_size,
'flush_interval': self.flush_interval,
'compression_enabled': self.compression,
'base_handler_stats': self.base_handler.get_stats()
})
return base_stats
4.3 完整的日志系統(tǒng)
class StructuredLogger:
"""結(jié)構(gòu)化日志記錄器"""
def __init__(
self,
name: str,
level: LogLevel = LogLevel.INFO,
handlers: Optional[List[LogHandler]] = None,
context: Optional[LogContext] = None,
capture_stacktrace: bool = False,
enable_performance_stats: bool = False
):
self.name = name
self.level = level
self.handlers = handlers or []
self.context = context or LogContext()
self.capture_stacktrace = capture_stacktrace
self.enable_performance_stats = enable_performance_stats
# 性能統(tǒng)計
self.stats = {
'log_count': defaultdict(int),
'last_log_time': None,
'total_log_time_ns': 0,
'error_count': 0
}
# 緩存調(diào)用者信息(性能優(yōu)化)
self._caller_cache = {}
def _get_caller_info(self, depth: int = 3) -> Dict[str, Any]:
"""獲取調(diào)用者信息"""
try:
# 使用緩存提高性能
cache_key = threading.get_ident()
if cache_key in self._caller_cache:
return self._caller_cache[cache_key]
# 獲取調(diào)用堆棧
frame = inspect.currentframe()
for _ in range(depth):
if frame is None:
break
frame = frame.f_back
if frame is None:
return {}
# 提取信息
info = {
'filename': frame.f_code.co_filename,
'function': frame.f_code.co_name,
'line_no': frame.f_lineno,
'module': frame.f_globals.get('__name__', '')
}
# 緩存
self._caller_cache[cache_key] = info
return info
except Exception:
return {}
finally:
# 清理引用
del frame
def _create_record(
self,
level: LogLevel,
message: str,
extra: Optional[Dict[str, Any]] = None,
error_info: Optional[Dict[str, Any]] = None
) -> LogRecord:
"""創(chuàng)建日志記錄"""
# 基礎(chǔ)時間
now = datetime.utcnow()
# 調(diào)用者信息
caller_info = self._get_caller_info() if self.capture_stacktrace else {}
# 構(gòu)建記錄
record = LogRecord(
timestamp=now.isoformat() + 'Z',
level=level.name,
message=message,
logger_name=self.name,
**caller_info
)
# 添加線程信息
record.thread_id = threading.get_ident()
record.thread_name = threading.current_thread().name
record.process_id = os.getpid()
# 添加上下文
context_data = self.context.get_all_context()
for key, value in context_data.items():
if hasattr(record, key):
setattr(record, key, value)
else:
record.extra[key] = value
# 添加額外字段
if extra:
record.extra.update(extra)
# 添加錯誤信息
if error_info:
record.error_type = error_info.get('type')
record.error_message = error_info.get('message')
record.stack_trace = error_info.get('stack_trace')
return record
def log(
self,
level: LogLevel,
message: str,
extra: Optional[Dict[str, Any]] = None,
**kwargs
):
"""記錄日志"""
start_time = time.time_ns() if self.enable_performance_stats else 0
try:
# 檢查級別
if level.value < self.level.value:
return
# 合并額外字段
all_extra = extra.copy() if extra else {}
all_extra.update(kwargs)
# 錯誤信息處理
error_info = None
if 'exc_info' in kwargs and kwargs['exc_info']:
exc_type, exc_value, exc_traceback = kwargs['exc_info']
if exc_type:
error_info = {
'type': exc_type.__name__,
'message': str(exc_value),
'stack_trace': traceback.format_exc()
}
# 創(chuàng)建記錄
record = self._create_record(level, message, all_extra, error_info)
# 處理記錄
for handler in self.handlers:
handler.handle(record)
# 更新統(tǒng)計
self.stats['log_count'][level.name] += 1
self.stats['last_log_time'] = record.timestamp
if level == LogLevel.ERROR or level == LogLevel.FATAL:
self.stats['error_count'] += 1
except Exception as e:
# 記錄器內(nèi)部錯誤處理
print(f"日志記錄錯誤: {e}")
self.stats['error_count'] += 1
finally:
# 性能統(tǒng)計
if self.enable_performance_stats and start_time:
duration_ns = time.time_ns() - start_time
self.stats['total_log_time_ns'] += duration_ns
# 便捷方法
def trace(self, message: str, **kwargs):
"""記錄TRACE級別日志"""
self.log(LogLevel.TRACE, message, **kwargs)
def debug(self, message: str, **kwargs):
"""記錄DEBUG級別日志"""
self.log(LogLevel.DEBUG, message, **kwargs)
def info(self, message: str, **kwargs):
"""記錄INFO級別日志"""
self.log(LogLevel.INFO, message, **kwargs)
def warn(self, message: str, **kwargs):
"""記錄WARN級別日志"""
self.log(LogLevel.WARN, message, **kwargs)
def error(self, message: str, **kwargs):
"""記錄ERROR級別日志"""
self.log(LogLevel.ERROR, message, **kwargs)
def fatal(self, message: str, **kwargs):
"""記錄FATAL級別日志"""
self.log(LogLevel.FATAL, message, **kwargs)
def exception(self, message: str, exc: Optional[Exception] = None, **kwargs):
"""記錄異常"""
if exc is None:
# 捕獲當(dāng)前異常
exc_info = sys.exc_info()
else:
exc_info = (type(exc), exc, exc.__traceback__)
kwargs['exc_info'] = exc_info
self.log(LogLevel.ERROR, message, **kwargs)
def with_context(self, **kwargs):
"""添加上下文"""
return LogContextManager(self.context, kwargs)
def add_handler(self, handler: LogHandler):
"""添加處理器"""
self.handlers.append(handler)
def remove_handler(self, handler: LogHandler):
"""移除處理器"""
if handler in self.handlers:
self.handlers.remove(handler)
def get_stats(self) -> Dict[str, Any]:
"""獲取統(tǒng)計信息"""
handler_stats = [h.get_stats() for h in self.handlers]
stats = {
'logger_name': self.name,
'level': self.level.name,
'handler_count': len(self.handlers),
'log_counts': dict(self.stats['log_count']),
'error_count': self.stats['error_count'],
'handler_stats': handler_stats
}
if self.enable_performance_stats:
total_logs = sum(self.stats['log_count'].values())
if total_logs > 0:
avg_time_ns = self.stats['total_log_time_ns'] / total_logs
stats['performance'] = {
'total_time_ns': self.stats['total_log_time_ns'],
'avg_time_ns': avg_time_ns,
'avg_time_ms': avg_time_ns / 1_000_000
}
return stats
class LogManager:
"""日志管理器"""
_instance = None
_lock = threading.Lock()
def __new__(cls):
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self):
if self._initialized:
return
self._loggers: Dict[str, StructuredLogger] = {}
self._default_config: Dict[str, Any] = {}
self._global_context = LogContext()
self._initialized = True
# 默認配置
self._setup_defaults()
def _setup_defaults(self):
"""設(shè)置默認配置"""
self._default_config = {
'level': LogLevel.INFO,
'handlers': [
ConsoleHandler(
level=LogLevel.INFO,
formatter=JSONFormatter(indent=None)
)
],
'capture_stacktrace': False,
'enable_performance_stats': False
}
# 設(shè)置全局上下文
import socket
self._global_context.set('hostname', socket.gethostname(), global_scope=True)
self._global_context.set('process_id', os.getpid(), global_scope=True)
def get_logger(
self,
name: str,
level: Optional[LogLevel] = None,
handlers: Optional[List[LogHandler]] = None,
capture_stacktrace: Optional[bool] = None,
enable_performance_stats: Optional[bool] = None
) -> StructuredLogger:
"""獲取或創(chuàng)建日志記錄器"""
if name in self._loggers:
return self._loggers[name]
# 使用配置或默認值
config = self._default_config.copy()
if level is not None:
config['level'] = level
if handlers is not None:
config['handlers'] = handlers
if capture_stacktrace is not None:
config['capture_stacktrace'] = capture_stacktrace
if enable_performance_stats is not None:
config['enable_performance_stats'] = enable_performance_stats
# 創(chuàng)建日志記錄器
logger = StructuredLogger(
name=name,
context=self._global_context,
**config
)
self._loggers[name] = logger
return logger
def configure(
self,
config: Dict[str, Any],
name: Optional[str] = None
):
"""配置日志記錄器"""
if name:
# 配置特定記錄器
if name in self._loggers:
logger = self._loggers[name]
if 'level' in config:
logger.level = LogLevel.from_string(config['level'])
if 'handlers' in config:
# 這里需要根據(jù)配置創(chuàng)建處理器
logger.handlers = self._create_handlers_from_config(config['handlers'])
if 'capture_stacktrace' in config:
logger.capture_stacktrace = config['capture_stacktrace']
if 'enable_performance_stats' in config:
logger.enable_performance_stats = config['enable_performance_stats']
else:
# 更新默認配置
self._default_config.update(config)
# 更新現(xiàn)有記錄器
for logger in self._loggers.values():
self.configure(config, logger.name)
def _create_handlers_from_config(self, handlers_config: List[Dict]) -> List[LogHandler]:
"""從配置創(chuàng)建處理器"""
handlers = []
for handler_config in handlers_config:
handler_type = handler_config.get('type', 'console')
try:
if handler_type == 'console':
handler = ConsoleHandler(
level=LogLevel.from_string(handler_config.get('level', 'info')),
formatter=self._create_formatter_from_config(
handler_config.get('formatter', {})
),
use_colors=handler_config.get('use_colors', True)
)
elif handler_type == 'file':
handler = FileHandler(
filename=handler_config['filename'],
level=LogLevel.from_string(handler_config.get('level', 'info')),
formatter=self._create_formatter_from_config(
handler_config.get('formatter', {})
)
)
elif handler_type == 'rotating_file':
handler = RotatingFileHandler(
filename=handler_config['filename'],
level=LogLevel.from_string(handler_config.get('level', 'info')),
formatter=self._create_formatter_from_config(
handler_config.get('formatter', {})
),
max_size_mb=handler_config.get('max_size_mb', 100),
backup_count=handler_config.get('backup_count', 5)
)
elif handler_type == 'async':
base_handler_config = handler_config.get('base_handler', {})
base_handler = self._create_handlers_from_config([base_handler_config])[0]
handler = AsyncHandler(
base_handler=base_handler,
max_queue_size=handler_config.get('max_queue_size', 10000),
worker_count=handler_config.get('worker_count', 1),
drop_when_full=handler_config.get('drop_when_full', False)
)
else:
raise ValueError(f"未知的處理器類型: {handler_type}")
# 添加過濾器
filters_config = handler_config.get('filters', [])
for filter_config in filters_config:
filter_type = filter_config.get('type', 'level')
if filter_type == 'level':
handler.filters.append(LevelFilter(
LogLevel.from_string(filter_config.get('min_level', 'info'))
))
elif filter_type == 'rate_limit':
handler.filters.append(RateLimitFilter(
max_per_second=filter_config.get('max_per_second', 10),
window_seconds=filter_config.get('window_seconds', 1)
))
elif filter_type == 'sensitive_data':
handler.filters.append(SensitiveDataFilter())
handlers.append(handler)
except Exception as e:
print(f"創(chuàng)建處理器失敗 {handler_type}: {e}")
continue
return handlers
def _create_formatter_from_config(self, formatter_config: Dict) -> StructuredFormatter:
"""從配置創(chuàng)建格式化器"""
formatter_type = formatter_config.get('type', 'json')
if formatter_type == 'json':
return JSONFormatter(
indent=formatter_config.get('indent'),
ensure_ascii=formatter_config.get('ensure_ascii', False),
sort_keys=formatter_config.get('sort_keys', False)
)
elif formatter_type == 'ndjson':
return NDJSONFormatter(
indent=formatter_config.get('indent'),
ensure_ascii=formatter_config.get('ensure_ascii', False),
sort_keys=formatter_config.get('sort_keys', False)
)
else:
# 默認使用JSON
return JSONFormatter()
def set_global_context(self, **kwargs):
"""設(shè)置全局上下文"""
self._global_context.update(kwargs, global_scope=True)
def get_global_context(self) -> Dict[str, Any]:
"""獲取全局上下文"""
return self._global_context.get_all_context()
def shutdown(self):
"""關(guān)閉所有日志記錄器"""
for logger in self._loggers.values():
for handler in logger.handlers:
if hasattr(handler, 'shutdown'):
handler.shutdown()
elif hasattr(handler, 'close'):
handler.close()
self._loggers.clear()
def get_all_stats(self) -> Dict[str, Any]:
"""獲取所有統(tǒng)計信息"""
logger_stats = {}
total_logs = 0
total_errors = 0
for name, logger in self._loggers.items():
stats = logger.get_stats()
logger_stats[name] = stats
total_logs += sum(stats['log_counts'].values())
total_errors += stats['error_count']
return {
'logger_count': len(self._loggers),
'total_logs': total_logs,
'total_errors': total_errors,
'loggers': logger_stats,
'global_context': self.get_global_context()
}
5. 高級特性實現(xiàn)
5.1 分布式追蹤集成
class DistributedTraceContext:
"""分布式追蹤上下文"""
def __init__(self):
self._local = threading.local()
@property
def current(self) -> Dict[str, Any]:
"""獲取當(dāng)前追蹤上下文"""
if not hasattr(self._local, 'trace_context'):
self._local.trace_context = self._generate_new_context()
return self._local.trace_context
def _generate_new_context(self) -> Dict[str, Any]:
"""生成新的追蹤上下文"""
return {
'trace_id': self._generate_trace_id(),
'span_id': self._generate_span_id(),
'parent_span_id': None,
'sampled': True,
'flags': 0
}
def _generate_trace_id(self) -> str:
"""生成追蹤ID"""
return uuid.uuid4().hex
def _generate_span_id(self) -> str:
"""生成跨度ID"""
return uuid.uuid4().hex[:16]
def start_span(self, name: str, **attributes) -> 'Span':
"""開始新的跨度"""
parent_context = self.current.copy()
new_context = parent_context.copy()
new_context['span_id'] = self._generate_span_id()
new_context['parent_span_id'] = parent_context['span_id']
new_context['span_name'] = name
new_context['start_time'] = time.time_ns()
new_context['attributes'] = attributes
# 保存父上下文
if not hasattr(self._local, 'trace_stack'):
self._local.trace_stack = []
self._local.trace_stack.append(parent_context)
# 設(shè)置新上下文
self._local.trace_context = new_context
return Span(self, new_context)
def end_span(self, context: Dict[str, Any], status: str = "OK", **attributes):
"""結(jié)束跨度"""
if not hasattr(self._local, 'trace_stack') or not self._local.trace_stack:
return
# 計算持續(xù)時間
end_time = time.time_ns()
start_time = context.get('start_time', end_time)
duration_ns = end_time - start_time
# 創(chuàng)建跨度記錄
span_record = {
'trace_id': context.get('trace_id'),
'span_id': context.get('span_id'),
'parent_span_id': context.get('parent_span_id'),
'name': context.get('span_name', 'unknown'),
'start_time': start_time,
'end_time': end_time,
'duration_ns': duration_ns,
'status': status,
'attributes': {**context.get('attributes', {}), **attributes}
}
# 恢復(fù)父上下文
self._local.trace_context = self._local.trace_stack.pop()
return span_record
def get_current_span_id(self) -> Optional[str]:
"""獲取當(dāng)前跨度ID"""
return self.current.get('span_id')
def get_current_trace_id(self) -> Optional[str]:
"""獲取當(dāng)前追蹤ID"""
return self.current.get('trace_id')
class Span:
"""追蹤跨度"""
def __init__(self, tracer: DistributedTraceContext, context: Dict[str, Any]):
self.tracer = tracer
self.context = context
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
status = "ERROR" if exc_type else "OK"
self.tracer.end_span(self.context, status)
def set_attribute(self, key: str, value: Any):
"""設(shè)置跨度屬性"""
if 'attributes' not in self.context:
self.context['attributes'] = {}
self.context['attributes'][key] = value
def set_status(self, status: str):
"""設(shè)置跨度狀態(tài)"""
self.context['status'] = status
class TracingLogger(StructuredLogger):
"""集成追蹤的日志記錄器"""
def __init__(
self,
name: str,
tracer: Optional[DistributedTraceContext] = None,
**kwargs
):
super().__init__(name, **kwargs)
self.tracer = tracer or DistributedTraceContext()
# 自動添加上下文
self.context.set('tracer', self.tracer)
def _create_record(self, *args, **kwargs) -> LogRecord:
"""創(chuàng)建記錄(添加追蹤信息)"""
record = super()._create_record(*args, **kwargs)
# 添加追蹤信息
record.trace_id = self.tracer.get_current_trace_id()
record.span_id = self.tracer.get_current_span_id()
return record
def trace_span(self, name: str, **attributes):
"""創(chuàng)建追蹤跨度上下文管理器"""
return self.tracer.start_span(name, **attributes)
def log_with_span(
self,
level: LogLevel,
message: str,
span_name: Optional[str] = None,
**kwargs
):
"""在追蹤跨度中記錄日志"""
if span_name:
# 創(chuàng)建新跨度
with self.tracer.start_span(span_name):
self.log(level, message, **kwargs)
else:
# 使用當(dāng)前跨度
self.log(level, message, **kwargs)
5.2 性能監(jiān)控集成
class PerformanceMonitor:
"""性能監(jiān)控器"""
def __init__(self, logger: StructuredLogger):
self.logger = logger
self.metrics = defaultdict(list)
self.thresholds = {}
def measure(self, operation: str):
"""測量操作性能"""
return PerformanceTimer(self, operation)
def record_metric(
self,
name: str,
value: float,
unit: str = "ms",
tags: Optional[Dict[str, str]] = None
):
"""記錄性能指標(biāo)"""
timestamp = time.time_ns()
metric_record = {
'name': name,
'value': value,
'unit': unit,
'timestamp': timestamp,
'tags': tags or {}
}
# 存儲指標(biāo)
self.metrics[name].append(metric_record)
# 檢查閾值
if name in self.thresholds:
threshold = self.thresholds[name]
if value > threshold:
self.logger.warn(
f"性能閾值超過: {name} = {value}{unit} > {threshold}{unit}",
metric=metric_record
)
# 記錄指標(biāo)日志
self.logger.debug(
f"性能指標(biāo): {name}",
metric=metric_record,
extra={'metric_type': 'performance'}
)
return metric_record
def set_threshold(self, metric_name: str, threshold: float):
"""設(shè)置性能閾值"""
self.thresholds[metric_name] = threshold
def get_statistics(self, metric_name: str) -> Dict[str, float]:
"""獲取統(tǒng)計信息"""
records = self.metrics.get(metric_name, [])
if not records:
return {}
values = [r['value'] for r in records]
return {
'count': len(values),
'mean': sum(values) / len(values),
'min': min(values),
'max': max(values),
'p50': self._percentile(values, 50),
'p95': self._percentile(values, 95),
'p99': self._percentile(values, 99)
}
def _percentile(self, values: List[float], p: float) -> float:
"""計算百分位數(shù)"""
if not values:
return 0
sorted_values = sorted(values)
k = (len(sorted_values) - 1) * (p / 100)
f = int(k)
c = k - f
if f + 1 < len(sorted_values):
return sorted_values[f] + c * (sorted_values[f + 1] - sorted_values[f])
else:
return sorted_values[f]
def report_summary(self):
"""報告性能摘要"""
summary = {}
for metric_name in self.metrics:
stats = self.get_statistics(metric_name)
summary[metric_name] = stats
self.logger.info(
"性能監(jiān)控摘要",
performance_summary=summary,
extra={'report_type': 'performance_summary'}
)
return summary
class PerformanceTimer:
"""性能計時器"""
def __init__(self, monitor: PerformanceMonitor, operation: str):
self.monitor = monitor
self.operation = operation
self.start_time = None
self.tags = {}
def __enter__(self):
self.start_time = time.time_ns()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self.start_time is None:
return
end_time = time.time_ns()
duration_ns = end_time - self.start_time
duration_ms = duration_ns / 1_000_000
self.monitor.record_metric(
name=self.operation,
value=duration_ms,
unit="ms",
tags=self.tags
)
def add_tag(self, key: str, value: str):
"""添加標(biāo)簽"""
self.tags[key] = value
return self
5.3 日志采樣與聚合
class LogSampler:
"""日志采樣器"""
def __init__(
self,
base_logger: StructuredLogger,
sample_rate: float = 1.0, # 采樣率 0.0-1.0
adaptive_sampling: bool = False,
min_sample_rate: float = 0.01,
max_sample_rate: float = 1.0
):
self.base_logger = base_logger
self.sample_rate = sample_rate
self.adaptive_sampling = adaptive_sampling
self.min_sample_rate = min_sample_rate
self.max_sample_rate = max_sample_rate
# 采樣統(tǒng)計
self.sampled_count = 0
self.total_count = 0
# 自適應(yīng)采樣狀態(tài)
self.current_rate = sample_rate
self.last_adjust_time = time.time()
def should_sample(self, level: LogLevel) -> bool:
"""決定是否采樣"""
self.total_count += 1
# 高等級日志總是采樣
if level in [LogLevel.ERROR, LogLevel.FATAL]:
self.sampled_count += 1
return True
# 計算當(dāng)前采樣率
if self.adaptive_sampling:
self._adjust_sample_rate()
# 隨機采樣
import random
if random.random() <= self.current_rate:
self.sampled_count += 1
return True
return False
def _adjust_sample_rate(self):
"""調(diào)整采樣率"""
current_time = time.time()
# 每分鐘調(diào)整一次
if current_time - self.last_adjust_time < 60:
return
# 計算當(dāng)前實際采樣率
if self.total_count == 0:
actual_rate = 0
else:
actual_rate = self.sampled_count / self.total_count
# 調(diào)整采樣率
target_rate = self.sample_rate
if actual_rate < target_rate * 0.8:
# 采樣不足,提高采樣率
self.current_rate = min(self.current_rate * 1.2, self.max_sample_rate)
elif actual_rate > target_rate * 1.2:
# 采樣過多,降低采樣率
self.current_rate = max(self.current_rate * 0.8, self.min_sample_rate)
# 重置統(tǒng)計
self.sampled_count = 0
self.total_count = 0
self.last_adjust_time = current_time
def log(self, level: LogLevel, message: str, **kwargs):
"""記錄日志(帶采樣)"""
if self.should_sample(level):
self.base_logger.log(level, message, **kwargs)
class LogAggregator:
"""日志聚合器"""
def __init__(
self,
base_logger: StructuredLogger,
aggregation_window: float = 5.0, # 聚合窗口(秒)
max_aggregation_count: int = 1000 # 最大聚合條數(shù)
):
self.base_logger = base_logger
self.aggregation_window = aggregation_window
self.max_aggregation_count = max_aggregation_count
# 聚合緩沖區(qū)
self.buffer: Dict[str, List[LogRecord]] = defaultdict(list)
self.last_flush_time = time.time()
# 啟動定時刷新
self.flush_thread = threading.Thread(target=self._flush_loop, daemon=True)
self.running = True
self.flush_thread.start()
def _get_aggregation_key(self, record: LogRecord) -> str:
"""獲取聚合鍵"""
# 基于消息和級別聚合
key_parts = [
record.level,
record.message,
record.logger_name,
str(record.error_type) if record.error_type else "",
]
return hashlib.md5("|".join(key_parts).encode()).hexdigest()
def log(self, level: LogLevel, message: str, **kwargs):
"""記錄日志(帶聚合)"""
# 創(chuàng)建記錄但不立即發(fā)送
record = self.base_logger._create_record(level, message, kwargs.get('extra'))
# 添加到緩沖區(qū)
aggregation_key = self._get_aggregation_key(record)
self.buffer[aggregation_key].append(record)
# 檢查是否達到聚合上限
total_count = sum(len(records) for records in self.buffer.values())
if total_count >= self.max_aggregation_count:
self._flush_buffer()
def _flush_buffer(self):
"""刷新緩沖區(qū)"""
if not self.buffer:
return
flushed_records = []
for aggregation_key, records in self.buffer.items():
if not records:
continue
# 取第一條記錄作為模板
template_record = records[0]
# 創(chuàng)建聚合記錄
aggregated_record = LogRecord(
timestamp=datetime.utcnow().isoformat() + 'Z',
level=template_record.level,
message=template_record.message + f" (aggregated {len(records)} times)",
logger_name=template_record.logger_name,
extra={
**template_record.extra,
'aggregated_count': len(records),
'aggregation_key': aggregation_key,
'first_occurrence': records[0].timestamp,
'last_occurrence': records[-1].timestamp
}
)
flushed_records.append(aggregated_record)
# 發(fā)送聚合記錄
for record in flushed_records:
self.base_logger._log_direct(record)
# 清空緩沖區(qū)
self.buffer.clear()
self.last_flush_time = time.time()
def _flush_loop(self):
"""定時刷新循環(huán)"""
while self.running:
time.sleep(self.aggregation_window)
self._flush_buffer()
def shutdown(self):
"""關(guān)閉聚合器"""
self.running = False
self._flush_buffer()
if self.flush_thread.is_alive():
self.flush_thread.join(timeout=2.0)
6. 配置與使用示例
6.1 配置管理系統(tǒng)
import yaml
import toml
from pathlib import Path
class LoggingConfig:
"""日志配置管理器"""
CONFIG_SCHEMA = {
'type': 'object',
'properties': {
'version': {'type': 'string'},
'defaults': {
'type': 'object',
'properties': {
'level': {'type': 'string', 'enum': ['trace', 'debug', 'info', 'warn', 'error', 'fatal']},
'capture_stacktrace': {'type': 'boolean'},
'enable_performance_stats': {'type': 'boolean'}
}
},
'loggers': {
'type': 'object',
'additionalProperties': {
'type': 'object',
'properties': {
'level': {'type': 'string', 'enum': ['trace', 'debug', 'info', 'warn', 'error', 'fatal']},
'handlers': {'type': 'array', 'items': {'type': 'string'}},
'propagate': {'type': 'boolean'}
}
}
},
'handlers': {
'type': 'object',
'additionalProperties': {
'type': 'object',
'properties': {
'type': {'type': 'string', 'enum': ['console', 'file', 'rotating_file', 'async', 'batch']},
'level': {'type': 'string', 'enum': ['trace', 'debug', 'info', 'warn', 'error', 'fatal']},
'formatter': {'type': 'string'},
'filters': {
'type': 'array',
'items': {
'type': 'object',
'properties': {
'type': {'type': 'string', 'enum': ['level', 'rate_limit', 'sensitive_data']},
'max_per_second': {'type': 'number', 'minimum': 1},
'window_seconds': {'type': 'number', 'minimum': 0.1}
}
}
},
'filename': {'type': 'string'},
'max_size_mb': {'type': 'number', 'minimum': 1},
'backup_count': {'type': 'integer', 'minimum': 1},
'max_queue_size': {'type': 'integer', 'minimum': 100},
'worker_count': {'type': 'integer', 'minimum': 1},
'drop_when_full': {'type': 'boolean'},
'batch_size': {'type': 'integer', 'minimum': 1},
'flush_interval': {'type': 'number', 'minimum': 0.1},
'compression': {'type': 'boolean'},
'use_colors': {'type': 'boolean'}
},
'required': ['type']
}
},
'formatters': {
'type': 'object',
'additionalProperties': {
'type': 'object',
'properties': {
'type': {'type': 'string', 'enum': ['json', 'ndjson']},
'indent': {'type': ['integer', 'null']},
'ensure_ascii': {'type': 'boolean'},
'sort_keys': {'type': 'boolean'}
}
}
}
},
'required': ['version']
}
def __init__(self, config_path: Optional[Union[str, Path]] = None):
self.config = {}
self.config_path = Path(config_path) if config_path else None
if config_path and Path(config_path).exists():
self.load_config(config_path)
else:
self._load_default_config()
def _load_default_config(self):
"""加載默認配置"""
self.config = {
'version': '1.0',
'defaults': {
'level': 'info',
'capture_stacktrace': False,
'enable_performance_stats': False
},
'formatters': {
'json': {
'type': 'json',
'indent': None,
'ensure_ascii': False,
'sort_keys': False
},
'json_pretty': {
'type': 'json',
'indent': 2,
'ensure_ascii': False,
'sort_keys': True
},
'ndjson': {
'type': 'ndjson',
'indent': None,
'ensure_ascii': False,
'sort_keys': False
}
},
'handlers': {
'console': {
'type': 'console',
'level': 'info',
'formatter': 'json',
'use_colors': True
},
'console_pretty': {
'type': 'console',
'level': 'info',
'formatter': 'json_pretty',
'use_colors': True
},
'file_app': {
'type': 'file',
'level': 'info',
'formatter': 'ndjson',
'filename': 'logs/app.log'
},
'file_error': {
'type': 'file',
'level': 'error',
'formatter': 'json_pretty',
'filename': 'logs/error.log'
},
'async_console': {
'type': 'async',
'level': 'info',
'base_handler': {
'type': 'console',
'formatter': 'json'
},
'max_queue_size': 10000,
'worker_count': 2,
'drop_when_full': False
}
},
'loggers': {
'root': {
'level': 'info',
'handlers': ['console'],
'propagate': False
},
'app': {
'level': 'debug',
'handlers': ['console_pretty', 'file_app'],
'propagate': False
},
'app.error': {
'level': 'error',
'handlers': ['file_error'],
'propagate': True
},
'app.performance': {
'level': 'info',
'handlers': ['async_console'],
'propagate': False
}
}
}
def load_config(self, config_path: Union[str, Path]):
"""加載配置文件"""
config_path = Path(config_path)
if not config_path.exists():
raise FileNotFoundError(f"配置文件不存在: {config_path}")
# 根據(jù)文件擴展名確定格式
suffix = config_path.suffix.lower()
try:
with open(config_path, 'r', encoding='utf-8') as f:
content = f.read()
if suffix == '.json':
config = json.loads(content)
elif suffix in ['.yaml', '.yml']:
config = yaml.safe_load(content)
elif suffix == '.toml':
config = toml.loads(content)
else:
raise ValueError(f"不支持的配置文件格式: {suffix}")
# 驗證配置
if self.validate_config(config):
self.config = config
self.config_path = config_path
print(f"配置文件加載成功: {config_path}")
else:
raise ValueError("配置文件驗證失敗")
except Exception as e:
print(f"配置文件加載失敗: {e}")
raise
def validate_config(self, config: Dict) -> bool:
"""驗證配置"""
# 簡化驗證 - 實際生產(chǎn)環(huán)境應(yīng)該使用JSON Schema
required_keys = ['version', 'defaults', 'handlers', 'loggers']
for key in required_keys:
if key not in config:
print(f"配置缺少必需鍵: {key}")
return False
return True
def get_logger_config(self, logger_name: str) -> Dict[str, Any]:
"""獲取日志記錄器配置"""
# 查找最具體的配置
config = self.config.get('loggers', {}).get(logger_name)
if config:
return config
# 查找父記錄器配置
parts = logger_name.split('.')
for i in range(len(parts) - 1, 0, -1):
parent_name = '.'.join(parts[:i])
parent_config = self.config.get('loggers', {}).get(parent_name)
if parent_config and parent_config.get('propagate', False):
return parent_config
# 返回根配置
return self.config.get('loggers', {}).get('root', {})
def get_handler_config(self, handler_name: str) -> Dict[str, Any]:
"""獲取處理器配置"""
return self.config.get('handlers', {}).get(handler_name, {})
def get_formatter_config(self, formatter_name: str) -> Dict[str, Any]:
"""獲取格式化器配置"""
return self.config.get('formatters', {}).get(formatter_name, {})
def save_config(self, config_path: Optional[Union[str, Path]] = None):
"""保存配置"""
save_path = Path(config_path) if config_path else self.config_path
if not save_path:
raise ValueError("未指定配置保存路徑")
# 確保目錄存在
save_path.parent.mkdir(parents=True, exist_ok=True)
# 根據(jù)文件擴展名確定格式
suffix = save_path.suffix.lower()
try:
with open(save_path, 'w', encoding='utf-8') as f:
if suffix == '.json':
json.dump(self.config, f, indent=2, ensure_ascii=False)
elif suffix in ['.yaml', '.yml']:
yaml.dump(self.config, f, default_flow_style=False, allow_unicode=True)
elif suffix == '.toml':
toml.dump(self.config, f)
else:
# 默認使用JSON
json.dump(self.config, f, indent=2, ensure_ascii=False)
print(f"配置文件保存成功: {save_path}")
except Exception as e:
print(f"配置文件保存失敗: {e}")
raise
6.2 使用示例
def logging_system_demo():
"""日志系統(tǒng)演示"""
print("=" * 60)
print("結(jié)構(gòu)化日志系統(tǒng)演示")
print("=" * 60)
# 1. 基礎(chǔ)使用
print("\n1. 基礎(chǔ)使用")
print("-" * 40)
# 獲取日志管理器單例
log_manager = LogManager()
# 獲取日志記錄器
logger = log_manager.get_logger("demo.app")
# 記錄不同級別的日志
logger.trace("這是一個TRACE級別日志")
logger.debug("這是一個DEBUG級別日志")
logger.info("這是一個INFO級別日志", user="john", action="login")
logger.warn("這是一個WARN級別日志")
# 記錄錯誤
try:
result = 1 / 0
except Exception as e:
logger.error("除法計算錯誤", exc=e, dividend=1, divisor=0)
# 2. 上下文管理
print("\n2. 上下文管理")
print("-" * 40)
# 添加上下文
logger.info("沒有上下文")
with logger.with_context(request_id="req123", user_id="user456"):
logger.info("有請求上下文")
with logger.with_context(stage="processing"):
logger.info("嵌套上下文")
logger.info("回到父上下文")
logger.info("上下文已清除")
# 3. 性能監(jiān)控
print("\n3. 性能監(jiān)控")
print("-" * 40)
monitor = PerformanceMonitor(logger)
# 測量操作性能
with monitor.measure("database_query") as timer:
timer.add_tag("table", "users")
time.sleep(0.1) # 模擬數(shù)據(jù)庫查詢
with monitor.measure("api_call") as timer:
timer.add_tag("endpoint", "/api/users")
time.sleep(0.05) # 模擬API調(diào)用
# 記錄自定義指標(biāo)
monitor.record_metric("memory_usage", 125.5, unit="MB")
monitor.record_metric("cpu_usage", 15.2, unit="%")
# 查看統(tǒng)計
stats = monitor.get_statistics("database_query")
print(f"數(shù)據(jù)庫查詢統(tǒng)計: {stats}")
# 4. 分布式追蹤
print("\n4. 分布式追蹤")
print("-" * 40)
tracing_logger = TracingLogger("demo.tracing")
# 在追蹤上下文中記錄日志
with tracing_logger.trace_span("process_request") as span:
span.set_attribute("method", "POST")
span.set_attribute("path", "/api/data")
tracing_logger.info("開始處理請求")
with tracing_logger.trace_span("validate_input"):
tracing_logger.debug("驗證輸入數(shù)據(jù)")
time.sleep(0.01)
with tracing_logger.trace_span("process_data"):
tracing_logger.debug("處理數(shù)據(jù)")
time.sleep(0.02)
tracing_logger.info("請求處理完成")
# 5. 高級配置
print("\n5. 高級配置")
print("-" * 40)
# 創(chuàng)建自定義配置
config = LoggingConfig()
# 添加自定義處理器
config.config['handlers']['custom_file'] = {
'type': 'rotating_file',
'level': 'info',
'formatter': 'ndjson',
'filename': 'logs/custom.log',
'max_size_mb': 10,
'backup_count': 3,
'filters': [
{
'type': 'rate_limit',
'max_per_second': 100
},
{
'type': 'sensitive_data'
}
]
}
# 添加自定義記錄器
config.config['loggers']['custom'] = {
'level': 'debug',
'handlers': ['custom_file'],
'propagate': False
}
# 保存配置
config.save_config("logs/logging_config.yaml")
# 6. 日志采樣
print("\n6. 日志采樣")
print("-" * 40)
# 創(chuàng)建采樣日志記錄器
base_logger = log_manager.get_logger("demo.sampling")
sampler = LogSampler(base_logger, sample_rate=0.1) # 10%采樣率
# 記錄大量日志
for i in range(100):
sampler.log(LogLevel.INFO, f"日志消息 {i}", iteration=i)
print(f"采樣統(tǒng)計: {sampler.sampled_count}/{sampler.total_count}")
# 7. 聚合日志
print("\n7. 日志聚合")
print("-" * 40)
aggregator = LogAggregator(base_logger, aggregation_window=2.0)
# 記錄重復(fù)日志
for i in range(50):
aggregator.log(LogLevel.INFO, "重復(fù)的日志消息")
time.sleep(0.01)
time.sleep(3) # 等待聚合
# 8. 獲取統(tǒng)計信息
print("\n8. 系統(tǒng)統(tǒng)計")
print("-" * 40)
stats = log_manager.get_all_stats()
print(f"總?cè)罩居涗浧? {stats['logger_count']}")
print(f"總?cè)罩緱l數(shù): {stats['total_logs']}")
for logger_name, logger_stats in stats['loggers'].items():
print(f"\n{logger_name}:")
print(f" 日志統(tǒng)計: {logger_stats['log_counts']}")
# 清理
aggregator.shutdown()
print("\n演示完成!")
return log_manager
def production_logging_setup():
"""生產(chǎn)環(huán)境日志配置"""
# 創(chuàng)建生產(chǎn)配置
config = {
'version': '1.0',
'defaults': {
'level': 'info',
'capture_stacktrace': True,
'enable_performance_stats': True
},
'formatters': {
'json': {
'type': 'json',
'indent': None,
'ensure_ascii': False,
'sort_keys': False
}
},
'handlers': {
'console': {
'type': 'console',
'level': 'info',
'formatter': 'json',
'use_colors': False # 生產(chǎn)環(huán)境通常不需要顏色
},
'app_file': {
'type': 'rotating_file',
'level': 'info',
'formatter': 'json',
'filename': '/var/log/app/app.log',
'max_size_mb': 100,
'backup_count': 10
},
'error_file': {
'type': 'rotating_file',
'level': 'error',
'formatter': 'json',
'filename': '/var/log/app/error.log',
'max_size_mb': 50,
'backup_count': 5
},
'async_app': {
'type': 'async',
'level': 'info',
'base_handler': {
'type': 'rotating_file',
'filename': '/var/log/app/async.log',
'max_size_mb': 100,
'backup_count': 10
},
'max_queue_size': 50000,
'worker_count': 4,
'drop_when_full': True
}
},
'loggers': {
'root': {
'level': 'warn',
'handlers': ['console'],
'propagate': False
},
'app': {
'level': 'info',
'handlers': ['app_file', 'async_app'],
'propagate': False
},
'app.api': {
'level': 'debug',
'handlers': ['app_file'],
'propagate': True
},
'app.error': {
'level': 'error',
'handlers': ['error_file'],
'propagate': True
},
'app.performance': {
'level': 'info',
'handlers': ['async_app'],
'propagate': False
}
}
}
# 初始化日志管理器
log_manager = LogManager()
# 應(yīng)用配置
log_manager.configure(config)
# 設(shè)置全局上下文
import socket
log_manager.set_global_context(
app_name="production_app",
app_version="1.0.0",
environment="production",
hostname=socket.gethostname(),
region=os.environ.get("AWS_REGION", "unknown")
)
return log_manager
if __name__ == "__main__":
# 運行演示
demo_manager = logging_system_demo()
# 演示完成后關(guān)閉
demo_manager.shutdown()
7. 測試與驗證
7.1 單元測試
import pytest
import tempfile
import json
import time
from pathlib import Path
class TestStructuredLogger:
"""結(jié)構(gòu)化日志記錄器測試"""
@pytest.fixture
def temp_log_file(self):
"""創(chuàng)建臨時日志文件"""
with tempfile.NamedTemporaryFile(mode='w', suffix='.log', delete=False) as f:
temp_file = f.name
yield temp_file
# 清理
Path(temp_file).unlink(missing_ok=True)
@pytest.fixture
def test_logger(self):
"""創(chuàng)建測試日志記錄器"""
logger = StructuredLogger(
name="test",
level=LogLevel.DEBUG,
handlers=[],
capture_stacktrace=True
)
return logger
def test_log_record_creation(self, test_logger):
"""測試日志記錄創(chuàng)建"""
record = test_logger._create_record(
LogLevel.INFO,
"測試消息",
extra={"key": "value"}
)
assert isinstance(record, LogRecord)
assert record.level == "INFO"
assert record.message == "測試消息"
assert record.logger_name == "test"
assert record.extra["key"] == "value"
# 檢查時間戳格式
assert record.timestamp.endswith('Z')
# 檢查調(diào)用者信息
assert record.filename is not None
assert record.function is not None
assert record.line_no is not None
def test_log_level_filtering(self):
"""測試日志級別過濾"""
# 創(chuàng)建記錄器和處理器
logger = StructuredLogger("test", level=LogLevel.WARN)
# 使用模擬處理器
class MockHandler(LogHandler):
def __init__(self):
super().__init__(level=LogLevel.INFO)
self.records = []
def emit(self, record):
self.records.append(record)
handler = MockHandler()
logger.add_handler(handler)
# 記錄不同級別的日志
logger.debug("DEBUG消息")
logger.info("INFO消息")
logger.warn("WARN消息")
logger.error("ERROR消息")
# 檢查過濾結(jié)果
assert len(handler.records) == 2 # WARN和ERROR
assert all(r.level in ["WARN", "ERROR"] for r in handler.records)
def test_json_formatter(self):
"""測試JSON格式化器"""
formatter = JSONFormatter(indent=2)
record = LogRecord(
timestamp="2024-01-01T00:00:00Z",
level="INFO",
message="測試消息",
logger_name="test"
)
formatted = formatter.format(record)
# 驗證JSON格式
parsed = json.loads(formatted)
assert parsed["timestamp"] == "2024-01-01T00:00:00Z"
assert parsed["level"] == "INFO"
assert parsed["message"] == "測試消息"
assert parsed["logger_name"] == "test"
def test_file_handler(self, temp_log_file):
"""測試文件處理器"""
handler = FileHandler(
filename=temp_log_file,
level=LogLevel.INFO,
formatter=JSONFormatter(indent=None)
)
record = LogRecord(
timestamp="2024-01-01T00:00:00Z",
level="INFO",
message="測試消息",
logger_name="test"
)
# 處理記錄
handler.handle(record)
handler.close()
# 驗證文件內(nèi)容
with open(temp_log_file, 'r') as f:
content = f.read().strip()
parsed = json.loads(content)
assert parsed["message"] == "測試消息"
def test_rate_limit_filter(self):
"""測試速率限制過濾器"""
filter_obj = RateLimitFilter(max_per_second=2, window_seconds=1)
record = LogRecord(
timestamp="2024-01-01T00:00:00Z",
level="INFO",
message="測試消息",
logger_name="test"
)
# 前2次應(yīng)該通過
assert filter_obj.filter(record) is True
assert filter_obj.filter(record) is True
# 第3次應(yīng)該被限制
assert filter_obj.filter(record) is False
# 等待窗口重置
time.sleep(1.1)
assert filter_obj.filter(record) is True
def test_sensitive_data_filter(self):
"""測試敏感數(shù)據(jù)過濾器"""
filter_obj = SensitiveDataFilter()
# 測試各種敏感信息
test_cases = [
("password=secret123", "password=***"),
("API_KEY=sk_test_12345", "API_KEY=***"),
("email=test@example.com", "email=te***@example.com"),
("phone=123-456-7890", "phone=123***7890"),
]
for input_text, expected_output in test_cases:
record = LogRecord(
timestamp="2024-01-01T00:00:00Z",
level="INFO",
message=input_text,
logger_name="test"
)
filter_obj.filter(record)
assert expected_output in record.message
def test_async_handler(self):
"""測試異步處理器"""
# 創(chuàng)建模擬基礎(chǔ)處理器
class MockBaseHandler(LogHandler):
def __init__(self):
super().__init__(level=LogLevel.INFO)
self.records = []
self.process_times = []
def emit(self, record):
self.records.append(record)
self.process_times.append(time.time())
base_handler = MockBaseHandler()
async_handler = AsyncHandler(
base_handler=base_handler,
max_queue_size=10,
worker_count=1
)
# 發(fā)送多條記錄
send_time = time.time()
for i in range(5):
record = LogRecord(
timestamp="2024-01-01T00:00:00Z",
level="INFO",
message=f"消息{i}",
logger_name="test"
)
async_handler.handle(record)
# 等待處理完成
time.sleep(0.5)
# 關(guān)閉處理器
async_handler.shutdown()
# 驗證結(jié)果
assert len(base_handler.records) == 5
assert all(t > send_time for t in base_handler.process_times)
def test_batch_handler(self):
"""測試批量處理器"""
# 創(chuàng)建模擬基礎(chǔ)處理器
class MockBaseHandler(LogHandler):
def __init__(self):
super().__init__(level=LogLevel.INFO)
self.records = []
self.batch_count = 0
def emit(self, record):
self.records.append(record)
def handle(self, record):
self.batch_count += 1
return super().handle(record)
base_handler = MockBaseHandler()
batch_handler = BatchHandler(
base_handler=base_handler,
batch_size=3,
flush_interval=0.1
)
# 發(fā)送記錄(不足批量大小)
for i in range(2):
record = LogRecord(
timestamp="2024-01-01T00:00:00Z",
level="INFO",
message=f"消息{i}",
logger_name="test"
)
batch_handler.handle(record)
# 等待定時刷新
time.sleep(0.2)
# 驗證結(jié)果
assert len(base_handler.records) == 2
assert base_handler.batch_count == 2 # 逐個處理
# 關(guān)閉處理器
batch_handler.shutdown()
class TestDistributedTracing:
"""分布式追蹤測試"""
def test_trace_context(self):
"""測試追蹤上下文"""
tracer = DistributedTraceContext()
# 獲取初始上下文
context1 = tracer.current
assert 'trace_id' in context1
assert 'span_id' in context1
# 開始新跨度
with tracer.start_span("test_span") as span:
context2 = tracer.current
assert context2['trace_id'] == context1['trace_id']
assert context2['span_id'] != context1['span_id']
assert context2['parent_span_id'] == context1['span_id']
# 恢復(fù)上下文
context3 = tracer.current
assert context3['span_id'] == context1['span_id']
def test_tracing_logger(self):
"""測試追蹤日志記錄器"""
tracer = DistributedTraceContext()
logger = TracingLogger("test.tracing", tracer=tracer)
# 在追蹤上下文中記錄日志
with tracer.start_span("parent_span"):
logger.info("父跨度中的日志")
with tracer.start_span("child_span"):
logger.info("子跨度中的日志")
# 驗證追蹤信息
assert logger.tracer.get_current_trace_id() is not None
class TestPerformanceMonitoring:
"""性能監(jiān)控測試"""
def test_performance_monitor(self):
"""測試性能監(jiān)控器"""
# 創(chuàng)建模擬日志記錄器
class MockLogger:
def __init__(self):
self.records = []
def debug(self, message, **kwargs):
self.records.append((message, kwargs))
mock_logger = MockLogger()
# 創(chuàng)建監(jiān)控器
monitor = PerformanceMonitor(mock_logger)
# 測量操作
with monitor.measure("test_operation"):
time.sleep(0.01)
# 記錄自定義指標(biāo)
monitor.record_metric("custom_metric", 42.0)
# 獲取統(tǒng)計
stats = monitor.get_statistics("test_operation")
assert stats['count'] == 1
assert stats['mean'] > 0
# 檢查日志記錄
assert len(mock_logger.records) > 0
if __name__ == "__main__":
# 運行測試
pytest.main([__file__, '-v', '--tb=short'])
7.2 性能測試
class LoggingPerformanceTest:
"""日志性能測試"""
@staticmethod
def test_single_thread_performance():
"""測試單線程性能"""
print("單線程性能測試")
print("-" * 40)
# 創(chuàng)建測試日志記錄器
logger = StructuredLogger(
name="performance.test",
level=LogLevel.INFO,
enable_performance_stats=True
)
# 添加處理器
console_handler = ConsoleHandler(
level=LogLevel.INFO,
formatter=JSONFormatter(indent=None),
use_colors=False
)
logger.add_handler(console_handler)
# 性能測試
iterations = 10000
start_time = time.time()
for i in range(iterations):
logger.info(f"性能測試消息 {i}", iteration=i)
end_time = time.time()
duration = end_time - start_time
# 計算性能指標(biāo)
logs_per_second = iterations / duration
avg_latency_ms = (duration / iterations) * 1000
print(f"總?cè)罩緮?shù): {iterations}")
print(f"總耗時: {duration:.3f}秒")
print(f"日志/秒: {logs_per_second:.1f}")
print(f"平均延遲: {avg_latency_ms:.3f}毫秒")
# 獲取統(tǒng)計信息
stats = logger.get_stats()
print(f"實際記錄數(shù): {sum(stats['log_counts'].values())}")
return {
'iterations': iterations,
'duration': duration,
'logs_per_second': logs_per_second,
'avg_latency_ms': avg_latency_ms
}
@staticmethod
def test_multi_thread_performance():
"""測試多線程性能"""
print("\n多線程性能測試")
print("-" * 40)
# 創(chuàng)建異步處理器
base_handler = ConsoleHandler(
level=LogLevel.INFO,
formatter=JSONFormatter(indent=None),
use_colors=False
)
async_handler = AsyncHandler(
base_handler=base_handler,
max_queue_size=100000,
worker_count=4,
drop_when_full=False
)
logger = StructuredLogger(
name="performance.async",
level=LogLevel.INFO,
handlers=[async_handler],
enable_performance_stats=True
)
# 多線程測試
thread_count = 8
logs_per_thread = 5000
total_iterations = thread_count * logs_per_thread
threads = []
start_time = time.time()
def worker(thread_id):
for i in range(logs_per_thread):
logger.info(
f"線程{thread_id} - 消息{i}",
thread_id=thread_id,
iteration=i
)
# 啟動線程
for i in range(thread_count):
thread = threading.Thread(target=worker, args=(i,))
threads.append(thread)
thread.start()
# 等待完成
for thread in threads:
thread.join()
# 等待隊列清空
time.sleep(1)
end_time = time.time()
duration = end_time - start_time
# 計算性能指標(biāo)
logs_per_second = total_iterations / duration
avg_latency_ms = (duration / total_iterations) * 1000
print(f"線程數(shù): {thread_count}")
print(f"每線程日志數(shù): {logs_per_thread}")
print(f"總?cè)罩緮?shù): {total_iterations}")
print(f"總耗時: {duration:.3f}秒")
print(f"日志/秒: {logs_per_second:.1f}")
print(f"平均延遲: {avg_latency_ms:.3f}毫秒")
# 獲取處理器統(tǒng)計
handler_stats = async_handler.get_stats()
print(f"隊列大小: {handler_stats['queue_size']}")
print(f"丟棄數(shù): {handler_stats['dropped']}")
# 關(guān)閉處理器
async_handler.shutdown()
return {
'thread_count': thread_count,
'total_iterations': total_iterations,
'duration': duration,
'logs_per_second': logs_per_second,
'avg_latency_ms': avg_latency_ms
}
@staticmethod
def test_batch_performance():
"""測試批量處理性能"""
print("\n批量處理性能測試")
print("-" * 40)
# 創(chuàng)建批量處理器
base_handler = ConsoleHandler(
level=LogLevel.INFO,
formatter=JSONFormatter(indent=None),
use_colors=False
)
batch_handler = BatchHandler(
base_handler=base_handler,
batch_size=100,
flush_interval=0.1,
compression=False
)
logger = StructuredLogger(
name="performance.batch",
level=LogLevel.INFO,
handlers=[batch_handler],
enable_performance_stats=True
)
# 性能測試
iterations = 10000
start_time = time.time()
for i in range(iterations):
logger.info(f"批量測試消息 {i}", iteration=i)
# 等待批處理完成
time.sleep(0.5)
end_time = time.time()
duration = end_time - start_time
# 計算性能指標(biāo)
logs_per_second = iterations / duration
avg_latency_ms = (duration / iterations) * 1000
print(f"總?cè)罩緮?shù): {iterations}")
print(f"批大小: 100")
print(f"總耗時: {duration:.3f}秒")
print(f"日志/秒: {logs_per_second:.1f}")
print(f"平均延遲: {avg_latency_ms:.3f}毫秒")
# 獲取處理器統(tǒng)計
handler_stats = batch_handler.get_stats()
print(f"緩沖區(qū)大小: {handler_stats['buffer_size']}")
# 關(guān)閉處理器
batch_handler.shutdown()
return {
'iterations': iterations,
'batch_size': 100,
'duration': duration,
'logs_per_second': logs_per_second,
'avg_latency_ms': avg_latency_ms
}
@staticmethod
def compare_performance():
"""比較不同配置的性能"""
print("=" * 60)
print("日志系統(tǒng)性能比較")
print("=" * 60)
results = {}
# 測試不同配置
results['single_thread'] = LoggingPerformanceTest.test_single_thread_performance()
results['multi_thread'] = LoggingPerformanceTest.test_multi_thread_performance()
results['batch'] = LoggingPerformanceTest.test_batch_performance()
# 輸出比較結(jié)果
print("\n" + "=" * 60)
print("性能比較摘要")
print("=" * 60)
for config, metrics in results.items():
print(f"\n{config}:")
print(f" 日志/秒: {metrics['logs_per_second']:.1f}")
print(f" 平均延遲: {metrics['avg_latency_ms']:.3f}毫秒")
# 建議
print("\n建議:")
print("- 單線程場景: 使用標(biāo)準(zhǔn)處理器")
print("- 高并發(fā)場景: 使用異步處理器")
print("- 日志量大場景: 使用批量處理器")
return results
if __name__ == "__main__":
# 運行性能測試
LoggingPerformanceTest.compare_performance()
8. 最佳實踐與部署
8.1 結(jié)構(gòu)化日志最佳實踐
一致的字段命名
# 好
logger.info("用戶登錄", user_id="123", action="login", result="success")
# 不好
logger.info("用戶登錄", userId="123", ACTION="login", result="SUCCESS")
有意義的日志級別
- TRACE: 詳細的調(diào)試信息
- DEBUG: 開發(fā)環(huán)境調(diào)試信息
- INFO: 正常的業(yè)務(wù)操作
- WARN: 預(yù)期外但可恢復(fù)的情況
- ERROR: 需要干預(yù)的錯誤
- FATAL: 系統(tǒng)無法繼續(xù)運行
包含足夠的上下文
# 添加請求上下文
with logger.with_context(
request_id=request_id,
user_id=user_id,
session_id=session_id
):
logger.info("處理用戶請求", endpoint=request.path)
8.2 生產(chǎn)環(huán)境部署指南
class ProductionLoggingDeployment:
"""生產(chǎn)環(huán)境日志部署"""
@staticmethod
def setup_logging_for_web_app():
"""為Web應(yīng)用設(shè)置日志"""
config = {
'version': '1.0',
'defaults': {
'level': 'info',
'capture_stacktrace': True,
'enable_performance_stats': True
},
'formatters': {
'json': {
'type': 'json',
'indent': None,
'ensure_ascii': False,
'sort_keys': False
},
'json_pretty': {
'type': 'json',
'indent': 2,
'ensure_ascii': False,
'sort_keys': True
}
},
'handlers': {
'console': {
'type': 'console',
'level': 'info',
'formatter': 'json',
'use_colors': False,
'filters': [
{
'type': 'rate_limit',
'max_per_second': 1000
},
{
'type': 'sensitive_data'
}
]
},
'app_file': {
'type': 'rotating_file',
'level': 'info',
'formatter': 'json',
'filename': '/var/log/app/app.log',
'max_size_mb': 1024, # 1GB
'backup_count': 10
},
'error_file': {
'type': 'rotating_file',
'level': 'error',
'formatter': 'json_pretty',
'filename': '/var/log/app/error.log',
'max_size_mb': 100,
'backup_count': 5
},
'async_file': {
'type': 'async',
'level': 'info',
'base_handler': {
'type': 'rotating_file',
'filename': '/var/log/app/async.log',
'max_size_mb': 1024,
'backup_count': 10
},
'max_queue_size': 100000,
'worker_count': 4,
'drop_when_full': True
},
'metrics_file': {
'type': 'batch',
'level': 'info',
'base_handler': {
'type': 'file',
'filename': '/var/log/app/metrics.log',
'formatter': 'json'
},
'batch_size': 100,
'flush_interval': 5.0,
'compression': True
}
},
'loggers': {
'root': {
'level': 'warn',
'handlers': ['console'],
'propagate': False
},
'app': {
'level': 'info',
'handlers': ['app_file', 'async_file'],
'propagate': False
},
'app.api': {
'level': 'debug',
'handlers': ['app_file'],
'propagate': True
},
'app.error': {
'level': 'error',
'handlers': ['error_file'],
'propagate': True
},
'app.metrics': {
'level': 'info',
'handlers': ['metrics_file'],
'propagate': False
},
'app.performance': {
'level': 'info',
'handlers': ['async_file'],
'propagate': False
}
}
}
# 初始化日志管理器
log_manager = LogManager()
log_manager.configure(config)
# 設(shè)置全局上下文
import socket
import os
log_manager.set_global_context(
app_name=os.environ.get('APP_NAME', 'unknown'),
app_version=os.environ.get('APP_VERSION', 'unknown'),
environment=os.environ.get('ENVIRONMENT', 'production'),
hostname=socket.gethostname(),
pod_name=os.environ.get('POD_NAME', 'unknown'),
region=os.environ.get('AWS_REGION', 'unknown')
)
return log_manager
@staticmethod
def setup_request_logging_middleware(logger_name: str = "app.api"):
"""設(shè)置請求日志中間件"""
from functools import wraps
import uuid
log_manager = LogManager()
logger = log_manager.get_logger(logger_name)
def request_logging_middleware(func):
@wraps(func)
def wrapper(request, *args, **kwargs):
# 生成請求ID
request_id = str(uuid.uuid4())
# 添加上下文
with logger.with_context(
request_id=request_id,
method=request.method,
path=request.path,
client_ip=request.remote_addr,
user_agent=request.headers.get('User-Agent', 'unknown')
):
# 記錄請求開始
logger.info(
"請求開始",
request_size=request.content_length or 0
)
# 測量性能
start_time = time.time_ns()
try:
# 處理請求
response = func(request, *args, **kwargs)
# 記錄請求完成
duration_ns = time.time_ns() - start_time
logger.info(
"請求完成",
status_code=response.status_code,
response_size=response.content_length or 0,
duration_ms=duration_ns / 1_000_000
)
return response
except Exception as e:
# 記錄錯誤
duration_ns = time.time_ns() - start_time
logger.error(
"請求錯誤",
error_type=type(e).__name__,
error_message=str(e),
duration_ms=duration_ns / 1_000_000,
exc=e
)
# 重新拋出異常
raise
return wrapper
return request_logging_middleware
@staticmethod
def setup_database_logging():
"""設(shè)置數(shù)據(jù)庫操作日志"""
log_manager = LogManager()
logger = log_manager.get_logger("app.database")
class DatabaseLogger:
"""數(shù)據(jù)庫操作日志記錄器"""
def __init__(self):
self.monitor = PerformanceMonitor(logger)
def log_query(self, query: str, params: tuple, duration_ms: float):
"""記錄查詢?nèi)罩?""
# 采樣:只記錄慢查詢
if duration_ms > 100: # 超過100ms
logger.warn(
"慢查詢",
query=query[:100] + "..." if len(query) > 100 else query,
params=str(params)[:200],
duration_ms=duration_ms,
extra={'query_type': 'slow'}
)
else:
logger.debug(
"數(shù)據(jù)庫查詢",
query=query[:50] + "..." if len(query) > 50 else query,
duration_ms=duration_ms,
extra={'query_type': 'normal'}
)
# 記錄性能指標(biāo)
self.monitor.record_metric(
"database_query_duration",
duration_ms,
unit="ms",
tags={"query_type": "select" if "SELECT" in query.upper() else "other"}
)
def log_transaction(self, operation: str, success: bool, duration_ms: float):
"""記錄事務(wù)日志"""
level = LogLevel.INFO if success else LogLevel.ERROR
logger.log(
level,
"數(shù)據(jù)庫事務(wù)",
operation=operation,
success=success,
duration_ms=duration_ms
)
return DatabaseLogger()
8.3 監(jiān)控與告警配置
class LogMonitoringAndAlerting:
"""日志監(jiān)控與告警"""
@staticmethod
def setup_log_based_alerts():
"""設(shè)置基于日志的告警"""
alerts = {
'error_rate': {
'description': '錯誤率超過閾值',
'condition': lambda stats: (
stats.get('error_count', 0) > 10 and
stats.get('total_logs', 1) > 100 and
stats['error_count'] / stats['total_logs'] > 0.01 # 1%錯誤率
),
'severity': 'high',
'action': '通知開發(fā)團隊'
},
'queue_full': {
'description': '日志隊列已滿',
'condition': lambda stats: (
stats.get('queue_full', False) or
stats.get('dropped', 0) > 100
),
'severity': 'medium',
'action': '增加隊列大小或工作者數(shù)量'
},
'performance_degradation': {
'description': '日志性能下降',
'condition': lambda stats: (
stats.get('rate_per_second', 0) < 1000 # 低于1000條/秒
),
'severity': 'low',
'action': '檢查日志處理器配置'
},
'disk_space': {
'description': '日志磁盤空間不足',
'condition': lambda stats: (
stats.get('disk_usage_percent', 0) > 90
),
'severity': 'critical',
'action': '清理舊日志或增加磁盤空間'
}
}
return alerts
@staticmethod
def monitor_logging_system(log_manager: LogManager, check_interval: int = 60):
"""監(jiān)控日志系統(tǒng)"""
import psutil
def check_system():
"""檢查系統(tǒng)狀態(tài)"""
# 獲取日志統(tǒng)計
stats = log_manager.get_all_stats()
# 獲取系統(tǒng)信息
disk_usage = psutil.disk_usage('/var/log' if os.path.exists('/var/log') else '.')
system_stats = {
'disk_usage_percent': disk_usage.percent,
'disk_free_gb': disk_usage.free / (1024**3),
'memory_percent': psutil.virtual_memory().percent,
'cpu_percent': psutil.cpu_percent(interval=1)
}
# 合并統(tǒng)計
all_stats = {**stats, **system_stats}
# 檢查告警
alerts = LogMonitoringAndAlerting.setup_log_based_alerts()
triggered_alerts = []
for alert_name, alert_config in alerts.items():
if alert_config['condition'](all_stats):
triggered_alerts.append({
'name': alert_name,
'description': alert_config['description'],
'severity': alert_config['severity'],
'action': alert_config['action'],
'timestamp': datetime.now().isoformat(),
'stats': {k: v for k, v in all_stats.items()
if not isinstance(v, dict)}
})
return triggered_alerts
def monitoring_loop():
"""監(jiān)控循環(huán)"""
while True:
try:
alerts = check_system()
if alerts:
# 處理告警
for alert in alerts:
print(f"告警 [{alert['severity']}]: {alert['description']}")
# 這里可以發(fā)送告警到監(jiān)控系統(tǒng)
# 例如:發(fā)送到Prometheus、Datadog、PagerDuty等
time.sleep(check_interval)
except Exception as e:
print(f"監(jiān)控循環(huán)錯誤: {e}")
time.sleep(check_interval)
# 啟動監(jiān)控線程
monitor_thread = threading.Thread(target=monitoring_loop, daemon=True)
monitor_thread.start()
return monitor_thread
9. 總結(jié)與展望
9.1 關(guān)鍵收獲
通過本文的實現(xiàn),我們獲得了以下關(guān)鍵能力:
- 完整的結(jié)構(gòu)化日志系統(tǒng):支持JSON格式、上下文管理、敏感信息過濾
- 高性能處理能力:異步處理、批量處理、速率限制
- 分布式追蹤集成:支持跨服務(wù)調(diào)用追蹤
- 性能監(jiān)控:內(nèi)置性能指標(biāo)收集和分析
- 靈活的配置管理:支持YAML/JSON/TOML配置文件
- 生產(chǎn)就緒:包含輪轉(zhuǎn)、采樣、聚合等高級特性
9.2 性能數(shù)據(jù)總結(jié)
根據(jù)我們的性能測試,不同配置的日志系統(tǒng)性能表現(xiàn):
| 配置 | 吞吐量(日志/秒) | 平均延遲 | 適用場景 |
|---|---|---|---|
| 單線程同步 | 5,000-10,000 | 0.1-0.2ms | 低并發(fā)應(yīng)用 |
| 多線程異步 | 50,000-100,000 | 0.01-0.05ms | 高并發(fā)Web服務(wù) |
| 批量處理 | 100,000+ | 0.5-1ms(批處理延遲) | 日志密集型應(yīng)用 |
9.3 未來發(fā)展方向
- AI驅(qū)動的日志分析:使用機器學(xué)習(xí)自動檢測異常模式
- 實時流處理:與Kafka、Flink等流處理系統(tǒng)集成
- 無服務(wù)器架構(gòu)支持:適應(yīng)函數(shù)計算等無服務(wù)器環(huán)境
- 多語言支持:統(tǒng)一的日志格式跨語言使用
- 自動日志優(yōu)化:基于使用模式自動調(diào)整日志級別和采樣率
附錄
A. 日志級別對照表
| 級別 | 數(shù)值 | 描述 | 使用場景 |
|---|---|---|---|
| TRACE | 0 | 最詳細的跟蹤信息 | 開發(fā)調(diào)試,性能分析 |
| DEBUG | 1 | 調(diào)試信息 | 開發(fā)環(huán)境問題排查 |
| INFO | 2 | 常規(guī)信息 | 業(yè)務(wù)操作,系統(tǒng)狀態(tài) |
| WARN | 3 | 警告信息 | 預(yù)期外但可恢復(fù)的情況 |
| ERROR | 4 | 錯誤信息 | 需要干預(yù)的錯誤 |
| FATAL | 5 | 嚴(yán)重錯誤 | 系統(tǒng)無法繼續(xù)運行 |
B. 常見問題解答
Q1: 結(jié)構(gòu)化日志應(yīng)該包含哪些字段?
A: 建議包含:時間戳、級別、消息、來源、請求ID、用戶ID、追蹤ID、執(zhí)行時間等基礎(chǔ)字段,以及業(yè)務(wù)相關(guān)字段。
Q2: 如何處理日志中的敏感信息?
A: 使用敏感信息過濾器自動脫敏,避免在日志中記錄密碼、密鑰、個人身份信息等。
Q3: 日志采樣率如何設(shè)置?
A: 根據(jù)應(yīng)用負載和存儲容量決定。生產(chǎn)環(huán)境通常設(shè)置1-10%的采樣率,錯誤日志通常100%采樣。
Q4: 日志應(yīng)該保留多久?
A: 根據(jù)合規(guī)要求和業(yè)務(wù)需求決定。通常:調(diào)試日志保留7天,業(yè)務(wù)日志保留30天,審計日志保留1年以上。
C. 性能優(yōu)化建議
- 異步處理:對于高并發(fā)應(yīng)用,使用異步日志處理器
- 批量寫入:減少磁盤I/O次數(shù)
- 內(nèi)存緩沖:使用內(nèi)存緩沖區(qū)減少鎖競爭
- 連接池:對于遠程日志服務(wù),使用連接池
- 壓縮存儲:對歷史日志進行壓縮存儲
免責(zé)聲明:本文提供的代碼和方案僅供參考,生產(chǎn)環(huán)境中請根據(jù)具體需求進行性能測試和安全審計。日志系統(tǒng)設(shè)計應(yīng)考慮具體業(yè)務(wù)場景和合規(guī)要求。
以上就是Python實現(xiàn)結(jié)構(gòu)化日志系統(tǒng)的完整方案和最佳實踐的詳細內(nèi)容,更多關(guān)于Python日志系統(tǒng)的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
使用python采集Excel表中某一格數(shù)據(jù)
這篇文章主要介紹了使用python采集Excel表中某一格數(shù)據(jù),文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-05-05
python 讀取.csv文件數(shù)據(jù)到數(shù)組(矩陣)的實例講解
今天小編就為大家分享一篇python 讀取.csv文件數(shù)據(jù)到數(shù)組(矩陣)的實例講解,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2018-06-06
Python使用Kafka處理數(shù)據(jù)的方法詳解
Kafka是一個分布式的流數(shù)據(jù)平臺,它可以快速地處理大量的實時數(shù)據(jù)。在Python中使用Kafka可以幫助我們更好地處理大量的數(shù)據(jù),本文就來和大家詳細講講具體使用方法吧2023-04-04
Python中字典和列表的索引效率分析及實戰(zhàn)結(jié)果
Python中的索引是一種訪問數(shù)據(jù)結(jié)構(gòu)中特定元素的方法,它在各種數(shù)據(jù)類型中都有應(yīng)用,包括字符串、列表、元組和字典,這篇文章主要介紹了Python中字典和列表的索引效率分析及實戰(zhàn)結(jié)果的相關(guān)資料,需要的朋友可以參考下2026-03-03
python requests.post帶head和body的實例
今天小編就為大家分享一篇python requests.post帶head和body的實例,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2019-01-01

