RSPEED-2477: Deduplicate logging setup across entry point, log utility, and uvicorn runner

src/constants.py

@@ -183,3 +183,5 @@
 LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR = "LIGHTSPEED_STACK_LOG_LEVEL"
 # Default log level when environment variable is not set
 DEFAULT_LOG_LEVEL = "INFO"
+# Default log format for plain-text logging in non-TTY environments
+DEFAULT_LOG_FORMAT = "%(asctime)s %(levelname)-8s %(name)s:%(lineno)d %(message)s"

src/lightspeed_stack.py

@@ -9,41 +9,31 @@
 import sys
 from argparse import ArgumentParser
 
-
-from rich.logging import RichHandler
-
-from log import get_logger
+from log import get_logger, resolve_log_level, create_log_handler
 from configuration import configuration
 from runners.uvicorn import start_uvicorn
 from runners.quota_scheduler import start_quota_scheduler
 from utils import schema_dumper
-from constants import LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, DEFAULT_LOG_LEVEL
-
-# Read log level from environment variable with validation
-log_level_str = os.environ.get(LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, DEFAULT_LOG_LEVEL)
-log_level = getattr(logging, log_level_str.upper(), None)
-if not isinstance(log_level, int):
-    print(
-        f"WARNING: Invalid log level '{log_level_str}', falling back to {DEFAULT_LOG_LEVEL}",
-        file=sys.stderr,
-    )
-    log_level = getattr(logging, DEFAULT_LOG_LEVEL)
+from constants import LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR
+
+# Resolve log level and handler from centralized logging utilities
+log_level = resolve_log_level()
 
-# RichHandler's columnar layout produces very narrow log output in containers
-# without a TTY (Rich falls back to 80 columns, columns consume ~40 of those).
-# Use a plain format when there's no terminal attached.
+# 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=[RichHandler()],
+        handlers=[handler],
         force=True,
     )
 else:
     logging.basicConfig(
         level=log_level,
-        format="%(asctime)s %(levelname)-8s %(name)s:%(lineno)d %(message)s",
+        handlers=[handler],
         force=True,
     )
 

src/log.py

@@ -6,7 +6,72 @@
 
 from rich.logging import RichHandler
 
-from constants import LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, DEFAULT_LOG_LEVEL
+from constants import (
+    LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR,
+    DEFAULT_LOG_LEVEL,
+    DEFAULT_LOG_FORMAT,
+)
+
+
+def resolve_log_level() -> int:
+    """
+    Resolve and validate the log level from environment variable.
+
+    Reads the LIGHTSPEED_STACK_LOG_LEVEL environment variable and validates
+    it against Python's logging module. If the environment variable is not set,
+    defaults to DEFAULT_LOG_LEVEL. If the value is invalid, logs a warning and
+    falls back to DEFAULT_LOG_LEVEL.
+
+    Parameters:
+        None
+
+    Returns:
+        int: A valid logging level constant (e.g., logging.INFO, logging.DEBUG).
+    """
+    level_str = os.environ.get(LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, DEFAULT_LOG_LEVEL)
+
+    # Validate the level string and convert to logging level constant
+    validated_level = getattr(logging, level_str.upper(), None)
+    if not isinstance(validated_level, int):
+        # Write directly to stderr instead of using a logger. This function is
+        # called at module-import time (before logging is configured), so routing
+        # through a logger produces inconsistent output depending on root-logger
+        # state.
+        print(
+            f"WARNING: Invalid log level '{level_str}', "
+            f"falling back to {DEFAULT_LOG_LEVEL}",
+            file=sys.stderr,
+        )
+        validated_level = getattr(logging, DEFAULT_LOG_LEVEL)
+
+    return validated_level
+
+
+def create_log_handler() -> logging.Handler:
+    """
+    Create and return a configured log handler based on TTY availability.
+
+    If stderr is connected to a terminal (TTY), returns a RichHandler for
+    rich-formatted console output. Otherwise, returns a StreamHandler with
+    plain-text formatting suitable for non-TTY environments (e.g., containers).
+
+    Parameters:
+        None
+
+    Returns:
+        logging.Handler: A configured handler instance (RichHandler or StreamHandler).
+    """
+    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:
@@ -15,8 +80,8 @@ def get_logger(name: str) -> logging.Logger:
 
     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
-    RichHandler for rich-formatted console output, and propagation to ancestor
-    loggers disabled.
+    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.
@@ -30,34 +95,7 @@ def get_logger(name: str) -> logging.Logger:
     if logger.handlers:
         return logger
 
-    # RichHandler's columnar layout (timestamp, level, right-aligned filename) assumes
-    # a real terminal. 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 when there's no TTY.
-    if sys.stderr.isatty():
-        logger.handlers = [RichHandler()]
-    else:
-        handler = logging.StreamHandler()
-        handler.setFormatter(
-            logging.Formatter(
-                "%(asctime)s %(levelname)-8s %(name)s:%(lineno)d %(message)s"
-            )
-        )
-        logger.handlers = [handler]
+    logger.handlers = [create_log_handler()]
     logger.propagate = False
-
-    # Read log level from environment variable with default fallback
-    level_str = os.environ.get(LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, DEFAULT_LOG_LEVEL)
-
-    # Validate the level string and convert to logging level constant
-    validated_level = getattr(logging, level_str.upper(), None)
-    if not isinstance(validated_level, int):
-        logger.warning(
-            "Invalid log level '%s', falling back to %s",
-            level_str,
-            DEFAULT_LOG_LEVEL,
-        )
-        validated_level = getattr(logging, DEFAULT_LOG_LEVEL)
-
-    logger.setLevel(validated_level)
+    logger.setLevel(resolve_log_level())
     return logger

src/runners/uvicorn.py

@@ -1,39 +1,15 @@
 """Uvicorn runner."""
 
 import logging
-import os
 
 import uvicorn
 
-from constants import DEFAULT_LOG_LEVEL, LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR
-from log import get_logger
+from log import get_logger, resolve_log_level
 from models.config import ServiceConfiguration
 
 logger = get_logger(__name__)
 
 
-def _resolve_log_level() -> int:
-    """Resolve the uvicorn log level from the environment.
-
-    Reads the LIGHTSPEED_STACK_LOG_LEVEL environment variable and converts it
-    to a Python logging level constant. Falls back to the default log level
-    when the variable is unset or contains an invalid value.
-
-    Returns:
-        The resolved logging level as an integer constant.
-    """
-    level_str = os.environ.get(LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, DEFAULT_LOG_LEVEL)
-    level = getattr(logging, level_str.upper(), None)
-    if not isinstance(level, int):
-        logger.warning(
-            "Invalid log level '%s', falling back to %s",
-            level_str,
-            DEFAULT_LOG_LEVEL,
-        )
-        level = getattr(logging, DEFAULT_LOG_LEVEL)
-    return level
-
-
 def start_uvicorn(configuration: ServiceConfiguration) -> None:
     """Start the Uvicorn server using the provided service configuration.
 
@@ -43,7 +19,7 @@ def start_uvicorn(configuration: ServiceConfiguration) -> None:
         `tls_certificate_path`, and `tls_key_password`). TLS fields may be None
         and will be forwarded to uvicorn.run as provided.
     """
-    log_level = _resolve_log_level()
+    log_level = resolve_log_level()
     logger.info("Starting Uvicorn with log level %s", logging.getLevelName(log_level))
 
     # please note:

tests/unit/runners/test_uvicorn_runner.py

@@ -7,8 +7,9 @@
 from pytest_mock import MockerFixture
 
 from constants import LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR
+from log import resolve_log_level
 from models.config import ServiceConfiguration, TLSConfiguration
-from runners.uvicorn import _resolve_log_level, start_uvicorn
+from runners.uvicorn import start_uvicorn
 
 
 def test_start_uvicorn(mocker: MockerFixture) -> None:
@@ -146,17 +147,17 @@ def test_start_uvicorn_with_root_path(mocker: MockerFixture) -> None:
 def test_resolve_log_level_from_env(
     monkeypatch: pytest.MonkeyPatch, env_value: str, expected_level: int
 ) -> None:
-    """Test that _resolve_log_level resolves env var values to logging constants."""
+    """Test that resolve_log_level resolves env var values to logging constants."""
     monkeypatch.setenv(LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, env_value)
-    assert _resolve_log_level() == expected_level
+    assert resolve_log_level() == expected_level
 
 
 def test_resolve_log_level_defaults_to_info(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """Test that _resolve_log_level falls back to INFO when the env var is unset."""
+    """Test that resolve_log_level falls back to INFO when the env var is unset."""
     monkeypatch.delenv(LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, raising=False)
-    assert _resolve_log_level() == logging.INFO
+    assert resolve_log_level() == logging.INFO
 
 
 def test_start_uvicorn_respects_debug_log_level(

tests/unit/test_log.py

@@ -1,10 +1,13 @@
 """Unit tests for functions defined in src/log.py."""
 
 import logging
+
 import pytest
+from pytest_mock import MockerFixture
+from rich.logging import RichHandler
 
-from log import get_logger
-from constants import LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR
+from log import get_logger, resolve_log_level, create_log_handler
+from constants import LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, DEFAULT_LOG_FORMAT
 
 
 def test_get_logger() -> None:
@@ -57,3 +60,59 @@ def test_get_logger_default_log_level(monkeypatch: pytest.MonkeyPatch) -> None:
 
     logger = get_logger("test_default")
     assert logger.level == logging.INFO
+
+
+@pytest.mark.parametrize(
+    ("level_name", "expected_level"),
+    [
+        ("DEBUG", logging.DEBUG),
+        ("debug", logging.DEBUG),
+        ("INFO", logging.INFO),
+        ("WARNING", logging.WARNING),
+        ("ERROR", logging.ERROR),
+        ("CRITICAL", logging.CRITICAL),
+        ("critical", logging.CRITICAL),
+    ],
+)
+def test_resolve_log_level(
+    monkeypatch: pytest.MonkeyPatch, level_name: str, expected_level: int
+) -> None:
+    """Test that resolve_log_level correctly resolves valid level names."""
+    monkeypatch.setenv(LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, level_name)
+    assert resolve_log_level() == expected_level
+
+
+def test_resolve_log_level_invalid_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Test that resolve_log_level falls back to INFO for invalid values."""
+    monkeypatch.setenv(LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, "BOGUS")
+    assert resolve_log_level() == logging.INFO
+
+
+def test_resolve_log_level_default(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Test that resolve_log_level defaults to INFO when env var is unset."""
+    monkeypatch.delenv(LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR, raising=False)
+    assert resolve_log_level() == logging.INFO
+
+
+def test_create_log_handler_tty(mocker: MockerFixture) -> None:
+    """Test that create_log_handler returns RichHandler when TTY is available."""
+    mocker.patch("sys.stderr.isatty", return_value=True)
+    handler = create_log_handler()
+    assert isinstance(handler, RichHandler)
+
+
+def test_create_log_handler_non_tty(mocker: MockerFixture) -> None:
+    """Test that create_log_handler returns StreamHandler when no TTY."""
+    mocker.patch("sys.stderr.isatty", return_value=False)
+    handler = create_log_handler()
+    assert isinstance(handler, logging.StreamHandler)
+    assert not isinstance(handler, RichHandler)
+
+
+def test_create_log_handler_non_tty_format(mocker: MockerFixture) -> None:
+    """Test that non-TTY handler uses DEFAULT_LOG_FORMAT."""
+    mocker.patch("sys.stderr.isatty", return_value=False)
+    handler = create_log_handler()
+    assert handler.formatter is not None
+    # pylint: disable=protected-access
+    assert handler.formatter._fmt == DEFAULT_LOG_FORMAT