From 479d263b883a278e102d2e1f564b5a6065b85e90 Mon Sep 17 00:00:00 2001 From: qing <1771677662@qq.com> Date: Sat, 25 Jul 2026 20:32:56 +0800 Subject: [PATCH] log service in db --- db/.gitignore | 224 ++++++++--------- db/README.md | 453 ++++++++++++++++++++++----------- db/_run.sh | 10 +- db/_set.sh | 12 +- db/create_sqlite3.sql | 61 +++-- db/log_client.py | 152 ++++++++++++ db/log_module.py | 511 ++++++++++++++++++++++++++++++++++++++ db/main.py | 23 +- db/security_middleware.py | 84 +++---- 9 files changed, 1194 insertions(+), 336 deletions(-) create mode 100644 db/log_client.py create mode 100644 db/log_module.py diff --git a/db/.gitignore b/db/.gitignore index 817ce2d..c682457 100644 --- a/db/.gitignore +++ b/db/.gitignore @@ -1,112 +1,112 @@ -#prjs -*.db -*.bin -*.build -*.dist -*.onefile-build -nuitka-crash-report.xml - -# Byte-compiled / optimized / DLL files -__pycache__/ -*.py[cod] -*$py.class - -# C extensions -*.so - -# Distribution / packaging -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -*.egg-info/ -.installed.cfg -*.egg -MANIFEST - -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - -# Installer logs -pip-log.txt -pip-delete-this-directory.txt - -# Unit test / coverage reports -htmlcov/ -.tox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*.cover -.hypothesis/ -.pytest_cache/ - -# Translations -*.mo -*.pot - -# Django stuff: -*.log -local_settings.py -db.sqlite3 - -# Flask stuff: -instance/ -.webassets-cache - -# Scrapy stuff: -.scrapy - -# Sphinx documentation -docs/_build/ - -# PyBuilder -target/ - -# Jupyter Notebook -.ipynb_checkpoints - -# pyenv -.python-version - -# celery beat schedule file -celerybeat-schedule - -# SageMath parsed files -*.sage.py - -# Environments -.env -.venv -env/ -venv/ -ENV/ -env.bak/ -venv.bak/ - -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site - -# mypy -.mypy_cache/ +#prjs +*.db +*.bin +*.build +*.dist +*.onefile-build +nuitka-crash-report.xml + +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +.hypothesis/ +.pytest_cache/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ diff --git a/db/README.md b/db/README.md index 1efe2c4..6e1a73c 100644 --- a/db/README.md +++ b/db/README.md @@ -1,147 +1,306 @@ -# 跨数据库通用查询服务（基于 SQLAlchemy Core） - -一个安全的 HTTP 服务，通过结构化 JSON 描述查询意图，后端自动转换为对应数据库的 SQL，无需手写 SQL，即可无缝切换 SQLite、PostgreSQL、MySQL 等数据库。 - -## 功能特点 - -- **跨数据库**：基于 SQLAlchemy Core 表达式语言，一套接口适配多种数据库 -- **零 SQL 注入**：所有条件、值均通过 SQLAlchemy 参数绑定处理 -- **统一 JSON 接口**：`action` + `table` + `conditions` 模式 -- **支持增删改查**：select, insert, update, delete 全覆盖 -- **单线程本地监听**：waitress + 127.0.0.1:8005，简单可靠 - -## 快速开始 - -### 安装依赖 -```bash -pip install flask sqlalchemy waitress -# 根据需要安装数据库驱动，例如： -# pip install psycopg2-binary # PostgreSQL -# pip install pymysql # MySQL -``` - -### 启动服务（默认 SQLite） -```bash -python app.py -``` -服务运行在 `http://127.0.0.1:8005`。 - -使用其他数据库： -```bash -export DATABASE_URL="postgresql://user:pass@localhost/dbname" -python app.py -``` - -**注意**：新接口依赖表已存在（通过反射获取结构）。首次使用时请先手动建表： -```bash -sqlite3 database.db "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)" -``` - -## API 文档 - -### 请求格式 - -**端点**：`POST /execute` -**Content-Type**：`application/json` - -```json -{ - "action": "操作类型", // select | insert | update | delete - "table": "表名", - "columns": ["列1", "列2"], // select/insert 时可选 - "values": { // insert/update 时必填 - "列名": "值" - }, - "where": { // 条件，支持组合操作符 - "age": {"ge": 18, "lt": 60}, - "name": "Alice" - }, - "order_by": [{"column": "age", "desc": true}], - "limit": 10, - "offset": 0 -} -``` - -### where 条件支持的运算符 - -| 运算符 | 说明 | 示例 | -|--------|------|------| -| 直接值 | 等于 | `"name": "Alice"` | -| eq | 等于 | `"age": {"eq": 30}` | -| ne | 不等于 | `"status": {"ne": "deleted"}` | -| gt | 大于 | `"score": {"gt": 80}` | -| ge | 大于等于 | `"score": {"ge": 60}` | -| lt | 小于 | `"price": {"lt": 100}` | -| le | 小于等于 | `"price": {"le": 200}` | -| like | 模糊匹配 | `"name": {"like": "%Smith%"}` | -| in | 包含于列表 | `"id": {"in": [1,2,3]}` | - -多个条件自动使用 **AND** 连接。 - -### 响应格式 - -成功： -```json -{ - "success": true, - "data": [ // 查询结果数组（insert/update/delete 时包含影响行数或 lastrowid） - {"id": 1, "name": "Alice"} - ] -} -``` - -失败： -```json -{ - "success": false, - "error": "错误描述" -} -``` - -## 使用示例（curl） - -**插入**： -```bash -curl -X POST http://127.0.0.1:8005/execute \ - -H "Content-Type: application/json" \ - -d '{"action":"insert","table":"users","values":{"name":"Bob","age":25}}' -``` - -**查询**： -```bash -curl -X POST http://127.0.0.1:8005/execute \ - -H "Content-Type: application/json" \ - -d '{"action":"select","table":"users","where":{"age":{"gt":20}},"order_by":[{"column":"id","desc":true}],"limit":5}' -``` - -**更新**： -```bash -curl -X POST http://127.0.0.1:8005/execute \ - -H "Content-Type: application/json" \ - -d '{"action":"update","table":"users","values":{"age":26},"where":{"name":"Bob"}}' -``` - -**删除**： -```bash -curl -X POST http://127.0.0.1:8005/execute \ - -H "Content-Type: application/json" \ - -d '{"action":"delete","table":"users","where":{"id":{"in":[1,2]}}}' -``` - -## 安全机制 - -- 强制使用 SQLAlchemy 参数绑定，杜绝 SQL 注入。 -- 仅反射已有表，无法创建或删除表（保证结构安全）。 -- 所有异常自动回滚事务。 -- 默认监听 `127.0.0.1`，仅本机可访问。 - -## 切换数据库 - -只需修改环境变量 `DATABASE_URL`，并安装对应驱动： - -- SQLite：`sqlite:///database.db` -- PostgreSQL：`postgresql://user:pass@localhost/dbname` （需 `pip install psycopg2`） -- MySQL：`mysql+pymysql://user:pass@localhost/dbname` （需 `pip install pymysql`） - -SQLAlchemy 会自动生成适配各数据库的 SQL 方言，无需修改接口调用。 - +# 跨数据库通用查询服务（基于 SQLAlchemy Core） + +一个安全的 HTTP 服务，通过结构化 JSON 描述查询意图，后端自动转换为对应数据库的 SQL，无需手写 SQL，即可无缝切换 SQLite、PostgreSQL、MySQL 等数据库。 + +## 功能特点 + +- **跨数据库**：基于 SQLAlchemy Core 表达式语言，一套接口适配多种数据库 +- **零 SQL 注入**：所有条件、值均通过 SQLAlchemy 参数绑定处理 +- **统一 JSON 接口**：`action` + `table` + `conditions` 模式 +- **支持增删改查**：select, insert, update, delete 全覆盖 +- **单线程本地监听**：waitress + 127.0.0.1:8005，简单可靠 + +## 快速开始 + +### 安装依赖 +```bash +pip install flask sqlalchemy waitress +# 根据需要安装数据库驱动，例如： +# pip install psycopg2-binary # PostgreSQL +# pip install pymysql # MySQL +``` + +### 启动服务（默认 SQLite） +```bash +python app.py +``` +服务运行在 `http://127.0.0.1:8005`。 + +使用其他数据库： +```bash +export DATABASE_URL="postgresql://user:pass@localhost/dbname" +python app.py +``` + +**注意**：新接口依赖表已存在（通过反射获取结构）。首次使用时请先手动建表： +```bash +sqlite3 database.db "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)" +``` + +## API 文档 + +### 请求格式 + +**端点**：`POST /execute` +**Content-Type**：`application/json` + +```json +{ + "action": "操作类型", // select | insert | update | delete + "table": "表名", + "columns": ["列1", "列2"], // select/insert 时可选 + "values": { // insert/update 时必填 + "列名": "值" + }, + "where": { // 条件，支持组合操作符 + "age": {"ge": 18, "lt": 60}, + "name": "Alice" + }, + "order_by": [{"column": "age", "desc": true}], + "limit": 10, + "offset": 0 +} +``` + +### where 条件支持的运算符 + +| 运算符 | 说明 | 示例 | +|--------|------|------| +| 直接值 | 等于 | `"name": "Alice"` | +| eq | 等于 | `"age": {"eq": 30}` | +| ne | 不等于 | `"status": {"ne": "deleted"}` | +| gt | 大于 | `"score": {"gt": 80}` | +| ge | 大于等于 | `"score": {"ge": 60}` | +| lt | 小于 | `"price": {"lt": 100}` | +| le | 小于等于 | `"price": {"le": 200}` | +| like | 模糊匹配 | `"name": {"like": "%Smith%"}` | +| in | 包含于列表 | `"id": {"in": [1,2,3]}` | + +多个条件自动使用 **AND** 连接。 + +### 响应格式 + +成功： +```json +{ + "success": true, + "data": [ // 查询结果数组（insert/update/delete 时包含影响行数或 lastrowid） + {"id": 1, "name": "Alice"} + ] +} +``` + +失败： +```json +{ + "success": false, + "error": "错误描述" +} +``` + +## 使用示例（curl） + +**插入**： +```bash +curl -X POST http://127.0.0.1:8005/execute \ + -H "Content-Type: application/json" \ + -d '{"action":"insert","table":"users","values":{"name":"Bob","age":25}}' +``` + +**查询**： +```bash +curl -X POST http://127.0.0.1:8005/execute \ + -H "Content-Type: application/json" \ + -d '{"action":"select","table":"users","where":{"age":{"gt":20}},"order_by":[{"column":"id","desc":true}],"limit":5}' +``` + +**更新**： +```bash +curl -X POST http://127.0.0.1:8005/execute \ + -H "Content-Type: application/json" \ + -d '{"action":"update","table":"users","values":{"age":26},"where":{"name":"Bob"}}' +``` + +**删除**： +```bash +curl -X POST http://127.0.0.1:8005/execute \ + -H "Content-Type: application/json" \ + -d '{"action":"delete","table":"users","where":{"id":{"in":[1,2]}}}' +``` + +## 安全机制 + +- 强制使用 SQLAlchemy 参数绑定，杜绝 SQL 注入。 +- 仅反射已有表，无法创建或删除表（保证结构安全）。 +- 所有异常自动回滚事务。 +- 默认监听 `127.0.0.1`，仅本机可访问。 + +## 内置日志模块 + +db 服务内置了统一日志收集能力，各业务服务通过 HTTP 推送日志到 `/log/*` 系列接口，db 自身的操作日志通过 `log_internal()` 直接写入（同进程，无 HTTP 往返，无循环依赖）。 + +### 为什么集成在 db 而不是独立服务 + +独立 logs 服务会导致**循环调用**：db 记日志→调 logs，logs 存库→调 db，无限递归。集成在 db 内部后，db 自己记日志直接写表，彻底消除循环依赖。 + +### 日志表 + +```sql +CREATE TABLE IF NOT EXISTS Logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + service TEXT NOT NULL, + level TEXT NOT NULL, + message TEXT NOT NULL, + trace_id TEXT, + extra TEXT +); +``` + +- `service`：来源服务名（`waf`/`gateway`/`static`/`dnfs`/`db`/`ai_review`/`captcha`/`core`） +- `level`：`DEBUG`/`INFO`/`WARNING`/`ERROR`/`CRITICAL` +- `trace_id`：可选，请求追踪 ID +- `extra`：可选，JSON 字符串，存放附加结构化字段 + +### 接口 + +| 接口 | 方法 | 说明 | +|------|------|------| +| `/log/ingest` | POST | 推送日志（支持批量） | +| `/log/query` | POST | 查询日志（按服务/级别/时间/关键字过滤） | +| `/log/stats` | POST | 统计聚合（按 service/level 分组计数） | +| `/log/purge` | POST | 清理过期日志 | +| `/log/health` | GET | 健康检查 | + +所有接口的响应格式与 `/execute` 统一：`{success, data, error}`。 + +#### POST /log/ingest + +```json +请求: +{ + "logs": [ + { + "service": "dnfs", + "level": "INFO", + "message": "节点新增成功 id=42", + "ts": "2026-07-25T08:30:00.123Z", + "trace_id": "req-abc-123", + "extra": {"path": "/add_note", "status": 200} + } + ] +} + +响应: +{"success": true, "data": {"inserted": 1}} +``` + +- `ts` 可选，缺省自动填当前 UTC 时间 +- 单次最多 100 条 +- `message` 上限 8KB，`extra` 上限 4KB + +#### POST /log/query + +```json +请求: +{ + "service": "dnfs", // 可选，支持数组匹配多个 + "level": "WARNING", // 可选，支持数组匹配多个 + "since": "2026-07-25T00:00:00Z", // 可选 + "until": "2026-07-25T23:59:59Z", // 可选 + "keyword": "节点", // 可选，message 模糊搜索 + "trace_id": "req-abc-123", // 可选 + "order_by": "ts", // 默认 ts，可选 id/service/level + "desc": true, // 默认 true（最新在前） + "limit": 50, // 默认 100，上限 1000 + "offset": 0 // 分页偏移 +} + +响应: +{ + "success": true, + "data": { + "total": 128, + "logs": [{"id":1024,"ts":"...","service":"dnfs","level":"INFO","message":"...","trace_id":"...","extra":{...}}] + } +} +``` + +#### POST /log/stats + +```json +请求: +{"since":"...","until":"...","group_by":"service"} + +响应: +{ + "success": true, + "data": { + "groups": [{"service":"dnfs","level":"INFO","count":320}], + "total": 325 + } +} +``` + +`group_by` 支持 `service` / `level` / `service,level`。 + +#### POST /log/purge + +```json +请求: {"before": "2026-07-18T00:00:00Z"} +响应: {"success": true, "data": {"deleted": 1523}} +``` + +#### GET /log/health + +```json +{"success": true, "data": {"status": "ok", "count": 10240}} +``` + +### 自动清理 + +- 服务启动时开启后台守护线程，每 1 小时清理一次 +- 默认保留 30 天日志，超期自动删除 +- 日志总条数超过 100 万时，按时间倒序保留最新 N 条 + +### 各服务接入方式 + +把 `db/log_client.py` 复制到各服务目录（或加入 `PYTHONPATH`），即可使用： + +```python +from log_client import logs + +# 替换原来的 print +logs.info("dnfs", "节点新增成功 id=42") +logs.error("dnfs", "数据库连接失败", extra={"errno": 2002}) + +# 带追踪 ID +logs.push("gateway", "WARNING", "登录失败次数超限", + trace_id="req-abc", extra={"ip": "1.2.3.4"}) +``` + +特性： +- 使用 `urllib.request`（标准库，不引入 requests） +- 异步非阻塞：内存队列 + 后台线程批量推送 +- fail-safe：db 服务不可用时静默丢弃，绝不影响业务 +- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush + +### 防递归设计 + +| 场景 | 调用路径 | 是否安全 | +|------|---------|---------| +| 外部服务记日志 | `服务 → POST /log/ingest → 写 Logs 表` | 安全 | +| db 自己记日志 | `/execute → log_internal() → 直接写 Logs 表` | 安全（无 HTTP） | +| /log/ingest 成功 | `写入成功 → 直接返回` | 安全（不再记日志） | +| /log/ingest 失败 | `写入失败 → log_internal() → 直接写表` | 安全（log_internal 不回调 ingest） | + +## 切换数据库 + +只需修改环境变量 `DATABASE_URL`，并安装对应驱动： + +- SQLite：`sqlite:///database.db` +- PostgreSQL：`postgresql://user:pass@localhost/dbname` （需 `pip install psycopg2`） +- MySQL：`mysql+pymysql://user:pass@localhost/dbname` （需 `pip install pymysql`） + +SQLAlchemy 会自动生成适配各数据库的 SQL 方言，无需修改接口调用。 + diff --git a/db/_run.sh b/db/_run.sh index f3f83e3..3bd152d 100644 --- a/db/_run.sh +++ b/db/_run.sh @@ -1,5 +1,5 @@ -#!/bin/bash -cd "$(dirname "$0")" || exit -source ./venv/bin/activate -#exec ./main.dist/main.bin -exec python3 main.py +#!/bin/bash +cd "$(dirname "$0")" || exit +source ./venv/bin/activate +#exec ./main.dist/main.bin +exec python3 main.py diff --git a/db/_set.sh b/db/_set.sh index 68c83b2..66947c7 100644 --- a/db/_set.sh +++ b/db/_set.sh @@ -1,6 +1,6 @@ -#!/bin/bash -cd "$(dirname "$0")" || exit -python3 -m venv venv -source venv/bin/activate -pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple -#nuitka --standalone main.py +#!/bin/bash +cd "$(dirname "$0")" || exit +python3 -m venv venv +source venv/bin/activate +pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple +#nuitka --standalone main.py diff --git a/db/create_sqlite3.sql b/db/create_sqlite3.sql index 28ba662..e43a1d3 100644 --- a/db/create_sqlite3.sql +++ b/db/create_sqlite3.sql @@ -1,22 +1,39 @@ --- Users 表创建脚本 --- 数据库类型: SQLite - -CREATE TABLE IF NOT EXISTS Users ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - username TEXT UNIQUE NOT NULL, - password_hash TEXT NOT NULL, - create_at DATETIME DEFAULT CURRENT_TIMESTAMP, - last_login DATETIME, - is_active BOOLEAN DEFAULT 1 -); - -CREATE TABLE IF NOT EXISTS Notes ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - pid INTEGER NOT NULL, - owner VARCHAR(128) DEFAULT 'Unknow', - title VARCHAR(128) NOT NULL, - content TEXT NOT NULL -); - --- 旧版增量 --- ALTER TABLE Notes ADD COLUMN owner VARCHAR(128) DEFAULT 'Unknow'; +-- Users 表创建脚本 +-- 数据库类型: SQLite + +CREATE TABLE IF NOT EXISTS Users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT UNIQUE NOT NULL, + password_hash TEXT NOT NULL, + create_at DATETIME DEFAULT CURRENT_TIMESTAMP, + last_login DATETIME, + is_active BOOLEAN DEFAULT 1 +); + +CREATE TABLE IF NOT EXISTS Notes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + pid INTEGER NOT NULL, + owner VARCHAR(128) DEFAULT 'Unknow', + title VARCHAR(128) NOT NULL, + content TEXT NOT NULL +); + +-- 旧版增量 +-- ALTER TABLE Notes ADD COLUMN owner VARCHAR(128) DEFAULT 'Unknow'; + +-- ── 日志表（db 内置日志模块）───────────────────────────────────────── + +CREATE TABLE IF NOT EXISTS Logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + service TEXT NOT NULL, + level TEXT NOT NULL, + message TEXT NOT NULL, + trace_id TEXT, + extra TEXT +); + +CREATE INDEX IF NOT EXISTS idx_logs_service_ts ON Logs(service, ts); +CREATE INDEX IF NOT EXISTS idx_logs_level_ts ON Logs(level, ts); +CREATE INDEX IF NOT EXISTS idx_logs_ts ON Logs(ts); +CREATE INDEX IF NOT EXISTS idx_logs_trace ON Logs(trace_id); diff --git a/db/log_client.py b/db/log_client.py new file mode 100644 index 0000000..d84d252 --- /dev/null +++ b/db/log_client.py @@ -0,0 +1,152 @@ +""" +统一日志客户端 — 各服务共用。 + +设计要点： + - 使用 urllib.request（标准库，不引入 requests） + - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 + - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 + - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush + +用法： + from log_client import logs + + logs.info("dnfs", "节点新增成功 id=42") + logs.error("dnfs", "数据库连接失败", extra={"errno": 2002}) + logs.push("gateway", "WARNING", "登录失败次数超限", + trace_id="req-abc", extra={"ip": "1.2.3.4"}) +""" + +import json +import threading +import queue +import time +import urllib.request +import urllib.error +import atexit + +# ── 配置 ────────────────────────────────────────────────────────────── + +DB_LOG_URL = 'http://127.0.0.1:8005/log/ingest' + +_BATCH_SIZE = 50 # 队列满 50 条时 flush +_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush +_TIMEOUT = 5 # HTTP 请求超时（秒） + + +class LogClient: + """异步批量日志推送客户端。""" + + def __init__(self, url=DB_LOG_URL, batch_size=_BATCH_SIZE, + flush_interval=_FLUSH_INTERVAL): + self._url = url + self._batch_size = batch_size + self._flush_interval = flush_interval + self._queue = queue.Queue() + self._stop = threading.Event() + self._thread = threading.Thread(target=self._flush_loop, + daemon=True, name='log-client') + self._thread.start() + atexit.register(self._shutdown) + + # ── 公开接口 ──────────────────────────────────────────────────────── + + def push(self, service, level, message, trace_id=None, extra=None): + """推送一条日志（非阻塞，立即返回）。""" + self._queue.put({ + 'service': service, + 'level': level, + 'message': message, + 'trace_id': trace_id, + 'extra': extra, + }) + + def debug(self, service, message, **kw): + self.push(service, 'DEBUG', message, **kw) + + def info(self, service, message, **kw): + self.push(service, 'INFO', message, **kw) + + def warning(self, service, message, **kw): + self.push(service, 'WARNING', message, **kw) + + def error(self, service, message, **kw): + self.push(service, 'ERROR', message, **kw) + + def critical(self, service, message, **kw): + self.push(service, 'CRITICAL', message, **kw) + + def flush(self): + """手动 flush 队列中剩余的日志。""" + batch = [] + while True: + try: + batch.append(self._queue.get_nowait()) + if len(batch) >= self._batch_size: + self._send(batch) + batch = [] + except queue.Empty: + break + if batch: + self._send(batch) + + # ── 内部实现 ──────────────────────────────────────────────────────── + + def _flush_loop(self): + """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。""" + batch = [] + last_flush = time.monotonic() + + while not self._stop.is_set(): + timeout = max(0, self._flush_interval - (time.monotonic() - last_flush)) + try: + item = self._queue.get(timeout=timeout) + batch.append(item) + # 继续非阻塞地取，直到满 batch_size + while len(batch) < self._batch_size: + try: + batch.append(self._queue.get_nowait()) + except queue.Empty: + break + except queue.Empty: + pass # 超时，没有新日志 + + # 满了或超时了就发 + should_flush = ( + len(batch) >= self._batch_size or + (batch and time.monotonic() - last_flush >= self._flush_interval) + ) + if should_flush: + self._send(batch) + batch = [] + last_flush = time.monotonic() + + # 退出前 flush 剩余 + if batch: + self._send(batch) + + def _send(self, batch): + """发送一批日志到 db 服务。失败时静默丢弃。""" + data = json.dumps({'logs': batch}).encode('utf-8') + req = urllib.request.Request( + self._url, data=data, + headers={'Content-Type': 'application/json'}, + method='POST' + ) + try: + with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp: + resp.read() # 消费响应体 + except (urllib.error.URLError, urllib.error.HTTPError, OSError): + # fail-safe：日志服务不可用，静默丢弃，不影响业务 + pass + except Exception: + pass + + def _shutdown(self): + """进程退出时 flush 剩余日志。""" + self._stop.set() + self.flush() + + +# ── 全局单例 ────────────────────────────────────────────────────────── + +logs = LogClient() diff --git a/db/log_module.py b/db/log_module.py new file mode 100644 index 0000000..5ad7d33 --- /dev/null +++ b/db/log_module.py @@ -0,0 +1,511 @@ +""" +db 服务的内置日志子模块。 + +提供两部分能力： + 1. log_internal() — db 进程内部直接写 Logs 表（无 HTTP 往返，无循环依赖） + 2. /log/* HTTP 路由 — 供外部服务推送 / 查询 / 统计 / 清理日志 + +设计要点（防递归）： + - log_internal() 直接操作数据库，不经过任何 HTTP 路由 + - /log/ingest 成功后不再调 log_internal + - /log/ingest 失败时才调 log_internal，而 log_internal 不会回调 /log/ingest +""" + +import json +import threading +import time +from datetime import datetime, timezone, timedelta + +from flask import Blueprint, request, jsonify +from sqlalchemy import ( + Table, insert, select, delete, func, and_, or_, MetaData +) + +bp = Blueprint('log', __name__, url_prefix='/log') + +# ── 常量 ────────────────────────────────────────────────────────────── + +VALID_SERVICES = { + 'waf', 'gateway', 'static', 'dnfs', 'db', + 'ai_review', 'captcha', 'core', +} +VALID_LEVELS = {'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'} + +MAX_BATCH = 100 # 单次 /log/ingest 最多 100 条 +MAX_MESSAGE_LEN = 8192 # message 上限 8KB +MAX_EXTRA_LEN = 4096 # extra 上限 4KB + +DEFAULT_RETENTION_DAYS = 30 +DEFAULT_MAX_RECORDS = 1_000_000 +DEFAULT_CLEANUP_INTERVAL = 3600 # 秒 + +# ── 表管理 ───────────────────────────────────────────────────────────── + +_LOGS_TABLE_CACHE = {} + +def _ensure_logs_table(metadata: MetaData, engine): + """确保 Logs 表已建好并已反射到 metadata（首次调用时建表 + 反射）。""" + if 'Logs' in _LOGS_TABLE_CACHE: + return _LOGS_TABLE_CACHE['Logs'] + + # 先用原生 SQL 建表（SQLAlchemy 反射要求表已存在） + with engine.connect() as conn: + conn.exec_driver_sql(""" + CREATE TABLE IF NOT EXISTS Logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + service TEXT NOT NULL, + level TEXT NOT NULL, + message TEXT NOT NULL, + trace_id TEXT, + extra TEXT + ) + """) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS idx_logs_service_ts ON Logs(service, ts)" + ) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS idx_logs_level_ts ON Logs(level, ts)" + ) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS idx_logs_ts ON Logs(ts)" + ) + conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS idx_logs_trace ON Logs(trace_id)" + ) + conn.commit() + + # 反射到 metadata + table = Table('Logs', metadata, autoload_with=engine) + _LOGS_TABLE_CACHE['Logs'] = table + return table + + +# ── 内部日志记录（db 自己用，无 HTTP）───────────────────────────────── + +def log_internal(engine, metadata, service, level, message, + trace_id=None, extra=None): + """ + db 进程内部记录日志 — 直接操作数据库，不走 HTTP，无循环依赖。 + + 供 db 的 /execute 路由内部调用。失败时静默吞掉，不影响主业务。 + """ + try: + table = _ensure_logs_table(metadata, engine) + extra_json = json.dumps(extra, ensure_ascii=False) if extra else None + if len(message) > MAX_MESSAGE_LEN: + message = message[:MAX_MESSAGE_LEN] + if extra_json and len(extra_json) > MAX_EXTRA_LEN: + extra_json = extra_json[:MAX_EXTRA_LEN] + + with engine.connect() as conn: + conn.execute(insert(table).values( + ts=datetime.now(timezone.utc).isoformat(), + service=service, + level=level, + message=message, + trace_id=trace_id, + extra=extra_json, + )) + conn.commit() + except Exception: + # 日志记录失败不应影响主业务 + pass + + +# ── HTTP 路由 ────────────────────────────────────────────────────────── + +def init_log_routes(app, engine, metadata, + retention_days=DEFAULT_RETENTION_DAYS, + max_records=DEFAULT_MAX_RECORDS, + cleanup_interval=DEFAULT_CLEANUP_INTERVAL): + """把日志蓝图注册到 Flask app，并注入 engine/metadata。""" + print('[logs] 正在初始化日志模块 ...') + + app.config['LOG_ENGINE'] = engine + app.config['LOG_METADATA'] = metadata + app.config['LOG_RETENTION_DAYS'] = retention_days + app.config['LOG_MAX_RECORDS'] = max_records + app.config['LOG_CLEANUP_INTERVAL'] = cleanup_interval + + # 建表 + 反射 + try: + table = _ensure_logs_table(metadata, engine) + with engine.connect() as conn: + count = conn.execute( + select(func.count()).select_from(table) + ).scalar() or 0 + print(f'[logs] Logs 表就绪，当前 {count} 条日志') + except Exception as e: + print(f'[logs] 警告: Logs 表初始化失败: {e}') + + # 注册路由 + app.register_blueprint(bp) + routes = [ + 'POST /log/ingest', + 'POST /log/query', + 'POST /log/stats', + 'POST /log/purge', + 'GET /log/health', + ] + for r in routes: + print(f'[logs] 路由注册: {r}') + + # 配置摘要 + print(f'[logs] 保留策略: {retention_days} 天 / 上限 {max_records} 条') + print(f'[logs] 自动清理: 每 {cleanup_interval} 秒执行一次') + + # 启动后台清理线程 + _start_cleanup_thread(engine, metadata, + retention_days, max_records, cleanup_interval) + print('[logs] 后台清理线程已启动') + + print('[logs] 日志模块初始化完成') + + +# ── 5.1 推送日志 ────────────────────────────────────────────────────── + +@bp.route('/ingest', methods=['POST']) +def ingest(): + """接收外部日志推送（单条或批量）。""" + data = request.get_json(silent=True) + if not data or 'logs' not in data: + return jsonify({'success': False, 'error': '请求体缺少 logs 字段'}), 400 + + logs = data['logs'] + if not isinstance(logs, list) or not logs: + return jsonify({'success': False, 'error': 'logs 必须是非空数组'}), 400 + if len(logs) > MAX_BATCH: + return jsonify({ + 'success': False, + 'error': f'单次最多 {MAX_BATCH} 条，当前 {len(logs)} 条' + }), 400 + + engine = app.config['LOG_ENGINE'] + metadata = app.config['LOG_METADATA'] + + try: + table = _ensure_logs_table(metadata, engine) + except Exception as e: + return jsonify({'success': False, 'error': f'表初始化失败: {e}'}), 500 + + # 校验 + 组装 + now = datetime.now(timezone.utc).isoformat() + rows = [] + for i, log in enumerate(logs): + if not isinstance(log, dict): + return jsonify({'success': False, + 'error': f'第 {i} 条日志不是有效对象'}), 400 + + service = log.get('service') + level = log.get('level') + message = log.get('message') + + if not service or service not in VALID_SERVICES: + return jsonify({'success': False, + 'error': f'第 {i} 条: service 无效或缺失'}), 400 + if not level or level not in VALID_LEVELS: + return jsonify({'success': False, + 'error': f'第 {i} 条: level 无效或缺失'}), 400 + if not message or not isinstance(message, str): + return jsonify({'success': False, + 'error': f'第 {i} 条: message 缺失或非字符串'}), 400 + + if len(message) > MAX_MESSAGE_LEN: + message = message[:MAX_MESSAGE_LEN] + + trace_id = log.get('trace_id') + extra = log.get('extra') + extra_json = None + if extra is not None: + extra_json = json.dumps(extra, ensure_ascii=False) + if len(extra_json) > MAX_EXTRA_LEN: + extra_json = extra_json[:MAX_EXTRA_LEN] + + rows.append({ + 'ts': log.get('ts') or now, + 'service': service, + 'level': level, + 'message': message, + 'trace_id': trace_id, + 'extra': extra_json, + }) + + # 写入（成功后不再调 log_internal，防递归） + try: + with engine.connect() as conn: + conn.execute(insert(table), rows) + conn.commit() + except Exception as e: + # 失败时可以安全地记日志（log_internal 不回调 ingest） + log_internal(engine, metadata, 'db', 'ERROR', + f'/log/ingest 写入失败: {e}') + return jsonify({'success': False, 'error': str(e)}), 500 + + return jsonify({'success': True, 'data': {'inserted': len(rows)}}) + + +# ── 5.2 查询日志 ─────────────────────────────────────────────────────── + +@bp.route('/query', methods=['POST']) +def query(): + """结构化查询日志。""" + data = request.get_json(silent=True) or {} + + engine = app.config['LOG_ENGINE'] + metadata = app.config['LOG_METADATA'] + + try: + table = _ensure_logs_table(metadata, engine) + except Exception as e: + return jsonify({'success': False, 'error': f'表初始化失败: {e}'}), 500 + + clauses = [] + + # service（支持单个字符串或数组） + svc = data.get('service') + if svc: + col = table.columns['service'] + if isinstance(svc, list): + clauses.append(col.in_(svc)) + else: + clauses.append(col == svc) + + # level（支持单个字符串或数组） + lvl = data.get('level') + if lvl: + col = table.columns['level'] + if isinstance(lvl, list): + clauses.append(col.in_(lvl)) + else: + clauses.append(col == lvl) + + # trace_id + tid = data.get('trace_id') + if tid: + clauses.append(table.columns['trace_id'] == tid) + + # 时间区间 + since = data.get('since') + until = data.get('until') + ts_col = table.columns['ts'] + if since: + clauses.append(ts_col >= since) + if until: + clauses.append(ts_col <= until) + + # 关键字模糊搜索 + keyword = data.get('keyword') + if keyword: + clauses.append(table.columns['message'].like(f'%{keyword}%')) + + where_cond = and_(*clauses) if clauses else None + + # 排序 + order_col_name = data.get('order_by', 'ts') + if order_col_name not in ('id', 'ts', 'service', 'level'): + order_col_name = 'ts' + order_col = table.columns[order_col_name] + desc = data.get('desc', True) + order_expr = order_col.desc() if desc else order_col.asc() + + # 分页 + limit = min(data.get('limit', 100), 1000) + offset = data.get('offset', 0) + + try: + with engine.connect() as conn: + # 总数 + count_stmt = select(func.count()).select_from(table) + if where_cond is not None: + count_stmt = count_stmt.where(where_cond) + total = conn.execute(count_stmt).scalar() or 0 + + # 查询 + stmt = select(table) + if where_cond is not None: + stmt = stmt.where(where_cond) + stmt = stmt.order_by(order_expr).limit(limit).offset(offset) + + rows = conn.execute(stmt).fetchall() + logs = [] + for row in rows: + d = dict(row._mapping) + if d.get('extra'): + try: + d['extra'] = json.loads(d['extra']) + except (json.JSONDecodeError, TypeError): + pass + logs.append(d) + + return jsonify({'success': True, 'data': {'total': total, 'logs': logs}}) + + except Exception as e: + log_internal(engine, metadata, 'db', 'ERROR', + f'/log/query 失败: {e}') + return jsonify({'success': False, 'error': str(e)}), 500 + + +# ── 5.3 统计聚合 ─────────────────────────────────────────────────────── + +@bp.route('/stats', methods=['POST']) +def stats(): + """日志分类统计。""" + data = request.get_json(silent=True) or {} + + engine = app.config['LOG_ENGINE'] + metadata = app.config['LOG_METADATA'] + + try: + table = _ensure_logs_table(metadata, engine) + except Exception as e: + return jsonify({'success': False, 'error': f'表初始化失败: {e}'}), 500 + + group_by = data.get('group_by', 'service') + group_fields = [g.strip() for g in group_by.split(',') if g.strip()] + for g in group_fields: + if g not in ('service', 'level'): + return jsonify({ + 'success': False, + 'error': f'group_by 仅支持 service / level，收到: {g}' + }), 400 + + # 时间过滤 + clauses = [] + since = data.get('since') + until = data.get('until') + ts_col = table.columns['ts'] + if since: + clauses.append(ts_col >= since) + if until: + clauses.append(ts_col <= until) + where_cond = and_(*clauses) if clauses else None + + try: + with engine.connect() as conn: + cols = [table.columns[g] for g in group_fields] + [func.count().label('count')] + stmt = select(*cols) + if where_cond is not None: + stmt = stmt.where(where_cond) + stmt = stmt.group_by(*[table.columns[g] for g in group_fields]) + + rows = conn.execute(stmt).fetchall() + groups = [dict(row._mapping) for row in rows] + + # 总数 + count_stmt = select(func.count()).select_from(table) + if where_cond is not None: + count_stmt = count_stmt.where(where_cond) + total = conn.execute(count_stmt).scalar() or 0 + + return jsonify({'success': True, 'data': {'groups': groups, 'total': total}}) + + except Exception as e: + log_internal(engine, metadata, 'db', 'ERROR', + f'/log/stats 失败: {e}') + return jsonify({'success': False, 'error': str(e)}), 500 + + +# ── 5.4 清理过期 ────────────────────────────────────────────────────── + +@bp.route('/purge', methods=['POST']) +def purge(): + """清理过期日志。""" + data = request.get_json(silent=True) or {} + before = data.get('before') + if not before: + return jsonify({'success': False, 'error': '缺少 before 字段'}), 400 + + engine = app.config['LOG_ENGINE'] + metadata = app.config['LOG_METADATA'] + + try: + table = _ensure_logs_table(metadata, engine) + except Exception as e: + return jsonify({'success': False, 'error': f'表初始化失败: {e}'}), 500 + + try: + with engine.connect() as conn: + stmt = delete(table).where(table.columns['ts'] < before) + result = conn.execute(stmt) + conn.commit() + deleted = result.rowcount + log_internal(engine, metadata, 'db', 'INFO', + f'/log/purge 清理 {deleted} 条 (before={before})') + return jsonify({'success': True, 'data': {'deleted': deleted}}) + + except Exception as e: + log_internal(engine, metadata, 'db', 'ERROR', + f'/log/purge 失败: {e}') + return jsonify({'success': False, 'error': str(e)}), 500 + + +# ── 5.5 健康检查 ────────────────────────────────────────────────────── + +@bp.route('/health', methods=['GET']) +def health(): + """日志模块健康检查。""" + engine = app.config['LOG_ENGINE'] + metadata = app.config['LOG_METADATA'] + + try: + table = _ensure_logs_table(metadata, engine) + with engine.connect() as conn: + count = conn.execute(select(func.count()).select_from(table)).scalar() or 0 + return jsonify({'success': True, 'data': {'status': 'ok', 'count': count}}) + except Exception as e: + return jsonify({'success': False, 'error': str(e)}), 500 + + +# ── 后台自动清理线程 ────────────────────────────────────────────────── + +def _start_cleanup_thread(engine, metadata, + retention_days=DEFAULT_RETENTION_DAYS, + max_records=DEFAULT_MAX_RECORDS, + interval=DEFAULT_CLEANUP_INTERVAL): + """启动后台守护线程，定期清理过期日志。""" + def _cleanup_loop(): + while True: + time.sleep(interval) + try: + _do_cleanup(engine, metadata, retention_days, max_records) + except Exception: + # 清理失败不影响服务运行 + pass + + t = threading.Thread(target=_cleanup_loop, daemon=True, name='logs-cleanup') + t.start() + + +def _do_cleanup(engine, metadata, retention_days, max_records): + """执行一次清理：删过期 + 裁剪超量。""" + table = _ensure_logs_table(metadata, engine) + + # 1. 删除过期日志 + cutoff = (datetime.now(timezone.utc) - timedelta(days=retention_days)).isoformat() + with engine.connect() as conn: + result = conn.execute(delete(table).where(table.columns['ts'] < cutoff)) + conn.commit() + expired_deleted = result.rowcount + + # 2. 裁剪超量记录（保留最新的 max_records 条） + count = conn.execute(select(func.count()).select_from(table)).scalar() or 0 + trimmed = 0 + if count > max_records: + excess = count - max_records + # 找到第 max_records 条的 ts（按时间正序的第 max_records 条） + cutoff_row = conn.execute( + select(table.columns['ts']) + .order_by(table.columns['ts'].asc()) + .limit(1).offset(max_records) + ).fetchone() + if cutoff_row: + trim_result = conn.execute( + delete(table).where(table.columns['ts'] < cutoff_row[0]) + ) + conn.commit() + trimmed = trim_result.rowcount + + print(f'[logs] 自动清理完成: 过期删除 {expired_deleted} 条, 超量裁剪 {trimmed} 条') + log_internal(engine, metadata, 'db', 'INFO', + f'自动清理: 过期删除 {expired_deleted} 条, 超量裁剪 {trimmed} 条') diff --git a/db/main.py b/db/main.py index c553bc7..a156a0e 100644 --- a/db/main.py +++ b/db/main.py @@ -4,10 +4,11 @@ from sqlalchemy import ( create_engine, MetaData, Table, select, insert, update, delete, text ) from sqlalchemy.exc import SQLAlchemyError -from waitress import serve +from waitress import serve from security_middleware import apply_security_headers +from log_module import init_log_routes, log_internal -app = Flask(__name__) +app = Flask(__name__) apply_security_headers(app) # 数据库配置 @@ -17,6 +18,9 @@ engine = create_engine(DATABASE_URL) # 缓存反射的元数据（避免每次请求都反射） metadata = MetaData() +# 注册日志模块（/log/* 路由 + 后台清理线程） +init_log_routes(app, engine, metadata) + def get_table(table_name): """反射表对象（若已缓存则直接返回）""" if table_name not in metadata.tables: @@ -134,20 +138,35 @@ def execute_structured(): else: return jsonify({'success': False, 'error': f'不支持的操作: {action}'}), 400 + # 记录操作日志（同进程直接写库，无 HTTP 往返，无循环依赖） + log_internal(engine, metadata, 'db', 'INFO', + f'{action} {table_name} OK', + extra={'rowcount': len(result) if isinstance(result, list) else 0, + 'client_ip': request.remote_addr}) + return jsonify({'success': True, 'data': result}) except SQLAlchemyError as e: if conn: conn.rollback() + log_internal(engine, metadata, 'db', 'ERROR', + f'{action} {table_name} FAILED: {e}', + extra={'client_ip': request.remote_addr}) return jsonify({'success': False, 'error': str(e)}), 400 except Exception as e: if conn: conn.rollback() + log_internal(engine, metadata, 'db', 'ERROR', + f'{action} {table_name} EXCEPTION: {e}', + extra={'client_ip': request.remote_addr}) return jsonify({'success': False, 'error': str(e)}), 400 finally: if conn: conn.close() if __name__ == '__main__': + # 日志模块的初始化输出在 init_log_routes 调用时已完成 + log_internal(engine, metadata, 'db', 'INFO', + '服务启动，监听 http://127.0.0.1:8005') # 单线程单进程，绑定 127.0.0.1 serve(app, host='127.0.0.1', port=8005, threads=1) diff --git a/db/security_middleware.py b/db/security_middleware.py index ea9e040..a180e14 100644 --- a/db/security_middleware.py +++ b/db/security_middleware.py @@ -1,42 +1,42 @@ -""" -Security headers middleware for Flask applications. -Usage: - from security_middleware import apply_security_headers - apply_security_headers(app) -""" -import secrets - -def apply_security_headers(app, csp_policy=None): - """Apply security response headers to a Flask app.""" - - @app.after_request - def set_security_headers(response): - # Prevent clickjacking - response.headers['X-Frame-Options'] = 'DENY' - # XSS protection (legacy browsers) - response.headers['X-XSS-Protection'] = '1; mode=block' - # Prevent MIME type sniffing - response.headers['X-Content-Type-Options'] = 'nosniff' - # Referrer policy - response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' - # Permissions policy (disable camera, mic, geolocation) - response.headers['Permissions-Policy'] = 'camera=(), microphone=(), geolocation=()' - - # Content Security Policy - if csp_policy: - response.headers['Content-Security-Policy'] = csp_policy - else: - response.headers['Content-Security-Policy'] = ( - "default-src 'self'; " - "script-src 'self' 'unsafe-inline'; " - "style-src 'self' 'unsafe-inline'; " - "img-src 'self' data:; " - "font-src 'self'; " - "connect-src 'self'; " - "frame-ancestors 'none'" - ) - - # Remove server header info - response.headers.pop('Server', None) - - return response +""" +Security headers middleware for Flask applications. +Usage: + from security_middleware import apply_security_headers + apply_security_headers(app) +""" +import secrets + +def apply_security_headers(app, csp_policy=None): + """Apply security response headers to a Flask app.""" + + @app.after_request + def set_security_headers(response): + # Prevent clickjacking + response.headers['X-Frame-Options'] = 'DENY' + # XSS protection (legacy browsers) + response.headers['X-XSS-Protection'] = '1; mode=block' + # Prevent MIME type sniffing + response.headers['X-Content-Type-Options'] = 'nosniff' + # Referrer policy + response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' + # Permissions policy (disable camera, mic, geolocation) + response.headers['Permissions-Policy'] = 'camera=(), microphone=(), geolocation=()' + + # Content Security Policy + if csp_policy: + response.headers['Content-Security-Policy'] = csp_policy + else: + response.headers['Content-Security-Policy'] = ( + "default-src 'self'; " + "script-src 'self' 'unsafe-inline'; " + "style-src 'self' 'unsafe-inline'; " + "img-src 'self' data:; " + "font-src 'self'; " + "connect-src 'self'; " + "frame-ancestors 'none'" + ) + + # Remove server header info + response.headers.pop('Server', None) + + return response -- 2.54.0