rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-10-10 09:06:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(web): use src.logging_config in the web process; routine requests to DEBUG

Browse Source 
The web interface had its own logging setup (web_interface/logging_config.py)
that replaced the root handlers with a plain stdout formatter. The web
service's journal lines therefore never carried a syslog priority, so
`journalctl -p err -u ledmatrix-web` returned nothing while errors were
logged, and the line shape differed from the display's (the log viewer's
prefix stripping only matched the display format). It also ran after the
module-level managers were built, so their INFO lines at import (including
"Re-removed N uninstalled plugin(s)") were dropped.

app.py now calls src.logging_config.setup_logging() first thing, the same as
run.py: journald priorities under systemd, LEDMATRIX_DEBUG honoured,
LEDMATRIX_JSON_LOGGING still selects JSON.

Per-request logging moves to web_interface/request_logging.py. Every request
used to be logged at INFO, so the UI's polling filled the journal
("GET /api/v3/errors/summary - 200" every minute per tab). Now a successful
GET/HEAD/OPTIONS is DEBUG, a successful write is INFO, 4xx WARNING, 5xx
ERROR. Durations use perf_counter and print to 0.1ms.

The duplicate module is deleted; nothing else imported it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-09-23 16:58:54 -04:00
co-authored by Claude Opus 5.5
parent f3894916a9
commit 70eab6fb25
4 changed files with 240 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/web_interface/test_web_logging.py
+164
View File
@@ -0,0 +1,164 @@
"""The web process logs the way the display process does.
web_interface/app.py used to call its own setup (web_interface/logging_config.py)
which replaced the root handlers with a plain stdout formatter. Under systemd
every line then reached the journal as PRIORITY=6, so
journalctl -p err -u ledmatrix-web
showed nothing while the web interface was logging errors. It also logged
every request at INFO, including what the UI polls: the journal on a Pi showed
``GET /api/v3/errors/summary - 200`` every minute per open tab.
"""
import logging
import os
import subprocess
import sys
import textwrap
from pathlib import Path
import pytest
from flask import Flask
from web_interface import request_logging
PROJECT_ROOT = Path(__file__).resolve().parents[2]
# ---------------------------------------------------------------------------
# The real app, imported the way systemd runs it
# ---------------------------------------------------------------------------
_CHILD = textwrap.dedent("""
import logging
import web_interface.app as web_app
# Startup reconciliation may try to reinstall plugins; not this test's job.
web_app._reconciliation_started = True
client = web_app.app.test_client()
client.get('/api/v3/errors/summary')
client.get('/favicon.ico')
client.get('/api/v3/no-such-endpoint')
logging.getLogger('web_interface.probe').error('probe error line')
logging.getLogger('web_interface.probe').info('probe info line')
""")
@pytest.fixture(scope="module")
def journal_output(tmp_path_factory):
"""Run the child with stdout as a file systemd would call the journal.
systemd sets JOURNAL_STREAM to the dev:ino of the stream it captures;
src.logging_config only adds priorities when stdout really is that stream,
so hand the child a file and name that file's dev:ino.
"""
out_path = tmp_path_factory.mktemp("journal") / "stdout.txt"
with open(out_path, "wb") as out:
st = os.fstat(out.fileno())
env = dict(os.environ)
env.update({
"JOURNAL_STREAM": f"{st.st_dev}:{st.st_ino}",
"PYTHONUTF8": "1",
"EMULATOR": "true",
"PYTHONPATH": str(PROJECT_ROOT),
})
env.pop("LEDMATRIX_DEBUG", None)
env.pop("LEDMATRIX_JSON_LOGGING", None)
proc = subprocess.run(
[sys.executable, "-c", _CHILD], cwd=str(PROJECT_ROOT), env=env,
stdout=out, stderr=subprocess.PIPE, timeout=180,
)
text = out_path.read_text(encoding="utf-8", errors="replace")
assert proc.returncode == 0, proc.stderr.decode(errors="replace")[-4000:]
return text.splitlines()
def test_error_reaches_the_journal_as_err(journal_output):
lines = [l for l in journal_output if "probe error line" in l]
assert lines, "\n".join(journal_output[-40:])
assert lines[0].startswith("<3>"), lines[0]
# Same readable shape as the display service (and what the log viewer strips).
assert " - ERROR - web_interface.probe - probe error line" in lines[0]
def test_info_reaches_the_journal_as_info(journal_output):
lines = [l for l in journal_output if "probe info line" in l]
assert lines and lines[0].startswith("<6>"), journal_output[-40:]
def test_polling_gets_are_not_logged_at_info(journal_output):
for path in ("/api/v3/errors/summary", "/favicon.ico"):
assert not [l for l in journal_output if f"GET {path} " in l], (
f"a successful GET {path} was logged by default")
def test_failed_request_is_still_logged(journal_output):
lines = [l for l in journal_output if "GET /api/v3/no-such-endpoint - 404" in l]
assert lines and lines[0].startswith("<4>"), journal_output[-40:]
# ---------------------------------------------------------------------------
# The level policy
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("method,status,level", [
("GET", 200, logging.DEBUG),
("GET", 304, logging.DEBUG),
("HEAD", 200, logging.DEBUG),
("OPTIONS", 204, logging.DEBUG),
("get", 200, logging.DEBUG),
("POST", 200, logging.INFO),
("PUT", 204, logging.INFO),
("DELETE", 200, logging.INFO),
("PATCH", 302, logging.INFO),
("GET", 404, logging.WARNING),
("POST", 400, logging.WARNING),
("GET", 500, logging.ERROR),
("POST", 503, logging.ERROR),
])
def test_request_log_level(method, status, level):
assert request_logging.request_log_level(method, status) == level
@pytest.fixture
def tiny_app():
app = Flask(__name__)
request_logging.init_app(app)
@app.route("/poll")
def poll():
return "ok"
@app.route("/save", methods=["POST"])
def save():
return "saved"
@app.route("/boom")
def boom():
return "no", 500
return app.test_client()
def test_hooks_log_each_request_once_at_its_level(tiny_app, caplog):
caplog.set_level(logging.DEBUG, logger="web_interface.api")
tiny_app.get("/poll")
tiny_app.post("/save")
tiny_app.get("/boom")
tiny_app.get("/missing")
got = [(r.levelno, r.getMessage().split(" (")[0]) for r in caplog.records
if r.name == "web_interface.api"]
assert got == [
(logging.DEBUG, "GET /poll - 200"),
(logging.INFO, "POST /save - 200"),
(logging.ERROR, "GET /boom - 500"),
(logging.WARNING, "GET /missing - 404"),
]
def test_duration_is_rounded(tiny_app, caplog):
caplog.set_level(logging.DEBUG, logger="web_interface.api")
tiny_app.post("/save")
msg = caplog.records[-1].getMessage()
duration = msg.rsplit("(", 1)[1]
assert duration.endswith("ms)") and len(duration.split(".")[1]) == len("0ms)"), msg
web_interface/app.py
+14 -35
View File
@@ -16,6 +16,17 @@ from datetime import datetime, timedelta
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
# Configure logging before anything below logs: the same setup as the display
# service (run.py), so this process's journal lines carry their real syslog
# priority too (`journalctl -p err -u ledmatrix-web`). LEDMATRIX_DEBUG=true
# turns on DEBUG, which includes the routine per-request lines.
from src.logging_config import setup_logging
setup_logging(format_type=(
'json' if os.environ.get('LEDMATRIX_JSON_LOGGING', 'false').lower() == 'true'
else 'readable'))
logging.getLogger('werkzeug').setLevel(logging.WARNING) # request_logging covers requests
logging.getLogger('urllib3').setLevel(logging.WARNING)
from src.config_manager import ConfigManager
from src.web_interface.error_handler import describe_exception
from src.common.path_safety import (
@@ -346,41 +357,9 @@ def success_txt():
return redirect(url_for('pages_v3.captive_setup'), code=302)
return 'success', 200
# Initialize logging
try:
from web_interface.logging_config import setup_web_interface_logging, log_api_request
# Use JSON logging in production, readable logs in development
use_json_logging = os.environ.get('LEDMATRIX_JSON_LOGGING', 'false').lower() == 'true'
setup_web_interface_logging(level='INFO', use_json=use_json_logging)
except ImportError:
# Logging config not available, use default
log_api_request = None
# Request timing and logging middleware
@app.before_request
def before_request():
"""Track request start time for logging."""
from flask import request
request.start_time = time.time()
@app.after_request
def after_request_logging(response):
"""Log API requests after response."""
if log_api_request:
try:
from flask import request
duration_ms = (time.time() - getattr(request, 'start_time', time.time())) * 1000
ip_address = request.remote_addr if hasattr(request, 'remote_addr') else None
log_api_request(
method=request.method,
path=request.path,
status_code=response.status_code,
duration_ms=duration_ms,
ip_address=ip_address
)
except Exception: # nosec B110 - request logging must never interrupt a live HTTP response
pass # Don't break response if logging fails
return response
# Request timing and logging (routine reads at DEBUG; see request_logging)
from web_interface import request_logging
request_logging.init_app(app)
# Global error handlers
@app.errorhandler(404)
web_interface/logging_config.py
-110
View File
@@ -1,110 +0,0 @@
"""
Structured logging configuration for the web interface.
Provides JSON-formatted logs for production and readable logs for development.
"""
import logging
import json
import sys
from datetime import datetime
from typing import Optional
class JSONFormatter(logging.Formatter):
"""Formatter that outputs logs as JSON for structured logging."""
def format(self, record: logging.LogRecord) -> str:
"""Format log record as JSON."""
log_data = {
'timestamp': datetime.utcnow().isoformat(),
'level': record.levelname,
'logger': record.name,
'message': record.getMessage(),
'module': record.module,
'function': record.funcName,
'line': record.lineno,
}
# Add exception info if present
if record.exc_info:
log_data['exception'] = self.formatException(record.exc_info)
# Add extra fields if present
if hasattr(record, 'request_id'):
log_data['request_id'] = record.request_id
if hasattr(record, 'user_id'):
log_data['user_id'] = record.user_id
if hasattr(record, 'ip_address'):
log_data['ip_address'] = record.ip_address
if hasattr(record, 'duration_ms'):
log_data['duration_ms'] = record.duration_ms
return json.dumps(log_data)
def setup_web_interface_logging(level: str = 'INFO', use_json: bool = False):
"""
Set up logging for the web interface.
Args:
level: Log level (DEBUG, INFO, WARNING, ERROR)
use_json: If True, use JSON formatting (for production)
"""
# Get root logger
logger = logging.getLogger()
logger.setLevel(getattr(logging, level.upper()))
# Remove existing handlers
logger.handlers.clear()
# Create console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(getattr(logging, level.upper()))
# Set formatter
if use_json:
formatter = JSONFormatter()
else:
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
# Set levels for specific loggers
logging.getLogger('werkzeug').setLevel(logging.WARNING) # Reduce Flask noise
logging.getLogger('urllib3').setLevel(logging.WARNING) # Reduce HTTP noise
def log_api_request(method: str, path: str, status_code: int, duration_ms: float,
ip_address: Optional[str] = None, **kwargs):
"""
Log an API request with structured data.
Args:
method: HTTP method
path: Request path
status_code: HTTP status code
duration_ms: Request duration in milliseconds
ip_address: Client IP address
**kwargs: Additional context
"""
logger = logging.getLogger('web_interface.api')
extra = {
'method': method,
'path': path,
'status_code': status_code,
'duration_ms': round(duration_ms, 2),
'ip_address': ip_address,
**kwargs
}
# Log at appropriate level based on status code
if status_code >= 500:
logger.error(f"{method} {path} - {status_code} ({duration_ms}ms)", extra=extra)
elif status_code >= 400:
logger.warning(f"{method} {path} - {status_code} ({duration_ms}ms)", extra=extra)
else:
logger.info(f"{method} {path} - {status_code} ({duration_ms}ms)", extra=extra)
web_interface/request_logging.py
+62
View File
@@ -0,0 +1,62 @@
"""
Per-request logging for the web interface.
Logging itself is configured by ``src.logging_config.setup_logging`` (the same
formatter and journald priorities as the display service); this module only
decides what one HTTP request is worth logging, and at which level.
The UI polls: the error summary, system status, display preview and log
streams are fetched every few seconds by every open tab. Logging each of those
at INFO buried everything else in the journal (``GET /api/v3/errors/summary -
200`` once a minute per tab, forever). So a request that only read something
and succeeded is DEBUG; one that changed something, or failed, is logged at a
level that shows up by default.
"""
import logging
import time
from flask import Flask, request
logger = logging.getLogger('web_interface.api')
#: Methods that do not change server state. A successful one is routine.
_READ_ONLY_METHODS = frozenset({'GET', 'HEAD', 'OPTIONS'})
def request_log_level(method: str, status_code: int) -> int:
"""The level a finished request is logged at."""
if status_code >= 500:
return logging.ERROR
if status_code >= 400:
return logging.WARNING
if method.upper() in _READ_ONLY_METHODS:
return logging.DEBUG
return logging.INFO
def log_request(method: str, path: str, status_code: int,
duration_ms: float) -> None:
"""Log one finished request."""
level = request_log_level(method, status_code)
if logger.isEnabledFor(level):
logger.log(level, "%s %s - %d (%.1fms)",
method, path, status_code, duration_ms)
def init_app(app: Flask) -> None:
"""Time every request and log it when its response is ready."""
@app.before_request
def _start_request_timer():
request.start_time = time.perf_counter()
@app.after_request
def _log_finished_request(response):
try:
started = getattr(request, 'start_time', None)
duration_ms = 0.0 if started is None else (time.perf_counter() - started) * 1000
log_request(request.method, request.path, response.status_code,
duration_ms)
except Exception: # nosec B110 - request logging must never interrupt a live HTTP response
pass
return response