diff --git a/.env.example b/.env.example index ccd6258..fa16bf3 100644 --- a/.env.example +++ b/.env.example @@ -12,9 +12,12 @@ LOG_LEVEL=info MAX_REQUEST_BODY_BYTES=2097152 MAX_INPUT_CHARS=200000 -# 最近处理日志保留条数：10-100，默认并建议保持 100；仅保存在当前进程内存 +# 最近处理详情保留条数：10-100，默认并建议保持 100 LOG_CAPACITY=100 +# SQLite 处理日志数据库路径；Compose 默认映射到当前目录 ./data/auditer.db +TRACE_DB_PATH=/data/auditer.db + # 上游 HTTP 连接池；通常无需修改 HTTP_MAX_CONNECTIONS=200 HTTP_MAX_KEEPALIVE=50 diff --git a/README.md b/README.md index 18e85eb..81d5ce5 100644 --- a/README.md +++ b/README.md @@ -20,11 +20,12 @@ Categories: None|Violent|Jailbreak|... - 固定使用网页配置的实际审核模型，不受 sub2api 请求中的 `model` 字段影响。 - 兼容 JSON、`flagged/confidence`、Markdown JSON 代码块和 Qwen3Guard 原生文本。 - 把模型判定归一化为标准 OpenAI Chat Completions 响应。 -- 最近处理日志采用内存环形缓冲区，默认并最多保留 100 条。 +- 最近 100 条已完成处理详情持久化到 SQLite；重启容器后自动恢复。 - 每条日志记录四个关键时间点和三个阶段耗时。 - 内置统计页：吞吐、成功率、平均/P50/P95/最大延迟、阶段耗时、判定分布、错误分布和最慢请求。 - 使用 Starlette、Uvicorn 和 httpx 异步 I/O，复用上游 HTTP 连接池。 -- 支持 Docker、Docker Compose、健康检查和两层可选令牌鉴权。 +- 官方 Compose 直接使用 `ghcr.io/coderdoubleflower/sub2apiauditer:latest`。 +- 支持 Docker Compose、健康检查和两层可选令牌鉴权。 ## 工作流程 @@ -57,7 +58,7 @@ http://服务器地址:8080/ 页面包含三个页签： -1. **运行统计**：显示最近日志窗口的性能、吞吐、判定和错误统计。 +1. **运行统计**：显示最近 100 条日志窗口的性能、吞吐、判定和错误统计。 2. **处理日志**：显示最近 100 条请求的四个时间点、阶段耗时、结果与错误详情。 3. **节点配置**：管理上游网关、模型、密钥和提示词，并执行连通性与格式测试。 @@ -112,9 +113,31 @@ window.addEventListener('message', (event) => { > 服务响应允许被 iframe 嵌入。公开部署时建议由反向代理限制可嵌入来源，并使用 HTTPS。 -## 处理日志与时间定义 +## 处理日志、SQLite 与时间定义 -日志默认且最多保留当前进程最近 **100 条**，按新到旧显示。数据仅保存在内存，重启容器或进程后清空；清空日志也会同步清空统计窗口。 +服务默认且最多保留最近 **100 条**处理详情。 + +运行过程中，正在处理的 Trace 先保存在内存中；当 Auditer 已经把 HTTP 响应完整发送给 sub2api 后，才把这一条完整记录一次性写入 SQLite。Docker Compose 默认数据库路径为： + +```text +/data/auditer.db +``` + +宿主机对应： + +```text +./data/auditer.db +``` + +SQLite 使用 `WAL` 日志模式与 `synchronous=NORMAL`。这样既能跨容器重启恢复最近 100 条记录，又避免在“收到请求 / 转发上游 / 收到 LLM 回复”等每个阶段同步写磁盘。 + +需要注意： + +- 正常完成或正常返回错误的请求，会在响应发送完成后持久化。 +- 如果进程被强制杀死、宿主机断电等情况发生在请求仍“处理中”，这一条尚未完成的 Trace 可能来不及写入 SQLite。 +- SQLite 只保存处理元数据，不保存完整 Prompt、API Key 或完整模型原始输出。 +- 数据库会自动删除第 101 条及更老的完成记录，始终只保留最近 100 条。 +- `DELETE /api/logs` 会同时清空内存窗口和 SQLite 中的处理记录。 每条记录包含以下四个时间点： @@ -134,7 +157,7 @@ window.addEventListener('message', (event) => { 总耗时 = sub2api_replied_at - received_at ``` -展示时间使用 UTC 墙钟时间并精确到毫秒；所有耗时均由 `time.perf_counter_ns()` 单调时钟计算，系统时间/NTP 调整不会制造负延迟。 +展示时间使用 UTC 墙钟时间并精确到毫秒；请求运行期间的耗时均由 `time.perf_counter_ns()` 单调时钟计算，系统时间/NTP 调整不会制造负延迟。完成后会把已经计算好的阶段耗时一并持久化，所以重启后仍可显示原始耗时。 对于在某个阶段之前失败的请求，后续时间点会保持为空。例如连接上游失败时，没有 `llm_replied_at`；服务仍会记录错误码、HTTP 状态和已经发生的阶段。 @@ -180,7 +203,7 @@ Authorization: Bearer } ``` -清空当前实例日志： +清空处理记录和统计窗口： ```http DELETE /api/logs @@ -189,7 +212,7 @@ Authorization: Bearer ## 统计页 -统计数据严格从当前实例的内存日志窗口实时聚合，不维护另一套数据库计数，因此统计页与日志页来源一致。 +统计数据直接从当前最近 100 条窗口实时聚合。服务启动时会先从 SQLite 恢复持久化完成记录，因此容器正常重启后统计页不会从零开始。 包括： @@ -211,13 +234,7 @@ GET /api/statistics Authorization: Bearer ``` -由于日志是进程内数据： - -- 单实例、单 Uvicorn worker 时，页面看到的是该服务实例完整的最近 100 条窗口； -- 多进程、多容器或多副本部署时，每个 worker/实例各自维护一套窗口； -- 如果需要跨实例长期统计，应接入外部指标系统或持久化存储，而不是增加同步磁盘写入影响审计热路径。 - -Docker 默认启动一个 Uvicorn worker，适合保持日志和统计视图一致。 +Docker 默认启动一个 Uvicorn worker，适合保持内存窗口、SQLite 恢复结果和统计视图一致。当前实现不面向多个独立副本共享同一个 SQLite 文件；需要水平扩容时，应改用集中式数据库或指标系统。 ## 模型输出格式 @@ -238,25 +255,7 @@ Safety: Unsafe Categories: Jailbreak, PII ``` -并包装成标准 OpenAI Chat Completions envelope： - -```json -{ - "id": "chatcmpl-audit-...", - "object": "chat.completion", - "model": "sub2api-auditer", - "choices": [ - { - "index": 0, - "message": { - "role": "assistant", - "content": "Safety: Unsafe\nCategories: Jailbreak, PII" - }, - "finish_reason": "stop" - } - ] -} -``` +并包装成标准 OpenAI Chat Completions envelope。 还兼容： @@ -310,12 +309,44 @@ error.code = audit_model_invalid_response ## Docker Compose 部署 +Compose 默认直接拉取 GitHub Container Registry 中的多架构镜像： + +```text +ghcr.io/coderdoubleflower/sub2apiauditer:latest +``` + +支持 `linux/amd64` 和 `linux/arm64`。 + +部署： + ```bash git clone https://github.com/CoderDoubleflower/Sub2apiAuditer.git cd Sub2apiAuditer cp .env.example .env + +mkdir -p data +sudo chown -R 10001:10001 data +chmod 700 data + +docker compose pull +docker compose up -d +``` + +运行后当前目录中的持久化数据为： + +```text +Sub2apiAuditer/ +├── docker-compose.yml +├── .env +└── data/ + ├── config.json + ├── auditer.db + ├── auditer.db-wal # 运行时可能存在 + └── auditer.db-shm # 运行时可能存在 ``` +`data/` 已在 `.gitignore` 中忽略。`config.json` 包含完整上游 API Key，应当按敏感配置文件保护；`auditer.db` 不保存完整 Prompt 或 API Key。 + 编辑 `.env`。生产环境建议设置两个不同的长随机令牌： ```dotenv @@ -323,12 +354,6 @@ ADMIN_TOKEN=用于保护管理接口的长随机字符串 AUDITER_TOKEN=用于保护sub2api审核调用的另一个长随机字符串 ``` -启动： - -```bash -docker compose up -d --build -``` - 查看日志： ```bash @@ -346,9 +371,12 @@ curl -i http://127.0.0.1:8080/readyz ```bash git pull -docker compose up -d --build +docker compose pull +docker compose up -d ``` +Compose 配置了 `pull_policy: always`，执行 `docker compose up -d` 时也会尝试检查 `latest`，但显式执行 `docker compose pull` 更容易确认是否成功拉到新镜像。 + ## sub2api 配置 在 sub2api 的提示词审计节点中填写： @@ -386,13 +414,14 @@ http://127.0.0.1:8080 | 变量 | 默认值 | 说明 | |---|---:|---| | `AUDITER_PORT` | `8080` | Docker 对外映射端口 | -| `CONFIG_PATH` | `./data/config.json` | 配置文件位置；Docker 中为 `/data/config.json` | +| `CONFIG_PATH` | `/data/config.json` | 网页配置持久化文件 | +| `TRACE_DB_PATH` | `/data/auditer.db` | 最近 100 条完成处理详情的 SQLite 数据库 | | `ADMIN_TOKEN` | 空 | 保护 `/api/config`、`/api/test`、日志和统计接口 | | `AUDITER_TOKEN` | 空 | 保护 `/v1/models` 和 `/v1/chat/completions` | | `LOG_LEVEL` | `info` | Uvicorn 日志级别 | | `MAX_REQUEST_BODY_BYTES` | `2097152` | Auditer 入站请求体上限 | | `MAX_INPUT_CHARS` | `200000` | 单次待审核文本字符上限 | -| `LOG_CAPACITY` | `100` | 内存日志条数，限制为 10–100 | +| `LOG_CAPACITY` | `100` | 日志/SQLite 窗口条数，限制为 10–100 | | `HTTP_MAX_CONNECTIONS` | `200` | 上游 httpx 最大连接数 | | `HTTP_MAX_KEEPALIVE` | `50` | 上游 keep-alive 连接数 | | `FORWARDED_ALLOW_IPS` | `127.0.0.1` | Uvicorn 信任代理头的来源 | @@ -420,7 +449,7 @@ http://127.0.0.1:8080 | `GET` | `/api/status` | `ADMIN_TOKEN` | 运行状态 | | `POST` | `/api/test` | `ADMIN_TOKEN` | 网页测试审核 | | `GET` | `/api/logs` | `ADMIN_TOKEN` | 最近处理日志 | -| `DELETE` | `/api/logs` | `ADMIN_TOKEN` | 清空日志与统计窗口 | +| `DELETE` | `/api/logs` | `ADMIN_TOKEN` | 清空内存与 SQLite 日志 | | `GET` | `/api/statistics` | `ADMIN_TOKEN` | 日志窗口统计 | | `GET` | `/v1/models` | `AUDITER_TOKEN` | sub2api 节点探测 | | `POST` | `/v1/chat/completions` | `AUDITER_TOKEN` | sub2api 审计请求 | @@ -436,7 +465,10 @@ Auditer 的固定处理开销主要包括 JSON 读取、文本提取、请求体 - 全异步 HTTP 请求处理； - 进程级 httpx 连接池复用； - 配置不可变内存快照，热路径不读磁盘； -- 日志固定大小内存环形缓冲区，不同步写数据库或日志文件； +- 处理中 Trace 使用内存结构； +- 响应完整发送给 sub2api 后，再由 Starlette `BackgroundTask` 一次性写 SQLite； +- SQLite 使用 WAL + `synchronous=NORMAL`； +- SQLite 只保留最近 100 条完成记录； - 极短临界区的内存锁； - 请求体与上游响应体增量限长读取； - 静态网页资源内存缓存； @@ -444,7 +476,7 @@ Auditer 的固定处理开销主要包括 JSON 读取、文本提取、请求体 - 不继承宿主机 `HTTP_PROXY` / `HTTPS_PROXY`； - 不对上游失败执行隐式重试，避免重复费用和额外尾延迟。 -实际端到端延迟通常主要由审核模型推理和网络往返决定；应以统计页中三个阶段的实测数据判断，不应仅凭实现语言推断瓶颈。 +因此 SQLite 持久化不位于 Auditer 向 sub2api 返回响应之前的关键路径。实际端到端延迟通常主要由审核模型推理和网络往返决定；应以统计页中三个阶段的实测数据判断，不应仅凭实现语言推断瓶颈。 ## 安全说明 @@ -453,6 +485,7 @@ Auditer 的固定处理开销主要包括 JSON 读取、文本提取、请求体 - 配置文件通过临时文件、`fsync` 和原子替换写入，并尝试设置为 `0600`。 - Docker 容器使用非 root 用户运行。 - 不记录完整请求正文、Prompt、API Key 或完整模型输出。 +- SQLite 处理日志只包含请求元数据、时间点、耗时、判定与错误信息。 - 上游响应大小受限，防止异常响应占用过多内存。 - 生产环境必须设置 `ADMIN_TOKEN` 和 `AUDITER_TOKEN`，并使用 HTTPS 或仅在可信内网开放。 - 允许 iframe 是本项目的明确用途；公网部署时可在 Nginx/Caddy 层覆盖 CSP，只允许你的 sub2api 域名嵌入。 @@ -469,6 +502,12 @@ pytest -q sub2api-auditer --host 127.0.0.1 --port 8080 ``` +如需本地启用 SQLite 日志持久化： + +```bash +export TRACE_DB_PATH=./data/auditer.db +``` + 前端为原生 HTML/CSS/JavaScript，不需要 Node 构建步骤。 ## 许可证与来源 diff --git a/docker-compose.yml b/docker-compose.yml index 0340bec..1810bb5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,7 +1,7 @@ services: sub2api-auditer: - build: . - image: sub2api-auditer:latest + image: ghcr.io/coderdoubleflower/sub2apiauditer:latest + pull_policy: always container_name: sub2api-auditer restart: unless-stopped ports: @@ -9,6 +9,7 @@ services: environment: # 服务运行配置 CONFIG_PATH: /data/config.json + TRACE_DB_PATH: ${TRACE_DB_PATH:-/data/auditer.db} ADMIN_TOKEN: ${ADMIN_TOKEN:-} AUDITER_TOKEN: ${AUDITER_TOKEN:-} LOG_LEVEL: ${LOG_LEVEL:-info} @@ -29,5 +30,5 @@ services: extra_hosts: - "host.docker.internal:host-gateway" volumes: - # 将配置文件持久化到 docker-compose.yml 同级的 ./data 目录。 + # 当前目录持久化：配置写入 config.json，最近 100 条处理详情写入 auditer.db。 - ./data:/data diff --git a/pyproject.toml b/pyproject.toml index b4fcf60..c06ff2d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "sub2api-auditer" -version = "1.1.0" +version = "1.2.0" description = "将任意 OpenAI 兼容审核模型适配为 sub2api Prompt Audit 节点" readme = "README.md" requires-python = ">=3.11" diff --git a/src/sub2api_auditer/__init__.py b/src/sub2api_auditer/__init__.py index 87b1c3b..271ad3e 100644 --- a/src/sub2api_auditer/__init__.py +++ b/src/sub2api_auditer/__init__.py @@ -2,4 +2,4 @@ __all__ = ["__version__"] -__version__ = "1.1.0" +__version__ = "1.2.0" diff --git a/src/sub2api_auditer/observability.py b/src/sub2api_auditer/observability.py index f5bc77f..9c86219 100644 --- a/src/sub2api_auditer/observability.py +++ b/src/sub2api_auditer/observability.py @@ -1,13 +1,21 @@ from __future__ import annotations +import json +import logging +import os +import sqlite3 import time import uuid from collections import Counter, deque from dataclasses import dataclass, field from datetime import datetime, timezone +from pathlib import Path from threading import RLock from typing import Any +LOGGER = logging.getLogger("sub2api_auditer") +_SCHEMA_VERSION = 1 + def _iso_now() -> tuple[str, int, float]: """Return a wall-clock timestamp plus monotonic and epoch clocks. @@ -73,15 +81,35 @@ class ProcessingTrace: categories: tuple[str, ...] = field(default_factory=tuple) error_code: str = "" error_message: str = "" + persisted_preprocess_ms: float | None = None + persisted_upstream_ms: float | None = None + persisted_response_ms: float | None = None + persisted_total_ms: float | None = None def snapshot(self, *, now_perf_ns: int | None = None) -> dict[str, Any]: now_perf_ns = now_perf_ns or time.perf_counter_ns() - preprocess_ms = _duration_ms(self.received_perf_ns, self.forwarded_perf_ns) - upstream_ms = _duration_ms(self.forwarded_perf_ns, self.llm_replied_perf_ns) - response_ms = _duration_ms(self.llm_replied_perf_ns, self.sub2api_replied_perf_ns) - total_ms = _duration_ms(self.received_perf_ns, self.sub2api_replied_perf_ns) + preprocess_ms = ( + _duration_ms(self.received_perf_ns, self.forwarded_perf_ns) + if self.forwarded_perf_ns + else self.persisted_preprocess_ms + ) + upstream_ms = ( + _duration_ms(self.forwarded_perf_ns, self.llm_replied_perf_ns) + if self.forwarded_perf_ns and self.llm_replied_perf_ns + else self.persisted_upstream_ms + ) + response_ms = ( + _duration_ms(self.llm_replied_perf_ns, self.sub2api_replied_perf_ns) + if self.llm_replied_perf_ns and self.sub2api_replied_perf_ns + else self.persisted_response_ms + ) + total_ms = ( + _duration_ms(self.received_perf_ns, self.sub2api_replied_perf_ns) + if self.received_perf_ns and self.sub2api_replied_perf_ns + else self.persisted_total_ms + ) elapsed_ms = total_ms - if elapsed_ms is None: + if elapsed_ms is None and self.received_perf_ns: elapsed_ms = _duration_ms(self.received_perf_ns, now_perf_ns) return { "id": self.id, @@ -113,14 +141,242 @@ def snapshot(self, *, now_perf_ns: int | None = None) -> dict[str, Any]: class TraceStore: - """A lock-protected, in-memory ring buffer for the latest processing traces.""" + """Latest processing traces with optional SQLite persistence. - def __init__(self, capacity: int = 100) -> None: + In-flight state stays in memory. A completed trace is written to SQLite only + after the HTTP response has been sent, so audit-request latency is not + coupled to disk I/O. On startup the latest completed rows are restored. + """ + + def __init__( + self, + capacity: int = 100, + db_path: str | Path | None = None, + ) -> None: self.capacity = max(10, min(int(capacity), 100)) + configured_path = os.getenv("TRACE_DB_PATH", "").strip() + if db_path is None and configured_path: + db_path = configured_path + self.db_path = Path(db_path) if db_path else None self._items: deque[ProcessingTrace] = deque() self._by_id: dict[str, ProcessingTrace] = {} self._lock = RLock() + if self.db_path is not None: + self._initialize_database() + self._restore_latest() + + @property + def persistence(self) -> str: + return "sqlite" if self.db_path is not None else "memory" + + def _connect(self) -> sqlite3.Connection: + assert self.db_path is not None + connection = sqlite3.connect(str(self.db_path), timeout=5.0) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA busy_timeout = 5000") + return connection + + def _initialize_database(self) -> None: + assert self.db_path is not None + self.db_path.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as connection: + connection.execute("PRAGMA journal_mode = WAL") + connection.execute("PRAGMA synchronous = NORMAL") + connection.execute( + """ + CREATE TABLE IF NOT EXISTS processing_traces ( + id TEXT PRIMARY KEY, + source TEXT NOT NULL, + received_at TEXT NOT NULL, + received_epoch REAL NOT NULL, + client_request_id TEXT NOT NULL DEFAULT '', + request_model TEXT NOT NULL DEFAULT '', + upstream_model TEXT NOT NULL DEFAULT '', + input_chars INTEGER NOT NULL DEFAULT 0, + input_bytes INTEGER NOT NULL DEFAULT 0, + forwarded_at TEXT, + llm_replied_at TEXT, + sub2api_replied_at TEXT NOT NULL, + status TEXT NOT NULL, + http_status INTEGER NOT NULL DEFAULT 0, + upstream_http_status INTEGER NOT NULL DEFAULT 0, + upstream_request_id TEXT NOT NULL DEFAULT '', + upstream_response_bytes INTEGER NOT NULL DEFAULT 0, + safety TEXT NOT NULL DEFAULT '', + categories_json TEXT NOT NULL DEFAULT '[]', + error_code TEXT NOT NULL DEFAULT '', + error_message TEXT NOT NULL DEFAULT '', + preprocess_ms REAL, + upstream_ms REAL, + response_ms REAL, + total_ms REAL NOT NULL + ) + """ + ) + connection.execute( + """ + CREATE INDEX IF NOT EXISTS idx_processing_traces_received_epoch + ON processing_traces(received_epoch DESC) + """ + ) + connection.execute(f"PRAGMA user_version = {_SCHEMA_VERSION}") + + def _restore_latest(self) -> None: + if self.db_path is None: + return + with self._connect() as connection: + rows = connection.execute( + """ + SELECT * + FROM processing_traces + ORDER BY received_epoch DESC, sub2api_replied_at DESC + LIMIT ? + """, + (self.capacity,), + ).fetchall() + + restored: list[ProcessingTrace] = [] + for row in reversed(rows): + try: + raw_categories = json.loads(row["categories_json"] or "[]") + except (TypeError, json.JSONDecodeError): + raw_categories = [] + categories = tuple( + str(value)[:80] + for value in raw_categories + if isinstance(value, (str, int, float)) + )[:12] + restored.append( + ProcessingTrace( + id=str(row["id"]), + source=str(row["source"]), + received_at=str(row["received_at"]), + received_perf_ns=0, + received_epoch=float(row["received_epoch"]), + client_request_id=str(row["client_request_id"] or ""), + request_model=str(row["request_model"] or ""), + upstream_model=str(row["upstream_model"] or ""), + input_chars=int(row["input_chars"] or 0), + input_bytes=int(row["input_bytes"] or 0), + forwarded_at=str(row["forwarded_at"] or ""), + llm_replied_at=str(row["llm_replied_at"] or ""), + sub2api_replied_at=str(row["sub2api_replied_at"] or ""), + status=str(row["status"] or "error"), + http_status=int(row["http_status"] or 0), + upstream_http_status=int(row["upstream_http_status"] or 0), + upstream_request_id=str(row["upstream_request_id"] or ""), + upstream_response_bytes=int(row["upstream_response_bytes"] or 0), + safety=str(row["safety"] or ""), + categories=categories, + error_code=str(row["error_code"] or ""), + error_message=str(row["error_message"] or ""), + persisted_preprocess_ms=( + float(row["preprocess_ms"]) + if row["preprocess_ms"] is not None + else None + ), + persisted_upstream_ms=( + float(row["upstream_ms"]) + if row["upstream_ms"] is not None + else None + ), + persisted_response_ms=( + float(row["response_ms"]) + if row["response_ms"] is not None + else None + ), + persisted_total_ms=float(row["total_ms"]), + ) + ) + + with self._lock: + for trace in restored: + self._items.append(trace) + self._by_id[trace.id] = trace + self._trim_memory_locked() + + def _trim_memory_locked(self) -> None: + while len(self._items) > self.capacity: + candidate = next( + (item for item in self._items if item.status != "processing"), + None, + ) + if candidate is None: + return + self._items.remove(candidate) + self._by_id.pop(candidate.id, None) + + def _persist_snapshot(self, snapshot: dict[str, Any], received_epoch: float) -> None: + if self.db_path is None: + return + try: + with self._connect() as connection: + connection.execute( + """ + INSERT OR REPLACE INTO processing_traces ( + id, source, received_at, received_epoch, + client_request_id, request_model, upstream_model, + input_chars, input_bytes, + forwarded_at, llm_replied_at, sub2api_replied_at, + status, http_status, upstream_http_status, + upstream_request_id, upstream_response_bytes, + safety, categories_json, error_code, error_message, + preprocess_ms, upstream_ms, response_ms, total_ms + ) VALUES ( + ?, ?, ?, ?, + ?, ?, ?, + ?, ?, + ?, ?, ?, + ?, ?, ?, + ?, ?, + ?, ?, ?, ?, + ?, ?, ?, ? + ) + """, + ( + snapshot["id"], + snapshot["source"], + snapshot["received_at"], + received_epoch, + snapshot["client_request_id"], + snapshot["request_model"], + snapshot["upstream_model"], + snapshot["input_chars"], + snapshot["input_bytes"], + snapshot["forwarded_at"], + snapshot["llm_replied_at"], + snapshot["sub2api_replied_at"], + snapshot["status"], + snapshot["http_status"] or 0, + snapshot["upstream_http_status"] or 0, + snapshot["upstream_request_id"], + snapshot["upstream_response_bytes"], + snapshot["safety"], + json.dumps(snapshot["categories"], ensure_ascii=False), + snapshot["error_code"], + snapshot["error_message"], + snapshot["preprocess_ms"], + snapshot["upstream_ms"], + snapshot["response_ms"], + snapshot["total_ms"] or 0.0, + ), + ) + connection.execute( + """ + DELETE FROM processing_traces + WHERE id NOT IN ( + SELECT id + FROM processing_traces + ORDER BY received_epoch DESC, sub2api_replied_at DESC + LIMIT ? + ) + """, + (self.capacity,), + ) + except sqlite3.Error: + LOGGER.exception("failed to persist processing trace trace_id=%s", snapshot["id"]) + def begin( self, *, @@ -139,11 +395,9 @@ def begin( request_model=request_model[:256], ) with self._lock: - if len(self._items) >= self.capacity: - removed = self._items.popleft() - self._by_id.pop(removed.id, None) self._items.append(trace) self._by_id[trace.id] = trace + self._trim_memory_locked() return trace.id def update_request( @@ -226,6 +480,8 @@ def mark_error( def mark_replied(self, trace_id: str, *, http_status: int) -> None: rendered, perf_ns, _ = _iso_now() + snapshot: dict[str, Any] | None = None + received_epoch = 0.0 with self._lock: trace = self._by_id.get(trace_id) if trace is None: @@ -236,19 +492,40 @@ def mark_replied(self, trace_id: str, *, http_status: int) -> None: trace.http_status = int(http_status) if trace.status == "processing": trace.status = "success" if 200 <= http_status < 400 else "error" + snapshot = trace.snapshot(now_perf_ns=perf_ns) + received_epoch = trace.received_epoch + self._trim_memory_locked() + + if snapshot["sub2api_replied_at"] and snapshot["total_ms"] is not None: + self._persist_snapshot(snapshot, received_epoch) def list(self, *, limit: int = 100) -> list[dict[str, Any]]: limit = max(1, min(int(limit), self.capacity)) with self._lock: now_ns = time.perf_counter_ns() - return [item.snapshot(now_perf_ns=now_ns) for item in list(self._items)[-limit:][::-1]] + return [ + item.snapshot(now_perf_ns=now_ns) + for item in list(self._items)[-limit:][::-1] + ] def clear(self) -> int: with self._lock: count = len(self._items) self._items.clear() self._by_id.clear() - return count + + database_count = 0 + if self.db_path is not None: + try: + with self._connect() as connection: + row = connection.execute( + "SELECT COUNT(*) AS count FROM processing_traces" + ).fetchone() + database_count = int(row["count"] if row is not None else 0) + connection.execute("DELETE FROM processing_traces") + except sqlite3.Error: + LOGGER.exception("failed to clear processing trace database") + return max(count, database_count) def runtime_stats(self) -> dict[str, Any]: snapshots = self.list(limit=self.capacity) @@ -256,16 +533,23 @@ def runtime_stats(self) -> dict[str, Any]: failed = [item for item in snapshots if item["status"] == "error"] in_flight = [item for item in snapshots if item["status"] == "processing"] last = snapshots[0] if snapshots else None - last_completed = next((item for item in snapshots if item["total_ms"] is not None), None) + last_completed = next( + (item for item in snapshots if item["total_ms"] is not None), + None, + ) return { "total": len(snapshots), "success": len(successful), "failed": len(failed), "in_flight": len(in_flight), "last_latency_ms": last_completed["total_ms"] if last_completed else 0, - "last_error_code": next((item["error_code"] for item in snapshots if item["error_code"]), ""), + "last_error_code": next( + (item["error_code"] for item in snapshots if item["error_code"]), + "", + ), "last_request_at": last["received_at"] if last else "", "capacity": self.capacity, + "persistence": self.persistence, } def statistics(self) -> dict[str, Any]: @@ -275,16 +559,40 @@ def statistics(self) -> dict[str, Any]: failed = [item for item in snapshots if item["status"] == "error"] processing = [item for item in snapshots if item["status"] == "processing"] - total_values = [float(item["total_ms"]) for item in completed if item["total_ms"] is not None] - preprocess_values = [float(item["preprocess_ms"]) for item in completed if item["preprocess_ms"] is not None] - upstream_values = [float(item["upstream_ms"]) for item in completed if item["upstream_ms"] is not None] - response_values = [float(item["response_ms"]) for item in completed if item["response_ms"] is not None] + total_values = [ + float(item["total_ms"]) + for item in completed + if item["total_ms"] is not None + ] + preprocess_values = [ + float(item["preprocess_ms"]) + for item in completed + if item["preprocess_ms"] is not None + ] + upstream_values = [ + float(item["upstream_ms"]) + for item in completed + if item["upstream_ms"] is not None + ] + response_values = [ + float(item["response_ms"]) + for item in completed + if item["response_ms"] is not None + ] - decision_counts = Counter((item["safety"] or "Unclassified") for item in snapshots) - error_counts = Counter(item["error_code"] for item in failed if item["error_code"]) + decision_counts = Counter( + (item["safety"] or "Unclassified") for item in snapshots + ) + error_counts = Counter( + item["error_code"] for item in failed if item["error_code"] + ) now_epoch = time.time() with self._lock: - rpm = sum(1 for item in self._items if item.received_epoch >= now_epoch - 60) + rpm = sum( + 1 + for item in self._items + if item.received_epoch >= now_epoch - 60 + ) chronological = list(reversed(completed[:30])) series = [ @@ -312,6 +620,7 @@ def statistics(self) -> dict[str, Any]: return { "generated_at": _iso_now()[0], "capacity": self.capacity, + "persistence": self.persistence, "window_size": len(snapshots), "completed": len(completed), "success": len(successful), diff --git a/tests/test_observability.py b/tests/test_observability.py index bce985d..9c9742d 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -1,10 +1,36 @@ from __future__ import annotations +import sqlite3 import time from sub2api_auditer.observability import TraceStore +def _complete(store: TraceStore, *, client_request_id: str = "") -> str: + trace_id = store.begin( + source="sub2api", + client_request_id=client_request_id, + request_model="requested", + ) + store.update_request( + trace_id, + request_model="requested", + upstream_model="actual", + input_chars=4, + input_bytes=4, + ) + store.mark_forwarded(trace_id) + store.mark_llm_replied( + trace_id, + upstream_http_status=200, + upstream_request_id="upstream-id", + response_bytes=128, + ) + store.mark_result(trace_id, safety="Unsafe", categories=("Jailbreak",)) + store.mark_replied(trace_id, http_status=200) + return trace_id + + def test_trace_store_keeps_latest_capacity_items(): store = TraceStore(capacity=100) ids = [store.begin(source="sub2api", client_request_id=str(index)) for index in range(105)] @@ -52,3 +78,59 @@ def test_trace_store_records_monotonic_phase_durations_and_statistics(): assert stats["decisions"]["Unsafe"] == 1 assert stats["latency"]["p95_ms"] == trace["total_ms"] assert stats["phases"]["upstream_average_ms"] == trace["upstream_ms"] + assert stats["persistence"] == "memory" + + +def test_sqlite_persistence_restores_completed_trace(tmp_path): + db_path = tmp_path / "auditer.db" + store = TraceStore(capacity=100, db_path=db_path) + trace_id = _complete(store) + + assert db_path.exists() + with sqlite3.connect(db_path) as connection: + assert connection.execute("PRAGMA journal_mode").fetchone()[0] == "wal" + + restored = TraceStore(capacity=100, db_path=db_path) + items = restored.list() + + assert len(items) == 1 + assert items[0]["id"] == trace_id + assert items[0]["status"] == "success" + assert items[0]["safety"] == "Unsafe" + assert items[0]["categories"] == ["Jailbreak"] + assert items[0]["upstream_request_id"] == "upstream-id" + assert items[0]["total_ms"] is not None + assert restored.statistics()["persistence"] == "sqlite" + + +def test_sqlite_persistence_keeps_only_latest_100(tmp_path): + db_path = tmp_path / "auditer.db" + store = TraceStore(capacity=100, db_path=db_path) + + ids = [] + for index in range(105): + ids.append(_complete(store, client_request_id=str(index))) + + restored = TraceStore(capacity=100, db_path=db_path) + items = restored.list(limit=100) + + assert len(items) == 100 + assert items[0]["id"] == ids[-1] + assert items[-1]["id"] == ids[5] + + with sqlite3.connect(db_path) as connection: + count = connection.execute("SELECT COUNT(*) FROM processing_traces").fetchone()[0] + assert count == 100 + + +def test_clear_removes_persisted_traces(tmp_path): + db_path = tmp_path / "auditer.db" + store = TraceStore(capacity=100, db_path=db_path) + _complete(store) + _complete(store) + + assert store.clear() == 2 + assert store.list() == [] + + restored = TraceStore(capacity=100, db_path=db_path) + assert restored.list() == []