Merge branch 'dev_qing' of https://git.cancerai.top/qing/micro_servers into dev_qing
This commit is contained in:
qing committed
commit
ce9337fe0d
4 files changed
+211
-374
No files matched your search
+7
-208
@@ -15,19 +15,9 @@ dbv2 — 基于 pymsg 协议（MsgServer）的数据库服务。
|
||||
"limit": 10
|
||||
}
|
||||
|
||||
{
|
||||
"_route": "log/ingest",
|
||||
"logs": [{"service": "waf", "level": "INFO", "message": "..."}]
|
||||
}
|
||||
|
||||
支持的路由::
|
||||
|
||||
execute → 原 /execute
|
||||
log/ingest → 原 /log/ingest
|
||||
log/query → 原 /log/query
|
||||
log/stats → 原 /log/stats
|
||||
log/purge → 原 /log/purge
|
||||
log/health → 原 /log/health
|
||||
execute → 通用 CRUD（select / insert / update / delete）
|
||||
"""
|
||||
|
||||
import json
|
||||
@@ -53,15 +43,8 @@ DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///database.db")
|
||||
|
||||
# ── 日志模块常量 ────────────────────────────────────────────────────────
|
||||
|
||||
VALID_SERVICES = {
|
||||
"waf", "gateway", "static", "dnfs", "db",
|
||||
"ai_review", "captcha", "core",
|
||||
}
|
||||
VALID_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
|
||||
|
||||
MAX_BATCH = 100
|
||||
MAX_MESSAGE_LEN = 8192
|
||||
MAX_EXTRA_LEN = 4096
|
||||
MAX_MESSAGE_LEN = 8192 # message 上限 8KB
|
||||
MAX_EXTRA_LEN = 4096 # extra 上限 4KB
|
||||
|
||||
DEFAULT_RETENTION_DAYS = 30
|
||||
DEFAULT_MAX_RECORDS = 1_000_000
|
||||
@@ -82,23 +65,19 @@ class DBServer:
|
||||
self.metadata = MetaData()
|
||||
self._server = MsgServer(addr, port)
|
||||
|
||||
# 日志模块内部状态
|
||||
# 日志表缓存（避免重复反射）
|
||||
self._logs_table_cache = {}
|
||||
|
||||
# 路由表
|
||||
self._routes = {
|
||||
"execute": self._handle_execute,
|
||||
"log/ingest": self._handle_log_ingest,
|
||||
"log/query": self._handle_log_query,
|
||||
"log/stats": self._handle_log_stats,
|
||||
"log/purge": self._handle_log_purge,
|
||||
"log/health": self._handle_log_health,
|
||||
"execute": self._handle_execute,
|
||||
}
|
||||
|
||||
# ── 启动 / 停止 ──────────────────────────────────────────────────
|
||||
def start(self) -> None:
|
||||
"""启动服务（阻塞当前线程）。"""
|
||||
print(f"[dbv2] 监听 tcp://{self.addr}:{self.port}")
|
||||
self._ensure_logs_table() # 确保 Logs 表已建好（外部 MsgClient 写日志前）
|
||||
self._start_cleanup_thread()
|
||||
self._server.start(self._dispatch)
|
||||
|
||||
@@ -251,7 +230,7 @@ class DBServer:
|
||||
return [{"rowcount": res.rowcount}]
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════
|
||||
# 路由处理 — log/*
|
||||
# 日志表管理 + 后台清理
|
||||
# ═══════════════════════════════════════════════════════════════════
|
||||
|
||||
# ── 日志表管理 ────────────────────────────────────────────────────
|
||||
@@ -313,186 +292,6 @@ class DBServer:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ── log/ingest ────────────────────────────────────────────────────
|
||||
def _handle_log_ingest(self, data, **kw):
|
||||
logs = data.get("logs")
|
||||
if not isinstance(logs, list) or not logs:
|
||||
raise ValueError("logs 必须是非空数组")
|
||||
if len(logs) > MAX_BATCH:
|
||||
raise ValueError(f"单次最多 {MAX_BATCH} 条，当前 {len(logs)} 条")
|
||||
|
||||
table = self._ensure_logs_table()
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
rows = []
|
||||
|
||||
for i, log in enumerate(logs):
|
||||
if not isinstance(log, dict):
|
||||
raise ValueError(f"第 {i} 条日志不是有效对象")
|
||||
|
||||
service = log.get("service")
|
||||
level = log.get("level")
|
||||
message = log.get("message")
|
||||
|
||||
if not service or service not in VALID_SERVICES:
|
||||
raise ValueError(f"第 {i} 条: service 无效或缺失")
|
||||
if not level or level not in VALID_LEVELS:
|
||||
raise ValueError(f"第 {i} 条: level 无效或缺失")
|
||||
if not message or not isinstance(message, str):
|
||||
raise ValueError(f"第 {i} 条: message 缺失或非字符串")
|
||||
|
||||
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,
|
||||
})
|
||||
|
||||
with self.engine.connect() as conn:
|
||||
conn.execute(insert(table), rows)
|
||||
conn.commit()
|
||||
|
||||
return {"inserted": len(rows)}
|
||||
|
||||
# ── log/query ─────────────────────────────────────────────────────
|
||||
def _handle_log_query(self, data, **kw):
|
||||
table = self._ensure_logs_table()
|
||||
clauses = []
|
||||
|
||||
svc = data.get("service")
|
||||
if svc:
|
||||
col = table.columns["service"]
|
||||
clauses.append(col.in_(svc) if isinstance(svc, list) else col == svc)
|
||||
|
||||
lvl = data.get("level")
|
||||
if lvl:
|
||||
col = table.columns["level"]
|
||||
clauses.append(col.in_(lvl) if isinstance(lvl, list) else col == lvl)
|
||||
|
||||
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]
|
||||
order_expr = order_col.desc() if data.get("desc", True) else order_col.asc()
|
||||
|
||||
limit = min(data.get("limit", 100), 1000)
|
||||
offset = data.get("offset", 0)
|
||||
|
||||
with self.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 {"total": total, "logs": logs}
|
||||
|
||||
# ── log/stats ─────────────────────────────────────────────────────
|
||||
def _handle_log_stats(self, data, **kw):
|
||||
table = self._ensure_logs_table()
|
||||
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"):
|
||||
raise ValueError(f"group_by 仅支持 service / level，收到: {g}")
|
||||
|
||||
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
|
||||
|
||||
with self.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 {"groups": groups, "total": total}
|
||||
|
||||
# ── log/purge ─────────────────────────────────────────────────────
|
||||
def _handle_log_purge(self, data, **kw):
|
||||
before = data.get("before")
|
||||
if not before:
|
||||
raise ValueError("缺少 before 字段")
|
||||
|
||||
table = self._ensure_logs_table()
|
||||
with self.engine.connect() as conn:
|
||||
stmt = delete(table).where(table.columns["ts"] < before)
|
||||
result = conn.execute(stmt)
|
||||
conn.commit()
|
||||
deleted = result.rowcount
|
||||
self._log_internal("db", "INFO",
|
||||
f"purge 清理 {deleted} 条 (before={before})")
|
||||
return {"deleted": deleted}
|
||||
|
||||
# ── log/health ────────────────────────────────────────────────────
|
||||
def _handle_log_health(self, data, **kw):
|
||||
table = self._ensure_logs_table()
|
||||
with self.engine.connect() as conn:
|
||||
count = conn.execute(
|
||||
select(func.count()).select_from(table)
|
||||
).scalar() or 0
|
||||
return {"status": "ok", "count": count}
|
||||
|
||||
# ── 后台自动清理线程 ──────────────────────────────────────────────
|
||||
def _start_cleanup_thread(self):
|
||||
def _loop():
|
||||
|
||||
-166
@@ -1,166 +0,0 @@
|
||||
import os
|
||||
from flask import Flask, request, jsonify, g
|
||||
from sqlalchemy import (
|
||||
create_engine, MetaData, Table, select, insert, update, delete, text
|
||||
)
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
from waitress import serve
|
||||
from security_middleware import apply_security_headers
|
||||
from log_module import init_log_routes, log_internal
|
||||
|
||||
app = Flask(__name__)
|
||||
apply_security_headers(app)
|
||||
|
||||
# 数据库配置
|
||||
DATABASE_URL = os.environ.get('DATABASE_URL', 'sqlite:///database.db')
|
||||
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:
|
||||
Table(table_name, metadata, autoload_with=engine)
|
||||
return metadata.tables[table_name]
|
||||
|
||||
def build_where(table, conditions):
|
||||
"""将条件字典转换为 SQLAlchemy 条件表达式"""
|
||||
if not conditions:
|
||||
return None
|
||||
clauses = []
|
||||
for col, val in conditions.items():
|
||||
column = table.columns[col]
|
||||
if isinstance(val, dict):
|
||||
# 支持多种比较操作
|
||||
if 'eq' in val:
|
||||
clauses.append(column == val['eq'])
|
||||
if 'ne' in val:
|
||||
clauses.append(column != val['ne'])
|
||||
if 'gt' in val:
|
||||
clauses.append(column > val['gt'])
|
||||
if 'lt' in val:
|
||||
clauses.append(column < val['lt'])
|
||||
if 'ge' in val:
|
||||
clauses.append(column >= val['ge'])
|
||||
if 'le' in val:
|
||||
clauses.append(column <= val['le'])
|
||||
if 'like' in val:
|
||||
clauses.append(column.like(val['like']))
|
||||
if 'in' in val:
|
||||
clauses.append(column.in_(val['in']))
|
||||
else:
|
||||
# 直接等值比较
|
||||
clauses.append(column == val)
|
||||
from sqlalchemy import and_
|
||||
return and_(*clauses)
|
||||
|
||||
@app.route('/execute', methods=['POST'])
|
||||
def execute_structured():
|
||||
data = request.get_json()
|
||||
if not data:
|
||||
return jsonify({'success': False, 'error': '请求体不能为空'}), 400
|
||||
|
||||
action = data.get('action', 'select')
|
||||
table_name = data.get('table')
|
||||
if not table_name:
|
||||
return jsonify({'success': False, 'error': '缺少 table 字段'}), 400
|
||||
|
||||
try:
|
||||
table = get_table(table_name)
|
||||
except Exception as e:
|
||||
return jsonify({'success': False, 'error': f'表加载失败: {str(e)}'}), 400
|
||||
|
||||
conn = None
|
||||
try:
|
||||
conn = engine.connect()
|
||||
result = []
|
||||
if action == 'select':
|
||||
columns = data.get('columns', None)
|
||||
if columns:
|
||||
cols = [table.columns[c] for c in columns]
|
||||
else:
|
||||
cols = [table] # select *
|
||||
|
||||
stmt = select(*cols)
|
||||
where_cond = build_where(table, data.get('where'))
|
||||
if where_cond is not None:
|
||||
stmt = stmt.where(where_cond)
|
||||
|
||||
# 排序
|
||||
order_by = data.get('order_by')
|
||||
if order_by:
|
||||
for item in order_by:
|
||||
col = table.columns[item['column']]
|
||||
stmt = stmt.order_by(col.desc() if item.get('desc') else col.asc())
|
||||
|
||||
# 分页
|
||||
limit = data.get('limit')
|
||||
if limit is not None:
|
||||
stmt = stmt.limit(limit)
|
||||
offset = data.get('offset')
|
||||
if offset is not None:
|
||||
stmt = stmt.offset(offset)
|
||||
|
||||
res = conn.execute(stmt)
|
||||
rows = res.fetchall()
|
||||
result = [dict(row._mapping) for row in rows]
|
||||
|
||||
elif action == 'insert':
|
||||
values = data.get('values', {})
|
||||
stmt = insert(table).values(**values)
|
||||
res = conn.execute(stmt)
|
||||
conn.commit()
|
||||
result = [{'lastrowid': res.lastrowid}] if hasattr(res, 'lastrowid') else []
|
||||
|
||||
elif action == 'update':
|
||||
values = data.get('values', {})
|
||||
where_cond = build_where(table, data.get('where'))
|
||||
stmt = update(table).values(**values)
|
||||
if where_cond is not None:
|
||||
stmt = stmt.where(where_cond)
|
||||
res = conn.execute(stmt)
|
||||
conn.commit()
|
||||
result = [{'rowcount': res.rowcount}]
|
||||
|
||||
elif action == 'delete':
|
||||
where_cond = build_where(table, data.get('where'))
|
||||
stmt = delete(table)
|
||||
if where_cond is not None:
|
||||
stmt = stmt.where(where_cond)
|
||||
res = conn.execute(stmt)
|
||||
conn.commit()
|
||||
result = [{'rowcount': res.rowcount}]
|
||||
|
||||
else:
|
||||
return jsonify({'success': False, 'error': f'不支持的操作: {action}'}), 400
|
||||
|
||||
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)
|
||||
@@ -10,6 +10,10 @@ from pathlib import Path
|
||||
import configparser
|
||||
import sys
|
||||
|
||||
# pymsg 日志客户端（集中式日志 → dbv2 Logs 表）
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
from log_client import logs # noqa: E402
|
||||
|
||||
# 基准目录：编译后取可执行文件所在目录，开发时取脚本所在目录
|
||||
# 解决 Nuitka onefile 解压到 /tmp 后相对路径失效的问题
|
||||
if getattr(sys, 'frozen', False):
|
||||
@@ -64,8 +68,11 @@ if FILESYSTEM_ENABLED:
|
||||
try:
|
||||
Path(HTML_ROOT_DIR).mkdir(parents=True, exist_ok=True)
|
||||
logger.info(f"Created directory: {HTML_ROOT_DIR}")
|
||||
logs.info("static", f"目录已创建: {HTML_ROOT_DIR}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create directory: {str(e)}")
|
||||
logs.error("static", f"目录创建失败: {str(e)}",
|
||||
extra={"path": HTML_ROOT_DIR})
|
||||
FILESYSTEM_ENABLED = False
|
||||
|
||||
SSL_ENABLED = file_config['ssl'].get('enabled', 'false').lower() == 'true'
|
||||
@@ -168,6 +175,8 @@ def get_file_list():
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"Get file list error: {str(e)}")
|
||||
logs.error("static", f"获取文件列表失败: {str(e)}",
|
||||
extra={"path": requested_path})
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
@app.route('/video/<path:file_path>', methods=['GET'])
|
||||
@@ -212,6 +221,8 @@ def stream_video(file_path):
|
||||
return send_file(full_path, mimetype=mime_type, as_attachment=False)
|
||||
except Exception as e:
|
||||
logger.error(f"Video streaming error: {str(e)}")
|
||||
logs.error("static", f"视频流传输失败: {str(e)}",
|
||||
extra={"file": file_path})
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
GATEWAY_DB = '/opt/micro_servers/gateway/sqlite3.db'
|
||||
@@ -239,6 +250,7 @@ def api_reset_password():
|
||||
conn.execute("UPDATE Users SET password_hash=? WHERE email=?", (new_hash, email))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
logs.info("static", "密码重置", extra={"email": email[:3] + "***"})
|
||||
return jsonify({"success": True})
|
||||
|
||||
|
||||
@@ -302,6 +314,7 @@ def yhdet_callback():
|
||||
fake_hash = hashlib.sha256(password.encode()).hexdigest()
|
||||
conn.execute("INSERT INTO Users (email, password_hash) VALUES (?, ?)", (email, fake_hash))
|
||||
conn.commit()
|
||||
logs.info("static", "OAuth 新用户注册", extra={"email": email})
|
||||
conn.close()
|
||||
try:
|
||||
login_data = _json.dumps({"username": email, "password": password}).encode()
|
||||
@@ -335,9 +348,15 @@ def yhdet_login():
|
||||
rows = conn.execute("SELECT email, password_hash FROM Users WHERE email=?", (email,)).fetchall()
|
||||
conn.close()
|
||||
if not rows:
|
||||
logs.warning("static", "yhdet 登录失败-用户不存在",
|
||||
extra={"email": email[:8] + "***"})
|
||||
return jsonify({"success": False, "error": "Invalid credentials"})
|
||||
if rows[0][1] != pwd_hash:
|
||||
logs.warning("static", "yhdet 登录失败-密码错误",
|
||||
extra={"email": email[:8] + "***"})
|
||||
return jsonify({"success": False, "error": "Invalid credentials"})
|
||||
logs.info("static", "yhdet 登录成功",
|
||||
extra={"email": email[:8] + "***"})
|
||||
token_data = f"{email}:{password}"
|
||||
token = base64.urlsafe_b64encode(token_data.encode()).decode()
|
||||
return jsonify({"success": True, "token": token})
|
||||
@@ -396,6 +415,8 @@ def proxy_get_notes():
|
||||
with _urlreq.urlopen(req, timeout=15) as r:
|
||||
return _cors(make_response(r.read().decode(), 200, {"Content-Type": "application/json"}))
|
||||
except Exception as e:
|
||||
logs.warning("static", "dnfs get_notes 代理失败",
|
||||
extra={"error": str(e)})
|
||||
return _cors(make_response(_json.dumps({"error": str(e)}), 502, {"Content-Type": "application/json"}))
|
||||
|
||||
@app.route('/add_note', methods=['POST', 'OPTIONS'])
|
||||
@@ -409,6 +430,7 @@ def proxy_add_note():
|
||||
with _urlreq.urlopen(req, timeout=15) as r:
|
||||
return _cors(make_response(r.read().decode(), 200, {"Content-Type": "application/json"}))
|
||||
except Exception as e:
|
||||
logs.warning("static", "dnfs add_note 代理失败", extra={"error": str(e)})
|
||||
return _cors(make_response(_json.dumps({"error": str(e)}), 502, {"Content-Type": "application/json"}))
|
||||
|
||||
@app.route('/delete_note', methods=['POST', 'OPTIONS'])
|
||||
@@ -422,6 +444,7 @@ def proxy_delete_note():
|
||||
with _urlreq.urlopen(req, timeout=15) as r:
|
||||
return _cors(make_response(r.read().decode(), 200, {"Content-Type": "application/json"}))
|
||||
except Exception as e:
|
||||
logs.warning("static", "dnfs delete_note 代理失败", extra={"error": str(e)})
|
||||
return _cors(make_response(_json.dumps({"error": str(e)}), 502, {"Content-Type": "application/json"}))
|
||||
|
||||
@app.route('/update_note', methods=['POST', 'OPTIONS'])
|
||||
@@ -435,6 +458,8 @@ def proxy_update_note():
|
||||
with _urlreq.urlopen(req, timeout=15) as r:
|
||||
return _cors(make_response(r.read().decode(), 200, {"Content-Type": "application/json"}))
|
||||
except Exception as e:
|
||||
logs.warning("static", "dnfs update_note 代理失败",
|
||||
extra={"error": str(e)})
|
||||
return _cors(make_response(_json.dumps({"error": str(e)}), 502, {"Content-Type": "application/json"}))
|
||||
|
||||
|
||||
@@ -520,6 +545,8 @@ def serve_file(path):
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error: {str(e)}")
|
||||
logs.error("static", f"静态文件服务错误: {str(e)}",
|
||||
extra={"path": path})
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return f"服务器内部错误: {str(e)}", 500
|
||||
|
||||
@@ -0,0 +1,177 @@
|
||||
"""
|
||||
统一日志客户端 — 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表。
|
||||
|
||||
设计要点：
|
||||
- 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
|
||||
- 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
|
||||
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务
|
||||
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
|
||||
|
||||
用法：
|
||||
from log_client import logs
|
||||
|
||||
logs.info("static", "服务启动成功")
|
||||
logs.error("static", "文件系统未启用", extra={"path": "/data"})
|
||||
"""
|
||||
|
||||
import json
|
||||
import os as _os
|
||||
import sys as _sys
|
||||
import threading
|
||||
import queue
|
||||
import time
|
||||
import atexit
|
||||
from datetime import datetime, timezone
|
||||
|
||||
_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), ".."))
|
||||
from pymsg import MsgClient # noqa: E402
|
||||
|
||||
# ── 配置 ──────────────────────────────────────────────────────────────
|
||||
|
||||
_DBV2_ADDR = '127.0.0.1'
|
||||
_DBV2_PORT = 8005
|
||||
|
||||
_BATCH_SIZE = 50 # 队列满 50 条时 flush
|
||||
_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush
|
||||
_TIMEOUT = 5 # socket 超时（秒）
|
||||
|
||||
|
||||
class LogClient:
|
||||
"""异步批量日志推送客户端。
|
||||
|
||||
通过 MsgClient（pymsg TCP 协议）逐条 INSERT 到 dbv2 的 Logs 表。
|
||||
"""
|
||||
|
||||
def __init__(self, service_name="static",
|
||||
batch_size=_BATCH_SIZE,
|
||||
flush_interval=_FLUSH_INTERVAL):
|
||||
self._service_name = service_name
|
||||
self._batch_size = batch_size
|
||||
self._flush_interval = flush_interval
|
||||
self._queue = queue.Queue()
|
||||
self._stop = threading.Event()
|
||||
self._flush_signal = 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）。"""
|
||||
self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch
|
||||
time.sleep(0.6) # 等待后台线程响应
|
||||
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)
|
||||
self._flush_signal.clear()
|
||||
|
||||
# ── 内部实现 ────────────────────────────────────────────────────────
|
||||
|
||||
def _flush_loop(self):
|
||||
"""后台线程：积攒到 batch_size 或超时 flush_interval 时发送。"""
|
||||
batch = []
|
||||
last_flush = time.monotonic()
|
||||
|
||||
while not self._stop.is_set():
|
||||
timeout = self._flush_interval - (time.monotonic() - last_flush)
|
||||
chk_timeout = max(0.0, min(timeout, 0.5))
|
||||
try:
|
||||
item = self._queue.get(timeout=chk_timeout)
|
||||
batch.append(item)
|
||||
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
|
||||
self._flush_signal.is_set() or
|
||||
(batch and time.monotonic() - last_flush >= self._flush_interval)
|
||||
)
|
||||
if should_flush and batch:
|
||||
self._send(batch)
|
||||
batch = []
|
||||
last_flush = time.monotonic()
|
||||
|
||||
# 退出前 flush 剩余
|
||||
if batch:
|
||||
self._send(batch)
|
||||
|
||||
def _send(self, batch):
|
||||
"""通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。"""
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
try:
|
||||
client = MsgClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
|
||||
for entry in batch:
|
||||
extra_val = entry.get("extra")
|
||||
payload = {
|
||||
"_route": "execute",
|
||||
"action": "insert",
|
||||
"table": "Logs",
|
||||
"values": {
|
||||
"ts": now,
|
||||
"service": entry.get("service"),
|
||||
"level": entry.get("level"),
|
||||
"message": entry.get("message"),
|
||||
"trace_id": entry.get("trace_id"),
|
||||
"extra": json.dumps(extra_val, ensure_ascii=False)
|
||||
if extra_val is not None else None,
|
||||
},
|
||||
}
|
||||
resp = client.call(json.dumps(payload), timeout=_TIMEOUT)
|
||||
resp_obj = json.loads(resp)
|
||||
if not resp_obj.get("success"):
|
||||
import sys
|
||||
print(f"[log_client] INSERT 失败: {resp_obj.get('error')}",
|
||||
file=sys.stderr)
|
||||
except Exception as e:
|
||||
import sys
|
||||
print(f"[log_client] 发送失败: {e}", file=sys.stderr)
|
||||
|
||||
def _shutdown(self):
|
||||
"""进程退出时 flush 剩余日志。"""
|
||||
self._stop.set()
|
||||
self.flush()
|
||||
|
||||
|
||||
# ── 全局单例 ──────────────────────────────────────────────────────────
|
||||
|
||||
logs = LogClient(service_name="static")
|
||||
Reference in new issue
Block a user