feat(confs): 统一配置服务（pymsg/TCP :8010）
- confs.py 重写为统一配置服务：get/list/reload/health 路由， 内存缓存 + mtime 自动重载，路径遍历防护，ini/json 按键提取 - 复制 log_client.py，新增 configs/example.ini 示例配置 - 注册到 core/init.txt（db 之后启动）与 _inst.sh 安装序列
This commit is contained in:
qing committed
1 parent
ae92a6edde
commit
55fbe2f1f1
6 files changed
+448
-124
No files matched your search
@@ -36,4 +36,5 @@ inst_module "dnfs"
|
||||
inst_module "db"
|
||||
inst_module "smtp"
|
||||
inst_module "captcha"
|
||||
inst_module "confs"
|
||||
inst_module "core" # core last
|
||||
+1
-2
@@ -1,7 +1,6 @@
|
||||
#!/bin/bash
|
||||
cd "$(dirname "$0")" || exit
|
||||
python3 -m venv venv
|
||||
#sudo chmod +x .env/bin/activate
|
||||
source venv/bin/activate
|
||||
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
|
||||
#nuitka --standalone smtp_v2.py
|
||||
#nuitka --standalone confs.py
|
||||
@@ -0,0 +1,10 @@
|
||||
# 示例配置 —— 统一配置服务 confs 的配置文件存放目录
|
||||
# 各微服务的配置文件集中放在本目录，由 confs 通过 pymsg get 路由分发给内部服务。
|
||||
# 支持类型：.ini / .conf / .cfg / .json / .yaml / .yml / .toml / .txt
|
||||
|
||||
[Server]
|
||||
address = 127.0.0.1
|
||||
port = 8010
|
||||
|
||||
[Feature]
|
||||
dark_mode = on
|
||||
+262
-122
@@ -1,158 +1,298 @@
|
||||
#!/usr/bin/env python3
|
||||
'''
|
||||
使用pymsg库的尝试
|
||||
'''
|
||||
"""
|
||||
confs — 统一配置服务（pymsg/TCP，默认 127.0.0.1:8010）
|
||||
|
||||
from pymsg import MsgServer
|
||||
from email.mime.text import MIMEText
|
||||
from email.header import Header
|
||||
from log_client import logs
|
||||
from dotenv import load_dotenv
|
||||
import smtplib
|
||||
import random
|
||||
import time
|
||||
import sys, os
|
||||
集中存放各微服务的配置文件，通过 pymsg 协议分发给内部服务，
|
||||
替代各服务本地散落的 ini/json/yaml 配置文件，实现配置的统一管理与下发。
|
||||
|
||||
pymsg 路由::
|
||||
|
||||
{"_route": "get", "name": "example.ini"} → 取配置文件全文
|
||||
{"_route": "get", "name": "example.ini", "key": "Server.port"} → 取单个键（ini/json）
|
||||
{"_route": "list"} → 列出全部配置
|
||||
{"_route": "reload", "name": "example.ini"} → 重载指定配置（缺省清空全部缓存）
|
||||
{"_route": "health"} → 健康检查
|
||||
|
||||
响应格式（与 db / smtp / captcha 一致）::
|
||||
|
||||
{"success": true, "data": {...}}
|
||||
{"success": false, "error": "..."}
|
||||
|
||||
配置文件存放于 confs/configs/ 目录（可用 --dir 覆盖），内存缓存 +
|
||||
文件 mtime 变化自动重载；配置名校验防路径遍历。仅监听 127.0.0.1，不对外暴露。
|
||||
"""
|
||||
|
||||
import configparser
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
# 从文件加载到环境变量？
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))
|
||||
SMTP_SERVER: str = os.getenv('SMTP_SERVER')
|
||||
SMTP_PORT: int = int(os.getenv("SMTP_PORT"))
|
||||
SMTP_USER: str = os.getenv('SMTP_USER')
|
||||
SMTP_PASSWORD: str = os.getenv('__PASSWORD')
|
||||
# pymsg / log_client 导入：生产环境 pymsg 由 venv（pip install ../pymsg）提供；
|
||||
# 附加 ../pymsg 源码路径作为无 venv 时的开发兜底
|
||||
_here = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, os.path.join(_here, ".."))
|
||||
sys.path.insert(0, os.path.join(_here, "..", "pymsg"))
|
||||
from pymsg import RpcServer # noqa: E402
|
||||
from log_client import logs # noqa: E402
|
||||
|
||||
# ── 配置 ────────────────────────────────────────────────────────────────
|
||||
|
||||
CONFIG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "configs")
|
||||
# 允许分发的配置文件类型（白名单）
|
||||
ALLOWED_EXT = {".ini", ".conf", ".cfg", ".json", ".yaml", ".yml", ".toml", ".txt"}
|
||||
|
||||
|
||||
print("【SMTP】")
|
||||
print("server: ", SMTP_SERVER)
|
||||
print("port: ", SMTP_PORT)
|
||||
print("user: ", SMTP_USER)
|
||||
print("password: ", SMTP_PASSWORD)
|
||||
# ── 异常 ────────────────────────────────────────────────────────────────
|
||||
|
||||
captchas: dict[str, list[int, int]] = {}
|
||||
class ConfigError(Exception):
|
||||
"""配置服务业务错误（由 RpcServer 包装为 success=false 返回）。"""
|
||||
|
||||
|
||||
class SmtpServer:
|
||||
'''SMTP服务器类'''
|
||||
class ConfigNotFound(ConfigError):
|
||||
"""配置文件不存在。"""
|
||||
|
||||
def __init__(self, addr="127.0.0.1", port=1234, workers=4):
|
||||
|
||||
class ConfigNameInvalid(ConfigError):
|
||||
"""配置名非法（路径遍历或类型不允许）。"""
|
||||
|
||||
|
||||
class KeyNotFound(ConfigError):
|
||||
"""配置键不存在或键格式错误。"""
|
||||
|
||||
|
||||
def _iso(ts: float) -> str:
|
||||
"""时间戳转 ISO 8601（UTC，与 db 服务的日期序列化约定一致）。"""
|
||||
return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════
|
||||
# ConfigStore — 配置文件仓库
|
||||
# ═══════════════════════════════════════════════════════════════════════
|
||||
|
||||
class ConfigStore:
|
||||
"""配置文件仓库：磁盘 configs/ 目录 + 内存缓存，mtime 变化自动重载。
|
||||
|
||||
所有公开方法持锁操作（配置文件小、内部调用频率低，简单持锁即可），
|
||||
保证 workers>0 并发回调下的缓存一致性。
|
||||
"""
|
||||
|
||||
def __init__(self, root: str):
|
||||
self.root = os.path.realpath(root)
|
||||
os.makedirs(self.root, exist_ok=True)
|
||||
self._cache = {} # name -> {"content": str, "mtime": float, "size": int}
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def _resolve(self, name: str) -> str:
|
||||
"""校验配置名并返回绝对路径（防路径遍历 + 扩展名白名单）。"""
|
||||
if not name or name != os.path.basename(name):
|
||||
raise ConfigNameInvalid(f"非法配置名: {name!r}")
|
||||
ext = os.path.splitext(name)[1].lower()
|
||||
if ext not in ALLOWED_EXT:
|
||||
raise ConfigNameInvalid(f"不支持的配置类型: {ext or '(无扩展名)'}")
|
||||
path = os.path.realpath(os.path.join(self.root, name))
|
||||
if not path.startswith(self.root + os.sep):
|
||||
raise ConfigNameInvalid(f"非法配置名: {name!r}")
|
||||
if not os.path.isfile(path):
|
||||
raise ConfigNotFound(f"配置不存在: {name}")
|
||||
return path
|
||||
|
||||
def get(self, name: str) -> dict:
|
||||
"""取配置内容（带缓存；文件 mtime 变化时自动重新读取）。"""
|
||||
path = self._resolve(name)
|
||||
with self._lock:
|
||||
try:
|
||||
mtime = os.path.getmtime(path)
|
||||
except OSError:
|
||||
self._cache.pop(name, None)
|
||||
raise ConfigNotFound(f"配置不存在或不可读: {name}") from None
|
||||
entry = self._cache.get(name)
|
||||
if entry is not None and entry["mtime"] == mtime:
|
||||
return entry
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
except (OSError, UnicodeDecodeError) as e:
|
||||
raise ConfigError(f"配置读取失败: {name}（{e}）") from None
|
||||
entry = {"content": content, "mtime": mtime,
|
||||
"size": len(content.encode("utf-8"))}
|
||||
self._cache[name] = entry
|
||||
logs.info("confs", f"配置已加载 {name}（{entry['size']}B）")
|
||||
return entry
|
||||
|
||||
def list(self) -> list:
|
||||
"""列出配置目录下全部允许类型的配置文件。"""
|
||||
out = []
|
||||
for name in sorted(os.listdir(self.root)):
|
||||
if os.path.splitext(name)[1].lower() not in ALLOWED_EXT:
|
||||
continue
|
||||
path = os.path.join(self.root, name)
|
||||
if not os.path.isfile(path):
|
||||
continue
|
||||
st = os.stat(path)
|
||||
out.append({"name": name, "size": st.st_size,
|
||||
"mtime": _iso(st.st_mtime),
|
||||
"cached": name in self._cache})
|
||||
return out
|
||||
|
||||
def reload(self, name: "str | None" = None) -> list:
|
||||
"""重载配置：指定 name 立即重新读取；缺省清空全部缓存（下次访问惰性重载）。
|
||||
|
||||
Returns:
|
||||
受影响（重载/移出缓存）的配置名列表。
|
||||
"""
|
||||
if name is not None:
|
||||
self._resolve(name) # 校验，文件不存在则抛 ConfigNotFound
|
||||
with self._lock:
|
||||
self._cache.pop(name, None)
|
||||
self.get(name)
|
||||
return [name]
|
||||
with self._lock:
|
||||
dropped = list(self._cache.keys())
|
||||
self._cache.clear()
|
||||
return dropped
|
||||
|
||||
def cached_count(self) -> int:
|
||||
with self._lock:
|
||||
return len(self._cache)
|
||||
|
||||
|
||||
# ── 按键提取（可选功能） ────────────────────────────────────────────────
|
||||
|
||||
def _extract_key(name: str, content: str, key: str):
|
||||
"""从配置内容中提取单个键。
|
||||
|
||||
- ini/conf/cfg：键格式 ``Section.option``，返回字符串值；
|
||||
- json：键格式为点号路径（如 ``server.port``，列表可用数字下标），
|
||||
返回原始 JSON 值；
|
||||
- 其他类型不支持按键查询，需取全文后自行解析。
|
||||
"""
|
||||
ext = os.path.splitext(name)[1].lower()
|
||||
|
||||
if ext in (".ini", ".conf", ".cfg"):
|
||||
parser = configparser.ConfigParser()
|
||||
try:
|
||||
parser.read_string(content)
|
||||
except configparser.Error as e:
|
||||
raise ConfigError(f"INI 解析失败: {name}（{e}）") from None
|
||||
section, _, option = key.partition(".")
|
||||
if not section or not option:
|
||||
raise KeyNotFound(f"INI 键格式应为 Section.option: {key!r}")
|
||||
if parser.has_option(section, option):
|
||||
return parser.get(section, option)
|
||||
raise KeyNotFound(f"键不存在: {name} -> {key}")
|
||||
|
||||
if ext == ".json":
|
||||
try:
|
||||
node = json.loads(content)
|
||||
except json.JSONDecodeError as e:
|
||||
raise ConfigError(f"JSON 解析失败: {name}（{e}）") from None
|
||||
for part in key.split("."):
|
||||
if isinstance(node, dict) and part in node:
|
||||
node = node[part]
|
||||
elif isinstance(node, list) and part.isdigit() and int(part) < len(node):
|
||||
node = node[int(part)]
|
||||
else:
|
||||
raise KeyNotFound(f"键不存在: {name} -> {key}")
|
||||
return node
|
||||
|
||||
raise ConfigError(f"{ext} 类型不支持按键查询，请取全文后自行解析")
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════
|
||||
# ConfsServer — 主服务类
|
||||
# ═══════════════════════════════════════════════════════════════════════
|
||||
|
||||
class ConfsServer:
|
||||
"""统一配置服务：集中管理并向内部微服务分发配置文件。"""
|
||||
|
||||
def __init__(self, addr="127.0.0.1", port=8010, workers=4, config_dir=CONFIG_DIR):
|
||||
self.addr = addr
|
||||
self.port = port
|
||||
self._server = MsgServer(addr, port, workers=workers)
|
||||
self._started_at = time.time()
|
||||
self.store = ConfigStore(config_dir)
|
||||
self._server = RpcServer(
|
||||
addr, port, workers=workers,
|
||||
on_error=lambda route, e: logs.error(
|
||||
"confs", f"路由 {route} 处理异常: {e}"),
|
||||
)
|
||||
self._server.add_route("get", self._handle_get)
|
||||
self._server.add_route("list", self._handle_list)
|
||||
self._server.add_route("reload", self._handle_reload)
|
||||
self._server.add_route("health", self._handle_health)
|
||||
|
||||
self._route = {
|
||||
"send_email": self._send_email,
|
||||
"verify_email": self._verify_email
|
||||
}
|
||||
# ── 启动 / 停止 ──────────────────────────────────────────────────
|
||||
|
||||
def start(self):
|
||||
self._server.start(self._analysis)
|
||||
def start(self) -> None:
|
||||
"""启动服务（阻塞当前线程）。"""
|
||||
print(f"[confs] 监听 tcp://{self.addr}:{self.port}，配置目录 {self.store.root}")
|
||||
logs.info("confs", f"开始监听 tcp://{self.addr}:{self.port}"
|
||||
f"（配置目录 {self.store.root}）")
|
||||
self._server.start()
|
||||
|
||||
def stop(self):
|
||||
def stop(self) -> None:
|
||||
"""从另一线程停止服务。"""
|
||||
self._server.stop()
|
||||
|
||||
def _analysis(self, sock, addr, data):
|
||||
print("""【recv】\n"""+str(addr)+'\n'+data) # 打印实际收到的消息
|
||||
# ── 路由处理 ─────────────────────────────────────────────────────
|
||||
|
||||
def _handle_get(self, req):
|
||||
"""取配置文件全文；带 key 时返回单个键的值（仅 ini/json）。"""
|
||||
name = str(req.get("name", "")).strip()
|
||||
if not name:
|
||||
raise ConfigError("缺少 name 字段")
|
||||
entry = self.store.get(name)
|
||||
|
||||
# 解包json
|
||||
try:
|
||||
request = json.loads(data)
|
||||
except json.JSONDecodeError as err:
|
||||
self._reply(sock, False, error=f"解析JSON失败: {err}")
|
||||
return
|
||||
key = req.get("key")
|
||||
if key is None:
|
||||
logs.info("confs", f"分发配置 {name}（{entry['size']}B）")
|
||||
return {"name": name, "content": entry["content"],
|
||||
"size": entry["size"], "mtime": _iso(entry["mtime"])}
|
||||
|
||||
# 路由
|
||||
route = request.pop("_route", "")
|
||||
func = self._route.get(route)
|
||||
if func is None:
|
||||
self._reply(sock, False, error=f"解析路由失败：没有找到路由{route}")
|
||||
return
|
||||
key = str(key).strip()
|
||||
value = _extract_key(name, entry["content"], key)
|
||||
logs.info("confs", f"分发配置键 {name} -> {key}")
|
||||
return {"name": name, "key": key, "value": value}
|
||||
|
||||
# 执行分支
|
||||
try:
|
||||
result = func(request)
|
||||
self._reply(sock, True, data=result)
|
||||
except Exception as err:
|
||||
self._reply(sock, False, error=f"执行操作失败: {err}")
|
||||
def _handle_list(self, _req):
|
||||
"""列出配置目录下全部配置文件及其元信息。"""
|
||||
return {"config_dir": self.store.root, "configs": self.store.list()}
|
||||
|
||||
def _reply(self, sock, success: bool, data: str = None, error: str = None):
|
||||
print("""【send】\n"""+str(data)+'\n'+str(error)) # 打印实际输出
|
||||
response = {"success": success}
|
||||
if data is not None:
|
||||
response["data"] = data
|
||||
if error is not None:
|
||||
response["error"] = error
|
||||
self._server.reply(sock, json.dumps(response, ensure_ascii=False))
|
||||
|
||||
|
||||
def _send_email(self, data):
|
||||
def build_captcha() -> str:
|
||||
captcha: str = ""
|
||||
for i in range(0,6):
|
||||
digit: int = random.randint(0,9)
|
||||
captcha += str(digit)
|
||||
return captcha
|
||||
|
||||
recipient: str = data.get("email")
|
||||
subject: str = "DNFS 邮箱验证"
|
||||
captcha: str = build_captcha()
|
||||
content: str = f"DNFS: 您的邮箱验证码为 {captcha}，一分钟内有效。"
|
||||
|
||||
captchas[recipient] = ["", ""]
|
||||
captchas[recipient][0] = captcha
|
||||
captchas[recipient][1] = str(int(time.time())+60)
|
||||
|
||||
message = MIMEText(content, "plain", "UTF-8")
|
||||
message["From"] = SMTP_USER
|
||||
message["To"] = recipient
|
||||
message["Subject"] = subject
|
||||
|
||||
# 这一行不用写到try里，因为如果它失败就没有这个类不用quit()
|
||||
# 反而因为写进了try，和finally不在一个作用域
|
||||
server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT, timeout=10)
|
||||
try:
|
||||
server.login(SMTP_USER, SMTP_PASSWORD)
|
||||
server.sendmail(SMTP_USER, recipient, message.as_string())
|
||||
logs.info("smtp", f"{recipient}: 验证码发送成功({captcha})")
|
||||
return {"success": True, "data": "发送成功"}
|
||||
|
||||
except Exception as err:
|
||||
logs.error("smtp", f"{recipient}: 验证码发送失败({err})")
|
||||
return {"success": False, "error": f"发送失败: {err}"}
|
||||
|
||||
finally:
|
||||
server.quit()
|
||||
|
||||
|
||||
def _verify_email(data):
|
||||
for i in list(captchas.keys()):
|
||||
if int(captchas[i][1]) < int(time.time()):
|
||||
del captchas[i]
|
||||
|
||||
email: str = data.get("email")
|
||||
input_captcha: str = data.get("code")
|
||||
|
||||
if captchas[email][0] == input_captcha:
|
||||
logs.info("smtp", f"{email}: 验证通过")
|
||||
return {"success": True, "data": "验证通过"}
|
||||
def _handle_reload(self, req):
|
||||
"""重载指定配置；缺省 name 时清空全部缓存（下次访问惰性重载）。"""
|
||||
name = req.get("name")
|
||||
names = self.store.reload(str(name).strip() if name else None)
|
||||
if name:
|
||||
logs.info("confs", f"重载配置 {names[0]}")
|
||||
else:
|
||||
logs.info("smtp", f"{email}: 验证失败")
|
||||
return {"success": False, "data": "验证失败"}
|
||||
logs.info("confs", f"清空配置缓存（{len(names)} 项）")
|
||||
return {"reloaded": names}
|
||||
|
||||
def _handle_health(self, _req):
|
||||
return {"status": "ok", "service": "confs",
|
||||
"config_dir": self.store.root,
|
||||
"cached": self.store.cached_count(),
|
||||
"uptime": int(time.time() - self._started_at)}
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════
|
||||
# 入口
|
||||
# ═══════════════════════════════════════════════════════════════════════
|
||||
|
||||
if __name__ == "__main__":
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description="smtp--pymsg邮箱验证码服务")
|
||||
parser = argparse.ArgumentParser(description="confs — 统一配置服务（pymsg/TCP）")
|
||||
parser.add_argument("--addr", default="127.0.0.1", help="监听地址")
|
||||
parser.add_argument("--port", type=int, default=8008, help="监听端口")
|
||||
parser.add_argument("--port", type=int, default=8010, help="监听端口")
|
||||
parser.add_argument("--workers", type=int, default=4,
|
||||
help="回调并发线程数（0=同步串行模式）")
|
||||
parser.add_argument("--dir", default=CONFIG_DIR, help="配置文件目录")
|
||||
args = parser.parse_args()
|
||||
|
||||
server = SmtpServer(args.addr, args.port)
|
||||
server = ConfsServer(args.addr, args.port,
|
||||
workers=args.workers, config_dir=args.dir)
|
||||
try:
|
||||
server.start()
|
||||
except KeyboardInterrupt:
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
"""统一日志客户端 — 各服务共用（dnfs / smtp / static 三份副本内容一致，改动请同步）。
|
||||
|
||||
设计要点：
|
||||
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
|
||||
- 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
|
||||
- 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 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 RpcClient, RpcError # noqa: E402
|
||||
|
||||
# ── 配置 ──────────────────────────────────────────────────────────────
|
||||
|
||||
_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1')
|
||||
_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005'))
|
||||
|
||||
_BATCH_SIZE = 50 # 队列满 50 条时 flush
|
||||
_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush
|
||||
_TIMEOUT = 5 # socket 超时（秒）
|
||||
|
||||
|
||||
class LogClient:
|
||||
"""异步批量日志推送客户端。"""
|
||||
|
||||
def __init__(self, service_name="log",
|
||||
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.1, 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 RPC（execute 路由）逐条 INSERT 到 dbv2 Logs 表。"""
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
|
||||
for entry in batch:
|
||||
extra_val = entry.get("extra")
|
||||
payload = {
|
||||
"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,
|
||||
},
|
||||
}
|
||||
try:
|
||||
client.call("execute", payload, timeout=_TIMEOUT)
|
||||
except RpcError as e:
|
||||
# 业务错误（如表结构不符）：打印后继续下一条
|
||||
print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr)
|
||||
except Exception as e:
|
||||
# 网络错误：db 不可用，放弃本批剩余（fail-safe）
|
||||
print(f"[log_client] 发送失败: {e}", file=_sys.stderr)
|
||||
break
|
||||
|
||||
def _shutdown(self):
|
||||
"""进程退出时 flush 剩余日志。"""
|
||||
self._stop.set()
|
||||
self.flush()
|
||||
|
||||
|
||||
# ── 全局单例 ──────────────────────────────────────────────────────────
|
||||
|
||||
logs = LogClient()
|
||||
@@ -1,4 +1,5 @@
|
||||
bash /opt/micro_servers/db/_run.sh
|
||||
bash /opt/micro_servers/confs/_run.sh
|
||||
bash /opt/micro_servers/ai_review/_run.sh
|
||||
bash /opt/micro_servers/dnfs/_run.sh
|
||||
bash /opt/micro_servers/smtp/_run.sh
|
||||
|
||||
Reference in new issue
Block a user