qing/blackbird
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

logs in db develop

Browse Source
This commit is contained in:
qing committed 2026-07-25 21:01:46 +08:00
1 parent a4479696eb
commit 013e130a08
4 files changed
+179 -28
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

db/log_module.py
+6 -22
View File
@@ -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} 条')
db/main.py
-6
View File
@@ -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:
dnfs/dnfs.py
+21
View File
@@ -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)
dnfs/log_client.py
+152
View File
@@ -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()