From a875fe9f6308a802a9212ecee06a3a3315eb6649 Mon Sep 17 00:00:00 2001 From: shay7sev Date: Wed, 22 Jul 2026 13:52:40 +0800 Subject: [PATCH] feat(observability): add structured application logging Emit correlated JSON logs from the API and FunASR services, propagate request IDs to the client, and configure bounded opt-in Docker logging. Document the host-wide Grafana/Loki query workflow and keep observability deployment ownership outside the application repository. --- DEPLOY_5700U.md | 10 + README.md | 6 + app.wxss | 23 + asr-service/Dockerfile | 10 +- asr-service/app.py | 146 ++++- asr-service/logging_config.py | 90 +++ compose.deploy.yml | 37 +- docs/observability-implementation-plan.md | 737 ++-------------------- docs/observability-runbook.md | 174 +++++ pages/defaults/index.ts | 19 +- pages/defaults/index.wxml | 8 +- pages/feedback/index.ts | 15 +- pages/feedback/index.wxml | 8 +- pages/profile/index.ts | 41 +- pages/profile/index.wxml | 8 + pages/profile/index.wxss | 4 + pages/records/index.ts | 17 +- pages/records/index.wxml | 8 +- pages/students/index.ts | 13 +- pages/students/index.wxml | 8 +- server/Cargo.lock | 17 +- server/Cargo.toml | 7 +- server/Dockerfile | 8 +- server/src/config.rs | 20 + server/src/error.rs | 8 +- server/src/main.rs | 315 ++++++++- server/src/recording_routes.rs | 8 +- server/src/speech.rs | 21 +- server/src/transcription_worker.rs | 237 ++++++- utils/api.ts | 55 +- 30 files changed, 1261 insertions(+), 817 deletions(-) create mode 100644 asr-service/logging_config.py create mode 100644 docs/observability-runbook.md diff --git a/DEPLOY_5700U.md b/DEPLOY_5700U.md index bb3807b..6f4a5f0 100644 --- a/DEPLOY_5700U.md +++ b/DEPLOY_5700U.md @@ -242,3 +242,13 @@ Rust 健康检查应包含： - 尚未完成的验证或残余风险。 不得在报告中输出数据库密码、完整 `DATABASE_URL`、API 密钥或其他敏感信息。部署文件验证通过后保留改动供用户审查，不要自行提交或推送，除非用户在该会话中明确要求。 + +## 可观测性日志栈 + +业务服务通过 `compose.deploy.yml` 运行，并通过 `observability.logs=true` 标签接入宿主机中央 +观测栈。中央 Loki、Alloy 和 Grafana 由 +`/home/shay/data/docker-service/observability/compose.yml` 独立管理，不在本项目构建或启停。 + +本项目只负责结构化日志、脱敏、关联 ID、容器采集标签和日志轮转。查询和接入方法见 +`docs/observability-runbook.md`；中央栈的凭据、数据卷、备份和升级遵循其部署目录中的 +`README.md`。停止中央观测栈不得影响 API 或 FunASR。 diff --git a/README.md b/README.md index e5a152d..55311ae 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,12 @@ npm run typecheck 项目 AppID 已配置在 `project.config.json` 中。 +## 日志查询与问题定位 + +生产 API 和 FunASR 日志已接入中央 Grafana。访问地址、登录方式、常用 LogQL、请求 ID/转录 +任务关联方法，以及从日志定位到具体代码文件的流程，统一维护在 +[中央日志接入与查询手册](./docs/observability-runbook.md)。 + ## Rust 后端 Rust API 位于 [`server/`](./server)。它使用 Axum 和 PostgreSQL 驱动，但不会安装、启动或自动创建本地 PostgreSQL。 diff --git a/app.wxss b/app.wxss index cf43bdb..e38474b 100644 --- a/app.wxss +++ b/app.wxss @@ -153,6 +153,29 @@ button[disabled]:active { font-weight: 700; } +.status-copy { + flex: 1; + min-width: 0; +} + +.status-request-id, +.request-id-row { + display: flex; + align-items: flex-start; + gap: 16rpx; + margin-top: 10rpx; + color: #6e6e73; + font-size: 22rpx; + line-height: 1.4; +} + +.status-request-id > text:first-child, +.request-id-row > text:first-child { + flex: 1; + min-width: 0; + word-break: break-all; +} + .loading-state { padding: 70rpx 32rpx; color: #6e6e73; diff --git a/asr-service/Dockerfile b/asr-service/Dockerfile index 7af612c..dab7f12 100644 --- a/asr-service/Dockerfile +++ b/asr-service/Dockerfile @@ -2,10 +2,16 @@ FROM python:3.10-slim ARG DEBIAN_MIRROR=http://deb.debian.org ARG PYPI_INDEX_URL=https://pypi.org/simple +ARG APP_VERSION=0.2.0 +ARG GIT_SHA=unknown ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 \ - PIP_NO_CACHE_DIR=1 + PIP_NO_CACHE_DIR=1 \ + APP_ENV=lan \ + APP_VERSION=${APP_VERSION} \ + GIT_SHA=${GIT_SHA} \ + LOG_FORMAT=json RUN sed -i "s|http://deb.debian.org|${DEBIAN_MIRROR}|g" /etc/apt/sources.list.d/debian.sources \ && apt-get update \ @@ -18,7 +24,7 @@ RUN pip install --index-url "${PYPI_INDEX_URL}" --upgrade pip RUN pip install torch==2.8.0 torchaudio==2.8.0 --index-url https://download.pytorch.org/whl/cpu RUN pip install --index-url "${PYPI_INDEX_URL}" -r requirements.txt -COPY app.py ./ +COPY app.py logging_config.py ./ EXPOSE 10095 CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "10095", "--workers", "1", "--no-access-log"] diff --git a/asr-service/app.py b/asr-service/app.py index a028e5f..ca34e23 100644 --- a/asr-service/app.py +++ b/asr-service/app.py @@ -4,16 +4,20 @@ import asyncio import logging import os import tempfile +import time import uuid from contextlib import asynccontextmanager from pathlib import Path from typing import Any -from fastapi import FastAPI, File, Form, HTTPException, UploadFile +from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile from funasr import AutoModel from pydantic import BaseModel -LOGGER = logging.getLogger("funasr-service") +from logging_config import configure_logging, log_event, redirect_standard_streams + +configure_logging() + MAX_AUDIO_BYTES = 15 * 1024 * 1024 CHUNK_BYTES = 1024 * 1024 SUPPORTED_FORMATS = {"mp3", "aac", "wav", "m4a", "amr"} @@ -53,12 +57,12 @@ def positive_int(name: str, default: int) -> int: def load_model(settings: Settings) -> Any: - LOGGER.info( - "loading FunASR model model=%s vad=%s punc=%s device=%s", - settings.model, - settings.vad_model, - settings.punc_model, - settings.device, + log_event( + logging.INFO, + "asr_model_loading", + "loading FunASR model", + model=settings.model, + device=settings.device, ) model = AutoModel( model=settings.model, @@ -67,12 +71,13 @@ def load_model(settings: Settings) -> Any: device=settings.device, disable_update=True, ) - LOGGER.info("FunASR model is ready") + log_event(logging.INFO, "asr_model_ready", "FunASR model is ready") return model @asynccontextmanager async def lifespan(app: FastAPI): + redirect_standard_streams() settings = Settings() app.state.settings = settings app.state.model = await asyncio.to_thread(load_model, settings) @@ -82,11 +87,31 @@ async def lifespan(app: FastAPI): app = FastAPI( title="Teaching Feedback FunASR Service", - version="1.0.0", + version=os.getenv("APP_VERSION", "0.2.0"), lifespan=lifespan, ) +@app.middleware("http") +async def request_id_middleware(request: Request, call_next: Any): + request_id = valid_request_id(request.headers.get("x-request-id")) + request.state.request_id = request_id + try: + response = await call_next(request) + except Exception: + log_event( + logging.ERROR, + "asr_http_request_failed", + "ASR HTTP request failed", + request_id=request_id, + status=500, + error_class="unhandled_error", + ) + raise + response.headers["X-Request-ID"] = request_id + return response + + @app.get("/health", response_model=HealthResponse) async def health() -> HealthResponse: settings: Settings = app.state.settings @@ -95,17 +120,32 @@ async def health() -> HealthResponse: @app.post("/v1/transcriptions", response_model=TranscriptionResponse) async def create_transcription( + request: Request, file: UploadFile = File(...), language: str = Form("zh"), duration_ms: int = Form(0), hotword: str = Form(""), ) -> TranscriptionResponse: - if language not in {"zh", "zh-cn", "cmn"}: - raise HTTPException(status_code=422, detail="this service is configured for Chinese audio") - audio_format = file_extension(file.filename) - temporary_path = await save_upload(file, audio_format) - request_id = str(uuid.uuid4()) + request_id = request.state.request_id + started_at = time.perf_counter() + audio_format = "unknown" + audio_size_bytes = 0 + temporary_path: Path | None = None + normalized_duration_ms = max(0, duration_ms) + log_event( + logging.INFO, + "asr_request_started", + "ASR request started", + request_id=request_id, + audio_duration_ms=normalized_duration_ms, + audio_format=audio_format, + ) + try: + if language not in {"zh", "zh-cn", "cmn"}: + raise HTTPException(status_code=422, detail="this service is configured for Chinese audio") + audio_format = file_extension(file.filename) + temporary_path, audio_size_bytes = await save_upload(file, audio_format) async with app.state.inference_slots: text = await asyncio.to_thread( run_inference, @@ -114,21 +154,79 @@ async def create_transcription( app.state.settings.batch_size_seconds, hotword.strip(), ) - except Exception as error: - LOGGER.exception("FunASR inference failed request_id=%s", request_id) - raise HTTPException(status_code=500, detail=f"speech recognition failed: {error}") from error + if not text: + raise HTTPException(status_code=422, detail="no clear speech was recognized") + except HTTPException as error: + log_event( + logging.WARNING, + "asr_request_failed", + "ASR request failed", + request_id=request_id, + status=error.status_code, + latency_ms=elapsed_ms(started_at), + audio_size_bytes=audio_size_bytes, + audio_duration_ms=normalized_duration_ms, + audio_format=audio_format, + error_class=http_error_class(error.status_code), + ) + raise + except Exception: + log_event( + logging.ERROR, + "asr_request_failed", + "ASR request failed", + request_id=request_id, + status=500, + latency_ms=elapsed_ms(started_at), + audio_size_bytes=audio_size_bytes, + audio_duration_ms=normalized_duration_ms, + audio_format=audio_format, + error_class="inference_failed", + ) + raise HTTPException(status_code=500, detail="speech recognition failed") from None finally: - temporary_path.unlink(missing_ok=True) + if temporary_path is not None: + temporary_path.unlink(missing_ok=True) - if not text: - raise HTTPException(status_code=422, detail="no clear speech was recognized") + log_event( + logging.INFO, + "asr_request_finished", + "ASR request finished", + request_id=request_id, + status=200, + latency_ms=elapsed_ms(started_at), + audio_size_bytes=audio_size_bytes, + audio_duration_ms=normalized_duration_ms, + audio_format=audio_format, + ) return TranscriptionResponse( text=text, - duration_ms=max(0, duration_ms), + duration_ms=normalized_duration_ms, request_id=request_id, ) +def valid_request_id(value: str | None) -> str: + candidate = (value or "").strip() + if candidate and len(candidate) <= 128 and candidate.isascii() and candidate.isprintable(): + return candidate + return str(uuid.uuid4()) + + +def elapsed_ms(started_at: float) -> int: + return int((time.perf_counter() - started_at) * 1000) + + +def http_error_class(status: int) -> str: + if status == 413: + return "audio_too_large" + if status == 415: + return "unsupported_audio_format" + if status == 422: + return "invalid_audio" + return "request_error" + + def file_extension(filename: str | None) -> str: suffix = Path(filename or "segment.mp3").suffix.lower().lstrip(".") if suffix not in SUPPORTED_FORMATS: @@ -136,7 +234,7 @@ def file_extension(filename: str | None) -> str: return suffix -async def save_upload(file: UploadFile, audio_format: str) -> Path: +async def save_upload(file: UploadFile, audio_format: str) -> tuple[Path, int]: handle = tempfile.NamedTemporaryFile(suffix=f".{audio_format}", delete=False) path = Path(handle.name) total_bytes = 0 @@ -155,7 +253,7 @@ async def save_upload(file: UploadFile, audio_format: str) -> Path: if total_bytes == 0: path.unlink(missing_ok=True) raise HTTPException(status_code=422, detail="audio file is empty") - return path + return path, total_bytes def run_inference(model: Any, path: Path, batch_size_seconds: int, hotword: str) -> str: diff --git a/asr-service/logging_config.py b/asr-service/logging_config.py new file mode 100644 index 0000000..614ae26 --- /dev/null +++ b/asr-service/logging_config.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import json +import logging +import os +import sys +from datetime import datetime, timezone +from typing import Any + +LOG_FIELDS = ( + "event", + "request_id", + "segment_id", + "feedback_session_id", + "status", + "latency_ms", + "audio_size_bytes", + "audio_duration_ms", + "audio_format", + "error_class", +) + + +class JsonFormatter(logging.Formatter): + def format(self, record: logging.LogRecord) -> str: + payload: dict[str, Any] = { + "timestamp": datetime.now(timezone.utc).isoformat(timespec="milliseconds"), + "level": record.levelname, + "service": "funasr", + "environment": os.getenv("APP_ENV", "development"), + "app_version": os.getenv("APP_VERSION", "unknown"), + "git_sha": os.getenv("GIT_SHA", "unknown"), + "message": record.getMessage(), + } + for field in LOG_FIELDS: + value = getattr(record, field, None) + if value is not None: + payload[field] = value + return json.dumps(payload, ensure_ascii=False, default=str) + + +class JsonLogStream: + encoding = "utf-8" + + def __init__(self, level: int) -> None: + self.level = level + self.buffer = "" + + def write(self, message: str) -> int: + self.buffer += message + while "\n" in self.buffer: + line, self.buffer = self.buffer.split("\n", 1) + self._emit(line) + return len(message) + + def flush(self) -> None: + if self.buffer: + self._emit(self.buffer) + self.buffer = "" + + def isatty(self) -> bool: + return False + + def _emit(self, message: str) -> None: + message = message.strip() + if message: + logging.getLogger("funasr-service").log(self.level, message) + + +def configure_logging() -> None: + handler = logging.StreamHandler(sys.__stdout__) + handler.setFormatter(JsonFormatter()) + root = logging.getLogger() + root.handlers.clear() + root.addHandler(handler) + root.setLevel(logging.INFO) + + for name, logger in logging.Logger.manager.loggerDict.items(): + if isinstance(logger, logging.Logger): + logger.handlers.clear() + logger.propagate = name != "uvicorn.access" + + +def redirect_standard_streams() -> None: + sys.stdout = JsonLogStream(logging.INFO) # type: ignore[assignment] + sys.stderr = JsonLogStream(logging.ERROR) # type: ignore[assignment] + + +def log_event(level: int, event: str, message: str, **fields: Any) -> None: + logging.getLogger("funasr-service").log(level, message, extra={"event": event, **fields}) diff --git a/compose.deploy.yml b/compose.deploy.yml index 9211beb..13ae7b7 100644 --- a/compose.deploy.yml +++ b/compose.deploy.yml @@ -1,7 +1,19 @@ name: teaching-feedback x-api-base: &api-base - image: teaching-feedback-api:0.1.0 + image: teaching-feedback-api:${APP_VERSION:-0.2.0} + +x-default-logging: &default-logging + driver: local + options: + max-size: "20m" + max-file: "5" + +x-observability-labels: &observability-labels + observability.logs: "true" + observability.application: "teaching-feedback" + observability.environment: "lan" + observability.app_version: "${APP_VERSION:-0.2.0}" services: funasr: @@ -12,7 +24,13 @@ services: args: DEBIAN_MIRROR: http://mirrors.ustc.edu.cn PYPI_INDEX_URL: https://mirrors.ustc.edu.cn/pypi/simple + APP_VERSION: ${APP_VERSION:-0.2.0} + GIT_SHA: ${GIT_SHA:-unknown} environment: + APP_ENV: lan + APP_VERSION: ${APP_VERSION:-0.2.0} + GIT_SHA: ${GIT_SHA:-unknown} + LOG_FORMAT: json ASR_MODEL: paraformer-zh ASR_VAD_MODEL: fsmn-vad ASR_PUNC_MODEL: ct-punc @@ -20,6 +38,8 @@ services: ASR_MAX_CONCURRENCY: "1" ASR_BATCH_SIZE_SECONDS: "300" MODELSCOPE_CACHE: /models + labels: *observability-labels + logging: *default-logging volumes: - funasr-model-cache:/models restart: unless-stopped @@ -40,6 +60,13 @@ services: <<: *api-base env_file: - ./server/.env + environment: + APP_ENV: lan + APP_VERSION: ${APP_VERSION:-0.2.0} + GIT_SHA: ${GIT_SHA:-unknown} + LOG_FORMAT: json + labels: *observability-labels + logging: *default-logging command: - migrate restart: "no" @@ -53,9 +80,15 @@ services: dockerfile: Dockerfile args: DEBIAN_MIRROR: http://mirrors.ustc.edu.cn + APP_VERSION: ${APP_VERSION:-0.2.0} + GIT_SHA: ${GIT_SHA:-unknown} env_file: - ./server/.env environment: + APP_ENV: lan + APP_VERSION: ${APP_VERSION:-0.2.0} + GIT_SHA: ${GIT_SHA:-unknown} + LOG_FORMAT: json HOST: 0.0.0.0 PORT: "8080" AUDIO_STORAGE_DIR: /data/audio @@ -66,6 +99,8 @@ services: ASR_JOB_LEASE_SECONDS: "3600" DATABASE_MAX_CONNECTIONS: "5" ALLOW_DEVELOPMENT_USER_HEADER: "false" + labels: *observability-labels + logging: *default-logging ports: - "39180:8080" volumes: diff --git a/docs/observability-implementation-plan.md b/docs/observability-implementation-plan.md index d151b5d..979bdf5 100644 --- a/docs/observability-implementation-plan.md +++ b/docs/observability-implementation-plan.md @@ -1,266 +1,29 @@ -# 局域网服务日志与可观测性实施计划 +# 应用日志与中央观测接入规范 -- 状态：待实施 -- 编写日期：2026-07-22 -- 代码基线：`v0.2.0` / `04e80b1` -- 目标部署机：`192.168.193.237`（实施时必须重新确认） -- 当前 API：`http://192.168.193.237:39180` +- 状态：已实施并通过中央 Loki 查询验收 +- 实施日期：2026-07-22 +- 应用：`teaching-feedback` +- 业务入口：`http://192.168.193.237:39180` +- 中央 Grafana：`http://192.168.193.237:39300` +- 中央栈目录：`/home/shay/data/docker-service/observability` -## 新会话启动方式 +## 职责边界 -新开 Codex 会话后，先发送下面的提示词： +本项目只负责应用日志质量和接入声明，不包含 Loki、Alloy 或 Grafana 的 Compose、凭据、 +provisioning、数据卷和备份配置。中央栈由宿主机独立管理，停止或升级中央栈不得影响 API、 +FunASR 或数据库迁移。 -```text -请完整阅读 docs/observability-implementation-plan.md、DEPLOY_5700U.md、 -compose.deploy.yml、server/src/main.rs、server/src/transcription_worker.rs、 -asr-service/app.py 和 utils/api.ts。 +应用侧职责： -把 observability-implementation-plan.md 作为本次任务的执行约束。 -先检查 git、目标部署机、Docker 日志驱动、磁盘和现有容器状态，再从文档中 -第一个未完成的阶段开始实施。每完成一个阶段就运行该阶段验收，不要跳过日志 -脱敏、请求 ID、持久化、回滚验证或文档更新。除非我明确要求，不要自行提交、 -推送或修改目标机防火墙。 -``` +1. API 和 FunASR 向标准输出、标准错误写结构化 JSON 日志。 +2. 容器提供稳定的应用、环境和版本标签，并限制 Docker 本地日志占用。 +3. HTTP 请求返回并记录 `X-Request-ID`，异步转录记录可关联的业务 ID。 +4. 日志不得包含凭据、请求正文、反馈正文、转录文本、音频路径或学生个人信息。 +5. 中央栈不可用时，应用继续提供业务服务。 -新会话开始后应先更新本文档顶部的代码基线、目标机 IP 和实际组件版本。本文中的 -镜像版本、端口和资源阈值是实施起点，不得替代实施时的现场检查。 +## 容器接入 -## 1. 目标 - -本计划要建立一条可从小程序故障追溯到具体代码版本的日志链路： - -```text -小程序请求 - -> Rust API 结构化请求日志 - -> 转录队列事件 - -> FunASR 结构化推理日志 - -> Docker 本地日志与轮转 - -> Grafana Alloy 采集 - -> Loki 持久化与检索 - -> Grafana 查询、面板和告警 -``` - -完成后应支持： - -1. 任意可信局域网电脑通过浏览器访问 Grafana。 -2. 按环境、Compose 项目、服务、容器、日志级别、时间和代码版本筛选日志。 -3. 用 `request_id`、`feedback_session_id`、`lesson_id` 或 `segment_id` 关联 API 与 FunASR。 -4. 判断故障发生在哪个镜像版本和 Git 提交，避免按错误版本修改代码。 -5. 保存至少 14 天日志，并限制 Docker 原始容器日志占用。 -6. 在 Loki/Grafana 不可用时，API 与 FunASR 仍可正常提供业务服务。 -7. 后续可在同一 Grafana 中逐步加入日志告警和 Prometheus 指标。 - -## 2. 非目标 - -第一轮不做以下工作： - -- 不部署 Kubernetes，也不把现有单机 Compose 改成集群。 -- 不部署 Promtail；它已从当前 Loki 版本线移除，新部署使用 Alloy。 -- 不在应用代码中直接调用 Loki API，应用只写标准输出和标准错误。 -- 不在第一轮部署 Tempo、全链路追踪、Mimir 或 Elasticsearch。 -- 不采集请求体、反馈正文、转录文本、音频内容、Bearer Token 或数据库连接串。 -- 不把 Grafana、Loki、Alloy 或 Docker API 直接暴露到公网。 -- 不用日志替代业务数据库、审计表或音频持久化卷。 - -## 3. 当前状态与缺口 - -### 3.1 已确认状态 - -- `compose.deploy.yml` 只包含 `funasr`、`migrate` 和 `api`。 -- API 暴露宿主机 `39180`，FunASR 只在 Compose 内部网络提供 `10095`。 -- 规划时 `GET /health` 返回健康，数据库、语音和微信鉴权均已配置。 -- Rust API 已使用 `tracing` 和 `tracing-subscriber`，转录 worker 已记录部分 - `worker_id`、`segment_id` 和错误事件。 -- Rust 使用 `TraceLayer::new_for_http()`，但默认请求/响应事件为 `DEBUG`；镜像当前 - `RUST_LOG=info`，不能依赖它提供完整访问日志。 -- Rust 日志当前是文本格式，没有稳定 JSON 字段、请求 ID 或代码版本字段。 -- FunASR 使用 Python `logging`，但 Uvicorn 通过 `--no-access-log` 关闭访问日志；当前主要 - 记录模型启动和推理异常，没有成功请求耗时。 -- Compose 没有服务级 `logging` 配置。若目标机未修改 Docker daemon，通常会使用默认 - `json-file` 且没有自动轮转，存在磁盘耗尽风险。 -- `compose.deploy.yml` 中 API 镜像仍标记为 `teaching-feedback-api:0.1.0`，与当前 - `v0.2.0` 不一致。 -- 仓库没有 Loki、Alloy、Grafana、日志持久化卷、数据源自动配置或日志运行手册。 - -### 3.2 必须解决的问题 - -| 问题 | 影响 | 解决阶段 | -| --- | --- | --- | -| Docker 日志无显式轮转 | 宿主机磁盘可能被占满 | 阶段 1 | -| API 请求日志不完整 | 无法定位具体失败请求 | 阶段 2 | -| API 与 ASR 缺少统一关联 ID | 跨服务排障依赖时间猜测 | 阶段 2 | -| 日志没有版本元数据 | 无法映射到正确代码 | 阶段 1、2 | -| 日志仅保存在单机 Docker 中 | 其他电脑无法方便检索历史 | 阶段 3 | -| 没有保留期和持久化验证 | 重启或磁盘增长不可控 | 阶段 3 | -| 没有基于实际基线的告警 | 故障只能人工发现 | 阶段 4 | -| 没有时序指标 | 无法可靠发现“没有日志”的故障 | 阶段 5 | - -## 4. 目标架构 - -应用栈和可观测性栈使用两份 Compose 文件独立管理： - -```text -compose.deploy.yml compose.observability.yml - -api -------- stdout/stderr ----+ alloy ----> loki ----> grafana:3000 -funasr ----- stdout/stderr -----+ | | | -migrate ---- stdout/stderr -----+--------+ loki-data grafana-data - | - Docker Engine API - /var/run/docker.sock -``` - -约束： - -- `api`、`funasr`、`migrate` 不依赖 Loki、Alloy 或 Grafana 启动。 -- Alloy 通过 Docker Engine API 发现带指定标签的容器并读取日志。 -- Loki 与 Alloy 不发布宿主机端口。 -- 只有 Grafana 发布到可信局域网，建议默认使用 `${GRAFANA_PORT:-39300}:3000`。 -- 局域网外访问使用 WireGuard、Tailscale 或 SSH 隧道，不做公网裸端口。 -- Grafana 必须启用登录；禁止匿名访问。 -- Loki 初始使用单体模式、TSDB 索引和本地文件系统卷，接受单机无高可用的现实。 -- Loki 默认保留 14 天；稳定运行并确认磁盘余量后再考虑 30 天。 - -## 5. 日志字段规范 - -### 5.1 公共字段 - -Rust API 和 FunASR 每条应用日志尽量使用一致字段： - -| 字段 | 示例 | 说明 | -| --- | --- | --- | -| `timestamp` | `2026-07-22T02:00:00.123Z` | RFC 3339 UTC 时间 | -| `level` | `INFO` | `TRACE/DEBUG/INFO/WARN/ERROR` | -| `service` | `api` | `api`、`funasr` 或 `migrate` | -| `environment` | `lan` | 固定低基数字段 | -| `app_version` | `0.2.0` | 发行版本 | -| `git_sha` | `04e80b1` | 构建或部署提交 | -| `event` | `http_request_finished` | 稳定、可查询的事件名 | -| `request_id` | UUID | 单次 HTTP 请求关联 ID，不作为 Loki 标签 | -| `message` | 简短英文或中文 | 人类可读说明 | - -容器、Compose 项目、Compose 服务、镜像等字段由 Alloy 从 Docker 元数据补充。 - -### 5.2 API 请求字段 - -- `method` -- `route`：优先路由模板，例如 `/api/v1/profiles/{profile_id}` -- `path`：必要时保留原始路径，但不得包含敏感 query 参数 -- `status` -- `latency_ms` -- `request_id` -- `error_class`：只在失败时记录稳定分类 - -不要记录完整请求头。尤其禁止记录 `Authorization`、Cookie、微信 code、AppSecret 和 -上传内容。 - -### 5.3 语音链路字段 - -- `feedback_session_id` -- `lesson_id` -- `segment_id` -- `asr_request_id` -- `worker_id` -- `audio_format` -- `audio_size_bytes` -- `audio_duration_ms` -- `queue_wait_ms` -- `inference_latency_ms` -- `provider` -- `attempt` -- `status` - -禁止记录本地音频绝对路径、热词原文、转录文字或反馈正文。 - -### 5.4 Loki 标签规范 - -仅将低基数字段设为 Loki 标签： - -- `environment` -- `compose_project` -- `compose_service` -- `container` -- `service` -- `level` -- `app_version` - -以下字段只能保留在 JSON 日志正文中，通过 `| json` 查询，不能成为标签： - -- `request_id` -- 用户 ID、学生 ID、会话 ID、课节 ID、片段 ID -- URL 原始路径 -- 错误消息 - -## 6. 计划新增或修改的文件 - -| 文件 | 动作 | 目的 | -| --- | --- | --- | -| `compose.deploy.yml` | 修改 | 日志轮转、容器筛选标签、版本环境变量、镜像版本 | -| `server/Cargo.toml` | 修改 | 开启 `tracing-subscriber/json` 和 `tower-http/request-id` | -| `server/src/main.rs` | 修改 | JSON 日志、请求 ID、INFO 请求完成日志、启动版本日志 | -| `server/src/config.rs` | 修改 | 读取 `LOG_FORMAT`、`APP_VERSION`、`GIT_SHA`、`APP_ENV` | -| `server/src/transcription_worker.rs` | 修改 | 增加队列等待、尝试次数、ASR 耗时和关联字段 | -| `server/src/speech.rs` | 修改 | 将 `segment_id`/请求 ID 传给 FunASR | -| `server/.env.example` | 修改 | 记录非敏感日志配置项 | -| `utils/api.ts` | 修改 | 保存响应 `X-Request-ID` 到 `ApiRequestError` | -| 相关小程序错误状态 | 修改 | 在可复制位置显示请求编号，不塞入短 Toast | -| `asr-service/app.py` | 修改 | 请求 ID 中间件、成功/失败耗时和结构化字段 | -| `asr-service/logging_config.py` | 新增 | 标准库 JSON formatter 与日志初始化 | -| `asr-service/Dockerfile` | 修改 | 传入版本环境，继续避免重复 Uvicorn access log | -| `compose.observability.yml` | 新增 | Alloy、Loki、Grafana 三个服务 | -| `observability/alloy/config.alloy` | 新增 | Docker 发现、筛选、标签、转发配置 | -| `observability/loki/loki.yml` | 新增 | 单体 TSDB、文件系统、保留期配置 | -| `observability/grafana/provisioning/datasources/loki.yml` | 新增 | 自动创建 Loki 数据源 | -| `observability/grafana/provisioning/dashboards/` | 后续新增 | 阶段 4 面板和告警 | -| `observability/.env.example` | 新增 | 端口、版本、管理员账号变量说明，不含密码 | -| `.gitignore` | 修改 | 忽略 `observability/.env`、本地数据和导出文件 | -| `docs/observability-runbook.md` | 新增 | 启停、查询、备份、升级、故障处理 | -| `DEPLOY_5700U.md` | 修改 | 把日志栈加入部署和验收流程 | - -## 7. 分阶段实施 - -### 阶段 0：现场基线与风险检查 - -目标：在修改配置前记录目标机真实状态。 - -在目标部署机执行： - -```bash -cd /path/to/teaching-feedback-assistant -git status --short --branch -git rev-parse HEAD -git tag --points-at HEAD - -docker version -docker compose version -docker info --format '{{.LoggingDriver}}' -docker compose -f compose.deploy.yml ps -docker compose -f compose.deploy.yml images -docker compose -f compose.deploy.yml logs --since 1h api funasr -docker inspect teaching-feedback-api-1 --format '{{json .HostConfig.LogConfig}}' || true - -df -h -docker system df -timedatectl status -ss -lntp | grep -E ':(39180|39300|3100|12345)\b' || true -``` - -实施要求： - -1. 不输出 `server/.env` 内容。 -2. 记录 Docker 当前日志驱动、容器名、镜像 ID、重启次数和日志占用。 -3. 确认 `39300` 是否可用；冲突时选择其他端口并更新本文档。 -4. 确认至少有 10 GB 可用空间给日志卷；不足时先清理或降低保留期。 -5. 确认目标机时钟同步。日志统一为 UTC，Grafana 按浏览器时区显示。 -6. 确认 API、FunASR 和真实转录链路在改造前正常。 - -验收：把基线命令和非敏感结果写入实施记录，任何失败先解释，不进入阶段 1。 - -### 阶段 1：Docker 日志轮转与版本元数据 - -目标：即使集中日志尚未部署，容器日志也不能无限增长。 - -在 `compose.deploy.yml` 增加可复用配置： +生产 Compose 中的所有业务服务使用以下公共配置： ```yaml x-default-logging: &default-logging @@ -268,451 +31,55 @@ x-default-logging: &default-logging options: max-size: "20m" max-file: "5" -``` -对 `funasr`、`migrate` 和 `api` 使用该配置。优先使用服务级配置，不修改 Docker daemon -全局默认，避免影响目标机的其他项目。 - -同时增加： - -```yaml -labels: +x-observability-labels: &observability-labels observability.logs: "true" + observability.application: "teaching-feedback" observability.environment: "lan" -environment: - APP_ENV: lan - APP_VERSION: ${APP_VERSION:-0.2.0} - GIT_SHA: ${GIT_SHA:-unknown} - LOG_FORMAT: ${LOG_FORMAT:-json} ``` -要求： +每个服务引用 `labels: *observability-labels` 和 `logging: *default-logging`。Docker 标签和日志 +驱动只在创建容器时生效；仅修改这些字段时使用 `--force-recreate --no-build`，不需要重建镜像。 -1. API 镜像标签不得继续硬编码为 `0.1.0`。 -2. 实施时从 Git tag/commit 注入 `APP_VERSION` 和 `GIT_SHA`。 -3. 镜像版本必须固定，不使用 `latest`。 -4. `local` 日志驱动的配置值必须是字符串。 -5. 修改日志驱动后必须重建容器；仅 restart 不会更新已有容器的日志驱动。 +## JSON 日志字段 -验证： +所有事件应包含： -```bash -docker compose -f compose.deploy.yml config -docker compose -f compose.deploy.yml up -d --build --force-recreate -docker compose -f compose.deploy.yml ps -docker compose -f compose.deploy.yml logs --since 10m api funasr -docker inspect --format '{{json .HostConfig.LogConfig}}' -docker inspect --format '{{json .Config.Labels}}' -curl -fsS http://127.0.0.1:39180/health -``` +- `timestamp`：UTC RFC 3339 时间。 +- `level`：`TRACE`、`DEBUG`、`INFO`、`WARN` 或 `ERROR`。 +- `service`、`environment`、`app_version`、`git_sha`。 +- `event`：稳定的机器可查询事件名。 +- `request_id`：HTTP 请求关联 ID；客户端传入合法值时继续传播，否则生成 UUID。 -阶段 1 完成标准：三个服务使用受控日志驱动，API 健康且真实转录仍成功。 +按业务阶段补充 `feedback_session_id`、`lesson_id`、`segment_id`、`worker_id`、`status`、 +`latency_ms` 和 `error_class`。这些高基数字段保留在 JSON 正文中，不能设置为 Loki 标签。 -### 阶段 2：结构化请求日志与跨服务关联 +允许作为 Loki 标签的字段仅限 `application`、`environment`、`compose_project`、 +`compose_service`、`service`、`container`、`level`、`app_version` 和固定宿主机标识。 -#### 2.1 Rust API +## 脱敏要求 -依赖调整： +禁止记录： -```toml -tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } -tower-http = { version = "0.6", features = ["cors", "trace", "request-id"] } -``` +- `Authorization`、Cookie、Token、API Key、数据库连接串和密码。 +- 请求体、反馈正文、总结内容、转录文本和热词正文。 +- 音频内容、临时文件路径和包含学生身份的文件名。 +- 学生姓名、联系方式、微信身份和其他个人信息。 -实现要求： +失败事件记录稳定的 `error_class`、关联 ID 和耗时，不直接输出第三方异常对象或原始响应。 -1. `LOG_FORMAT=json` 时使用 `tracing_subscriber::fmt().json()`，关闭 ANSI，并保留当前 span。 -2. 本地开发允许 `LOG_FORMAT=pretty`，但生产 Compose 固定 `json`。 -3. 使用 `tower_http::request_id`： - - 接受合法的客户端 `X-Request-ID`。 - - 缺失时用 `MakeRequestUuid` 生成。 - - 在响应头中传播同一个 `X-Request-ID`。 -4. 中间件顺序按 tower-http 要求设置：先 Set Request ID，再 Trace，最后 Propagate。 -5. `TraceLayer` 的请求开始、请求结束和失败级别显式设置为 `INFO/ERROR`，不依赖默认 - `DEBUG`。 -6. `make_span_with` 不记录 headers；记录 `method`、`route/path` 和 `request_id`。 -7. `on_response` 记录 `status` 和 `latency_ms`。 -8. 5xx/超时记录 `ERROR`，可预期 4xx 不得全部当成系统错误。 -9. 启动时记录 `service`、`environment`、`app_version`、`git_sha`、监听地址和语音提供方。 -10. 不在日志中输出 `DATABASE_URL`、微信密钥、Summary API Key 或任何 Token。 +## 验收标准 -补充单元/集成测试： +发布后至少完成以下检查： -- 无请求 ID 时响应包含合法 UUID。 -- 带合法请求 ID 时响应原样传播。 -- 日志中不包含 Authorization 值。 -- 200、400、401、500 都产生一次完成事件。 -- JSON 日志每行可被 `jq` 解析。 +1. API 与 FunASR 容器健康，日志驱动为 `local/20m/5`。 +2. 容器包含 `observability.logs=true`、`application=teaching-feedback` 和 `environment=lan`。 +3. Grafana/Loki 可按 `application`、`service`、版本和时间筛选日志。 +4. 一次 HTTP 请求的请求头、响应头和 JSON 日志使用同一 `request_id`。 +5. API 与 FunASR 的异步事件可通过 `segment_id` 等字段关联。 +6. 查询结果不包含上述敏感数据。 +7. 中央栈停止时，`GET /health` 仍然成功。 -#### 2.2 转录 worker 与 Rust -> FunASR +具体查询与故障定位步骤见 `docs/observability-runbook.md`。中央栈运维只在 +`/home/shay/data/docker-service/observability/README.md` 中维护。 -1. 在 claim job 时取得 `processing_started_at`、`transcription_attempts` 等必要字段。 -2. 以 `segment_id` 作为跨 API/ASR 的稳定业务关联字段。 -3. Rust 请求 FunASR 时设置 `X-Request-ID`；可直接使用 `segment_id`，或生成新的 - `asr_request_id` 并同时记录两者。 -4. 记录排队等待、音频读取、HTTP 调用、推理完成和数据库更新的分段耗时。 -5. 成功日志不得包含 transcript；失败日志只记录经过清理的错误分类和消息。 - -#### 2.3 FunASR - -1. 新增标准 JSON formatter，不因日志系统引入另一套应用框架。 -2. 中间件读取/生成 `X-Request-ID`，并写回响应头。 -3. 记录 `asr_request_started`、`asr_request_finished` 和 `asr_request_failed`。 -4. 成功和失败均记录 `status`、`latency_ms`、`audio_size_bytes`、`audio_duration_ms`、 - `audio_format` 和 `request_id`。 -5. 保留 `--no-access-log`，避免 Uvicorn 默认访问日志和自定义日志重复。 -6. 不记录临时路径、文件名、hotword、模型输出文本或上传内容。 - -#### 2.4 小程序错误编号 - -1. `ApiRequestError` 增加可选 `requestId`。 -2. `wx.request` 与 `wx.uploadFile` 都从响应头读取 `X-Request-ID`，兼容响应头大小写。 -3. 短 Toast 不塞入长 UUID；需要持久显示错误的状态区增加“请求编号”与复制入口。 -4. 开发者控制台可以记录请求编号，但不得打印 Token、上传路径或请求正文。 - -阶段 2 验收： - -```bash -cd server -cargo fmt --check -cargo test -cargo clippy --all-targets --all-features -- -D warnings - -cd .. -npm run typecheck -docker compose -f compose.deploy.yml config -``` - -再用一个固定 `X-Request-ID` 调用 `/health` 和一个授权业务接口，确认 API 日志可按该 ID -检索。使用真实音频验证 `segment_id` 能关联 Rust 与 FunASR，且日志中没有转录正文。 - -### 阶段 3：部署 Alloy、Loki、Grafana - -#### 3.1 版本选择 - -实施时查阅官方 release notes，选择互相兼容的稳定版本并固定完整版本号： - -- `grafana/alloy:` -- `grafana/loki:` -- `grafana/grafana:` - -禁止使用 `latest`。将选定版本写入 `observability/.env.example` 和实施记录。 - -#### 3.2 Compose 约束 - -新增 `compose.observability.yml`，项目名建议为 `teaching-feedback-observability`。 - -`loki`： - -- 单体模式。 -- 配置文件只读挂载。 -- `/var/loki` 使用 `loki-data` 持久化卷。 -- 不发布 `3100` 到宿主机。 -- 提供 readiness 健康检查。 -- `restart: unless-stopped`。 - -`alloy`： - -- 配置文件只读挂载。 -- `/var/lib/alloy/data` 使用 `alloy-data`，保存 Docker 读取位置。 -- 初始部署读取 `/var/run/docker.sock`。 -- Docker Socket 挂载即使标记只读，也不是完整的 API 权限隔离；容器必须使用固定镜像、 - 最小权限且不对外开放 Alloy UI。安全要求提高时改为受限 Docker Socket Proxy。 -- 不发布 Alloy `12345` 到宿主机。 -- `restart: unless-stopped`。 - -`grafana`： - -- `/var/lib/grafana` 使用 `grafana-data`。 -- 默认发布 `${GRAFANA_PORT:-39300}:3000`。 -- `GF_AUTH_ANONYMOUS_ENABLED=false`。 -- 管理员密码仅放在未提交的 `observability/.env` 或 Docker Secret。 -- 自动挂载 Loki 数据源 provisioning。 -- `restart: unless-stopped`。 - -#### 3.3 Loki 配置 - -`observability/loki/loki.yml` 至少满足： - -- `auth_enabled: false`，但 Loki 只存在于内部 Docker 网络。 -- 单体 `target=all`。 -- TSDB schema，不使用已弃用 BoltDB index。 -- chunks、rules、WAL 和 compactor 目录全部落在 `/var/loki`。 -- compactor retention 开启。 -- 初始 `retention_period: 336h`（14 天）。 -- 禁用匿名 usage analytics（若当前版本支持对应配置）。 -- 设定合理 ingestion/query 限制，防止一次错误查询拖垮单机。 - -文件系统存储没有副本，也不会按磁盘剩余空间自动删除；保留期之外仍要监控磁盘。 - -#### 3.4 Alloy 配置 - -`observability/alloy/config.alloy` 应包含： - -1. `discovery.docker` 连接 `unix:///var/run/docker.sock`。 -2. `discovery.relabel` 只保留 `observability.logs=true` 的容器。 -3. 从 Docker 元数据映射 `compose_project`、`compose_service`、`container`、`image`。 -4. `loki.source.docker` 读取容器日志并保存 positions。 -5. `loki.process` 添加 `environment=lan`，解析 JSON 日志级别；解析失败时保留原始行。 -6. 不把 `request_id`、用户或业务 ID 提升为标签。 -7. `loki.write` 发送到 `http://loki:3100/loki/api/v1/push`。 - -#### 3.5 Grafana 自动配置 - -Loki 数据源通过 provisioning 创建： - -```yaml -apiVersion: 1 -datasources: - - name: Loki - type: loki - access: proxy - url: http://loki:3100 - isDefault: true - editable: false -``` - -首次只创建数据源和一个最小“日志浏览”面板，不提前加入未经验证的告警阈值。 - -#### 3.6 启动与验证 - -```bash -docker compose -f compose.observability.yml config -docker compose -f compose.observability.yml up -d -docker compose -f compose.observability.yml ps -docker compose -f compose.observability.yml logs --since 10m alloy loki grafana -curl -fsS http://127.0.0.1:39300/api/health -``` - -验收： - -1. 浏览器可从另一台局域网电脑打开 `http://192.168.193.237:39300`。 -2. 未登录不能查看日志。 -3. Grafana Explore 能看到 `api` 和 `funasr`。 -4. 以下查询可用，实际标签名以 Alloy 配置为准： - -```logql -{compose_service="api"} | json -{compose_service="api"} | json | request_id="" -{compose_service="funasr"} | json | segment_id="" -{compose_service="api"} | json | status >= 500 -``` - -5. 重启 Alloy 后不大量重复采集旧日志。 -6. 重启整套可观测性 Compose 后，Grafana 数据源、账号配置和 Loki 历史仍存在。 -7. 停止 Loki/Alloy/Grafana 后，API 和 FunASR 继续正常工作。 -8. Loki、Alloy 的宿主机端口不可访问。 - -### 阶段 4：基于实际日志建立面板与告警 - -先运行至少 3–7 天，观察正常流量、错误数量和转录耗时，再设置阈值。 - -首批面板： - -- API 请求总量、4xx、5xx。 -- API `latency_ms` 的 p50/p95/p99。 -- 各路由错误日志列表。 -- 转录成功、失败和重试数量。 -- ASR 推理耗时和音频时长比值。 -- `ASR service is unavailable` 与恢复事件。 -- 按 `app_version` 对比错误率。 - -首批日志告警候选： - -- 5 分钟内 API 5xx 达到 3 次。 -- 10 分钟内出现转录失败。 -- 5 分钟内持续出现数据库查询失败。 -- ASR 不可用持续超过 5 分钟。 -- Loki ingestion 或 Alloy target unhealthy。 - -告警要求： - -1. 每条告警必须提供可直接打开的 Grafana 查询。 -2. 配置 `for` 持续时间，避免瞬时启动日志触发。 -3. 通知内容包含服务、环境、版本和时间范围，不包含用户数据。 -4. 先发送到低风险通知渠道进行一周观察，再升级为正式告警。 -5. 告警规则和 dashboard JSON 纳入 Git，禁止只在网页里手工保存。 - -Grafana 可直接基于 Loki 日志创建告警，但“服务完全无日志”“CPU/内存异常”“磁盘即将 -耗尽”更适合指标，留到阶段 5。 - -### 阶段 5：按实际故障增加指标采集 - -日志稳定后再引入 Prometheus 指标。Alloy 可以抓取指标，但它不是长期指标存储；自托管 -方案需要增加 Prometheus，或把指标 remote write 到已有兼容后端。 - -建议指标： - -Rust API： - -- `http_requests_total{route,method,status_class}` -- `http_request_duration_seconds{route,method}` -- `transcription_queue_jobs{status}` -- `transcription_job_duration_seconds` -- `transcription_job_retries_total` -- `database_pool_connections{state}` -- `asr_available` - -FunASR： - -- `asr_requests_total{status}` -- `asr_inference_duration_seconds` -- `asr_audio_duration_seconds` -- `asr_inflight_requests` -- `asr_failures_total{reason}` - -宿主机与容器： - -- node-exporter：CPU、内存、磁盘、负载。 -- cAdvisor 或 Alloy Docker integration：容器 CPU、内存、重启和网络。 -- `up`：API、FunASR、Loki、Alloy、Grafana 抓取状态。 - -指标标签禁止包含请求 ID、学生 ID、会话 ID 或原始 URL。路由必须使用模板路径，避免 -高基数。 - -阶段 5 告警优先级： - -1. 磁盘剩余小于 15%。 -2. API/FunASR `up == 0` 持续 2 分钟。 -3. 转录队列持续增长或最老任务等待超阈值。 -4. API p95 延迟持续异常。 -5. 容器反复重启或内存逼近限制。 - -只有在日志和指标仍无法解释跨服务延迟时，才评估 OpenTelemetry + Tempo tracing。 - -## 8. 安全与隐私要求 - -### 禁止进入日志 - -- `Authorization`、Cookie、微信登录 code、session key。 -- `WECHAT_APP_SECRET`、数据库密码、完整 `DATABASE_URL`。 -- Summary/LLM API Key。 -- 学生姓名、家长称呼、反馈正文、转录全文、hotword。 -- 音频内容、完整本地路径、上传临时文件名。 -- 完整请求/响应 body。 - -### 允许但不作为标签 - -- 内部 UUID：profile/session/lesson/segment。 -- request ID。 -- 清理后的错误消息。 - -### 网络与权限 - -- Grafana 只允许可信局域网 CIDR 或 VPN 访问。 -- Loki、Alloy 和 Docker API 不发布到宿主机。 -- Alloy 的 Docker Socket 权限视为高权限，固定镜像版本并限制可访问人员。 -- Grafana 管理员密码不得提交；首次登录后更换默认密码。 -- Scalar 和 Grafana 都属于运维入口，未来若局域网信任边界扩大，应统一加反向代理、 - HTTPS 和认证。 - -## 9. 容量、保留与备份 - -初始值： - -- Docker 原始日志：每容器 `20 MB x 5`。 -- Loki：14 天保留。 -- 每日检查：`df -h`、`docker system df`、Loki 卷大小。 -- 预警：磁盘剩余 20% 开始观察，15% 告警，10% 紧急处理。 - -备份： - -- `grafana-data`：保存数据源、用户和本地状态；虽然 provisioning 在 Git 中，仍需备份。 -- `loki-data`：若历史日志不是关键资产，可不做强一致备份，但必须明确接受丢失风险。 -- 配置文件和 dashboard/alert provisioning 必须进入 Git。 -- 不备份 Alloy positions 时会导致重采或漏采风险，因此 `alloy-data` 应持久化。 - -备份和恢复命令在实施时写入 `docs/observability-runbook.md`，并实际演练一次 Grafana 配置 -恢复和 Loki 重启持久性。 - -## 10. 回滚方案 - -可观测性栈必须可独立回滚： - -```bash -docker compose -f compose.observability.yml down -``` - -该操作不能停止 `compose.deploy.yml` 的业务服务。默认不加 `-v`，避免误删日志和 Grafana -数据。 - -应用日志回滚： - -1. 将 `LOG_FORMAT=pretty` 可暂时恢复文本输出，不需要改业务代码。 -2. 若请求中间件引发问题，回退对应提交并重建 API 镜像。 -3. 若 `local` 日志驱动与采集不兼容，回退 Compose 日志块后 `--force-recreate` 容器； - restart 不足以恢复日志驱动。 -4. 每阶段使用独立提交，禁止把业务功能改动混进日志基础设施提交。 - -删除数据前必须先确认： - -```bash -docker volume ls | grep teaching-feedback -docker compose -f compose.observability.yml down -# 只有用户明确确认后才允许删除 loki-data/grafana-data/alloy-data -``` - -## 11. 建议提交拆分 - -1. `ops(logging): configure container log rotation and version labels` -2. `feat(logging): add structured request and transcription logs` -3. `ops(observability): add alloy loki and grafana stack` -4. `docs(observability): add operations and troubleshooting runbook` -5. `feat(observability): add validated dashboards and alerts` -6. `feat(metrics): expose service and transcription metrics`（阶段 5，单独实施） - -每个提交前运行与其影响范围相匹配的测试。部署修改未经目标机验收不得直接打 release tag。 - -## 12. 最终验收清单 - -- [ ] 目标机 Docker 日志驱动和轮转参数已记录。 -- [ ] API、FunASR、migrate 均使用受控日志轮转。 -- [ ] Rust 与 FunASR 每行输出有效 JSON。 -- [ ] API 响应返回 `X-Request-ID`。 -- [ ] 小程序能保留并展示失败请求编号。 -- [ ] 真实录音可通过 `segment_id/request_id` 关联 Rust 与 FunASR。 -- [ ] 日志中未发现 Token、密钥、学生信息、反馈正文或转录全文。 -- [ ] Alloy 只采集明确标记的容器。 -- [ ] Loki 与 Alloy 没有宿主机端口。 -- [ ] Grafana 需要登录且只在可信网络可达。 -- [ ] Loki 数据源由 provisioning 自动创建。 -- [ ] 14 天 retention 生效，磁盘增长已观察。 -- [ ] 重启可观测性栈后历史和配置仍存在。 -- [ ] 停止可观测性栈不影响 API 和 FunASR。 -- [ ] `cargo fmt --check`、`cargo test`、严格 Clippy 和 `npm run typecheck` 通过。 -- [ ] `docs/observability-runbook.md` 包含启停、查询、备份、升级和回滚命令。 -- [ ] 至少 3–7 天基线后再启用正式告警。 -- [ ] 指标阶段明确使用 Prometheus 或兼容存储，不误把 Alloy 当成指标数据库。 - -## 13. 官方参考资料 - -实施时重新核对最新稳定版本和配置语法： - -- Docker logging drivers： -- Grafana Alloy 工作方式： -- Alloy Docker 日志采集： -- Alloy Docker 容器安装： -- Loki Docker/Compose 安装： -- Loki 存储和 retention： -- Loki 配置参考： -- Grafana Docker 安装： -- Grafana Alerting： -- Alloy Prometheus scrape： -- tower-http request ID： -- tower-http trace： - -## 14. 实施完成后的交付报告 - -新会话完成每个阶段后应报告： - -- 实际提交和部署版本。 -- 新增/修改文件与 `git diff --stat`。 -- 目标机日志驱动、容器、端口、网络和卷状态。 -- API、FunASR、Alloy、Loki、Grafana 健康检查结果。 -- 一条真实请求和一条真实转录的关联查询结果。 -- 日志脱敏检查结果。 -- 数据保留、磁盘使用和备份/恢复验证。 -- 已配置告警及其基线依据。 -- 未完成项、已知风险和明确回滚命令。 - -报告不得输出密码、Token、完整数据库地址、学生信息、反馈正文或转录内容。 diff --git a/docs/observability-runbook.md b/docs/observability-runbook.md new file mode 100644 index 0000000..3c6c45c --- /dev/null +++ b/docs/observability-runbook.md @@ -0,0 +1,174 @@ +# 中央日志接入与查询手册 + +教学反馈服务通过容器标签接入宿主机中央观测栈。本项目不包含也不启停 Loki、Alloy 或 +Grafana；中央栈位于 `/home/shay/data/docker-service/observability`，Grafana 地址为 +`http://192.168.193.237:39300`。 + +## 登录 Grafana + +1. 在可信局域网或已接入该主机网络的浏览器中打开 `http://192.168.193.237:39300`。 +2. 用户名使用 `admin`。 +3. 初始密码读取中央栈的 + `/home/shay/data/docker-service/observability/.env` 中的 `GRAFANA_ADMIN_PASSWORD`。 + +需要在目标机终端确认初始密码时使用： + +```bash +sed -n 's/^GRAFANA_ADMIN_PASSWORD=//p' /home/shay/data/docker-service/observability/.env +``` + +不要把密码粘贴到 issue、提交信息、聊天记录或截图中。如果已经在 Grafana 页面修改过密码， +应使用修改后的密码；`.env` 只保证记录首次创建管理员时使用的值。 + +## 发布与接入检查 + +业务镜像构建时注入版本信息： + +```bash +export APP_VERSION="$(git describe --tags --always)" +export GIT_SHA="$(git rev-parse --short HEAD)" +docker compose -f compose.deploy.yml config -q +docker compose -f compose.deploy.yml up -d --build --force-recreate +docker compose -f compose.deploy.yml ps +``` + +仅修改采集标签或日志轮转时不需要构建镜像： + +```bash +docker compose -f compose.deploy.yml up -d --force-recreate --no-build api funasr +``` + +确认运行容器配置： + +```bash +docker inspect teaching-feedback-api-1 --format '{{json .HostConfig.LogConfig}}' +docker inspect teaching-feedback-api-1 --format '{{json .Config.Labels}}' +docker inspect teaching-feedback-funasr-1 --format '{{json .HostConfig.LogConfig}}' +docker inspect teaching-feedback-funasr-1 --format '{{json .Config.Labels}}' +curl -fsS http://127.0.0.1:39180/health +``` + +预期日志驱动为 `local`，`max-size=20m`、`max-file=5`；两个服务都应包含 +`observability.logs=true`、`observability.application=teaching-feedback` 和 +`observability.environment=lan`。 + +## 使用预置仪表盘 + +登录后打开 `Dashboards`，进入 `Host Observability / Host Service Logs`，然后设置： + +- `Application`：`teaching-feedback` +- `Environment`：`lan` +- `Service`：按需选择 `api`、`funasr` 或 `All` +- 右上角时间范围：复现问题时优先选择最近 5 至 30 分钟 + +仪表盘适合浏览和筛选；需要精确关联请求或比较版本时，打开 `Explore`，数据源选择 `Loki`。 + +## 常用 LogQL + +查看教学反馈的全部日志： + +```logql +{application="teaching-feedback"} +``` + +按服务、级别和文本筛选： + +```logql +{application="teaching-feedback", service="api"} | json +{application="teaching-feedback", service="funasr"} | json +{application="teaching-feedback", level=~"WARN|ERROR"} +{application="teaching-feedback"} |= "database" +``` + +按 HTTP 状态和耗时寻找优化目标： + +```logql +{application="teaching-feedback", service="api"} | json | status >= 500 +{application="teaching-feedback", service="api"} | json | latency_ms >= 1000 +{application="teaching-feedback", service="api"} | json | event="http_request_finished" +``` + +按关联 ID 追踪一次请求或转录任务： + +```logql +{application="teaching-feedback", service="api"} | json | request_id="" +{application="teaching-feedback", service="api"} | json | feedback_session_id="" +{application="teaching-feedback", service="api"} | json | lesson_id="" +{application="teaching-feedback", service="api"} | json | segment_id="" +{application="teaching-feedback", service="funasr"} | json | segment_id="" +``` + +按部署版本确认问题发生在哪份代码： + +```logql +{application="teaching-feedback"} | json | app_version="0.2.0" +{application="teaching-feedback"} | json | git_sha="" +``` + +`request_id`、`segment_id`、`git_sha` 等字段位于 JSON 正文，不能直接放进流选择器的 `{...}`， +也不能新增为 Loki 标签。查询无结果时先扩大时间范围，再从只包含 `application` 的查询逐步增加条件。 + +## 用日志定位代码 + +优化或修复代码时采用以下流程： + +1. 复现问题，记录发生时间和小程序错误界面显示的 `request_id`。 +2. 用 `request_id` 查询 API 的完整事件序列，确认状态码、`event`、`latency_ms` 和 + `error_class`。 +3. 涉及录音时，从 API 日志取得 `segment_id`，分别查询 API worker 和 FunASR，判断耗时发生在 + 排队、上传、推理还是结果写回阶段。 +4. 用 `app_version` 和 `git_sha` 确认运行代码版本，避免根据旧镜像日志修改新代码。 +5. 修改后使用相同路径复现并比较事件数量、失败分类和耗时，不以“没有 ERROR”代替性能验收。 + +日志事件主要对应以下代码： + +- HTTP 请求开始、结束、请求 ID：`server/src/main.rs` +- API 错误分类和响应：`server/src/error.rs`、`server/src/recording_routes.rs` +- 转录队列、租约、重试和状态写回：`server/src/transcription_worker.rs` +- API 到本地 ASR 的请求关联：`server/src/speech.rs` +- FunASR 请求和推理耗时：`asr-service/app.py` +- 小程序请求 ID 展示与复制：`utils/api.ts` 及各页面的错误处理代码 + +## 历史范围 + +中央栈首次接入时回采了当时仍存在的 Docker 容器日志。2026-07-22 验收时可查询到的最早记录为： + +- API：2026-07-22 11:48:42（北京时间） +- FunASR：2026-07-22 11:53:42（北京时间） + +已删除容器中更早的日志无法恢复。Loki 当前保留期为 14 天，Docker 本地日志还受到 +`20m x 5` 轮转限制，因此长期比较应记录 `git_sha`、问题时间和关键查询结果，而不能假设日志永久存在。 + +## 故障定位 + +应用健康但 Grafana 无日志： + +1. 检查业务容器是否具有采集标签。 +2. 检查业务容器是否持续向 stdout/stderr 输出日志。 +3. 检查中央栈运行状态；命令见中央目录 `README.md`。 +4. 使用 `{application="teaching-feedback"}` 查询，避免一开始叠加过多过滤条件。 + +Grafana/Loki 不可用时，不要重启或回滚业务容器来尝试修复中央栈。先确认： + +```bash +curl -fsS http://127.0.0.1:39180/health +docker compose -f compose.deploy.yml ps +docker compose -f compose.deploy.yml logs --since 10m api funasr +``` + +如果本地 Docker 日志仍正常，则问题属于中央观测栈。中央栈的启停、数据卷、备份、恢复和升级 +统一由 `/home/shay/data/docker-service/observability/README.md` 管理。 + +Grafana 不可用时，可以先用 Docker 本地日志继续定位： + +```bash +docker compose -f compose.deploy.yml logs --since 30m api funasr +``` + +不要为了临时查询而把 Loki `3100` 或 Alloy `12345` 发布到宿主机。 + +## 应用侧回滚 + +结构化日志代码出现回归时，回滚对应代码并重建业务镜像。仅日志驱动不兼容时，修改 +`compose.deploy.yml` 后必须 `--force-recreate`；普通 restart 不会更新容器日志驱动。任何应用侧 +回滚都不得删除中央 Loki 或 Grafana 数据卷。 diff --git a/pages/defaults/index.ts b/pages/defaults/index.ts index 93ac5d6..b644fde 100644 --- a/pages/defaults/index.ts +++ b/pages/defaults/index.ts @@ -1,4 +1,10 @@ -import { getApiErrorMessage, getProfileDefaults, updateProfileDefaults } from '../../utils/api' +import { + copyApiRequestId, + getApiErrorMessage, + getApiRequestId, + getProfileDefaults, + updateProfileDefaults +} from '../../utils/api' import type { StudentProfileDefaults } from '../../utils/types' const grades = ['初一', '初二', '初三'] @@ -30,7 +36,8 @@ Page({ maximumDuration, loading: false, saving: false, - errorMessage: '' + errorMessage: '', + errorRequestId: '' }, onShow() { @@ -38,7 +45,7 @@ Page({ }, async loadDefaults(showSuccess = false) { - this.setData({ loading: true, errorMessage: '' }) + this.setData({ loading: true, errorMessage: '', errorRequestId: '' }) try { const defaults = await getProfileDefaults() this.setData({ @@ -49,7 +56,7 @@ Page({ }) if (showSuccess) wx.showToast({ title: '预设已刷新', icon: 'success' }) } catch (error) { - this.setData({ errorMessage: getApiErrorMessage(error) }) + this.setData({ errorMessage: getApiErrorMessage(error), errorRequestId: getApiRequestId(error) }) } finally { this.setData({ loading: false }) } @@ -59,6 +66,10 @@ Page({ void this.loadDefaults(true) }, + copyRequestId() { + copyApiRequestId(this.data.errorRequestId) + }, + onGradeChange(event: PickerEvent) { this.setData({ 'defaults.grade': grades[Number(event.detail.value)] }) }, diff --git a/pages/defaults/index.wxml b/pages/defaults/index.wxml index 6191c45..aa9efd1 100644 --- a/pages/defaults/index.wxml +++ b/pages/defaults/index.wxml @@ -13,7 +13,13 @@ - {{errorMessage}} + + {{errorMessage}} + + 请求编号：{{errorRequestId}} + 复制 + + 重试 diff --git a/pages/feedback/index.ts b/pages/feedback/index.ts index 23fccb0..3e9cdf7 100644 --- a/pages/feedback/index.ts +++ b/pages/feedback/index.ts @@ -1,4 +1,5 @@ import { + copyApiRequestId, createFeedbackRecord, createVoiceLesson, ensureFeedbackSession, @@ -6,6 +7,7 @@ import { generateFeedbackFromVoice, getActiveFeedbackSession, getApiErrorMessage, + getApiRequestId, listProfiles, markVoiceLessonApplied, retryVoiceSegment, @@ -114,7 +116,8 @@ Page({ voiceStatus: '语音录入', voiceActionable: false, primaryActionText: '保存反馈', - errorMessage: '' + errorMessage: '', + errorRequestId: '' }, onLoad() { @@ -189,7 +192,7 @@ Page({ }, async loadProfiles() { - this.setData({ loading: true, errorMessage: '' }) + this.setData({ loading: true, errorMessage: '', errorRequestId: '' }) try { const profiles = await listProfiles() const matchedProfileIndex = profiles.findIndex((profile) => profile.id === this.data.draft.profileId) @@ -203,7 +206,7 @@ Page({ }) await this.loadActiveSession(profileId) } catch (error) { - this.setData({ errorMessage: getApiErrorMessage(error) }) + this.setData({ errorMessage: getApiErrorMessage(error), errorRequestId: getApiRequestId(error) }) } finally { this.setData({ loading: false }) } @@ -227,7 +230,7 @@ Page({ this.syncVoicePresentation(session) this.scheduleVoicePolling(session) } catch (error) { - this.setData({ errorMessage: getApiErrorMessage(error) }) + this.setData({ errorMessage: getApiErrorMessage(error), errorRequestId: getApiRequestId(error) }) } }, @@ -309,6 +312,10 @@ Page({ void this.loadProfiles() }, + copyRequestId() { + copyApiRequestId(this.data.errorRequestId) + }, + onProfileChange(event: PickerEvent) { this.clearDraftSaveTimer() const selectedProfileIndex = Number(event.detail.value) diff --git a/pages/feedback/index.wxml b/pages/feedback/index.wxml index 1550255..50b2c0d 100644 --- a/pages/feedback/index.wxml +++ b/pages/feedback/index.wxml @@ -6,7 +6,13 @@ 填写课堂表现并保存到教学反馈记录。 - {{errorMessage}} + + {{errorMessage}} + + 请求编号：{{errorRequestId}} + 复制 + + 重试 diff --git a/pages/profile/index.ts b/pages/profile/index.ts index e25b7e1..36d08cc 100644 --- a/pages/profile/index.ts +++ b/pages/profile/index.ts @@ -1,7 +1,9 @@ import { + copyApiRequestId, ensureAuthentication, getApiBaseUrl, getApiErrorMessage, + getApiRequestId, getHealth, setApiBaseUrl } from '../../utils/api' @@ -13,10 +15,12 @@ Page({ apiBaseUrl: getApiBaseUrl(), connectionText: '尚未检测', connectionTone: 'idle', + connectionRequestId: '', testing: false, authText: '尚未登录', authTone: 'idle', - authUserId: '' + authUserId: '', + authRequestId: '' }, onShow() { @@ -37,12 +41,21 @@ Page({ await this.checkConnection() await this.checkAuthentication() } catch (error) { - this.setData({ connectionText: getApiErrorMessage(error), connectionTone: 'error' }) + this.setData({ + connectionText: getApiErrorMessage(error), + connectionTone: 'error', + connectionRequestId: getApiRequestId(error) + }) } }, async checkConnection() { - this.setData({ testing: true, connectionText: '正在检测服务...', connectionTone: 'idle' }) + this.setData({ + testing: true, + connectionText: '正在检测服务...', + connectionTone: 'idle', + connectionRequestId: '' + }) try { const health = await getHealth() let connectionText = '服务、数据库和语音转录正常' @@ -62,30 +75,42 @@ Page({ } this.setData({ connectionText, - connectionTone + connectionTone, + connectionRequestId: '' }) } catch (error) { - this.setData({ connectionText: getApiErrorMessage(error), connectionTone: 'error' }) + this.setData({ + connectionText: getApiErrorMessage(error), + connectionTone: 'error', + connectionRequestId: getApiRequestId(error) + }) } finally { this.setData({ testing: false }) } }, async checkAuthentication() { - this.setData({ authText: '正在登录...', authTone: 'idle', authUserId: '' }) + this.setData({ authText: '正在登录...', authTone: 'idle', authUserId: '', authRequestId: '' }) try { const session = await ensureAuthentication() this.setData({ authText: '微信身份已登录', authTone: 'success', - authUserId: session.userId + authUserId: session.userId, + authRequestId: '' }) } catch (error) { this.setData({ authText: getApiErrorMessage(error), authTone: 'error', - authUserId: '' + authUserId: '', + authRequestId: getApiRequestId(error) }) } + }, + + copyRequestId(event: WechatMiniprogram.TouchEvent) { + const { requestId } = event.currentTarget.dataset as { requestId: string } + copyApiRequestId(requestId) } }) diff --git a/pages/profile/index.wxml b/pages/profile/index.wxml index 5b6d991..de43e49 100644 --- a/pages/profile/index.wxml +++ b/pages/profile/index.wxml @@ -13,6 +13,10 @@ {{connectionText}} + + 请求编号：{{connectionRequestId}} + 复制 + @@ -22,6 +26,10 @@ {{authText}} + + 请求编号：{{authRequestId}} + 复制 + {{authUserId}} diff --git a/pages/profile/index.wxss b/pages/profile/index.wxss index b063fa3..1162d10 100644 --- a/pages/profile/index.wxss +++ b/pages/profile/index.wxss @@ -70,6 +70,10 @@ font-size: 25rpx; } +.request-id-row { + margin-left: 28rpx; +} + .check-button { margin-top: 24rpx; } diff --git a/pages/records/index.ts b/pages/records/index.ts index e0b86bc..37cf5ef 100644 --- a/pages/records/index.ts +++ b/pages/records/index.ts @@ -1,6 +1,8 @@ import { + copyApiRequestId, deleteFeedbackRecord, getApiErrorMessage, + getApiRequestId, listFeedbackRecords, listProfiles } from '../../utils/api' @@ -17,7 +19,8 @@ Page({ selectedProfileIndex: 0, loading: false, deletingId: '', - errorMessage: '' + errorMessage: '', + errorRequestId: '' }, onShow() { @@ -26,7 +29,7 @@ Page({ }, async loadData() { - this.setData({ loading: true, errorMessage: '' }) + this.setData({ loading: true, errorMessage: '', errorRequestId: '' }) try { const profiles = await listProfiles() const selectedProfile = profiles[this.data.selectedProfileIndex - 1] @@ -38,7 +41,7 @@ Page({ records: this.withProfileNames(records, profiles) }) } catch (error) { - this.setData({ errorMessage: getApiErrorMessage(error) }) + this.setData({ errorMessage: getApiErrorMessage(error), errorRequestId: getApiRequestId(error) }) } finally { this.setData({ loading: false }) } @@ -55,15 +58,19 @@ Page({ void this.loadData() }, + copyRequestId() { + copyApiRequestId(this.data.errorRequestId) + }, + async onProfileChange(event: PickerEvent) { const selectedProfileIndex = Number(event.detail.value) const profile = this.data.profiles[selectedProfileIndex - 1] - this.setData({ selectedProfileIndex, loading: true, errorMessage: '' }) + this.setData({ selectedProfileIndex, loading: true, errorMessage: '', errorRequestId: '' }) try { const records = await listFeedbackRecords(profile?.id) this.setData({ records: this.withProfileNames(records, this.data.profiles) }) } catch (error) { - this.setData({ errorMessage: getApiErrorMessage(error) }) + this.setData({ errorMessage: getApiErrorMessage(error), errorRequestId: getApiRequestId(error) }) } finally { this.setData({ loading: false }) } diff --git a/pages/records/index.wxml b/pages/records/index.wxml index 83d7fe7..441897e 100644 --- a/pages/records/index.wxml +++ b/pages/records/index.wxml @@ -6,7 +6,13 @@ 按学生查阅已保存的课堂反馈。 - {{errorMessage}} + + {{errorMessage}} + + 请求编号：{{errorRequestId}} + 复制 + + 重试 diff --git a/pages/students/index.ts b/pages/students/index.ts index 0c56f9e..9c4a1ae 100644 --- a/pages/students/index.ts +++ b/pages/students/index.ts @@ -1,7 +1,9 @@ import { + copyApiRequestId, createProfile, deleteProfile as deleteProfileRequest, getApiErrorMessage, + getApiRequestId, getProfileDefaults, listFeedbackRecords, listProfiles @@ -107,7 +109,8 @@ Page({ loading: false, saving: false, deletingId: '', - errorMessage: '' + errorMessage: '', + errorRequestId: '' }, onShow() { @@ -119,7 +122,7 @@ Page({ }, async refreshProfiles(showSuccess = false) { - this.setData({ loading: true, errorMessage: '' }) + this.setData({ loading: true, errorMessage: '', errorRequestId: '' }) try { const profiles = (await listProfiles()).map(withAcademicTermLabel) const filteredProfiles = filterProfiles( @@ -139,7 +142,7 @@ Page({ }) if (showSuccess) wx.showToast({ title: '档案已刷新', icon: 'success' }) } catch (error) { - this.setData({ errorMessage: getApiErrorMessage(error) }) + this.setData({ errorMessage: getApiErrorMessage(error), errorRequestId: getApiRequestId(error) }) } finally { this.setData({ loading: false }) } @@ -153,6 +156,10 @@ Page({ void this.refreshProfiles() }, + copyRequestId() { + copyApiRequestId(this.data.errorRequestId) + }, + onSearch(event: InputEvent) { const query = event.detail.value this.setData({ diff --git a/pages/students/index.wxml b/pages/students/index.wxml index 9eeb79d..259a783 100644 --- a/pages/students/index.wxml +++ b/pages/students/index.wxml @@ -6,7 +6,13 @@ 管理学生默认信息，用于快速生成课堂反馈 - {{errorMessage}} + + {{errorMessage}} + + 请求编号：{{errorRequestId}} + 复制 + + 重试 diff --git a/server/Cargo.lock b/server/Cargo.lock index ce2098a..c1bc830 100644 --- a/server/Cargo.lock +++ b/server/Cargo.lock @@ -1890,7 +1890,7 @@ dependencies = [ [[package]] name = "teaching-feedback-api" -version = "0.1.0" +version = "0.2.0" dependencies = [ "axum", "base64", @@ -1906,6 +1906,7 @@ dependencies = [ "sqlx", "thiserror", "tokio", + "tower", "tower-http", "tracing", "tracing-subscriber", @@ -2051,6 +2052,7 @@ dependencies = [ "tower-service", "tracing", "url", + "uuid", ] [[package]] @@ -2109,6 +2111,16 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + [[package]] name = "tracing-subscriber" version = "0.3.23" @@ -2119,12 +2131,15 @@ dependencies = [ "nu-ansi-term", "once_cell", "regex-automata", + "serde", + "serde_json", "sharded-slab", "smallvec", "thread_local", "tracing", "tracing-core", "tracing-log", + "tracing-serde", ] [[package]] diff --git a/server/Cargo.toml b/server/Cargo.toml index 26e6525..47bf9a6 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "teaching-feedback-api" -version = "0.1.0" +version = "0.2.0" edition = "2024" rust-version = "1.85" default-run = "teaching-feedback-api" @@ -20,9 +20,10 @@ sha2 = "0.10" sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio-rustls", "postgres", "chrono", "uuid", "migrate", "macros"] } thiserror = "2" tokio = { version = "1", features = ["fs", "macros", "rt-multi-thread", "net", "signal"] } -tower-http = { version = "0.6", features = ["cors", "trace"] } +tower = { version = "0.5", features = ["util"] } +tower-http = { version = "0.6", features = ["cors", "trace", "request-id"] } tracing = "0.1" -tracing-subscriber = { version = "0.3", features = ["env-filter"] } +tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } url = "2" utoipa = { version = "5.5", features = ["chrono", "uuid"] } utoipa-axum = "0.2" diff --git a/server/Dockerfile b/server/Dockerfile index eec5c6b..b5b6c36 100644 --- a/server/Dockerfile +++ b/server/Dockerfile @@ -28,6 +28,8 @@ RUN --mount=type=cache,id=teaching-feedback-cargo-registry,target=/usr/local/car FROM debian:bookworm-slim AS runtime ARG DEBIAN_MIRROR=http://deb.debian.org +ARG APP_VERSION=0.2.0 +ARG GIT_SHA=unknown RUN sed -i "s|http://deb.debian.org|${DEBIAN_MIRROR}|g" /etc/apt/sources.list.d/debian.sources \ && apt-get update \ @@ -44,7 +46,11 @@ COPY --from=builder /build/output/migrate-user-owner /usr/local/bin/migrate-user ENV HOST=0.0.0.0 \ PORT=8080 \ AUDIO_STORAGE_DIR=/data/audio \ - RUST_LOG=info + RUST_LOG=info \ + APP_ENV=lan \ + APP_VERSION=${APP_VERSION} \ + GIT_SHA=${GIT_SHA} \ + LOG_FORMAT=json USER app diff --git a/server/src/config.rs b/server/src/config.rs index a3c1642..646321a 100644 --- a/server/src/config.rs +++ b/server/src/config.rs @@ -1,5 +1,11 @@ use std::{env, net::IpAddr, path::PathBuf}; +#[derive(Clone, Copy, Eq, PartialEq)] +pub enum LogFormat { + Json, + Pretty, +} + #[derive(Clone)] pub enum SpeechProviderConfig { Local { api_url: String }, @@ -50,6 +56,10 @@ pub struct Config { pub database_url: Option, pub database_max_connections: u32, pub cors_allowed_origin: Option, + pub log_format: LogFormat, + pub app_env: String, + pub app_version: String, + pub git_sha: String, pub audio_storage_dir: PathBuf, pub speech: Option, pub summary: Option, @@ -70,6 +80,11 @@ impl Config { .unwrap_or_else(|_| "5".to_owned()) .parse() .map_err(|_| "DATABASE_MAX_CONNECTIONS must be a valid u32 value".to_owned())?; + let log_format = match optional_env("LOG_FORMAT").as_deref().unwrap_or("pretty") { + "json" => LogFormat::Json, + "pretty" => LogFormat::Pretty, + _ => return Err("LOG_FORMAT must be json or pretty".to_owned()), + }; let speech = speech_config()?; let summary = summary_config()?; @@ -85,6 +100,11 @@ impl Config { cors_allowed_origin: env::var("CORS_ALLOWED_ORIGIN") .ok() .filter(|value| !value.trim().is_empty()), + log_format, + app_env: optional_env("APP_ENV").unwrap_or_else(|| "development".to_owned()), + app_version: optional_env("APP_VERSION") + .unwrap_or_else(|| env!("CARGO_PKG_VERSION").to_owned()), + git_sha: optional_env("GIT_SHA").unwrap_or_else(|| "unknown".to_owned()), audio_storage_dir: env::var("AUDIO_STORAGE_DIR") .map(PathBuf::from) .unwrap_or_else(|_| PathBuf::from("./data/audio")), diff --git a/server/src/error.rs b/server/src/error.rs index f2da806..e786ef5 100644 --- a/server/src/error.rs +++ b/server/src/error.rs @@ -92,8 +92,12 @@ impl IntoResponse for ApiError { } impl From for ApiError { - fn from(error: sqlx::Error) -> Self { - tracing::error!(?error, "database query failed"); + fn from(_: sqlx::Error) -> Self { + tracing::error!( + event = "database_query_failed", + error_class = "database_query_failed", + "database query failed" + ); Self::Internal } } diff --git a/server/src/main.rs b/server/src/main.rs index 9dc4d91..7b98580 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -7,12 +7,25 @@ mod speech; mod summary; mod transcription_worker; -use std::{path::PathBuf, sync::Arc}; +use std::{path::PathBuf, sync::Arc, time::Duration}; -use axum::{Json, Router, extract::DefaultBodyLimit, http::HeaderValue, routing::get}; -use config::Config; +use axum::{ + Json, Router, + extract::{DefaultBodyLimit, MatchedPath}, + http::{HeaderName, HeaderValue, Request}, + response::Response, + routing::get, +}; +use config::{Config, LogFormat}; use sqlx::{PgPool, postgres::PgPoolOptions}; -use tower_http::{cors::CorsLayer, trace::TraceLayer}; +use tower::ServiceBuilder; +use tower_http::{ + classify::ServerErrorsFailureClass, + cors::CorsLayer, + request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer}, + trace::TraceLayer, +}; +use tracing::Span; use tracing_subscriber::EnvFilter; use utoipa_scalar::{Scalar, Servable}; @@ -25,15 +38,30 @@ pub struct AppState { pub auth: auth::AuthService, } +#[derive(Clone)] +pub(crate) struct LogContext { + pub(crate) environment: String, + pub(crate) app_version: String, + pub(crate) git_sha: String, +} + +impl From<&Config> for LogContext { + fn from(config: &Config) -> Self { + Self { + environment: config.app_env.clone(), + app_version: config.app_version.clone(), + git_sha: config.git_sha.clone(), + } + } +} + #[tokio::main] async fn main() -> Result<(), Box> { dotenvy::dotenv().ok(); - tracing_subscriber::fmt() - .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into())) - .init(); - let config = Config::from_env().map_err(std::io::Error::other)?; - let pool = connect_database(&config).await?; + init_tracing(&config); + let log_context = LogContext::from(&config); + let pool = connect_database(&config, &log_context).await?; let speech = speech::SpeechService::new(config.speech.clone()); let auth = auth::AuthService::new(config.auth.clone()); let state = AppState { @@ -45,21 +73,57 @@ async fn main() -> Result<(), Box> { }; if let Some(pool) = state.pool.clone() { if speech.configured() { - transcription_worker::spawn(pool, speech); + transcription_worker::spawn(pool, speech.clone(), log_context.clone()); } } - let app = build_app(state, config.cors_allowed_origin.as_deref())?; + let app = build_app( + state, + config.cors_allowed_origin.as_deref(), + log_context.clone(), + )?; let address = std::net::SocketAddr::from((config.host, config.port)); let listener = tokio::net::TcpListener::bind(address).await?; - tracing::info!(%address, scalar = %format!("http://{address}/scalar"), "API server is listening"); + tracing::info!( + service = "api", + environment = %log_context.environment, + app_version = %log_context.app_version, + git_sha = %log_context.git_sha, + event = "api_started", + %address, + scalar = %format!("http://{address}/scalar"), + provider = ?speech.provider_name(), + "API server is listening" + ); axum::serve(listener, app) - .with_graceful_shutdown(shutdown_signal()) + .with_graceful_shutdown(shutdown_signal(log_context)) .await?; Ok(()) } -pub fn build_app(state: AppState, cors_allowed_origin: Option<&str>) -> Result { +fn init_tracing(config: &Config) { + let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()); + match config.log_format { + LogFormat::Json => tracing_subscriber::fmt() + .json() + .flatten_event(true) + .with_ansi(false) + .with_current_span(true) + .with_span_list(true) + .with_env_filter(filter) + .init(), + LogFormat::Pretty => tracing_subscriber::fmt() + .with_env_filter(filter) + .pretty() + .init(), + } +} + +fn build_app( + state: AppState, + cors_allowed_origin: Option<&str>, + log_context: LogContext, +) -> Result { let (api_router, openapi) = routes::router(); let openapi_json = openapi.clone(); let mut app = api_router @@ -72,21 +136,142 @@ pub fn build_app(state: AppState, cors_allowed_origin: Option<&str>) -> Result| { + make_http_request_span(request, &request_span_context) + }) + .on_request(move |request: &Request<_>, span: &Span| { + tracing::info!( + parent: span, + service = "api", + environment = %request_started_context.environment, + app_version = %request_started_context.app_version, + git_sha = %request_started_context.git_sha, + event = "http_request_started", + request_id = request_id_from_headers(request), + "HTTP request started" + ); + }) + .on_response(move |response: &Response, latency: Duration, span: &Span| { + let status = response.status(); + let latency_ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX); + let request_id = response + .headers() + .get("x-request-id") + .and_then(|value| value.to_str().ok()) + .unwrap_or("unknown"); + if status.is_server_error() { + tracing::error!( + parent: span, + service = "api", + environment = %request_finished_context.environment, + app_version = %request_finished_context.app_version, + git_sha = %request_finished_context.git_sha, + event = "http_request_finished", + status = status.as_u16(), + latency_ms, + request_id, + error_class = "server_error", + "HTTP request finished" + ); + } else { + tracing::info!( + parent: span, + service = "api", + environment = %request_finished_context.environment, + app_version = %request_finished_context.app_version, + git_sha = %request_finished_context.git_sha, + event = "http_request_finished", + status = status.as_u16(), + latency_ms, + request_id, + "HTTP request finished" + ); + } + }) + .on_failure( + move |failure: ServerErrorsFailureClass, latency: Duration, span: &Span| { + let latency_ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX); + tracing::error!( + parent: span, + service = "api", + environment = %request_failed_context.environment, + app_version = %request_failed_context.app_version, + git_sha = %request_failed_context.git_sha, + event = "http_request_failed", + latency_ms, + error_class = ?failure, + "HTTP request failed" + ); + }, + ), + ) + .layer(PropagateRequestIdLayer::x_request_id()), + )) } -async fn connect_database(config: &Config) -> Result, sqlx::Error> { +fn make_http_request_span(request: &Request, context: &LogContext) -> Span { + let route = request + .extensions() + .get::() + .map(MatchedPath::as_str) + .unwrap_or_else(|| request.uri().path()); + let request_id = request_id_from_headers(request); + + tracing::info_span!( + "http_request", + service = "api", + environment = %context.environment, + app_version = %context.app_version, + git_sha = %context.git_sha, + method = %request.method(), + route, + path = %request.uri().path(), + request_id, + ) +} + +fn request_id_from_headers(request: &Request) -> &str { + request + .headers() + .get("x-request-id") + .and_then(|value| value.to_str().ok()) + .unwrap_or("invalid") +} + +async fn connect_database( + config: &Config, + log_context: &LogContext, +) -> Result, sqlx::Error> { let Some(database_url) = &config.database_url else { - tracing::warn!("DATABASE_URL is not set; data endpoints will return HTTP 503"); + tracing::warn!( + service = "api", + environment = %log_context.environment, + app_version = %log_context.app_version, + git_sha = %log_context.git_sha, + event = "database_not_configured", + "DATABASE_URL is not set; data endpoints will return HTTP 503" + ); return Ok(None); }; @@ -94,11 +279,95 @@ async fn connect_database(config: &Config) -> Result, sqlx::Error .max_connections(config.database_max_connections) .connect(database_url) .await?; - tracing::info!("connected to PostgreSQL"); + tracing::info!( + service = "api", + environment = %log_context.environment, + app_version = %log_context.app_version, + git_sha = %log_context.git_sha, + event = "database_connected", + "connected to PostgreSQL" + ); Ok(Some(pool)) } -async fn shutdown_signal() { +async fn shutdown_signal(log_context: LogContext) { let _ = tokio::signal::ctrl_c().await; - tracing::info!("shutdown signal received"); + tracing::info!( + service = "api", + environment = %log_context.environment, + app_version = %log_context.app_version, + git_sha = %log_context.git_sha, + event = "api_shutdown", + "shutdown signal received" + ); +} + +#[cfg(test)] +mod tests { + use axum::{body::Body, http::Request}; + use tower::ServiceExt; + use uuid::Uuid; + + use super::{AppState, LogContext, build_app}; + use crate::{ + auth::AuthService, config::AuthConfig, speech::SpeechService, summary::SummaryService, + }; + + fn app() -> axum::Router { + build_app( + AppState { + pool: None, + audio_storage_dir: std::path::PathBuf::from("./data/audio"), + speech: SpeechService::new(None), + summary: SummaryService::new(None), + auth: AuthService::new(AuthConfig { + wechat: None, + session_ttl_days: 30, + allow_development_user_header: false, + }), + }, + None, + LogContext { + environment: "test".to_owned(), + app_version: "test".to_owned(), + git_sha: "test".to_owned(), + }, + ) + .unwrap() + } + + #[tokio::test] + async fn health_generates_a_uuid_request_id() { + let response = app() + .oneshot( + Request::builder() + .uri("/health") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let request_id = response + .headers() + .get("x-request-id") + .unwrap() + .to_str() + .unwrap(); + assert!(Uuid::parse_str(request_id).is_ok()); + } + + #[tokio::test] + async fn health_propagates_a_client_request_id() { + let response = app() + .oneshot( + Request::builder() + .uri("/health") + .header("x-request-id", "test-request-id") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.headers()["x-request-id"], "test-request-id"); + } } diff --git a/server/src/recording_routes.rs b/server/src/recording_routes.rs index 1cca5ac..db795a8 100644 --- a/server/src/recording_routes.rs +++ b/server/src/recording_routes.rs @@ -970,8 +970,12 @@ async fn touch_session(pool: &PgPool, session_id: Uuid) -> Result<(), ApiError> Ok(()) } -fn internal_io_error(error: std::io::Error) -> ApiError { - tracing::error!(?error, "audio file operation failed"); +fn internal_io_error(_: std::io::Error) -> ApiError { + tracing::error!( + event = "audio_storage_io_failed", + error_class = "audio_storage_io_failed", + "audio file operation failed" + ); ApiError::Internal } diff --git a/server/src/speech.rs b/server/src/speech.rs index aa3843d..0d4c145 100644 --- a/server/src/speech.rs +++ b/server/src/speech.rs @@ -107,6 +107,7 @@ impl SpeechService { audio: Vec, audio_format: &str, duration_ms: i32, + request_id: &str, ) -> Result { let config = self .config @@ -114,7 +115,7 @@ impl SpeechService { .ok_or_else(|| "语音转录服务尚未配置".to_owned())?; match &config.provider { SpeechProviderConfig::Local { api_url } => { - self.transcribe_local(api_url, audio, audio_format, duration_ms) + self.transcribe_local(api_url, audio, audio_format, duration_ms, request_id) .await } SpeechProviderConfig::Tencent(config) => { @@ -129,6 +130,7 @@ impl SpeechService { audio: Vec, audio_format: &str, duration_ms: i32, + request_id: &str, ) -> Result { let part = Part::bytes(audio) .file_name(format!("segment.{audio_format}")) @@ -141,6 +143,7 @@ impl SpeechService { let response = self .client .post(local_transcription_url(api_url)) + .header("X-Request-ID", request_id) .multipart(form) .send() .await @@ -151,10 +154,7 @@ impl SpeechService { .await .map_err(|error| format!("读取本地语音转录响应失败：{error}"))?; if !status.is_success() { - return Err(format!( - "本地语音转录服务返回 HTTP {status}：{}", - compact_error(&body) - )); + return Err(format!("本地语音转录服务返回 HTTP {status}")); } let result: LocalResponse = serde_json::from_str(&body) .map_err(|error| format!("本地语音转录响应格式无效：{error}"))?; @@ -239,11 +239,6 @@ fn audio_content_type(audio_format: &str) -> &'static str { } } -fn compact_error(body: &str) -> String { - let value = body.split_whitespace().collect::>().join(" "); - value.chars().take(240).collect() -} - fn signed_request( config: &TencentSpeechConfig, audio_format: &str, @@ -279,6 +274,7 @@ fn signed_request( mod tests { use axum::{ Json, Router, + http::HeaderMap, routing::{get, post}, }; use serde_json::json; @@ -318,7 +314,8 @@ mod tests { .route("/health", get(|| async { Json(json!({"status": "ok"})) })) .route( "/v1/transcriptions", - post(|| async { + post(|headers: HeaderMap| async move { + assert_eq!(headers["x-request-id"], "segment-request-id"); Json(json!({ "text": "课堂练习完成。", "duration_ms": 12_345, @@ -340,7 +337,7 @@ mod tests { assert!(service.available().await); let result = service - .transcribe(vec![1, 2, 3], "mp3", 10_000) + .transcribe(vec![1, 2, 3], "mp3", 10_000, "segment-request-id") .await .unwrap(); assert_eq!(result.text, "课堂练习完成。"); diff --git a/server/src/transcription_worker.rs b/server/src/transcription_worker.rs index 36b325f..f065fe6 100644 --- a/server/src/transcription_worker.rs +++ b/server/src/transcription_worker.rs @@ -1,9 +1,14 @@ -use std::time::Duration; +use std::time::{Duration, Instant}; +use chrono::{DateTime, Utc}; use sqlx::{FromRow, PgPool}; +use tracing::Instrument; use uuid::Uuid; -use crate::speech::{SpeechService, Transcription}; +use crate::{ + LogContext, + speech::{SpeechService, Transcription}, +}; const IDLE_POLL_INTERVAL: Duration = Duration::from_millis(750); const UNAVAILABLE_POLL_INTERVAL: Duration = Duration::from_secs(5); @@ -12,18 +17,29 @@ const UNAVAILABLE_POLL_INTERVAL: Duration = Duration::from_secs(5); struct TranscriptionJob { id: Uuid, lesson_session_id: Uuid, + feedback_session_id: Uuid, duration_ms: i32, audio_format: String, audio_path: String, + created_at: DateTime, + processing_started_at: DateTime, + transcription_attempts: i32, } -pub fn spawn(pool: PgPool, speech: SpeechService) { +pub fn spawn(pool: PgPool, speech: SpeechService, log_context: LogContext) { for worker_id in 1..=speech.worker_concurrency() { let pool = pool.clone(); let speech = speech.clone(); - tokio::spawn(async move { - run(worker_id, pool, speech).await; - }); + let log_context = log_context.clone(); + let span = tracing::info_span!( + "transcription_worker", + service = "api", + environment = %log_context.environment, + app_version = %log_context.app_version, + git_sha = %log_context.git_sha, + worker_id, + ); + tokio::spawn(async move { run(worker_id, pool, speech).await }.instrument(span)); } } @@ -36,8 +52,13 @@ async fn run(worker_id: usize, pool: PgPool, speech: SpeechService) { continue; } Ok(true) => {} - Err(error) => { - tracing::error!(worker_id, error = %error, "failed to inspect transcription queue"); + Err(_) => { + tracing::error!( + event = "transcription_queue_inspection_failed", + worker_id, + error_class = "database_query_failed", + "failed to inspect transcription queue" + ); tokio::time::sleep(UNAVAILABLE_POLL_INTERVAL).await; continue; } @@ -45,28 +66,49 @@ async fn run(worker_id: usize, pool: PgPool, speech: SpeechService) { let available = speech.available().await; if !available { if service_available { - tracing::warn!(worker_id, provider = ?speech.provider_name(), "ASR service is unavailable; queued audio will wait"); + tracing::warn!( + event = "asr_service_unavailable", + worker_id, + provider = speech.provider_name().unwrap_or("unknown"), + "ASR service is unavailable; queued audio will wait" + ); } service_available = false; tokio::time::sleep(UNAVAILABLE_POLL_INTERVAL).await; continue; } if !service_available { - tracing::info!(worker_id, provider = ?speech.provider_name(), "ASR service is available again"); + tracing::info!( + event = "asr_service_recovered", + worker_id, + provider = speech.provider_name().unwrap_or("unknown"), + "ASR service is available again" + ); } service_available = true; match claim_next_job(&pool, speech.job_lease_seconds()).await { Ok(Some(job)) => { - tracing::info!(worker_id, segment_id = %job.id, "transcribing audio segment"); - if let Err(error) = process_job(&pool, &speech, &job).await { - tracing::error!(worker_id, segment_id = %job.id, error = %error, "transcription job update failed"); + if process_job(&pool, &speech, worker_id, &job).await.is_err() { + tracing::error!( + event = "transcription_job_update_failed", + worker_id, + segment_id = %job.id, + feedback_session_id = %job.feedback_session_id, + error_class = "database_update_failed", + "transcription job update failed" + ); tokio::time::sleep(IDLE_POLL_INTERVAL).await; } } Ok(None) => tokio::time::sleep(IDLE_POLL_INTERVAL).await, - Err(error) => { - tracing::error!(worker_id, error = %error, "failed to claim transcription job"); + Err(_) => { + tracing::error!( + event = "transcription_job_claim_failed", + worker_id, + error_class = "database_query_failed", + "failed to claim transcription job" + ); tokio::time::sleep(UNAVAILABLE_POLL_INTERVAL).await; } } @@ -104,10 +146,13 @@ async fn claim_next_job( SET processing_started_at = now(), \ transcription_attempts = transcription_attempts + 1, \ updated_at = now() \ - FROM next_job \ + FROM next_job, lesson_sessions AS lesson \ WHERE segment.id = next_job.id \ - RETURNING segment.id, segment.lesson_session_id, segment.duration_ms, \ - segment.audio_format, segment.audio_path", + AND lesson.id = segment.lesson_session_id \ + RETURNING segment.id, segment.lesson_session_id, lesson.feedback_session_id, \ + segment.duration_ms, segment.audio_format, segment.audio_path, \ + segment.created_at, segment.processing_started_at, \ + segment.transcription_attempts", ) .bind(lease_seconds) .fetch_optional(pool) @@ -117,29 +162,96 @@ async fn claim_next_job( async fn process_job( pool: &PgPool, speech: &SpeechService, + worker_id: usize, job: &TranscriptionJob, ) -> Result<(), sqlx::Error> { - let result = match tokio::fs::read(&job.audio_path).await { - Ok(audio) => { - speech - .transcribe(audio, &job.audio_format, job.duration_ms) - .await + let queue_wait_ms = elapsed_since(job.created_at); + tracing::info!( + event = "transcription_job_started", + worker_id, + segment_id = %job.id, + feedback_session_id = %job.feedback_session_id, + lesson_id = %job.lesson_session_id, + attempt = job.transcription_attempts, + queue_wait_ms, + audio_duration_ms = job.duration_ms, + audio_format = %job.audio_format, + status = "processing", + "transcribing audio segment" + ); + + let audio_read_started = Instant::now(); + let audio = match tokio::fs::read(&job.audio_path).await { + Ok(audio) => audio, + Err(_) => { + fail_job( + pool, + job, + worker_id, + queue_wait_ms, + elapsed_ms(audio_read_started), + 0, + 0, + "audio_read_failed", + "unable to read audio for transcription", + ) + .await?; + refresh_lesson(pool, job.lesson_session_id).await?; + return Ok(()); } - Err(error) => Err(format!("读取录音文件失败：{error}")), }; + let audio_read_latency_ms = elapsed_ms(audio_read_started); + let audio_size_bytes = u64::try_from(audio.len()).unwrap_or(u64::MAX); + let asr_request_id = job.id.to_string(); + let inference_started = Instant::now(); + let result = speech + .transcribe(audio, &job.audio_format, job.duration_ms, &asr_request_id) + .await; + let inference_latency_ms = elapsed_ms(inference_started); match result { - Ok(transcription) => complete_job(pool, job, transcription).await?, - Err(message) => fail_job(pool, job, &message).await?, + Ok(transcription) => { + complete_job( + pool, + job, + worker_id, + transcription, + queue_wait_ms, + audio_read_latency_ms, + inference_latency_ms, + audio_size_bytes, + ) + .await? + } + Err(message) => { + fail_job( + pool, + job, + worker_id, + queue_wait_ms, + audio_read_latency_ms, + inference_latency_ms, + audio_size_bytes, + transcription_error_class(&message), + &message, + ) + .await? + } } refresh_lesson(pool, job.lesson_session_id).await?; Ok(()) } +#[allow(clippy::too_many_arguments)] async fn complete_job( pool: &PgPool, job: &TranscriptionJob, + worker_id: usize, transcription: Transcription, + queue_wait_ms: i64, + audio_read_latency_ms: u64, + inference_latency_ms: u64, + audio_size_bytes: u64, ) -> Result<(), sqlx::Error> { let duration_ms = if transcription.duration_ms > 0 { transcription.duration_ms @@ -158,12 +270,40 @@ async fn complete_job( .bind(&transcription.request_id) .execute(pool) .await?; - tracing::info!(segment_id = %job.id, "audio segment transcription is ready"); + tracing::info!( + event = "transcription_job_finished", + worker_id, + segment_id = %job.id, + feedback_session_id = %job.feedback_session_id, + lesson_id = %job.lesson_session_id, + asr_request_id = %transcription.request_id, + attempt = job.transcription_attempts, + queue_wait_ms, + audio_read_latency_ms, + inference_latency_ms, + processing_elapsed_ms = elapsed_since(job.processing_started_at), + audio_size_bytes, + audio_duration_ms = duration_ms, + audio_format = %job.audio_format, + provider = "local-funasr", + status = "ready", + "audio segment transcription is ready" + ); Ok(()) } -async fn fail_job(pool: &PgPool, job: &TranscriptionJob, message: &str) -> Result<(), sqlx::Error> { - tracing::warn!(segment_id = %job.id, error = %message, "audio transcription failed"); +#[allow(clippy::too_many_arguments)] +async fn fail_job( + pool: &PgPool, + job: &TranscriptionJob, + worker_id: usize, + queue_wait_ms: i64, + audio_read_latency_ms: u64, + inference_latency_ms: u64, + audio_size_bytes: u64, + error_class: &str, + message: &str, +) -> Result<(), sqlx::Error> { sqlx::query( "UPDATE audio_segments \ SET status = 'failed', error_message = $2, processing_started_at = NULL, updated_at = now() \ @@ -173,9 +313,48 @@ async fn fail_job(pool: &PgPool, job: &TranscriptionJob, message: &str) -> Resul .bind(message) .execute(pool) .await?; + tracing::warn!( + event = "transcription_job_failed", + worker_id, + segment_id = %job.id, + feedback_session_id = %job.feedback_session_id, + lesson_id = %job.lesson_session_id, + asr_request_id = %job.id, + attempt = job.transcription_attempts, + queue_wait_ms, + audio_read_latency_ms, + inference_latency_ms, + processing_elapsed_ms = elapsed_since(job.processing_started_at), + audio_size_bytes, + audio_duration_ms = job.duration_ms, + audio_format = %job.audio_format, + status = "failed", + error_class, + "audio transcription failed" + ); Ok(()) } +fn transcription_error_class(message: &str) -> &'static str { + if message.contains("连接失败") || message.contains("unavailable") { + "asr_unavailable" + } else if message.contains("HTTP") { + "asr_http_error" + } else if message.contains("读取录音") { + "audio_read_failed" + } else { + "transcription_failed" + } +} + +fn elapsed_ms(start: Instant) -> u64 { + u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX) +} + +fn elapsed_since(start: DateTime) -> i64 { + (Utc::now() - start).num_milliseconds().max(0) +} + async fn refresh_lesson(pool: &PgPool, lesson_id: Uuid) -> Result<(), sqlx::Error> { sqlx::query( "UPDATE lesson_sessions AS lesson \ diff --git a/utils/api.ts b/utils/api.ts index af00596..f7f2b91 100644 --- a/utils/api.ts +++ b/utils/api.ts @@ -132,7 +132,8 @@ type HttpMethod = 'GET' | 'POST' | 'PUT' | 'DELETE' export class ApiRequestError extends Error { constructor( message: string, - readonly statusCode?: number + readonly statusCode?: number, + readonly requestId?: string ) { super(message) this.name = 'ApiRequestError' @@ -173,6 +174,20 @@ export function getApiErrorMessage(error: unknown): string { return error.message } +export function getApiRequestId(error: unknown): string { + return error instanceof ApiRequestError ? error.requestId || '' : '' +} + +export function copyApiRequestId(requestId: string): void { + if (!requestId) return + wx.setClipboardData({ + data: requestId, + success() { + wx.showToast({ title: '请求编号已复制', icon: 'none' }) + } + }) +} + export function ensureAuthentication(): Promise { return ensureAuthSession(getApiBaseUrl()) } @@ -223,7 +238,13 @@ function requestOnce( } const body = response.data as ApiErrorBody - reject(new ApiRequestError(body?.error || `服务返回 HTTP ${response.statusCode}`, response.statusCode)) + reject( + new ApiRequestError( + body?.error || `服务返回 HTTP ${response.statusCode}`, + response.statusCode, + responseRequestId(response.header) + ) + ) }, fail(error) { reject(new ApiRequestError(`无法连接后端：${error.errMsg}`)) @@ -489,14 +510,26 @@ function uploadVoiceSegmentOnce( try { body = JSON.parse(response.data) as RawVoiceAudioSegment | ApiErrorBody } catch { - reject(new ApiRequestError('服务返回了无效的录音响应', response.statusCode)) + reject( + new ApiRequestError( + '服务返回了无效的录音响应', + response.statusCode, + responseRequestId(uploadResponseHeaders(response)) + ) + ) return } if (response.statusCode >= 200 && response.statusCode < 300) { resolve(mapVoiceSegment(body as RawVoiceAudioSegment)) return } - reject(new ApiRequestError((body as ApiErrorBody).error || `服务返回 HTTP ${response.statusCode}`, response.statusCode)) + reject( + new ApiRequestError( + (body as ApiErrorBody).error || `服务返回 HTTP ${response.statusCode}`, + response.statusCode, + responseRequestId(uploadResponseHeaders(response)) + ) + ) }, fail(error) { reject(new ApiRequestError(`录音上传失败：${error.errMsg}`)) @@ -505,6 +538,20 @@ function uploadVoiceSegmentOnce( }) } +function responseRequestId(headers: Record | undefined): string | undefined { + for (const [name, value] of Object.entries(headers || {})) { + if (name.toLowerCase() !== 'x-request-id' || typeof value !== 'string') continue + const requestId = value.trim() + if (requestId) return requestId + } + return undefined +} + +function uploadResponseHeaders(response: unknown): Record | undefined { + const headers = (response as { header?: unknown }).header + return headers && typeof headers === 'object' ? (headers as Record) : undefined +} + export async function finishVoiceLesson(lessonId: string): Promise { const result = await request( `/api/v1/feedback-lessons/${encodeURIComponent(lessonId)}/finish`,