Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 84 additions & 0 deletions app/core/logging_setup.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
"""File logging for the stemdeck logger tree (#291).

Until now the app logged to stdout only (via uvicorn's root handler): server
and Docker deployments kept no log file at all, and LOGS_DIR existed but was
never written to. This module attaches a rotating file handler to the
"stemdeck" logger so every deployment keeps a bounded on-disk trail:

LOGS_DIR/stemdeck.log (5 MB x 3 backups, UTF-8, timestamped)

Level control:
- STEMDECK_LOG_LEVEL=DEBUG|INFO|WARNING (default INFO)
- STEMDECK_DEBUG=1 (shorthand for DEBUG; enables the
per-job analyze diagnostics: "chroma:", "key candidates:")

Everything here is best-effort: a read-only filesystem (locked-down Docker)
must never prevent startup, so failures degrade to stdout-only logging.
"""

from __future__ import annotations

import logging
import os
import sys
from logging.handlers import RotatingFileHandler

from app.core.config import LOGS_DIR

# Module-level so tests can shrink them to exercise rotation.
_MAX_BYTES = 5 * 1024 * 1024
_BACKUP_COUNT = 3

# Marker attribute so repeat calls (uvicorn --reload re-imports app.main)
# don't stack duplicate handlers.
_HANDLER_MARK = "_stemdeck_file_handler"

_LEVELS = {"DEBUG": logging.DEBUG, "INFO": logging.INFO, "WARNING": logging.WARNING}


def _resolve_level() -> int:
if os.environ.get("STEMDECK_DEBUG", "").strip() == "1":
return logging.DEBUG
name = os.environ.get("STEMDECK_LOG_LEVEL", "").strip().upper()
return _LEVELS.get(name, logging.INFO)


def configure_logging() -> None:
"""Set the stemdeck logger level and attach the rotating file handler.

Propagation stays on, so records continue to flow to uvicorn's stdout
handler exactly as before -- the file is additive.
"""
root = logging.getLogger("stemdeck")
root.setLevel(_resolve_level())

if any(getattr(h, _HANDLER_MARK, False) for h in root.handlers):
return # already configured (reload / repeated import)

try:
LOGS_DIR.mkdir(parents=True, exist_ok=True)
# delay=True: don't open the file until the first record, so a
# read-only FS fails at emit time (swallowed by logging's internal
# error handling) instead of at startup.
handler = RotatingFileHandler(
LOGS_DIR / "stemdeck.log",
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
delay=True,
)
except OSError:
print(
f"stemdeck: file logging disabled (cannot use logs dir {LOGS_DIR})",
file=sys.stderr,
)
return

handler.setFormatter(
logging.Formatter(
"%(asctime)s %(levelname).1s %(name)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
)
setattr(handler, _HANDLER_MARK, True)
root.addHandler(handler)
12 changes: 7 additions & 5 deletions app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
configure_portable_environment,
ensure_runtime_dirs,
)
from app.core.logging_setup import configure_logging
from app.core.registry import restore as restore_registry
from app.core.settings import (
get_allow_network,
Expand All @@ -45,11 +46,12 @@
)
from app.pipeline.collect import sweep_old_jobs

# Show our INFO-level logs through uvicorn's root handler. Without this,
# Python's default root level (WARNING) silently drops every
# logger.info(...) call across the app, including the analyze
# diagnostics ("chroma:", "key candidates:").
logging.getLogger("stemdeck").setLevel(logging.INFO)
# Set the stemdeck logger level (Python's default root level of WARNING would
# silently drop every logger.info(...) call) and attach the rotating file log
# at LOGS_DIR/stemdeck.log. The analyze diagnostics ("chroma:", "key
# candidates:") are DEBUG-level -- set STEMDECK_DEBUG=1 (or
# STEMDECK_LOG_LEVEL=DEBUG) to see them.
configure_logging()
logging.getLogger("stemdeck").info(
"demucs config: model=%s device=%s", DEMUCS_MODEL, get_demucs_device()
)
Expand Down
6 changes: 4 additions & 2 deletions app/pipeline/analyze.py
Original file line number Diff line number Diff line change
Expand Up @@ -325,7 +325,9 @@ def analyze(job: Job, source: Path) -> tuple[int | None, str | None]:
stage="Analysis complete",
)
return bpm, key
except Exception as e:
except Exception:
# Full traceback goes to the log; the UI stage line stays generic --
# raw exception reprs (paths, library internals) must not reach it.
logger.exception("analyze failed for job %s", job.id)
_set(job, stage=f"Analysis skipped ({e})")
_set(job, stage="Analysis skipped")
return None, None
95 changes: 95 additions & 0 deletions tests/test_logging_setup.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
"""Tests for the rotating file log setup (#291, #292)."""

from __future__ import annotations

import logging

import pytest

from app.core import logging_setup


def _strip_our_handlers(root: logging.Logger) -> None:
for h in list(root.handlers):
if getattr(h, logging_setup._HANDLER_MARK, False):
root.removeHandler(h)
h.close()


@pytest.fixture()
def _clean_logger(monkeypatch, tmp_path):
"""Point LOGS_DIR at a temp dir and strip our handler around each test.

Stripping BEFORE the test matters: importing app.main anywhere in the
suite already ran configure_logging() against the real LOGS_DIR, and the
idempotence guard would otherwise skip attaching a handler here."""
monkeypatch.setattr(logging_setup, "LOGS_DIR", tmp_path / "logs")
root = logging.getLogger("stemdeck")
saved_level = root.level
_strip_our_handlers(root)
yield root
_strip_our_handlers(root)
root.setLevel(saved_level)


def _our_handlers(root: logging.Logger) -> list[logging.Handler]:
return [h for h in root.handlers if getattr(h, logging_setup._HANDLER_MARK, False)]


def test_creates_log_file_on_first_record(_clean_logger, tmp_path):
logging_setup.configure_logging()
log_file = tmp_path / "logs" / "stemdeck.log"
assert not log_file.exists() # delay=True: nothing written yet
logging.getLogger("stemdeck.test").info("hello file log")
assert log_file.is_file()
text = log_file.read_text(encoding="utf-8")
assert "hello file log" in text
assert "stemdeck.test" in text


def test_idempotent_across_repeat_calls(_clean_logger):
logging_setup.configure_logging()
logging_setup.configure_logging() # uvicorn --reload re-imports app.main
assert len(_our_handlers(_clean_logger)) == 1


def test_rotation_keeps_bounded_backups(_clean_logger, monkeypatch, tmp_path):
monkeypatch.setattr(logging_setup, "_MAX_BYTES", 200)
monkeypatch.setattr(logging_setup, "_BACKUP_COUNT", 2)
logging_setup.configure_logging()
log = logging.getLogger("stemdeck.test")
for i in range(30):
log.info("filler record %03d %s", i, "x" * 40)
logs_dir = tmp_path / "logs"
assert (logs_dir / "stemdeck.log").is_file()
assert (logs_dir / "stemdeck.log.1").is_file()
assert not (logs_dir / "stemdeck.log.3").exists() # bounded at backupCount


def test_level_from_env(_clean_logger, monkeypatch):
monkeypatch.setenv("STEMDECK_LOG_LEVEL", "DEBUG")
logging_setup.configure_logging()
assert _clean_logger.level == logging.DEBUG


def test_debug_shorthand_wins(_clean_logger, monkeypatch):
monkeypatch.setenv("STEMDECK_LOG_LEVEL", "WARNING")
monkeypatch.setenv("STEMDECK_DEBUG", "1")
logging_setup.configure_logging()
assert _clean_logger.level == logging.DEBUG


def test_bogus_level_falls_back_to_info(_clean_logger, monkeypatch):
monkeypatch.setenv("STEMDECK_LOG_LEVEL", "SHOUTING")
logging_setup.configure_logging()
assert _clean_logger.level == logging.INFO


def test_unwritable_logs_dir_degrades_gracefully(_clean_logger, monkeypatch, tmp_path):
# LOGS_DIR path occupied by a *file*: mkdir raises, startup must not.
blocker = tmp_path / "blocked"
blocker.write_text("not a dir", encoding="utf-8")
monkeypatch.setattr(logging_setup, "LOGS_DIR", blocker)
logging_setup.configure_logging() # must not raise
assert _our_handlers(_clean_logger) == []
assert _clean_logger.level == logging.INFO # level still applied