Coverage for open_webui/utils/logger.py: 50%
87 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1import json
2import logging
3import sys
4import traceback
5from typing import TYPE_CHECKING
7from loguru import logger
8from open_webui.env import (
9 _LEVEL_MAP,
10 AUDIT_LOG_FILE_ROTATION_SIZE,
11 AUDIT_LOG_LEVEL,
12 AUDIT_LOGS_FILE_PATH,
13 AUDIT_UVICORN_LOGGER_NAMES,
14 ENABLE_AUDIT_LOGS_FILE,
15 ENABLE_AUDIT_STDOUT,
16 ENABLE_OTEL,
17 ENABLE_OTEL_LOGS,
18 GLOBAL_LOG_LEVEL,
19 LOG_FORMAT,
20 LOGURU_DIAGNOSE,
21)
22from open_webui.utils.json_codec import JSONCodec
24if TYPE_CHECKING: 24 ↛ 25line 24 didn't jump to line 25 because the condition on line 24 was never true
25 from loguru import Message, Record
28def stdout_format(record: 'Record') -> str:
29 """
30 Generates a formatted string for log records that are output to the console. This format includes a timestamp, log level, source location (module, function, and line), the log message, and any extra data (serialized as JSON).
32 Parameters:
33 record (Record): A Loguru record that contains logging details including time, level, name, function, line, message, and any extra context.
34 Returns:
35 str: A formatted log string intended for stdout.
36 """
37 if record['extra']: 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true
38 record['extra']['extra_json'] = JSONCodec.dumps(record['extra'])
39 extra_format = ' - {extra[extra_json]}'
40 else:
41 extra_format = ''
42 return (
43 '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | '
44 '<level>{level: <8}</level> | '
45 '<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - '
46 '<level>{message}</level>' + extra_format + '\n{exception}'
47 )
50def _json_sink(message: 'Message') -> None:
51 """Write log records as single-line JSON to stdout.
53 Used as a Loguru sink when LOG_FORMAT is set to "json".
54 """
55 try:
56 record = message.record
57 log_entry = {
58 'ts': record['time'].isoformat(timespec='milliseconds'),
59 'level': _LEVEL_MAP.get(record['level'].name, record['level'].name.lower()),
60 'msg': record['message'],
61 'caller': f'{record["name"]}:{record["function"]}:{record["line"]}',
62 }
64 if record['extra']:
65 log_entry['extra'] = record['extra']
67 exc = record['exception']
68 if exc is not None:
69 log_entry['error'] = {
70 'type': exc.type.__name__ if exc.type else None,
71 'message': str(exc.value) if exc.value else None,
72 'stacktrace': ''.join(traceback.format_exception(exc.type, exc.value, exc.traceback)).rstrip(),
73 }
75 sys.stdout.write(json.dumps(log_entry, ensure_ascii=False, default=str) + '\n')
76 sys.stdout.flush()
77 except Exception:
78 # Last-resort fallback: never let a logging failure crash the application.
79 # Emit a minimal valid JSON line so the structured logging pipeline stays intact.
80 try:
81 fallback = {
82 'ts': message.record['time'].isoformat(timespec='milliseconds'),
83 'level': 'error',
84 'msg': f'[logging error] failed to serialize log record: {message}',
85 }
86 sys.stdout.write(json.dumps(fallback, ensure_ascii=False, default=str) + '\n')
87 sys.stdout.flush()
88 except Exception:
89 sys.stderr.write(f'[logging error] _json_sink failed: {message}\n')
90 sys.stderr.flush()
93class InterceptHandler(logging.Handler):
94 """
95 Intercepts log records from Python's standard logging module
96 and redirects them to Loguru's logger.
97 """
99 def emit(self, record):
100 """
101 Called by the standard logging module for each log event.
102 It transforms the standard `LogRecord` into a format compatible with Loguru
103 and passes it to Loguru's logger.
104 """
105 try:
106 level = logger.level(record.levelname).name
107 except ValueError:
108 level = record.levelno
110 frame, depth = sys._getframe(6), 6
111 while frame and frame.f_code.co_filename == logging.__file__:
112 frame = frame.f_back
113 depth += 1
115 message = record.getMessage()
116 logger.opt(depth=depth, exception=record.exc_info).bind(**self._get_extras()).log(level, message)
117 if ENABLE_OTEL and ENABLE_OTEL_LOGS: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true
118 from open_webui.utils.telemetry.logs import otel_handler
120 # reuse the message we built so %-args format once; a non-str msg is left alone, otel exports it structured
121 if isinstance(record.msg, str):
122 record.msg, record.args = message, None
123 otel_handler.emit(record)
125 def _get_extras(self):
126 if not ENABLE_OTEL: 126 ↛ 129line 126 didn't jump to line 129 because the condition on line 126 was always true
127 return {}
129 from opentelemetry import trace
131 extras = {}
132 context = trace.get_current_span().get_span_context()
133 if context.is_valid:
134 extras['trace_id'] = trace.format_trace_id(context.trace_id)
135 extras['span_id'] = trace.format_span_id(context.span_id)
136 return extras
139def file_format(record: 'Record'):
140 """
141 Formats audit log records into a structured JSON string for file output.
143 Parameters:
144 record (Record): A Loguru record containing extra audit data.
145 Returns:
146 str: A JSON-formatted string representing the audit data.
147 """
149 audit_data = {
150 'id': record['extra'].get('id', ''),
151 'timestamp': int(record['time'].timestamp()),
152 'user': record['extra'].get('user', dict()),
153 'audit_level': record['extra'].get('audit_level', ''),
154 'verb': record['extra'].get('verb', ''),
155 'request_uri': record['extra'].get('request_uri', ''),
156 'response_status_code': record['extra'].get('response_status_code', 0),
157 'source_ip': record['extra'].get('source_ip', ''),
158 'user_agent': record['extra'].get('user_agent', ''),
159 'request_object': record['extra'].get('request_object', b''),
160 'response_object': record['extra'].get('response_object', b''),
161 'extra': record['extra'].get('extra', {}),
162 }
164 record['extra']['file_extra'] = json.dumps(audit_data, default=str)
165 return '{extra[file_extra]}\n'
168def start_logger():
169 """
170 Initializes and configures Loguru's logger with distinct handlers:
172 A console (stdout) handler for general log messages (excluding those marked as auditable).
173 An optional file handler for audit logs if audit logging is enabled.
174 Additionally, this function reconfigures Python’s standard logging to route through Loguru and adjusts logging levels for Uvicorn.
176 Parameters:
177 enable_audit_logging (bool): Determines whether audit-specific log entries should be recorded to file.
178 """
179 logger.remove()
181 audit_filter = lambda record: True if ENABLE_AUDIT_STDOUT else 'auditable' not in record['extra']
182 if LOG_FORMAT == 'json': 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 logger.add(
184 _json_sink,
185 level=GLOBAL_LOG_LEVEL,
186 filter=audit_filter,
187 diagnose=LOGURU_DIAGNOSE,
188 )
189 else:
190 logger.add(
191 sys.stdout,
192 level=GLOBAL_LOG_LEVEL,
193 format=stdout_format,
194 filter=audit_filter,
195 diagnose=LOGURU_DIAGNOSE,
196 )
197 if AUDIT_LOG_LEVEL != 'NONE' and ENABLE_AUDIT_LOGS_FILE: 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true
198 try:
199 logger.add(
200 AUDIT_LOGS_FILE_PATH,
201 level='INFO',
202 rotation=AUDIT_LOG_FILE_ROTATION_SIZE,
203 compression='zip',
204 format=file_format,
205 filter=lambda record: record['extra'].get('auditable') is True,
206 diagnose=LOGURU_DIAGNOSE,
207 )
208 except Exception as e:
209 logger.error(f'Failed to initialize audit log file handler: {str(e)}')
211 logging.basicConfig(handlers=[InterceptHandler()], level=GLOBAL_LOG_LEVEL, force=True)
213 for uvicorn_logger_name in ['uvicorn', 'uvicorn.error']:
214 uvicorn_logger = logging.getLogger(uvicorn_logger_name)
215 uvicorn_logger.setLevel(GLOBAL_LOG_LEVEL)
216 uvicorn_logger.handlers = []
218 for uvicorn_logger_name in AUDIT_UVICORN_LOGGER_NAMES:
219 uvicorn_logger = logging.getLogger(uvicorn_logger_name)
220 uvicorn_logger.setLevel(GLOBAL_LOG_LEVEL)
221 uvicorn_logger.handlers = [InterceptHandler()]
223 logger.info(f'GLOBAL_LOG_LEVEL: {GLOBAL_LOG_LEVEL}')