Coverage for src/qdrant_loader_mcp_server/utils/logging.py: 100%

1"""Centralized logging configuration for the MCP Server application.""" 

2 

3import logging 

4import os 

5import re 

6import sys 

7from pathlib import Path 

8 

9import structlog 

10 

11 

12class QdrantVersionFilter(logging.Filter): 

13 """Filter to suppress Qdrant version check warnings.""" 

14 

15 def filter(self, record): 

16 return "Failed to obtain server version" not in str(record.msg) 

17 

18 

19class ApplicationFilter(logging.Filter): 

20 """Filter to only show logs from our application.""" 

21 

22 def filter(self, record): 

23 # Show logs from our application and related modules 

24 return ( 

25 record.name.startswith("mcp_server") 

26 or record.name.startswith("src.") 

27 or record.name == "uvicorn" 

28 or record.name == "fastapi" 

29 or record.name == "__main__" # Allow logs from main module 

30 or record.name == "asyncio" # Allow logs from asyncio 

31 or record.name == "main" # Allow logs when started as a script 

32 or record.name == "qdrant_loader_mcp_server" # Allow logs from the package 

33 ) 

34 

35 

36class CleanFormatter(logging.Formatter): 

37 """Formatter that removes ANSI color codes.""" 

38 

39 def format(self, record): 

40 # Get the formatted message 

41 message = super().format(record) 

42 # Remove ANSI color codes 

43 ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") 

44 return ansi_escape.sub("", message) 

45 

46 

47class LoggingConfig: 

48 """Centralized logging configuration.""" 

49 

50 _initialized = False 

51 _current_config = None 

52 

53 @classmethod 

54 def setup( 

55 cls, 

56 level: str = "INFO", 

57 format: str = "console", 

58 file: str | None = None, 

59 suppress_qdrant_warnings: bool = True, 

60 ) -> None: 

61 """Setup logging configuration. 

62 

63 Args: 

64 level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) 

65 format: Log format (json or text) 

66 file: Path to log file (optional) 

67 suppress_qdrant_warnings: Whether to suppress Qdrant version check warnings 

68 """ 

69 # Check if console logging is disabled first 

70 disable_console_logging = ( 

71 os.getenv("MCP_DISABLE_CONSOLE_LOGGING", "").lower() == "true" 

72 ) 

73 

74 try: 

75 # Get log level from environment variable or use default 

76 level = os.getenv("MCP_LOG_LEVEL", level) 

77 # Convert string level to logging level 

78 numeric_level = getattr(logging, level.upper()) 

79 except AttributeError: 

80 raise ValueError(f"Invalid log level: {level}") from None 

81 

82 # Reset logging configuration 

83 logging.getLogger().handlers = [] 

84 structlog.reset_defaults() 

85 

86 # Create a list of handlers 

87 handlers = [] 

88 

89 # Add console handler for stderr only if console logging is not disabled 

90 if not disable_console_logging: 

91 stderr_handler = logging.StreamHandler(sys.stderr) 

92 stderr_handler.setFormatter(logging.Formatter("%(message)s")) 

93 stderr_handler.addFilter( 

94 ApplicationFilter() 

95 ) # Only show our application logs 

96 handlers.append(stderr_handler) 

97 

98 # Add file handler if file is configured 

99 if file: 

100 file_handler = logging.FileHandler(file) 

101 file_handler.setFormatter(CleanFormatter("%(message)s")) 

102 handlers.append(file_handler) 

103 

104 # Add clean log file handler at configured path 

105 log_file = os.getenv("MCP_LOG_FILE") 

106 if log_file: 

107 log_path = Path(log_file) 

108 log_path.parent.mkdir(parents=True, exist_ok=True) 

109 clean_log_handler = logging.FileHandler(log_path) 

110 clean_log_handler.setFormatter(CleanFormatter("%(message)s")) 

111 clean_log_handler.addFilter(ApplicationFilter()) 

112 handlers.append(clean_log_handler) 

113 

114 # Configure standard logging 

115 logging.basicConfig( 

116 level=numeric_level, 

117 format="%(message)s", 

118 handlers=handlers, 

119 force=True, # Force reconfiguration 

120 ) 

121 

122 # Add filter to suppress Qdrant version check warnings 

123 if suppress_qdrant_warnings: 

124 qdrant_logger = logging.getLogger("qdrant_client") 

125 qdrant_logger.addFilter(QdrantVersionFilter()) 

126 

127 # Configure structlog processors based on format 

128 processors = [ 

129 structlog.stdlib.filter_by_level, 

130 structlog.stdlib.add_logger_name, 

131 structlog.stdlib.add_log_level, 

132 structlog.processors.TimeStamper(fmt="iso"), 

133 structlog.processors.StackInfoRenderer(), 

134 structlog.processors.format_exc_info, 

135 structlog.processors.UnicodeDecoder(), 

136 structlog.processors.CallsiteParameterAdder( 

137 [ 

138 structlog.processors.CallsiteParameter.FILENAME, 

139 structlog.processors.CallsiteParameter.FUNC_NAME, 

140 structlog.processors.CallsiteParameter.LINENO, 

141 ] 

142 ), 

143 ] 

144 

145 if format == "json": 

146 processors.append(structlog.processors.JSONRenderer()) 

147 else: 

148 processors.append(structlog.dev.ConsoleRenderer(colors=True)) 

149 

150 # Configure structlog 

151 structlog.configure( 

152 processors=processors, 

153 wrapper_class=structlog.make_filtering_bound_logger(numeric_level), 

154 logger_factory=structlog.stdlib.LoggerFactory(), 

155 cache_logger_on_first_use=False, # Disable caching to ensure new configuration is used 

156 ) 

157 

158 cls._initialized = True 

159 cls._current_config = (level, format, file, suppress_qdrant_warnings) 

160 

161 @classmethod 

162 def get_logger(cls, name: str | None = None) -> structlog.BoundLogger: 

163 """Get a logger instance. 

164 

165 Args: 

166 name: Logger name. If None, will use the calling module's name. 

167 

168 Returns: 

169 structlog.BoundLogger: Logger instance 

170 """ 

171 if not cls._initialized: 

172 # Initialize with default settings if not already initialized 

173 cls.setup() 

174 return structlog.get_logger(name)