Coverage for src / mysingle / core / logging.py: 0%

1""" 

2통합 로깅 시스템 (Structured + Traditional Logging) 

3 

4이 모듈은 기존 logging_config.py와 structured_logging.py를 통합하여 

5다음 기능을 제공합니다: 

6 

71. 구조화된 로깅 (structlog 기반) 

8 - Correlation ID, User ID, Request ID 컨텍스트 변수 

9 - JSON 출력 지원 

10 - 서비스명 자동 태깅 

11 - 편의 함수들 (log_user_action, log_service_call, log_database_operation) 

12 

132. 전통적인 로깅 (logging 기반) 

14 - 컬러 출력 (colorlog) 

15 - 파일 로깅 (app.log, error.log) 

16 - 외부 라이브러리 로그 레벨 조정 

17 

183. 통합 설정 

19 - 환경별 설정 (development, production) 

20 - 자동 서비스명 감지 

21 - FastAPI 미들웨어 통합 지원 

22""" 

23 

24import logging 

25import sys 

26from contextvars import ContextVar 

27from pathlib import Path 

28from typing import Optional 

29 

30import structlog 

31 

32try: 

33 import colorlog 

34 

35 HAS_COLORLOG = True 

36except ImportError: 

37 HAS_COLORLOG = False 

38 

39 

40# Correlation ID, User ID, Request ID 컨텍스트 변수 

41correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="") 

42user_id_var: ContextVar[str] = ContextVar("user_id", default="") 

43request_id_var: ContextVar[str] = ContextVar("request_id", default="") 

44 

45 

46# ============================================================================= 

47# 구조화된 로깅 시스템 (structlog) 

48# ============================================================================= 

49 

50 

51class CorrelationIdProcessor: 

52 """Correlation ID를 로그에 추가하는 프로세서""" 

53 

54 def __call__(self, logger, method_name, event_dict): 

55 correlation_id = correlation_id_var.get() 

56 if correlation_id: 

57 # 로그 메시지에 correlation ID 프리픽스 추가 

58 event_dict["event"] = ( 

59 f"[{correlation_id[:8]}] {event_dict.get('event', '')}" 

60 ) 

61 return event_dict 

62 

63 

64class ServiceNameProcessor: 

65 """서비스명을 로그에 추가하는 프로세서""" 

66 

67 def __init__(self, service_name: str): 

68 self.service_name = service_name 

69 

70 def __call__(self, logger, method_name, event_dict): 

71 event_dict["service"] = self.service_name 

72 return event_dict 

73 

74 

75class UserContextProcessor: 

76 """User ID와 Request ID를 로그에 추가하는 프로세서""" 

77 

78 def __call__(self, logger, method_name, event_dict): 

79 user_id = user_id_var.get() 

80 request_id = request_id_var.get() 

81 

82 if user_id: 

83 event_dict["user_id"] = user_id 

84 

85 if request_id: 

86 event_dict["request_id"] = request_id 

87 

88 return event_dict 

89 

90 

91def configure_structured_logging( 

92 service_name: str, 

93 log_level: str = "INFO", 

94 enable_json: bool = False, 

95 enable_correlation_id: bool = True, 

96 enable_user_context: bool = True, 

97): 

98 """ 

99 구조화된 로깅 설정 

100 

101 Args: 

102 service_name: 서비스명 

103 log_level: 로그 레벨 (DEBUG, INFO, WARNING, ERROR, CRITICAL) 

104 enable_json: JSON 형식 출력 활성화 

105 enable_correlation_id: Correlation ID 추가 활성화 

106 enable_user_context: User/Request ID 컨텍스트 추가 활성화 

107 """ 

108 processors = [ 

109 structlog.processors.TimeStamper(fmt="ISO"), 

110 structlog.processors.add_log_level, 

111 ServiceNameProcessor(service_name), 

112 ] 

113 

114 if enable_correlation_id: 

115 processors.append(CorrelationIdProcessor()) 

116 

117 if enable_user_context: 

118 processors.append(UserContextProcessor()) 

119 

120 if enable_json: 

121 processors.append(structlog.processors.JSONRenderer()) 

122 else: 

123 processors.append(structlog.dev.ConsoleRenderer()) 

124 

125 structlog.configure( 

126 processors=processors, 

127 wrapper_class=structlog.make_filtering_bound_logger( 

128 getattr(logging, log_level.upper()) 

129 ), 

130 context_class=dict, 

131 logger_factory=structlog.WriteLoggerFactory(), 

132 cache_logger_on_first_use=True, 

133 ) 

134 

135 logger = get_structured_logger(__name__) 

136 logger.info( 

137 "Structured logging configured", 

138 extra={ 

139 "json_logging": enable_json, 

140 "correlation_id_enabled": enable_correlation_id, 

141 }, 

142 ) 

143 

144 

145def get_structured_logger(name: str): 

146 """구조화된 로거 인스턴스 획득""" 

147 return structlog.get_logger(name) 

148 

149 

150# ============================================================================= 

151# 컨텍스트 변수 관리 

152# ============================================================================= 

153 

154 

155def set_correlation_id(correlation_id: str): 

156 """Correlation ID 설정""" 

157 correlation_id_var.set(correlation_id) 

158 

159 

160def set_user_id(user_id: str): 

161 """User ID 설정""" 

162 user_id_var.set(user_id) 

163 

164 

165def set_request_id(request_id: str): 

166 """Request ID 설정""" 

167 request_id_var.set(request_id) 

168 

169 

170def get_correlation_id() -> str: 

171 """현재 Correlation ID 획득""" 

172 return correlation_id_var.get() 

173 

174 

175def get_user_id() -> str: 

176 """현재 User ID 획득""" 

177 return user_id_var.get() 

178 

179 

180def get_request_id() -> str: 

181 """현재 Request ID 획득""" 

182 return request_id_var.get() 

183 

184 

185def clear_logging_context(): 

186 """로깅 컨텍스트 초기화""" 

187 correlation_id_var.set("") 

188 user_id_var.set("") 

189 request_id_var.set("") 

190 

191 

192# ============================================================================= 

193# 편의 함수들 (구조화된 로깅) 

194# ============================================================================= 

195 

196 

197def log_user_action( 

198 action: str, 

199 resource_type: str, 

200 resource_id: Optional[str] = None, 

201 details: Optional[dict] = None, 

202 success: bool = True, 

203 error: Optional[str] = None, 

204): 

205 """사용자 액션 로깅""" 

206 logger = get_structured_logger(__name__) 

207 

208 log_data = { 

209 "action": action, 

210 "resource_type": resource_type, 

211 "success": success, 

212 } 

213 

214 if resource_id: 

215 log_data["resource_id"] = resource_id 

216 

217 if details: 

218 log_data["details"] = details 

219 

220 if error: 

221 log_data["error"] = error 

222 logger.error("User action failed", extra=log_data) 

223 else: 

224 logger.info("User action completed", extra=log_data) 

225 

226 

227def log_service_call( 

228 service_name: str, 

229 method: str, 

230 endpoint: str, 

231 duration: float, 

232 status_code: Optional[int] = None, 

233 error: Optional[str] = None, 

234): 

235 """서비스 호출 로깅""" 

236 logger = get_structured_logger(__name__) 

237 

238 log_data = { 

239 "target_service": service_name, 

240 "http_method": method, 

241 "endpoint": endpoint, 

242 "duration_ms": round(duration * 1000, 2), 

243 } 

244 

245 if status_code: 

246 log_data["status_code"] = status_code 

247 

248 if error: 

249 log_data["error"] = error 

250 logger.error("Service call failed", extra=log_data) 

251 else: 

252 logger.info("Service call completed", extra=log_data) 

253 

254 

255def log_database_operation( 

256 operation: str, 

257 collection: str, 

258 duration: float, 

259 document_count: Optional[int] = None, 

260 error: Optional[str] = None, 

261): 

262 """데이터베이스 작업 로깅""" 

263 logger = get_structured_logger(__name__) 

264 

265 log_data = { 

266 "operation": operation, 

267 "collection": collection, 

268 "duration_ms": round(duration * 1000, 2), 

269 } 

270 

271 if document_count is not None: 

272 log_data["document_count"] = document_count 

273 

274 if error: 

275 log_data["error"] = error 

276 logger.error("Database operation failed", extra=log_data) 

277 else: 

278 logger.info("Database operation completed", extra=log_data) 

279 

280 

281# ============================================================================= 

282# 전통적인 로깅 시스템 (기존 logging_config.py 통합) 

283# ============================================================================= 

284 

285 

286def setup_traditional_logging(): 

287 """전통적인 파일/콘솔 로깅 설정""" 

288 

289 # 로그 디렉토리 생성 

290 log_dir = Path("logs") 

291 log_dir.mkdir(exist_ok=True) 

292 

293 # 루트 로거 설정 

294 root_logger = logging.getLogger() 

295 root_logger.setLevel(logging.INFO) 

296 

297 # 기존 핸들러 제거 

298 for handler in root_logger.handlers[:]: 

299 root_logger.removeHandler(handler) 

300 

301 # 컬러 로그 포맷 (colorlog 사용) 

302 if HAS_COLORLOG: 

303 color_format = ( 

304 "%(log_color)s%(asctime)s%(reset)s | " 

305 "%(log_color)s%(levelname)-8s%(reset)s | " 

306 "%(cyan)s%(name)-30s%(reset)s | " 

307 "%(message_log_color)s%(message)s%(reset)s" 

308 ) 

309 date_format = "%H:%M:%S" 

310 

311 console_formatter = colorlog.ColoredFormatter( 

312 color_format, 

313 datefmt=date_format, 

314 log_colors={ 

315 "DEBUG": "blue", 

316 "INFO": "green", 

317 "WARNING": "yellow", 

318 "ERROR": "red", 

319 "CRITICAL": "red,bg_white", 

320 }, 

321 secondary_log_colors={ 

322 "message": { 

323 "DEBUG": "white", 

324 "INFO": "white", 

325 "WARNING": "yellow", 

326 "ERROR": "red", 

327 "CRITICAL": "red", 

328 } 

329 }, 

330 ) 

331 else: 

332 # colorlog 없을 때 기본 포맷 

333 log_format = "%(asctime)s | %(levelname)-8s | %(name)-30s | %(message)s" 

334 date_format = "%H:%M:%S" 

335 console_formatter = logging.Formatter(log_format, datefmt=date_format) 

336 

337 # 콘솔 핸들러 

338 console_handler = logging.StreamHandler(sys.stdout) 

339 console_handler.setLevel(logging.INFO) 

340 console_handler.setFormatter(console_formatter) 

341 root_logger.addHandler(console_handler) 

342 

343 # 파일 핸들러 (일반 로그) - 컬러 없이 

344 file_format = "%(asctime)s | %(levelname)-8s | %(name)-30s | %(message)s" 

345 file_date_format = "%Y-%m-%d %H:%M:%S" 

346 

347 file_handler = logging.FileHandler(log_dir / "app.log", encoding="utf-8") 

348 file_handler.setLevel(logging.INFO) 

349 file_formatter = logging.Formatter(file_format, datefmt=file_date_format) 

350 file_handler.setFormatter(file_formatter) 

351 root_logger.addHandler(file_handler) 

352 

353 # 에러 로그 파일 핸들러 

354 error_handler = logging.FileHandler(log_dir / "error.log", encoding="utf-8") 

355 error_handler.setLevel(logging.ERROR) 

356 error_formatter = logging.Formatter(file_format, datefmt=file_date_format) 

357 error_handler.setFormatter(error_formatter) 

358 root_logger.addHandler(error_handler) 

359 

360 # 외부 라이브러리 로그 레벨 조정 

361 _configure_external_loggers() 

362 

363 

364def _configure_external_loggers(): 

365 """외부 라이브러리 로거 설정""" 

366 external_loggers = { 

367 "uvicorn.access": logging.WARNING, 

368 "uvicorn.error": logging.INFO, 

369 "httpx": logging.WARNING, 

370 "watchfiles": logging.WARNING, 

371 "watchfiles.main": logging.WARNING, 

372 "pymongo": logging.WARNING, 

373 "pymongo.serverSelection": logging.WARNING, 

374 "pymongo.connection": logging.WARNING, 

375 "pymongo.command": logging.WARNING, 

376 "pymongo.topology": logging.WARNING, 

377 # Email 및 User 관리 로거 (서비스별) 

378 "app.utils.email": logging.INFO, 

379 "app.services.user_manager": logging.INFO, 

380 } 

381 

382 for logger_name, level in external_loggers.items(): 

383 logging.getLogger(logger_name).setLevel(level) 

384 

385 

386# ============================================================================= 

387# 통합 로깅 설정 (권장) 

388# ============================================================================= 

389 

390 

391def setup_logging( 

392 service_name: str = "unknown-service", 

393 log_level: str = "INFO", 

394 environment: str = "development", 

395 enable_structured: bool = True, 

396 enable_traditional: bool = True, 

397 enable_json: bool = False, 

398): 

399 """ 

400 통합 로깅 시스템 설정 (구조화된 + 전통적인 로깅) 

401 

402 Args: 

403 service_name: 서비스명 

404 log_level: 로그 레벨 

405 environment: 환경 (development, production) 

406 enable_structured: 구조화된 로깅 활성화 

407 enable_traditional: 전통적인 로깅 활성화 

408 enable_json: JSON 출력 활성화 (production 권장) 

409 """ 

410 # 환경별 기본 설정 

411 if environment == "production": 

412 enable_json = True 

413 log_level = "INFO" 

414 elif environment == "development": 

415 enable_json = False 

416 log_level = "DEBUG" 

417 

418 # 전통적인 로깅 설정 (파일, 콘솔) 

419 if enable_traditional: 

420 setup_traditional_logging() 

421 

422 # 구조화된 로깅 설정 

423 if enable_structured: 

424 configure_structured_logging( 

425 service_name=service_name, 

426 log_level=log_level, 

427 enable_json=enable_json, 

428 ) 

429 

430 # 설정 완료 로그 

431 logger = get_structured_logger(__name__) 

432 logger.info(f"✅ Integrated logging configured for {service_name}") 

433 logger.info(f"Environment: {environment}, Level: {log_level}, JSON: {enable_json}") 

434 

435 

436# ============================================================================= 

437# 레거시 호환성 함수 

438# ============================================================================= 

439 

440 

441def setup_logging_legacy(): 

442 """기존 setup_logging 함수 (호환성 유지)""" 

443 setup_traditional_logging() 

444 

445 

446# 기본 설정 함수 (서비스명 없이 호출하는 경우) 

447def configure_logging_for_service(service_name: str): 

448 """서비스별 로깅 설정 (간단한 인터페이스)""" 

449 setup_logging(service_name=service_name)