vllm.logger ¶

def _configure_vllm_root_logger(config: "LoggingConfig | None" = None) -> None:
    """Configure logging from explicit config or bootstrap environment values."""
    logging_config: dict[str, dict[str, Any] | Any] = {}
    if config is None:
        configure_logging = envs.VLLM_CONFIGURE_LOGGING
        log_level = envs.VLLM_LOGGING_LEVEL
        log_config_file = envs.VLLM_LOGGING_CONFIG_PATH
    else:
        configure_logging = config.configure_logging
        log_level = config.log_level
        log_config_file = config.pylogging_config_file

    if not configure_logging and log_config_file:
        raise RuntimeError(
            "Logging configuration is disabled, but a Python logging config "
            "file was given. pylogging_config_file requires "
            "configure_logging to be enabled."
        )

    if configure_logging:
        logging.setLogRecordFactory(_vllm_log_record_factory)
        logging_config = deepcopy(DEFAULT_LOGGING_CONFIG)

        vllm_handler = logging_config["handlers"]["vllm"]
        # Refresh these values in case env vars have changed.
        vllm_handler["level"] = log_level
        vllm_handler["stream"] = envs.VLLM_LOGGING_STREAM
        vllm_handler["formatter"] = "vllm_color" if _use_color() else "vllm"

        vllm_loggers = logging_config["loggers"]["vllm"]
        vllm_loggers["level"] = log_level
        for formatter in logging_config["formatters"].values():
            formatter["log_level"] = log_level

    if log_config_file:
        if not path.exists(log_config_file):
            raise RuntimeError(
                "Could not load logging config. File does not exist: %s",
                log_config_file,
            )
        with open(log_config_file, encoding="utf-8") as file:
            custom_config = json.loads(file.read())

        if not isinstance(custom_config, dict):
            raise ValueError(
                "Invalid logging config. Expected dict, got %s.",
                type(custom_config).__name__,
            )
        logging_config = custom_config

    for formatter in logging_config.get("formatters", {}).values():
        # This provides backwards compatibility after #10134.
        if formatter.get("class") == "vllm.logging.NewLineFormatter":
            formatter["class"] = "vllm.logging_utils.NewLineFormatter"

    if logging_config:
        dictConfig(logging_config)

    # Transformers uses httpx to access the Hugging Face Hub. httpx is quite verbose,
    # so we set its logging level to WARNING when vLLM's logging level is INFO.
    # httpx2 is the successor huggingface_hub switches to in its 2.x releases.
    if log_level == "INFO":
        logging.getLogger("httpx").setLevel(logging.WARNING)
        logging.getLogger("httpx2").setLevel(logging.WARNING)

def _log_platform_warnings(config: "LoggingConfig") -> None:
    """Emit platform diagnostics only after an enabled config is active."""
    if not config.configure_logging:
        return

    # Import lazily because platform modules use init_logger during import.
    from vllm.platforms import current_platform

    current_platform.log_warnings()

def _should_log_with_scope(scope: LogScope) -> bool:
    """Decide whether to log based on scope."""
    if scope == "global":
        from vllm.distributed.parallel_state import is_global_first_rank

        return is_global_first_rank()
    if scope == "local":
        from vllm.distributed.parallel_state import is_local_first_rank

        return is_local_first_rank()
    return True

def _vllm_log_record_factory(*args: Any, **kwargs: Any) -> logging.LogRecord:
    """Add vLLM process metadata to each log record."""
    record = _base_log_record_factory(*args, **kwargs)
    process_info = _vllm_process_info
    pid = os.getpid()
    if process_info is None or process_info[1] != pid:
        record.vllm_process_name = record.processName
    else:
        record.vllm_process_name = process_info[0]
    return record

def configure_logging(config: "LoggingConfig") -> None:
    """Apply a logging configuration in the current process."""
    _configure_vllm_root_logger(config)
    global _last_configured_logging_config
    _last_configured_logging_config = config
    _log_platform_warnings(config)

def configure_logging_from_args(args: Any) -> "LoggingConfig":
    """Apply parsed logging arguments and retain them for child processes."""
    from vllm.config.logging import LoggingConfig

    config = getattr(args, "logging_config", None) or LoggingConfig()
    if hasattr(args, "log_level"):
        config = replace(config, log_level=args.log_level)
    if hasattr(args, "log_config_file"):
        config = replace(config, pylogging_config_file=args.log_config_file)

    configure_logging_if_needed(config)
    args.log_config_file = config.pylogging_config_file
    args.logging_config = config
    return config

def configure_logging_if_needed(config: "LoggingConfig") -> None:
    """Apply a logging configuration unless it is already active in this process."""
    if config != _last_configured_logging_config:
        configure_logging(config)

def enable_trace_function_call(log_file_path: str, root_dir: str | None = None):
    """Enable tracing of every function call in code under `root_dir`.
    This is useful for debugging hangs or crashes.
    `log_file_path` is the path to the log file.
    `root_dir` is the root directory of the code to trace. If None, it is the
    vllm root directory.

    Note that this call is thread-level, any threads calling this function
    will have the trace enabled. Other threads will not be affected.
    """
    logger.warning(
        "VLLM_TRACE_FUNCTION is enabled. It will record every"
        " function executed by Python. This will slow down the code. It "
        "is suggested to be used for debugging hang or crashes only."
    )
    logger.info("Trace frame log is saved to %s", log_file_path)
    if root_dir is None:
        # by default, this is the vllm root directory
        root_dir = os.path.dirname(os.path.dirname(__file__))
    sys.settrace(partial(_trace_calls, log_file_path, root_dir))

def init_logger(name: str) -> _VllmLogger:
    """Retrieve a logger and add vLLM's convenience logging methods."""
    logger = logging.getLogger(name)

    for method_name, method in _METHODS_TO_PATCH.items():
        setattr(logger, method_name, MethodType(method, logger))

    return cast(_VllmLogger, logger)

def set_vllm_process_name(process_name: str, *, skip_if_set: bool = False) -> None:
    """Set the vLLM process name added to subsequent log records."""
    global _vllm_process_info
    pid = os.getpid()
    if skip_if_set and _vllm_process_info is not None and _vllm_process_info[1] == pid:
        return
    _vllm_process_info = (process_name, pid)