Skip to content

Memory handler

In-memory logging handler for dashboard log viewing.

This module provides a custom logging handler that stores log records in memory for retrieval via the /logs API endpoint. It's designed to be used when ENABLE_IN_MEMORY_LOGS environment variable is set to 'true'.

MemoryLogHandler

Bases: Handler

Custom log handler that stores log records in memory for dashboard access

Source code in inference/core/logging/memory_handler.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class MemoryLogHandler(logging.Handler):
    """Custom log handler that stores log records in memory for dashboard access"""

    def emit(self, record):
        try:
            # Format the log entry for JSON serialization
            log_entry = {
                "timestamp": datetime.fromtimestamp(record.created).isoformat(),
                "level": record.levelname,
                "logger": record.name,
                "message": self.format(record),
                "module": record.module or "",
                "line": record.lineno,
            }

            with _log_lock:
                _log_entries.append(log_entry)
        except Exception:
            # Silently handle any errors in logging to prevent recursion
            pass

get_recent_logs(limit=100, level=None, since=None)

Get recent log entries from memory

Source code in inference/core/logging/memory_handler.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def get_recent_logs(
    limit: int = 100, level: str = None, since: str = None
) -> List[Dict[str, Any]]:
    """Get recent log entries from memory"""
    with _log_lock:
        logs = list(_log_entries)

    # Filter by log level if specified
    if level:
        level_upper = level.upper()
        logs = [log for log in logs if log["level"] == level_upper]

    # Filter by timestamp if specified
    if since:
        try:
            since_dt = datetime.fromisoformat(since.replace("Z", "+00:00"))
            logs = [
                log
                for log in logs
                if datetime.fromisoformat(log["timestamp"]) > since_dt
            ]
        except ValueError:
            pass  # Invalid since timestamp, ignore filter

    # Limit results
    return logs[-limit:] if limit else logs

setup_memory_logging()

Set up memory logging handler for the current logger hierarchy

Source code in inference/core/logging/memory_handler.py
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def setup_memory_logging() -> None:
    """Set up memory logging handler for the current logger hierarchy"""
    if not is_memory_logging_enabled():
        return
    logger.info("Setting up memory logging")
    memory_handler = MemoryLogHandler()
    memory_handler.setLevel(logging.DEBUG)  # Capture all levels
    memory_formatter = logging.Formatter(
        "%(asctime)s %(levelname)s %(name)s: %(message)s"
    )
    memory_handler.setFormatter(memory_formatter)

    # Add to root logger to capture all logs immediately
    root_logger = logging.getLogger()
    if memory_handler not in root_logger.handlers:
        root_logger.addHandler(memory_handler)

    # Specifically add to uvicorn.access logger to ensure access logs are captured now
    access_logger = logging.getLogger("uvicorn.access")
    if memory_handler not in access_logger.handlers:
        access_logger.addHandler(memory_handler)

    # Also patch uvicorn's default LOGGING_CONFIG so when uvicorn applies dictConfig,
    # our in-memory handler remains attached
    global _uvicorn_config_patched
    if not _uvicorn_config_patched:
        try:
            from uvicorn.config import LOGGING_CONFIG as UVICORN_LOGGING_CONFIG

            # Modify in-place (safe: uvicorn makes a deep copy later)
            log_config = UVICORN_LOGGING_CONFIG

            log_config.setdefault("formatters", {})
            if "default" not in log_config["formatters"]:
                log_config["formatters"]["default"] = {
                    "()": "uvicorn.logging.DefaultFormatter",
                    "fmt": "%(levelprefix)s %(message)s",
                    "use_colors": None,
                }

            log_config.setdefault("handlers", {})["inmemory"] = {
                "class": "inference.core.logging.memory_handler.MemoryLogHandler",
                "level": "DEBUG",
                "formatter": "default",
            }

            log_config.setdefault("loggers", {})
            log_config["loggers"].setdefault(
                "uvicorn.access",
                {
                    "handlers": ["default"],
                    "level": "INFO",
                    "propagate": False,
                },
            )
            if "inmemory" not in log_config["loggers"]["uvicorn.access"]["handlers"]:
                log_config["loggers"]["uvicorn.access"]["handlers"].append("inmemory")

            log_config["loggers"].setdefault(
                "uvicorn", {"handlers": ["default"], "level": "INFO"}
            )
            log_config["loggers"].setdefault("uvicorn.error", {"level": "INFO"})

            root_cfg = log_config.setdefault(
                "root", {"handlers": ["default"], "level": "INFO"}
            )
            if "inmemory" not in root_cfg.get("handlers", []):
                root_cfg.setdefault("handlers", []).append("inmemory")

            _uvicorn_config_patched = True
            logger.info("Patched uvicorn LOGGING_CONFIG to include MemoryLogHandler")
        except Exception:
            # Avoid hard failure if uvicorn is not available
            pass

    return memory_handler