From 18ac5724426dc36d7bc4c74fd0868d13da307d29 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Mon, 27 Jul 2026 10:22:16 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0dbv2=E3=80=81=E5=88=A0?= =?UTF-8?q?=E9=99=A4=E6=97=A7=E7=9A=84db1=E3=80=81=E4=B8=BAstatic=E6=B7=BB?= =?UTF-8?q?=E5=8A=A0log?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db/dbv2.py | 215 ++----------------------------------------- db/main.py | 166 --------------------------------- static/fsys.py | 27 ++++++ static/log_client.py | 177 +++++++++++++++++++++++++++++++++++ 4 files changed, 211 insertions(+), 374 deletions(-) delete mode 100644 db/main.py create mode 100644 static/log_client.py diff --git a/db/dbv2.py b/db/dbv2.py index bd5afb8..00cd52b 100644 --- a/db/dbv2.py +++ b/db/dbv2.py @@ -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(): diff --git a/db/main.py b/db/main.py deleted file mode 100644 index 9efeb55..0000000 --- a/db/main.py +++ /dev/null @@ -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) diff --git a/static/fsys.py b/static/fsys.py index 218516d..17b9ff9 100644 --- a/static/fsys.py +++ b/static/fsys.py @@ -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/', 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 diff --git a/static/log_client.py b/static/log_client.py new file mode 100644 index 0000000..422097d --- /dev/null +++ b/static/log_client.py @@ -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")