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

Add structured per-request access logging with a request id

Browse Source 
Each request now emits one access line (bgfilter.access) carrying a
request_id, the effective params (source file/url, seg_model, cross_check,
background_mode, screen_color, auth), input/output sizes, and stage
timings (t_queue/fetch/decode/process/total_ms). A middleware binds the id
(reusing a sanitized inbound X-Request-ID, else uuid), a logging filter
injects it into every record, and it is echoed back as the X-Request-ID
response header on every response -- including errors, which also log the
detail. Requests that raise still produce a line (fields accumulate on
request.state; the middleware finalizes status/total/client).

request_id lives in a contextvar (not a thread-local) so a pooled worker
never inherits the previous request's id; propagating it into the executor
for deep pipeline logs is left for later. BGFILTER_LOG_FORMAT=text|json
switches human vs shipper-friendly output. /healthz is not logged.

Verified locally via TestClient (fake manager, no models): 200 path logs
all fields + timings with rid == header; inbound id echoed; over-long id
sanitized; XOR 400s log status + error.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
孔令弘
2026-07-15 05:55:23 +00:00
parent 6506bdd643
commit dc4badd0e0
5 changed files with 194 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+188 -75
View File
@@ -7,12 +7,16 @@ and a global lock. All model handling lives in ``bgfilter.service.PipelineManage
from __future__ import annotations
import asyncio
import contextvars
import datetime
import hmac
import io
import json
import logging
import os
import re
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from pathlib import Path
@@ -20,17 +24,72 @@ from urllib.parse import urlparse
import httpx
import numpy as np
from fastapi import Depends, FastAPI, File, Form, Header, HTTPException, Response, UploadFile
from fastapi import Depends, FastAPI, File, Form, Header, HTTPException, Request, Response, UploadFile
from PIL import Image, ImageOps, UnidentifiedImageError
from bgfilter.config import load_settings, override_settings
from bgfilter.service import SEG_MODELS, PipelineManager
logging.basicConfig(
level=os.environ.get("BGFILTER_LOG_LEVEL", "INFO"),
format="%(asctime)s %(levelname)s %(name)s %(message)s",
# 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", "text").strip().lower()
_request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")
_RID_RE = re.compile(r"^[A-Za-z0-9._-]{1,64}$")
def _make_request_id(incoming: str | None) -> str:
"""Reuse a sanitized inbound X-Request-ID (e.g. nginx $request_id), else generate."""
if incoming and _RID_RE.match(incoming):
return incoming
return uuid.uuid4().hex[:16]
class _RequestIdFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
record.request_id = _request_id_var.get()
return True
class _JsonFormatter(logging.Formatter):
"""One JSON object per record; per-request access fields ride in `extra={"fields"}`."""
def format(self, record: logging.LogRecord) -> str:
out = {
"ts": datetime.datetime.fromtimestamp(record.created, datetime.timezone.utc)
.isoformat(timespec="milliseconds")
.replace("+00:00", "Z"),
"level": record.levelname,
"logger": record.name,
"request_id": getattr(record, "request_id", "-"),
"msg": record.getMessage(),
}
fields = getattr(record, "fields", None)
if isinstance(fields, dict):
out.update(fields)
if record.exc_info:
out["exc"] = self.formatException(record.exc_info)
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")
)
logging.basicConfig(level=os.environ.get("BGFILTER_LOG_LEVEL", "INFO"), handlers=[_log_handler])
logger = logging.getLogger("bgfilter.app")
access_logger = logging.getLogger("bgfilter.access")
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"))
VERSION = "0.1.0"
_HEX_RE = re.compile(r"^#[0-9a-fA-F]{6}$")
@@ -167,6 +226,33 @@ async def lifespan(app: FastAPI):
app = FastAPI(title="BGfilter", version=VERSION, lifespan=lifespan)
@app.middleware("http")
async def _access_log_mw(request: Request, call_next):
"""Bind a request id, echo it back, and emit one structured access line per request.
The handler accumulates its fields on ``request.state.fields``; here we add the
id, client, status and total time -- so even a request that errors out (the
handler raised before finishing) still produces a log line.
"""
rid = _make_request_id(request.headers.get("x-request-id"))
_request_id_var.set(rid)
request.state.fields = {}
t0 = time.perf_counter()
status = 500
try:
response = await call_next(request)
status = response.status_code
response.headers["X-Request-ID"] = rid
return response
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)
@app.get("/healthz")
def healthz():
if _manager is None:
@@ -294,6 +380,7 @@ def _encode_png(rgb: np.ndarray, alpha: np.ndarray) -> bytes:
@app.post("/remove-background")
async def remove_background(
request: Request,
file: UploadFile | None = File(None),
image_url: str | None = Form(None),
screen_color: str | None = Form(None),
@@ -305,76 +392,102 @@ async def remove_background(
if _manager is None:
raise HTTPException(status_code=503, detail="service not ready")
# Exactly one image source. (Auth already ran, so only authenticated callers can
# trigger an outbound fetch.)
url = image_url.strip() if image_url else None
if (file is None) == (not url):
raise HTTPException(
status_code=400, detail="provide exactly one of 'file' or 'image_url'"
)
normalized_color = _normalize_screen_color(screen_color)
normalized_seg = _normalize_seg_model(seg_model) or _manager.default_seg_model
normalized_cc = _normalize_cross_check(cross_check)
# None/"flat" -> flat (chroma on); "complex" -> non-flat (chroma off).
normalized_bg = _normalize_background_mode(background_mode)
chroma_enabled = normalized_bg != "complex"
effective_bg = "complex" if not chroma_enabled else "flat"
# Complex mode defaults the cross-check veto OFF (it costs the HR-matting
# forward and only clears residue near strands); an explicit cross_check=on
# still enables its hue-free gate.
if not chroma_enabled and normalized_cc is None:
normalized_cc = False
# Honors the lane lock (a locked cpu-fast box reports/runs "off" even on cross_check=on).
effective_cc = _manager.effective_cross_check(normalized_cc)
# Acquire the raw bytes BEFORE the compute gate: a slow fetch must not hold a
# pool/compute slot. The upload body is already buffered by Starlette (fast);
# a URL is fetched under its own looser bound (_fetch_sem).
if url is not None:
async with _fetch_sem:
raw = await _fetch_image(url)
else:
raw = await file.read()
# Bound in-flight requests to the pool size, then run on the pool. The GPU
# sections are serialized inside the pipeline (a shared GPU mutex); the CPU
# sections of concurrent requests overlap across the pool threads.
async with _sem:
rgb = _decode_image(raw)
start = time.perf_counter()
loop = asyncio.get_running_loop()
try:
result = await loop.run_in_executor(
_executor,
lambda: _manager.process(
rgb,
screen_color=normalized_color,
seg_model=normalized_seg,
cross_check=normalized_cc,
chroma=chroma_enabled,
),
# Access-log fields accumulate here; the middleware emits the line (with status /
# total time) even if we raise below. Record the error detail on the way out.
log = request.state.fields
try:
# Exactly one image source. (Auth already ran, so only authenticated callers
# can trigger an outbound fetch.)
url = image_url.strip() if image_url else None
if (file is None) == (not url):
raise HTTPException(
status_code=400, detail="provide exactly one of 'file' or 'image_url'"
)
except RuntimeError as exc:
# No clean flat background and no screen_color given -> client-actionable.
if "auto-detect" in str(exc).lower():
raise HTTPException(
status_code=400,
detail="could not auto-detect background colour; pass screen_color",
)
logger.exception("inference failed")
raise HTTPException(status_code=500, detail="inference failed")
except Exception:
logger.exception("inference failed")
raise HTTPException(status_code=500, detail="inference failed")
elapsed_ms = int((time.perf_counter() - start) * 1000)
png = _encode_png(result.rgb, result.alpha)
headers = {
"X-BGFilter-Elapsed-Ms": str(elapsed_ms),
"X-BGFilter-Seg-Model": normalized_seg,
"X-BGFilter-Screen-Color": normalized_color if normalized_color else "auto",
"X-BGFilter-Cross-Check": "on" if effective_cc else "off",
"X-BGFilter-Background-Mode": effective_bg,
}
return Response(content=png, media_type="image/png", headers=headers)
normalized_color = _normalize_screen_color(screen_color)
normalized_seg = _normalize_seg_model(seg_model) or _manager.default_seg_model
normalized_cc = _normalize_cross_check(cross_check)
# None/"flat" -> flat (chroma on); "complex" -> non-flat (chroma off).
normalized_bg = _normalize_background_mode(background_mode)
chroma_enabled = normalized_bg != "complex"
effective_bg = "complex" if not chroma_enabled else "flat"
# Complex mode defaults the cross-check veto OFF (it costs the HR-matting
# forward and only clears residue near strands); an explicit cross_check=on
# still enables its hue-free gate.
if not chroma_enabled and normalized_cc is None:
normalized_cc = False
# Honors the lane lock (a locked cpu-fast box reports/runs "off" even on cross_check=on).
effective_cc = _manager.effective_cross_check(normalized_cc)
log["source"] = "url" if url else "file"
if url:
log["url_host"] = urlparse(url).hostname or "-"
log["seg_model"] = normalized_seg
log["cross_check"] = "on" if effective_cc else "off"
log["background_mode"] = effective_bg
log["screen_color"] = normalized_color if normalized_color else "auto"
log["auth"] = "ok" if AUTH_TOKEN else "open"
# Acquire the raw bytes BEFORE the compute gate: a slow fetch must not hold a
# pool/compute slot. The upload body is already buffered by Starlette (fast);
# a URL is fetched under its own looser bound (_fetch_sem).
if url is not None:
t = time.perf_counter()
async with _fetch_sem:
raw = await _fetch_image(url)
log["t_fetch_ms"] = round((time.perf_counter() - t) * 1000)
else:
raw = await file.read()
# Bound in-flight requests to the pool size, then run on the pool. The GPU
# sections are serialized inside the pipeline (a shared GPU mutex); the CPU
# sections of concurrent requests overlap across the pool threads.
t_wait = time.perf_counter()
async with _sem:
log["t_queue_ms"] = round((time.perf_counter() - t_wait) * 1000)
t_decode = time.perf_counter()
rgb = _decode_image(raw)
log["t_decode_ms"] = round((time.perf_counter() - t_decode) * 1000)
log["input_bytes"] = len(raw)
log["input_h"], log["input_w"] = int(rgb.shape[0]), int(rgb.shape[1])
start = time.perf_counter()
loop = asyncio.get_running_loop()
try:
result = await loop.run_in_executor(
_executor,
lambda: _manager.process(
rgb,
screen_color=normalized_color,
seg_model=normalized_seg,
cross_check=normalized_cc,
chroma=chroma_enabled,
),
)
except RuntimeError as exc:
# No clean flat background and no screen_color given -> client-actionable.
if "auto-detect" in str(exc).lower():
raise HTTPException(
status_code=400,
detail="could not auto-detect background colour; pass screen_color",
)
logger.exception("inference failed")
raise HTTPException(status_code=500, detail="inference failed")
except Exception:
logger.exception("inference failed")
raise HTTPException(status_code=500, detail="inference failed")
elapsed_ms = int((time.perf_counter() - start) * 1000)
log["t_process_ms"] = elapsed_ms
png = _encode_png(result.rgb, result.alpha)
log["output_bytes"] = len(png)
headers = {
"X-BGFilter-Elapsed-Ms": str(elapsed_ms),
"X-BGFilter-Seg-Model": normalized_seg,
"X-BGFilter-Screen-Color": normalized_color if normalized_color else "auto",
"X-BGFilter-Cross-Check": "on" if effective_cc else "off",
"X-BGFilter-Background-Mode": effective_bg,
}
return Response(content=png, media_type="image/png", headers=headers)
except HTTPException as exc:
log["error"] = str(exc.detail)[:200]
raise