lhk229/BGfilter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Split logging: prose event stream vs structured JSONL access stream

Browse Source 
Rule: bulk per-request data never lands in the prose log; everything with
analysis value goes to a structured file.

- Event stream (stderr, text): startup summary, warnings, tracebacks --
  what supervisord/journald already captures. Human-readable, low volume.
- Access stream: one JSONL record per request (plus a "startup" meta
  record with the instance config) written to BGFILTER_ACCESS_LOG,
  app-rotated 50MB x 5; falls back to stderr when unset (dev).
  Failed requests carry a truncated `exc` copy so the access file is
  self-contained; the full traceback stays in the event stream, joined
  by request_id.
- BGFILTER_LOG_FORMAT retired: format is now a property of the stream,
  not a global switch.
- uvicorn runs with --no-access-log everywhere (deploy script + unit
  examples): its prose per-request lines duplicated a subset of ours.
- deploy_autodl.sh --log-dir now provisions both files; docs updated
  (EN+ZH: env table, AutoDL section, run/systemd examples).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
孔令弘
2026-07-17 13:40:33 +00:00
parent 3a8a1ef38a
commit 43797eedeb
4 changed files with 98 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+64 -16
View File
@@ -16,9 +16,11 @@ import logging
import os
import re
import time
import traceback
import uuid
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from logging.handlers import RotatingFileHandler
from pathlib import Path
from urllib.parse import urlparse
@@ -33,7 +35,6 @@ from bgfilter.service import SEG_MODELS, PipelineManager
# Per-request correlation id. A middleware sets it; a logging filter injects it into
# every record so all logs for one request share the id. contextvars (not thread-
# locals) so a pooled worker thread never inherits the previous request's id.
LOG_FORMAT = os.environ.get("BGFILTER_LOG_FORMAT", "json").strip().lower()
_request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")
_RID_RE = re.compile(r"^[A-Za-z0-9._-]{1,64}$")
@@ -72,24 +73,45 @@ class _JsonFormatter(logging.Formatter):
return json.dumps(out, ensure_ascii=False)
_log_handler = logging.StreamHandler()
_log_handler.addFilter(_RequestIdFilter())
_log_handler.setFormatter(
_JsonFormatter()
if LOG_FORMAT == "json"
else logging.Formatter("%(asctime)s %(levelname)s %(name)s [%(request_id)s] %(message)s")
# Two log streams with distinct jobs (never mixed into one file):
#
# * Event stream -- prose for humans, low volume: startup summary, warnings, error
# tracebacks. Goes to stderr (text format) where supervisord/journald captures it.
# Per-request records must NOT be logged here: bulk structured content does not
# belong in the prose file.
# * Access stream -- structured JSONL, one record per request (plus one "startup"
# meta record): everything with analysis value. Written to the file named by
# BGFILTER_ACCESS_LOG (rotated by the app itself: supervisord can't rotate a file
# it doesn't own); when unset (dev), the JSON lines fall back to stderr.
#
# Both streams stamp the same request_id, so a traceback in the event stream joins
# back to its access record (which also carries a truncated `exc` copy).
_event_handler = logging.StreamHandler()
_event_handler.addFilter(_RequestIdFilter())
_event_handler.setFormatter(
logging.Formatter("%(asctime)s %(levelname)s %(name)s [%(request_id)s] %(message)s")
)
logging.basicConfig(level=os.environ.get("BGFILTER_LOG_LEVEL", "INFO"), handlers=[_log_handler])
logging.basicConfig(level=os.environ.get("BGFILTER_LOG_LEVEL", "INFO"), handlers=[_event_handler])
logger = logging.getLogger("bgfilter.app")
ACCESS_LOG_PATH = os.environ.get("BGFILTER_ACCESS_LOG", "").strip()
access_logger = logging.getLogger("bgfilter.access")
access_logger.propagate = False # never leak per-request records into the prose stream
if ACCESS_LOG_PATH:
Path(ACCESS_LOG_PATH).parent.mkdir(parents=True, exist_ok=True)
_access_handler: logging.Handler = RotatingFileHandler(
ACCESS_LOG_PATH, maxBytes=50 * 1024 * 1024, backupCount=5, encoding="utf-8"
)
else:
_access_handler = logging.StreamHandler()
_access_handler.addFilter(_RequestIdFilter())
_access_handler.setFormatter(_JsonFormatter())
access_logger.addHandler(_access_handler)
def _emit_access(fields: dict) -> None:
"""One access line per request. JSON mode merges the fields; text mode joins them."""
if LOG_FORMAT == "json":
access_logger.info("request", extra={"fields": fields})
else:
access_logger.info(" ".join(f"{k}={v}" for k, v in fields.items() if k != "request_id"))
def _emit_access(event: str, fields: dict) -> None:
"""One JSONL record on the access stream (`msg` = record type)."""
access_logger.info(event, extra={"fields": fields})
VERSION = "0.1.0"
_HEX_RE = re.compile(r"^#[0-9a-fA-F]{6}$")
@@ -200,6 +222,7 @@ async def lifespan(app: FastAPI):
timeout=httpx.Timeout(connect=5.0, read=FETCH_TIMEOUT_S, write=10.0, pool=None),
limits=httpx.Limits(max_connections=fetch_conc),
)
gpu_concurrency = max(1, _int_env("BGFILTER_GPU_CONCURRENCY", settings.server.gpu_concurrency))
logger.info(
"starting bgfilter service: device=%s default_seg=%s cross_check=%s preload=%s "
"cpu_workers=%d gpu_concurrency=%d max_pixels=%d auth=%s",
@@ -208,10 +231,27 @@ async def lifespan(app: FastAPI):
settings.cross_check.enabled,
preload,
cpu_workers,
max(1, _int_env("BGFILTER_GPU_CONCURRENCY", settings.server.gpu_concurrency)),
gpu_concurrency,
MAX_IMAGE_PIXELS,
"enabled" if AUTH_TOKEN else "DISABLED(open)",
)
# Same facts as a structured record on the access stream, so per-instance config
# is queryable next to the requests it served (no prose parsing).
_emit_access(
"startup",
{
"version": VERSION,
"device": settings.model.device,
"default_seg": settings.segmentation.backend,
"cross_check": settings.cross_check.enabled,
"reuse_as_seg": settings.cross_check.reuse_as_seg,
"preload": preload,
"cpu_workers": cpu_workers,
"gpu_concurrency": gpu_concurrency,
"max_pixels": MAX_IMAGE_PIXELS,
"auth": "enabled" if AUTH_TOKEN else "open",
},
)
if not AUTH_TOKEN:
logger.warning(
"BGFILTER_AUTH_TOKEN not set: /remove-background is OPEN (no app-level auth)"
@@ -244,13 +284,19 @@ async def _access_log_mw(request: Request, call_next):
status = response.status_code
response.headers["X-Request-ID"] = rid
return response
except Exception:
# Unhandled escape (HTTPExceptions were already turned into responses by the
# inner exception middleware). setdefault: a more precise traceback recorded
# at the raise site wins over this generic one.
request.state.fields.setdefault("exc", traceback.format_exc()[-2000:])
raise
finally:
if request.url.path != "/healthz":
fields = {"request_id": rid, "client": request.client.host if request.client else "-"}
fields.update(getattr(request.state, "fields", {}) or {})
fields["status"] = status
fields["t_total_ms"] = round((time.perf_counter() - t0) * 1000)
_emit_access(fields)
_emit_access("request", fields)
@app.get("/healthz")
@@ -475,9 +521,11 @@ async def remove_background(
detail="could not auto-detect background colour; pass screen_color",
)
logger.exception("inference failed")
log["exc"] = traceback.format_exc()[-2000:]
raise HTTPException(status_code=500, detail="inference failed")
except Exception:
logger.exception("inference failed")
log["exc"] = traceback.format_exc()[-2000:]
raise HTTPException(status_code=500, detail="inference failed")
elapsed_ms = int((time.perf_counter() - start) * 1000)
log["t_process_ms"] = elapsed_ms