Coverage for src / mysingle / core / metrics / middleware.py: 0%

1"""Enhanced metrics middleware with performance optimizations.""" 

2 

3import time 

4from typing import Any 

5 

6from fastapi import Request, Response 

7from starlette.middleware.base import BaseHTTPMiddleware 

8 

9from ..logging import get_structured_logger 

10from .collector import MetricsCollector, MetricsConfig 

11 

12logger = get_structured_logger(__name__) 

13 

14# Global metrics collector 

15_metrics_collector: MetricsCollector | None = None 

16 

17 

18def get_metrics_collector() -> MetricsCollector: 

19 """Get the global metrics collector.""" 

20 global _metrics_collector 

21 if _metrics_collector is None: 

22 raise RuntimeError("Metrics collector not initialized.") 

23 return _metrics_collector 

24 

25 

26class MetricsMiddleware(BaseHTTPMiddleware): 

27 """Enhanced middleware to collect HTTP request metrics with performance optimizations.""" 

28 

29 def __init__( 

30 self, 

31 app: Any, 

32 collector: MetricsCollector, 

33 exclude_paths: set[str] | None = None, 

34 include_response_headers: bool = True, 

35 track_user_agents: bool = False, 

36 ) -> None: 

37 super().__init__(app) 

38 self.collector = collector 

39 # 성능을 위해 메트릭에서 제외할 경로들 (health check 등) 

40 self.exclude_paths = exclude_paths or { 

41 "/health", 

42 "/metrics", 

43 "/docs", 

44 "/redoc", 

45 "/openapi.json", 

46 } 

47 self.include_response_headers = include_response_headers 

48 self.track_user_agents = track_user_agents 

49 

50 def _should_track_request(self, request: Request) -> bool: 

51 """Determine if request should be tracked.""" 

52 path = request.url.path 

53 

54 # 제외 경로 확인 

55 if path in self.exclude_paths: 

56 return False 

57 

58 # 정적 파일 제외 (성능 최적화) 

59 return not path.startswith(("/static/", "/assets/", "/favicon")) 

60 

61 def _extract_route_pattern(self, request: Request) -> str: 

62 """Extract normalized route pattern from request.""" 

63 try: 

64 # FastAPI route pattern 추출 

65 if hasattr(request, "scope") and "route" in request.scope: 

66 route = request.scope["route"] 

67 if hasattr(route, "path"): 

68 return str(route.path) 

69 

70 # 경로에서 ID 패턴 정규화 (성능 최적화) 

71 path = request.url.path 

72 

73 # UUID 패턴 정규화 

74 import re 

75 

76 uuid_pattern = ( 

77 r"/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}" 

78 ) 

79 path = re.sub(uuid_pattern, "/{uuid}", path, flags=re.IGNORECASE) 

80 

81 # 숫자 ID 패턴 정규화 

82 numeric_pattern = r"/\d+" 

83 path = re.sub(numeric_pattern, "/{id}", path) 

84 

85 return path 

86 

87 except Exception as e: 

88 logger.debug(f"Error extracting route pattern: {e}") 

89 return request.url.path 

90 

91 async def dispatch(self, request: Request, call_next: Any) -> Response: 

92 """Process request and collect metrics with enhanced performance.""" 

93 # 성능 최적화: 추적하지 않을 요청은 빠르게 처리 

94 if not self._should_track_request(request): 

95 early_response: Response = await call_next(request) 

96 return early_response 

97 

98 start_time = time.time() 

99 

100 # 경로 패턴 추출 

101 route_path = self._extract_route_pattern(request) 

102 

103 try: 

104 # 요청 처리 

105 response: Response = await call_next(request) 

106 except Exception as e: 

107 # 예외 발생 시에도 메트릭 기록 

108 duration = time.time() - start_time 

109 self.collector.record_request_sync( 

110 method=request.method, 

111 path=route_path, 

112 status_code=500, 

113 duration=duration, 

114 ) 

115 logger.error(f"Error processing request {request.method} {route_path}: {e}") 

116 raise 

117 

118 # 지속 시간 계산 

119 duration = time.time() - start_time 

120 

121 # 메트릭 기록 (비동기로 처리하여 응답 지연 최소화) 

122 try: 

123 self.collector.record_request_sync( 

124 method=request.method, 

125 path=route_path, 

126 status_code=response.status_code, 

127 duration=duration, 

128 ) 

129 except Exception as e: 

130 logger.warning(f"Error recording metrics: {e}") 

131 

132 # 응답 헤더 추가 (선택적) 

133 if self.include_response_headers: 

134 response.headers["X-Response-Time"] = f"{duration:.4f}s" 

135 response.headers["X-Service-Name"] = self.collector.service_name 

136 

137 return response 

138 

139 

140def create_metrics_middleware( 

141 service_name: str, 

142 config: MetricsConfig | None = None, 

143 exclude_paths: set[str] | None = None, 

144) -> None: 

145 """Create and configure metrics middleware for the given service. 

146 

147 Args: 

148 service_name: Name of the service 

149 config: Metrics configuration 

150 exclude_paths: Paths to exclude from metrics collection 

151 

152 Returns: 

153 None (sets up global collector) 

154 """ 

155 global _metrics_collector 

156 

157 try: 

158 # 설정 기본값 

159 metrics_config = config or MetricsConfig() 

160 

161 # 메트릭 컬렉터 초기화 

162 _metrics_collector = MetricsCollector(service_name, metrics_config) 

163 

164 logger.info(f"✅ Metrics collector initialized for {service_name}") 

165 logger.debug(f"Metrics config: {metrics_config}") 

166 

167 except Exception as e: 

168 logger.error(f"❌ Failed to create metrics middleware for {service_name}: {e}") 

169 raise