| import json |
| import logging |
| import sys |
| import traceback |
| from typing import TYPE_CHECKING |
|
|
| from loguru import logger |
| from open_webui.env import ( |
| _LEVEL_MAP, |
| AUDIT_LOG_FILE_ROTATION_SIZE, |
| AUDIT_LOG_LEVEL, |
| AUDIT_LOGS_FILE_PATH, |
| AUDIT_UVICORN_LOGGER_NAMES, |
| ENABLE_AUDIT_LOGS_FILE, |
| ENABLE_AUDIT_STDOUT, |
| ENABLE_OTEL, |
| ENABLE_OTEL_LOGS, |
| GLOBAL_LOG_LEVEL, |
| LOG_FORMAT, |
| ) |
|
|
| if TYPE_CHECKING: |
| from loguru import Message, Record |
|
|
|
|
| def stdout_format(record: 'Record') -> str: |
| """ |
| 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). |
| |
| Parameters: |
| record (Record): A Loguru record that contains logging details including time, level, name, function, line, message, and any extra context. |
| Returns: |
| str: A formatted log string intended for stdout. |
| """ |
| if record['extra']: |
| record['extra']['extra_json'] = json.dumps(record['extra']) |
| extra_format = ' - {extra[extra_json]}' |
| else: |
| extra_format = '' |
| return ( |
| '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | ' |
| '<level>{level: <8}</level> | ' |
| '<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - ' |
| '<level>{message}</level>' + extra_format + '\n{exception}' |
| ) |
|
|
|
|
| def _json_sink(message: 'Message') -> None: |
| """Write log records as single-line JSON to stdout. |
| |
| Used as a Loguru sink when LOG_FORMAT is set to "json". |
| """ |
| try: |
| record = message.record |
| log_entry = { |
| 'ts': record['time'].isoformat(timespec='milliseconds'), |
| 'level': _LEVEL_MAP.get(record['level'].name, record['level'].name.lower()), |
| 'msg': record['message'], |
| 'caller': f'{record["name"]}:{record["function"]}:{record["line"]}', |
| } |
|
|
| if record['extra']: |
| log_entry['extra'] = record['extra'] |
|
|
| exc = record['exception'] |
| if exc is not None: |
| log_entry['error'] = { |
| 'type': exc.type.__name__ if exc.type else None, |
| 'message': str(exc.value) if exc.value else None, |
| 'stacktrace': ''.join(traceback.format_exception(exc.type, exc.value, exc.traceback)).rstrip(), |
| } |
|
|
| sys.stdout.write(json.dumps(log_entry, ensure_ascii=False, default=str) + '\n') |
| sys.stdout.flush() |
| except Exception: |
| |
| |
| try: |
| fallback = { |
| 'ts': message.record['time'].isoformat(timespec='milliseconds'), |
| 'level': 'error', |
| 'msg': f'[logging error] failed to serialize log record: {message}', |
| } |
| sys.stdout.write(json.dumps(fallback, ensure_ascii=False, default=str) + '\n') |
| sys.stdout.flush() |
| except Exception: |
| sys.stderr.write(f'[logging error] _json_sink failed: {message}\n') |
| sys.stderr.flush() |
|
|
|
|
| class InterceptHandler(logging.Handler): |
| """ |
| Intercepts log records from Python's standard logging module |
| and redirects them to Loguru's logger. |
| """ |
|
|
| def emit(self, record): |
| """ |
| Called by the standard logging module for each log event. |
| It transforms the standard `LogRecord` into a format compatible with Loguru |
| and passes it to Loguru's logger. |
| """ |
| try: |
| level = logger.level(record.levelname).name |
| except ValueError: |
| level = record.levelno |
|
|
| frame, depth = sys._getframe(6), 6 |
| while frame and frame.f_code.co_filename == logging.__file__: |
| frame = frame.f_back |
| depth += 1 |
|
|
| logger.opt(depth=depth, exception=record.exc_info).bind(**self._get_extras()).log(level, record.getMessage()) |
| if ENABLE_OTEL and ENABLE_OTEL_LOGS: |
| from open_webui.utils.telemetry.logs import otel_handler |
|
|
| otel_handler.emit(record) |
|
|
| def _get_extras(self): |
| if not ENABLE_OTEL: |
| return {} |
|
|
| from opentelemetry import trace |
|
|
| extras = {} |
| context = trace.get_current_span().get_span_context() |
| if context.is_valid: |
| extras['trace_id'] = trace.format_trace_id(context.trace_id) |
| extras['span_id'] = trace.format_span_id(context.span_id) |
| return extras |
|
|
|
|
| def file_format(record: 'Record'): |
| """ |
| Formats audit log records into a structured JSON string for file output. |
| |
| Parameters: |
| record (Record): A Loguru record containing extra audit data. |
| Returns: |
| str: A JSON-formatted string representing the audit data. |
| """ |
|
|
| audit_data = { |
| 'id': record['extra'].get('id', ''), |
| 'timestamp': int(record['time'].timestamp()), |
| 'user': record['extra'].get('user', dict()), |
| 'audit_level': record['extra'].get('audit_level', ''), |
| 'verb': record['extra'].get('verb', ''), |
| 'request_uri': record['extra'].get('request_uri', ''), |
| 'response_status_code': record['extra'].get('response_status_code', 0), |
| 'source_ip': record['extra'].get('source_ip', ''), |
| 'user_agent': record['extra'].get('user_agent', ''), |
| 'request_object': record['extra'].get('request_object', b''), |
| 'response_object': record['extra'].get('response_object', b''), |
| 'extra': record['extra'].get('extra', {}), |
| } |
|
|
| record['extra']['file_extra'] = json.dumps(audit_data, default=str) |
| return '{extra[file_extra]}\n' |
|
|
|
|
| def start_logger(): |
| """ |
| Initializes and configures Loguru's logger with distinct handlers: |
| |
| A console (stdout) handler for general log messages (excluding those marked as auditable). |
| An optional file handler for audit logs if audit logging is enabled. |
| Additionally, this function reconfigures Python’s standard logging to route through Loguru and adjusts logging levels for Uvicorn. |
| |
| Parameters: |
| enable_audit_logging (bool): Determines whether audit-specific log entries should be recorded to file. |
| """ |
| logger.remove() |
|
|
| audit_filter = lambda record: True if ENABLE_AUDIT_STDOUT else 'auditable' not in record['extra'] |
| if LOG_FORMAT == 'json': |
| logger.add( |
| _json_sink, |
| level=GLOBAL_LOG_LEVEL, |
| filter=audit_filter, |
| ) |
| else: |
| logger.add( |
| sys.stdout, |
| level=GLOBAL_LOG_LEVEL, |
| format=stdout_format, |
| filter=audit_filter, |
| ) |
| if AUDIT_LOG_LEVEL != 'NONE' and ENABLE_AUDIT_LOGS_FILE: |
| try: |
| logger.add( |
| AUDIT_LOGS_FILE_PATH, |
| level='INFO', |
| rotation=AUDIT_LOG_FILE_ROTATION_SIZE, |
| compression='zip', |
| format=file_format, |
| filter=lambda record: record['extra'].get('auditable') is True, |
| ) |
| except Exception as e: |
| logger.error(f'Failed to initialize audit log file handler: {str(e)}') |
|
|
| logging.basicConfig(handlers=[InterceptHandler()], level=GLOBAL_LOG_LEVEL, force=True) |
|
|
| for uvicorn_logger_name in ['uvicorn', 'uvicorn.error']: |
| uvicorn_logger = logging.getLogger(uvicorn_logger_name) |
| uvicorn_logger.setLevel(GLOBAL_LOG_LEVEL) |
| uvicorn_logger.handlers = [] |
|
|
| for uvicorn_logger_name in AUDIT_UVICORN_LOGGER_NAMES: |
| uvicorn_logger = logging.getLogger(uvicorn_logger_name) |
| uvicorn_logger.setLevel(GLOBAL_LOG_LEVEL) |
| uvicorn_logger.handlers = [InterceptHandler()] |
|
|
| logger.info(f'GLOBAL_LOG_LEVEL: {GLOBAL_LOG_LEVEL}') |
|
|
