From dc4badd0e0eff4b799b071e1c54de9c06e3fc42b Mon Sep 17 00:00:00 2001 From: Linghong Date: Wed, 15 Jul 2026 05:55:23 +0000 Subject: [PATCH] Add structured per-request access logging with a request id 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 --- DEPLOY.md | 2 + DEPLOY_ZH.md | 2 + README.md | 1 + README_ZH.md | 1 + app.py | 263 ++++++++++++++++++++++++++++++++++++--------------- 5 files changed, 194 insertions(+), 75 deletions(-) diff --git a/DEPLOY.md b/DEPLOY.md index f78405f..8a5847c 100644 --- a/DEPLOY.md +++ b/DEPLOY.md @@ -220,6 +220,8 @@ redirect back to `huggingface.co`, which recent `huggingface_hub` rejects with | `BGFILTER_FETCH_MAX_BYTES` | `26214400` | `image_url` mode: cap the downloaded body (~25 MB) → `413`. No SSRF filtering — keep auth on / callers trusted. | | `BGFILTER_FETCH_TIMEOUT` | `15` | `image_url` mode: read timeout (s) for the fetch (connect fixed at 5 s). Outbound fetch concurrency is fixed at `2 × cpu_workers` (hard cap 64). | | `BGFILTER_PRELOAD` | `1` | Load default models at startup (first request isn't cold) | +| `BGFILTER_LOG_FORMAT` | `text` | `text` (human) or `json` (one structured object per line, for log shippers). Each request emits one access line with a `request_id`, params, and stage timings (`t_queue/fetch/decode/process/total_ms`); the id is also echoed as the `X-Request-ID` response header. | +| `BGFILTER_LOG_LEVEL` | `INFO` | Root log level. | | `BGFILTER_CROSS_CHECK` | config (`on`) | Force the cross-check veto on/off (`0` disables; needs `BiRefNet_HR-matting`) | | `BGFILTER_CPU_WORKERS` | config `server.cpu_workers` (`1`) | Thread-pool size for request pipelining. `1` = strictly serial (one request at a time). `>1` lets that many requests run concurrently, overlapping their CPU stages while a GPU mutex serializes the model forwards — keeps the GPU fed instead of idle. Biggest win when the pipeline is CPU-bound (e.g. cross-check off). Set it in the config (`server:` section) or override per box with this env var. | | `BGFILTER_GPU_CONCURRENCY` | config `server.gpu_concurrency` (`1`) | How many requests may be inside a GPU forward at once. `1` = single-stream (safe). `2` lets two forwards co-schedule on the SMs (batch-1 underfills them), raising GPU-bound throughput — but concurrent activations multiply VRAM, so size it as `weights + N × peak_forward ≤ VRAM` (on a 32 GB card, `2` peaks ~19 GB at 2048 cross-check; keep `1` on ≤16 GB). Needs `cpu_workers>1` to matter. Config key `server.gpu_concurrency`; this env var overrides it. | diff --git a/DEPLOY_ZH.md b/DEPLOY_ZH.md index eaef015..0eb707f 100644 --- a/DEPLOY_ZH.md +++ b/DEPLOY_ZH.md @@ -200,6 +200,8 @@ export HF_HUB_DISABLE_XET=1 # 这些 repo 是 Xet 存储；强 | `BGFILTER_CONFIG` | `configs/default.yaml` | 管线配置文件 | | `BGFILTER_DEVICE` | `cpu` | 同时覆盖 model 和 segmentation 的 device（GPU 用 `cuda`）| | `BGFILTER_MAX_IMAGE_PIXELS` | `4194304` | 超过则返回 `413`（约 4 MP）| +| `BGFILTER_LOG_FORMAT` | `text` | `text`（人读）或 `json`（每行一条结构化对象,便于日志采集）。每请求打一行访问日志,含 `request_id`、参数、分阶段耗时（`t_queue/fetch/decode/process/total_ms`）;该 id 同时以 `X-Request-ID` 响应头返回。 | +| `BGFILTER_LOG_LEVEL` | `INFO` | 根日志级别。 | | `BGFILTER_FETCH_MAX_BYTES` | `26214400` | `image_url` 模式：下载体上限（约 25 MB）→ `413`。无 SSRF 过滤——务必开鉴权 / 调用方可信。 | | `BGFILTER_FETCH_TIMEOUT` | `15` | `image_url` 模式：fetch 读超时（秒；连接超时固定 5 秒）。出站 fetch 并发固定为 `2 × cpu_workers`（硬顶 64）。 | | `BGFILTER_PRELOAD` | `1` | 启动时预加载默认模型（首个请求不用冷加载）| diff --git a/README.md b/README.md index 189d3a5..408775e 100644 --- a/README.md +++ b/README.md @@ -223,6 +223,7 @@ is expected to authenticate). `/healthz` is always open. See [DEPLOY.md](DEPLOY. | `X-BGFilter-Screen-Color` | `#CFEFFF` / `auto` | Background colour used. | | `X-BGFilter-Cross-Check` | `on` / `off` | Effective cross-check state. | | `X-BGFilter-Background-Mode` | `flat` / `complex` | Background mode actually used. | +| `X-Request-ID` | `42efe4e0d76f4b83` | Correlation id (echoes an inbound `X-Request-ID` if you send one, else generated). Quote it to correlate with server logs. Returned on every response, including errors. | ### Status codes diff --git a/README_ZH.md b/README_ZH.md index 2fa9db0..57f2099 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -202,6 +202,7 @@ python -m uvicorn app:app --host 127.0.0.1 --port 18083 --workers 1 | `X-BGFilter-Screen-Color` | `#CFEFFF` / `auto` | 实际使用的背景色。 | | `X-BGFilter-Cross-Check` | `on` / `off` | 本次实际的 cross-check 状态。 | | `X-BGFilter-Background-Mode` | `flat` / `complex` | 本次实际的背景模式。 | +| `X-Request-ID` | `42efe4e0d76f4b83` | 关联 id（你若传入 `X-Request-ID` 则回显,否则自动生成）。报障时报这个 id 即可对上服务端日志。每个响应都返回,含错误响应。 | ### 状态码 diff --git a/app.py b/app.py index deb7133..8bffe56 100644 --- a/app.py +++ b/app.py @@ -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