From 013e130a082c8f4d2f70a1320ed6041994def521 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sat, 25 Jul 2026 21:01:46 +0800 Subject: [PATCH] logs in db develop --- db/log_module.py | 28 ++------- db/main.py | 6 -- dnfs/dnfs.py | 21 +++++++ dnfs/log_client.py | 152 +++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 179 insertions(+), 28 deletions(-) create mode 100644 dnfs/log_client.py diff --git a/db/log_module.py b/db/log_module.py index 5ad7d33..4041067 100644 --- a/db/log_module.py +++ b/db/log_module.py @@ -120,7 +120,6 @@ def init_log_routes(app, engine, metadata, 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 @@ -129,38 +128,23 @@ def init_log_routes(app, engine, metadata, app.config['LOG_CLEANUP_INTERVAL'] = cleanup_interval # 建表 + 反射 + count = 0 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}') + log_internal(engine, metadata, 'db', 'ERROR', + f'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 推送日志 ────────────────────────────────────────────────────── @@ -506,6 +490,6 @@ def _do_cleanup(engine, metadata, retention_days, max_records): conn.commit() trimmed = trim_result.rowcount - print(f'[logs] 自动清理完成: 过期删除 {expired_deleted} 条, 超量裁剪 {trimmed} 条') - log_internal(engine, metadata, 'db', 'INFO', - f'自动清理: 过期删除 {expired_deleted} 条, 超量裁剪 {trimmed} 条') + if expired_deleted or trimmed: + log_internal(engine, metadata, 'db', 'INFO', + f'自动清理: 过期删除 {expired_deleted} 条, 超量裁剪 {trimmed} 条') diff --git a/db/main.py b/db/main.py index a156a0e..9efeb55 100644 --- a/db/main.py +++ b/db/main.py @@ -138,12 +138,6 @@ 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: diff --git a/dnfs/dnfs.py b/dnfs/dnfs.py index 2c397cc..700e2e9 100755 --- a/dnfs/dnfs.py +++ b/dnfs/dnfs.py @@ -27,6 +27,7 @@ import urllib.request import urllib.error from typing import Dict, Any from security_middleware import apply_security_headers +from log_client import logs DB_SERVER_URL = 'http://127.0.0.1:8005/execute' def call_db_service(url: str, payload: dict, timeout: int = 10) -> dict: @@ -184,6 +185,7 @@ def add_note(): if pid is None or not title or not content: + logs.warning('dnfs', 'add_note 参数缺失', extra={'ip': request.remote_addr}) return jsonify({"error": "Pid, title and content are required"}), 400 info = None @@ -193,6 +195,7 @@ def add_note(): info = decode_token(token) if info is None: + logs.warning('dnfs', 'add_note 未授权访问', extra={'ip': request.remote_addr}) return jsonify({"error": "Warning: no Authorization"}), 401 if pid != 0: @@ -209,6 +212,7 @@ def add_note(): # 不存在的父节点 if len( notes['data']) == 0: + logs.warning('dnfs', f'add_note 父节点不存在 pid={pid}', extra={'user': info['user']}) return jsonify({'error': 'the new node need a parent'}), 400 @@ -216,6 +220,7 @@ def add_note(): review = review_content(title + '\n' + content, 10) if True != review: #return jsonify({'error': 'review failed', 'data': review}), 400 + logs.warning('dnfs', f'add_note 内容审核未通过 title={title}', extra={'user': info['user']}) return jsonify({'error': 'review failed'}), 400 insert_payload = { @@ -234,8 +239,10 @@ def add_note(): success = result['success'] if success is True: + logs.info('dnfs', f'add_note 成功 title={title} pid={pid}', extra={'user': info['user']}) return jsonify({'success': success}), 200 else: + logs.error('dnfs', f'add_note 写入失败 title={title}', extra={'user': info['user']}) return jsonify({'success': success}), 400 @@ -262,6 +269,7 @@ def get_notes(): result = notes['data'] + logs.info('dnfs', f'get_notes 查询 pid={pid} 返回{len(result)}条', extra={'ip': request.remote_addr}) return jsonify(result), 200 @@ -276,6 +284,7 @@ def delete_note(): if not note_id: + logs.warning('dnfs', 'delete_note 参数缺失 id', extra={'ip': request.remote_addr}) return jsonify({"error": "Note ID is required"}), 400 info = None @@ -285,6 +294,7 @@ def delete_note(): info = decode_token(token) if info is None: + logs.warning('dnfs', 'delete_note 未授权访问', extra={'ip': request.remote_addr}) return jsonify({"error": "Warning: no Authorization"}), 401 #with app.app_context(): @@ -320,6 +330,7 @@ def delete_note(): # 有子节点的情况不可删除 if len(notes['data']) > 0: + logs.warning('dnfs', f'delete_note 存在子节点 id={note_id}', extra={'user': info['user']}) return jsonify({'error': 'this node has a child '}), 400 # 检查所有权 @@ -335,6 +346,7 @@ def delete_note(): notes = data['data'] if not info['user'] == 'admin': if not len(notes) == 1 or not notes[0]['owner'] == info['user']: + logs.warning('dnfs', f'delete_note 所有权拒绝 id={note_id}', extra={'user': info['user']}) return jsonify({"error": 'It\'s not your note'}), 403 @@ -350,10 +362,12 @@ def delete_note(): if result is True: + logs.info('dnfs', f'delete_note 成功 id={note_id}', extra={'user': info['user']}) return jsonify({'success': True}), 200 else: + logs.error('dnfs', f'delete_note 失败 id={note_id}', extra={'user': info['user']}) return jsonify({'success': False}), 400 @@ -379,6 +393,7 @@ def update_note(): if id is None or pid is None or not title or not content: + logs.warning('dnfs', 'update_note 参数缺失', extra={'ip': request.remote_addr}) return jsonify({"error": "Invalid param"}), 400 # 获取用户名 @@ -389,6 +404,7 @@ def update_note(): info = decode_token(token) if info is None: + logs.warning('dnfs', 'update_note 未授权访问', extra={'ip': request.remote_addr}) return jsonify({"error": "Warning: no Authorization"}), 401 @@ -404,12 +420,14 @@ def update_note(): data = call_db_service(DB_SERVER_URL, select_payload) notes = data['data'] if not len(notes) == 1 or not notes[0]['owner'] == info['user']: + logs.warning('dnfs', f'update_note 所有权拒绝 id={id}', extra={'user': info['user']}) return jsonify({"error": 'It\'s not your note'}), 403 # 内容审查 review = review_content(title + '\n' + content, 10) if True != review: #return jsonify({'error': 'review failed', 'data': review}), 400 + logs.warning('dnfs', f'update_note 内容审核未通过 id={id} title={title}', extra={'user': info['user']}) return jsonify({'error': 'review failed'}), 400 #with app.app_context(): @@ -443,10 +461,12 @@ def update_note(): if result is True: + logs.info('dnfs', f'update_note 成功 id={id} title={title}', extra={'user': info['user']}) return jsonify({'success': True}), 200 else: + logs.error('dnfs', f'update_note 失败 id={id}', extra={'user': info['user']}) return jsonify({'success': False}), 400 @@ -460,6 +480,7 @@ def update_note(): ADDR='127.0.0.1' PORT=8006 if __name__ == "__main__": + logs.info('dnfs', f'服务启动，监听 http://{ADDR}:{PORT}') print(f'Listen at http://{ADDR}:{PORT}') #app.run(host="127.0.0.1", port=8006) serve(app, host='127.0.0.1', port=8006) diff --git a/dnfs/log_client.py b/dnfs/log_client.py new file mode 100644 index 0000000..d84d252 --- /dev/null +++ b/dnfs/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()