LCORE- Unify logging configuration and setup

src/constants.py

@@ -230,10 +230,11 @@
 # Environment variable name for configurable log level
 LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR: Final[str] = "LIGHTSPEED_STACK_LOG_LEVEL"
 # Default log level when environment variable is not set
+DEFAULT_LOGGER_NAME: Final[str] = "lightspeed_stack"
 DEFAULT_LOG_LEVEL: Final[str] = "INFO"
 # Default log format for plain-text logging in non-TTY environments
 DEFAULT_LOG_FORMAT: Final[str] = (
-    "%(asctime)s %(levelname)-8s %(name)s:%(lineno)d %(message)s"
+    "%(asctime)s.%(msecs)03d %(levelprefix)s %(message)s  [%(name)s:%(lineno)d]"
 )
 # Environment variable to force StreamHandler instead of RichHandler
 # Set to any non-empty value to disable RichHandler

src/lightspeed_stack.py

@@ -4,39 +4,17 @@
 main() function.
 """
 
-import logging
 import os
-import sys
 from argparse import ArgumentParser
 
 from configuration import configuration
 from constants import LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR
-from log import create_log_handler, get_logger, resolve_log_level
+from log import get_logger, setup_logging
 from runners.quota_scheduler import start_quota_scheduler
 from runners.uvicorn import start_uvicorn
 from utils import schema_dumper
 
-# Resolve log level and handler from centralized logging utilities
-log_level = resolve_log_level()
-
-# Configure root logger. basicConfig(force=True) is intentionally root-logger-specific.
-# RichHandler needs format="%(message)s" to prevent double-formatting by the root Formatter.
-handler = create_log_handler()
-if sys.stderr.isatty():
-    logging.basicConfig(
-        level=log_level,
-        format="%(message)s",
-        datefmt="[%X]",
-        handlers=[handler],
-        force=True,
-    )
-else:
-    logging.basicConfig(
-        level=log_level,
-        handlers=[handler],
-        force=True,
-    )
-
+setup_logging()
 logger = get_logger(__name__)
 
 
@@ -119,11 +97,8 @@ def main() -> None:
 
     if args.verbose:
         os.environ[LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR] = "DEBUG"
-        logging.getLogger().setLevel(logging.DEBUG)
-        for logger_name in logging.Logger.manager.loggerDict:
-            existing_logger = logging.getLogger(logger_name)
-            if isinstance(existing_logger, logging.Logger):
-                existing_logger.setLevel(logging.DEBUG)
+        setup_logging.cache_clear()
+        setup_logging()
 
     configuration.load_configuration(args.config_file)
     logger.info("Configuration: %s", configuration.configuration)

src/log.py

@@ -1,19 +1,45 @@
 """Log utilities."""
 
 import logging
+import logging.config
 import os
 import sys
+import typing as t
+from copy import deepcopy
+from datetime import datetime
+from functools import lru_cache
 
-from rich.logging import RichHandler
+import uvicorn.config
+from rich.text import Text
 
 from constants import (
     DEFAULT_LOG_FORMAT,
     DEFAULT_LOG_LEVEL,
+    DEFAULT_LOGGER_NAME,
     LIGHTSPEED_STACK_DISABLE_RICH_HANDLER_ENV_VAR,
     LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR,
 )
 
 
+def _ms_time_format(dt: datetime) -> Text:
+    """Format datetime object with zero padded milliseconds."""
+    return Text(dt.strftime("%Y-%m-%d %H:%M:%S.") + f"{dt.microsecond // 1000:03d}")
+
+
+def _deep_merge(
+    mapping: dict[t.Any, t.Any], updates: dict[t.Any, t.Any]
+) -> dict[t.Any, t.Any]:
+    """Recursively merge updates into mapping."""
+    merged = mapping.copy()
+    for k, v in updates.items():
+        if k in merged and isinstance(merged[k], dict) and isinstance(v, dict):
+            merged[k] = _deep_merge(merged[k], v)
+        else:
+            merged[k] = v
+
+    return merged
+
+
 def resolve_log_level() -> int:
     """
     Resolve and validate the log level from environment variable.
@@ -50,62 +76,70 @@ def resolve_log_level() -> int:
     return validated_level
 
 
-def create_log_handler() -> logging.Handler:
-    """
-    Create and return a configured log handler based on TTY availability and environment settings.
-
-    If LIGHTSPEED_STACK_DISABLE_RICH_HANDLER is set to any non-empty value,
-    returns a StreamHandler with plain-text formatting. Otherwise, if stderr
-    is connected to a terminal (TTY), returns a RichHandler for rich-formatted
-    console output. If neither condition is met, returns a StreamHandler with
-    plain-text formatting suitable for non-TTY environments (e.g., containers).
-
-    Returns:
-        logging.Handler: A configured handler instance (RichHandler or StreamHandler).
-    """
-    # Check if RichHandler is explicitly disabled via environment variable
-    if os.environ.get(LIGHTSPEED_STACK_DISABLE_RICH_HANDLER_ENV_VAR):
-        handler = logging.StreamHandler()
-        handler.setFormatter(logging.Formatter(DEFAULT_LOG_FORMAT))
-        return handler
-
-    if sys.stderr.isatty():
-        # RichHandler's columnar layout assumes a real terminal.
-        # RichHandler handles its own formatting, so no formatter is set.
-        return RichHandler()
-
-    # In containers without a TTY, Rich falls back to 80 columns and
-    # the columns consume most of that width, leaving ~40 chars for the actual message.
-    # Tracebacks become nearly unreadable. Use a plain StreamHandler instead.
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter(DEFAULT_LOG_FORMAT))
-    return handler
-
-
 def get_logger(name: str) -> logging.Logger:
-    """
-    Get a logger configured for Rich console output.
-
-    The returned logger has its level set based on the LIGHTSPEED_STACK_LOG_LEVEL
-    environment variable (defaults to INFO), its handlers replaced with a single
-    handler (RichHandler for TTY or StreamHandler for non-TTY), and propagation
-    to ancestor loggers disabled.
-
-    Parameters:
-    ----------
-        name (str): Name of the logger to retrieve or create.
-
-    Returns:
-    -------
-        logging.Logger: The configured logger instance.
-    """
-    logger = logging.getLogger(name)
+    """Create a common logger for all modules in this package."""
+    # The need for this function should be removed in the future.
+    #
+    # Normally this is derived from the package name (__name__).
+    #
+    # Since this program is sometimes called from from the entrypoint and
+    # sometimes called from src/lightspeed_stack.py, the value for __name__
+    # does not contain a consistent root value.
+    #
+    # How the application is installed and run needs to be streamlined so that
+    # __name__ provides the expected value in all cases.
+    return logging.getLogger(f"{DEFAULT_LOGGER_NAME}.{name}")
+
+
+@lru_cache
+def setup_logging() -> dict[t.Any, t.Any]:
+    """Create logging configuration."""
+    handler = "default"
+    log_level = resolve_log_level()
+    if sys.stderr.isatty() and not os.environ.get(
+        LIGHTSPEED_STACK_DISABLE_RICH_HANDLER_ENV_VAR
+    ):
+        handler = "rich"
+
+    logging_conf = {
+        "version": 1,
+        "disable_existing_loggers": False,
+        "handlers": {
+            "rich": {
+                "()": "rich.logging.RichHandler",
+                "show_time": True,
+                "log_time_format": _ms_time_format,
+                "level": log_level,
+            },
+        },
+        "loggers": {
+            DEFAULT_LOGGER_NAME: {
+                "handlers": [handler],
+                "level": log_level,
+                "propagate": False,
+            },
+            "llama_stack_client": {
+                "handlers": [handler],
+                "level": log_level,
+                "propagate": False,
+            },
+        },
+    }
+
+    # Create a deep copy of uvicorn's logging config to avoid mutating global state.
+    merged_config = _deep_merge(deepcopy(uvicorn.config.LOGGING_CONFIG), logging_conf)
+
+    if handler == "rich":
+        merged_config["loggers"]["uvicorn"]["handlers"] = [handler]
+        merged_config["loggers"]["uvicorn.access"]["handlers"] = [handler]
+    else:
+        merged_config["formatters"]["access"]["fmt"] = (
+            "%(asctime)s.%(msecs)03d %(levelprefix)s "
+            '%(client_addr)s - "%(request_line)s" %(status_code)s'
+        )
+        merged_config["formatters"]["default"]["fmt"] = DEFAULT_LOG_FORMAT
+        merged_config["formatters"]["default"]["datefmt"] = "%Y-%m-%d %H:%M:%S"
 
-    # Skip reconfiguration if logger already has handlers from a prior call
-    if logger.handlers:
-        return logger
+    logging.config.dictConfig(merged_config)
 
-    logger.handlers = [create_log_handler()]
-    logger.propagate = False
-    logger.setLevel(resolve_log_level())
-    return logger
+    return merged_config

src/runners/uvicorn.py

@@ -4,24 +4,31 @@
 
 import uvicorn
 
-from log import get_logger, resolve_log_level
+from log import get_logger, resolve_log_level, setup_logging
 from models.config import ServiceConfiguration
 
 logger = get_logger(__name__)
 
 
-def start_uvicorn(configuration: ServiceConfiguration) -> None:
+def start_uvicorn(
+    configuration: ServiceConfiguration,
+    log_config: dict | None = None,
+) -> None:
     """Start the Uvicorn server using the provided service configuration.
 
     Parameters:
     ----------
         configuration (ServiceConfiguration): Configuration providing host,
-        port, workers, and `tls_config` (including `tls_key_path`,
-        `tls_certificate_path`, and `tls_key_password`). TLS fields may be None
-        and will be forwarded to uvicorn.run as provided.
+            port, workers, and `tls_config` (including `tls_key_path`,
+            `tls_certificate_path`, and `tls_key_password`). TLS fields may be None
+            and will be forwarded to uvicorn.run as provided.
+        log_config (dict | None): Logging configuration dictionary passed to
+            uvicorn.run. When None, defaults to the output of setup_logging().
     """
     log_level = resolve_log_level()
     logger.info("Starting Uvicorn with log level %s", logging.getLevelName(log_level))
+    if log_config is None:
+        log_config = setup_logging()
 
     # please note:
     # TLS fields can be None, which means we will pass those values as None to uvicorn.run
@@ -30,6 +37,7 @@ def start_uvicorn(configuration: ServiceConfiguration) -> None:
         host=configuration.host,
         port=configuration.port,
         workers=configuration.workers,
+        log_config=log_config,
         log_level=log_level,
         ssl_keyfile=configuration.tls_config.tls_key_path,
         ssl_certfile=configuration.tls_config.tls_certificate_path,