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

Compare commits

base: qing/micro_servers:9708fbc87ee28a3048e82843ecc40b6bae4b6674
qing/micro_servers:main
...
compare: qing/micro_servers:3be9e78335d5553b8953da8dd567cf85bcf2d078
qing/micro_servers:main
4 Commits
9708fbc87e ... 3be9e78335
Author SHA1 Message Date
qing 3be9e78335 docs(aibox): update README for v2.2 prompt system fixes 
- Expand prompts directory listing to show all 10 skill files explicitly
- Update prompt system table: '1_*.md' → 'N_*.md (N=1~10)' with numeric ordering
- Document load order guarantee (identity → skills 1-10 → soul → status)
- Document is_continuation behavior (no repeated pre_prompt on recursive calls)
- Add v2.2 changelog: prompt ordering fix, token waste reduction,
  recursion pre_prompt dedup, large-file threshold, memory search tuning
 2026-08-01 03:24:39 +08:00
qing 1b9c727e80 fix(log_client): harden background flush thread against exceptions 
- Wrap _flush_loop main body in outer try/except to prevent any
  unexpected exception (network errors, serialization errors, etc.)
  from killing the background thread (which would cause unbounded
  queue growth and eventual OOM)
- Wrap _send() call in inner try/except as final safety net
- Wrap RpcClient() construction in try/except (connection failures
  during client creation were previously unhandled)
- Add _safe_json() helper: unserializable extra fields fall back to
  repr()/str() instead of raising TypeError that could kill thread
- Generate timestamp at push() (enqueue time) instead of _send() time,
  so timestamps reflect actual event time even when queue is backed up
- Replace unreliable time.sleep(0.6) in flush() with _flush_done Event
  for proper synchronization between caller and background thread
- Add _QUEUE_MAXSIZE=10000 with drop-oldest policy to prevent unbounded
  memory growth even in pathological cases
- Use put_nowait with Full handling to ensure push() never blocks
  business threads
- Sync fix to all 4 copies: confs/static/smtp/dnfs
 2026-08-01 03:23:19 +08:00
qing e8fa26d79f fix(aibox): fix prompt ordering, reduce per-turn token waste, and improve defaults 
Prompt system bugs and improvements:

1. **SKILL_PREFIX bug fixed**: SKILL_PREFIX = "1_" only matched files starting
   with "1_", causing 2_ through 10_ skill prompt files to be misclassified
   as "other" files and sorted in wrong order (e.g., 10_task_handling.md
   sorted before 2_memory_save.md due to lexicographic ordering). Replaced
   with regex _SKILL_FILE_RE = re.compile(r'^(\d+)_') and _skill_sort_key()
   that extracts the numeric prefix for proper numeric sorting.

2. **_init_system_prompts used unsorted list**: get_prompts_list() returned
   prompts in os.listdir() lexicographic order, which put 100_STATUS.md
   before 1_memory_search.md and 10_task_handling.md before 1_memory_search.md.
   Now uses get_sorted_prompts_list() for correct ordering:
   identity → skill prompts (1-10 by number) → soul → status.

3. **Verbose pre_prompt injected every turn (~30 lines)**: The pre_prompt
   contained detailed workflow instructions (search_skills first, memory
   search, ask_user, etc.) that duplicated what system skill prompts
   already say, wasting ~300+ tokens per user message. Worse, the same
   verbose prompt was injected on recursive think_and_respond calls after
   tool execution, confusing the model with repeated "第一步搜索技能"
   instructions when it was already mid-conversation with tool results.
   - Reduced pre_prompt to 5 concise bullet points as a reminder
   - Added is_continuation parameter: recursive calls after tool execution
     skip pre_prompt entirely and just pass "请基于以上工具执行结果继续回答。"
   - Removed post_prompt (save_memory/update_soul instructions already in
     2_memory_save.md and 8_identity_management.md)

4. **Unreasonable large-file thresholds**: 3_file_operations.md default
   said >50KB or >200 lines is "too large" — too conservative for code
   files. Raised to 500KB/2000 lines; added tail=true and binary file
   guidance.

5. **Overly rigid memory search instruction**: 1_memory_search.md said
   "每次都要做！" (must do every time), causing unnecessary memory searches
   for simple greetings/acknowledgments. Softened to "根据话题相关性判断",
   with explicit examples of when to skip (greetings, confirmations).

New/changed methods:
- PromptManager._skill_sort_key(cls, filename) - extract numeric prefix
- PromptManager._prompt_sort_key(self, p) - tuple-based sort key
- PromptManager.get_sorted_prompts_list() - returns correctly ordered list
- PromptManager.get_skill_prompts() - uses regex instead of startswith("1_")
- DeepSeekChat.think_and_respond() - added is_continuation parameter
 2026-08-01 03:19:58 +08:00
qing da7f11d352 fix(confs): add MAX_CONFIG_SIZE (1 MiB) to prevent OOM on oversized configs 
- Add MAX_CONFIG_SIZE constant (1 MiB) to limit single config file size
- Check file size via os.stat() before reading into memory
- Oversized files raise ConfigError with descriptive message instead of
  being fully read (which could cause OOM with multi-GB files)
- Use single os.stat() call to get both mtime and size atomically
- Update README: document size limit in security mechanisms and notes
- Existing small configs (ini/json/yaml, typically <10KB) are unaffected
 2026-08-01 03:15:06 +08:00
8 changed files with 1090 additions and 794 deletions
Expand all files Collapse all files

No files matched your search

aibox/README.md
+26 -7
View File
@@ -85,11 +85,20 @@ aibox/
```
~/.aibox/
├── config.json # 配置文件
├── prompts/ # 系统提示词目录
│ ├── 00_IDENTITY.md # 用户身份信息
│ ├── 1_*.md # 基础技能提示词（文件操作、网页访问等）
├── prompts/ # 系统提示词目录（按文件名数字前缀排序加载）
│ ├── 00_IDENTITY.md # 用户身份信息（最先加载）
│ ├── 1_memory_search.md # 基础技能提示词（1-10，按数字升序加载）
│ ├── 2_memory_save.md
│ ├── 3_file_operations.md
│ ├── 4_web_access.md
│ ├── 5_memo_management.md
│ ├── 6_command_execution.md
│ ├── 7_extension_skills.md
│ ├── 8_identity_management.md
│ ├── 9_language_instruction.md
│ ├── 10_task_handling.md
│ ├── 99_SOUL.md # AI 灵魂定义（角色、性格、能力边界）
│ └── 100_STATUS.md # AI 状态记录
│ └── 100_STATUS.md # AI 状态记录（最后加载）
├── skills/ # 扩展技能目录（运行时从仓库 skills/ 复制或用户自建）
├── memories.db # 记忆和任务数据库（SQLite）
├── current_conversation.json # 当前对话
@@ -364,15 +373,17 @@ AI 可以通过 `update_soul` 工具动态更新 `99_SOUL.md`，通过 `update_i
## 🧩 提示词系统
提示词文件位于 `~/.aibox/prompts/`：
提示词文件位于 `~/.aibox/prompts/`，按文件名**数字前缀升序**加载（identity → 技能 1~10 → soul → status）：
| 文件 | 用途 | 更新方式 |
|-----|------|---------|
| `00_IDENTITY.md` | 用户身份信息 | AI 通过 `update_identity` 工具更新 |
| `1_*.md` | 基础技能提示词 | 手动编辑，AI 通过 `reload_prompts` 重载 |
| `99_SOUL.md` | AI 灵魂定义（角色、性格） | AI 通过 `update_soul` 工具更新 |
| `N_*.md`（N=1~10） | 基础技能提示词（记忆搜索/保存、文件/网页/命令/任务操作、技能扩展、身份管理、语言、任务规则） | 手动编辑，AI 通过 `/reload_prompts` 重载 |
| `99_SOUL.md` | AI 灵魂定义（角色、性格、能力边界、行为准则） | AI 通过 `update_soul` 工具更新 |
| `100_STATUS.md` | AI 状态记录 | AI 通过 `save_memory_and_end_conversation` 更新 |
> **加载顺序保证**：`00_IDENTITY.md` 最先加载（AI 先知道用户是谁），然后是 1~10 号技能提示词（按数字升序），接着是 `99_SOUL.md`（AI 的角色定位），最后是 `100_STATUS.md`（当前状态）。每轮用户输入前会附加一段简洁的工作流程提醒（不重复 system prompt 中的详细说明），工具调用后递归继续时不再重复注入。
## ⚙️ 配置说明
主要配置项（`~/.aibox/config.json`，首次启动自动生成）：
@@ -673,6 +684,14 @@ elif command == "/mycommand":
## 📄 更新日志
### v2.2
- ✅ 修复提示词系统排序 bug：`SKILL_PREFIX = "1_"` 只匹配 `1_` 开头文件，导致 `2_`~`10_` 技能提示词排序错乱（`10_task_handling.md` 排在 `1_memory_search.md` 前面）；改用正则按数字前缀排序
- ✅ 修复 `_init_system_prompts` 使用未排序列表的问题，确保 system prompt 按 identity → 技能 → soul → status 正确顺序加载
- ✅ 精简每轮用户输入的 pre_prompt：原来约 30 行重复指令（与 system prompt 内容重复），每轮浪费 ~300 token；精简为 5 条简洁提醒
- ✅ 修复工具调用后递归继续时重复注入工作流指令的问题：新增 `is_continuation` 参数，递归时跳过 pre_prompt
- ✅ 优化默认大文件阈值：50KB/200行 → 500KB/2000行，增加 `tail=true` 和二进制文件处理指导
- ✅ 优化记忆搜索习惯："每次都要做"→"根据话题相关性判断"，避免简单问候也触发无意义搜索
### v2.1
- ✅ 修复工具调用结果破坏对话导致 API 持续请求失败的问题
- 不再回传 `reasoning_content`（API 响应字段）到请求中
aibox/aibox.py
+77 -94
View File
@@ -436,7 +436,7 @@ class PromptManager:
文件命名规范:
- 00_IDENTITY.md: 用户身份信息
- 1_*.md/txt: 基础技能（必须从1_开头）
- N_*.md/txt: 基础技能提示词（N 为数字，如 1_memory_search.md、10_task_handling.md），按数字升序加载
- 99_SOUL.md: 智能体灵魂文件（角色定位、身份定义、性格等）
- 100_STATUS.md: 智能体状态记录
"""
@@ -450,8 +450,8 @@ class PromptManager:
STATUS_FILE = "100_STATUS.md"
SOUL_FILE = "99_SOUL.md" # 智能体灵魂文件 - 所有角色定位内容
# 基础技能前缀
SKILL_PREFIX = "1_" # 基础技能文件必须以这个开头
# 基础技能文件名模式：N_xxx.md（N 为数字，如 1_memory_search.md、10_task_handling.md）
_SKILL_FILE_RE = re.compile(r'^(\d+)_')
def __new__(cls):
if cls._instance is None:
@@ -642,14 +642,17 @@ class PromptManager:
print(f"⚠️ 创建默认灵魂文件失败: {e}")
def _ensure_skill_files(self):
"""确保基础技能文件存在 - 以1_开头的文件"""
"""确保基础技能文件存在 - N_ 开头的文件（N 为数字，按数字顺序加载）"""
skill_files = [
{
'filename': '1_memory_search.md',
'content': """# 记忆搜索习惯
## 重要习惯
在每次回答用户问题前，**必须**先使用 `memory search` 工具搜索记忆库中与当前话题相关的记忆。
## 核心习惯
在回答用户问题前，**应根据话题相关性判断是否需要搜索记忆**：
- 涉及过往讨论、用户偏好、之前做过的事 → 必须搜索
- 全新话题、简单问候（"你好"）、确认回应（"好的""明白了"）→ 无需搜索
- 不确定时，搜索一下也无妨
## 为什么要这样做？
- 保持对话的连贯性
@@ -665,11 +668,9 @@ memory search keywords=["关键词1", "关键词2"]
## 示例
如果用户说："上次我们讨论的那个项目怎么样了？"
你应该：
1. 先搜索 "项目" 相关的记忆
1. 搜索 "项目" 相关的记忆
2. 根据搜索结果回答
3. 如果找不到相关记忆，诚实告知用户
记住：这是你的核心习惯，每次都要做！
"""
},
{
@@ -711,19 +712,22 @@ save_memory_and_end_conversation(
- **list_files**: 列出目录
## 大文件处理规则
当遇到以下情况时，**必须先告知用户文件过大**：
1. 文件大小超过 50KB
2. 文件行数超过 200 行
当遇到以下情况时，**必须分页读取**，不要一次性读取整个文件：
1. 文件大小超过 500KB
2. 文件行数超过 2000 行
对于大文件，应该：
- 使用 head/tail 参数只读取开头或结尾
- 先用 head=true 读取前50行了解文件结构
- 使用 start_line/end_line 参数指定行范围
- 建议用户提供更具体的查询条件
- 使用 tail=true 读取文件末尾
- 对于二进制文件或特殊格式文件，使用 run_command 配合 head/strings 等命令查看
- 建议用户提供更具体的查询条件以缩小范围
## 示例
```
read_file filepath="/path/to/file.txt" head=true max_lines=50
read_file filepath="/path/to/file.txt" start_line=10 end_line=20
read_file filepath="/path/to/file.txt" start_line=100 end_line=200
read_file filepath="/path/to/file.txt" tail=true max_lines=50
```
"""
},
@@ -1047,42 +1051,55 @@ update_soul content="我是...（新的自我定义）"
"""创建默认提示词（已由各 _ensure_* 方法处理）"""
pass
@classmethod
def _skill_sort_key(cls, filename: str):
"""技能文件按前缀数字排序，非技能文件返回 None。"""
m = cls._SKILL_FILE_RE.match(filename)
return int(m.group(1)) if m else None
def _prompt_sort_key(self, p):
"""排序规则：00_IDENTITY → 技能文件(按数字升序) → 99_SOUL → 100_STATUS → 其他"""
file = p['file']
if file == self.IDENTITY_FILE:
return (0,)
skill_num = self._skill_sort_key(file)
if skill_num is not None:
return (1, skill_num)
if file == self.SOUL_FILE:
return (2,)
if file == self.STATUS_FILE:
return (4,)
return (3, file) # 其他文件按名字排序
def get_prompts(self, join_with: str = "\n\n") -> str:
"""获取所有提示词，按顺序拼接"""
"""获取所有提示词，按顺序拼接（identity → 技能文件 → soul → status）"""
if not self._prompts:
return ""
# 排序规则：先基础技能（1_开头），再其他，最后灵魂文件
def sort_key(p):
file = p['file']
if file == self.SOUL_FILE:
return 3 # 灵魂文件最后加载
elif file.startswith(self.SKILL_PREFIX):
return 1 # 基础技能
elif file == self.IDENTITY_FILE:
return 0 # 身份文件最先
elif file == self.STATUS_FILE:
return 4 # 状态文件最后
else:
return 2
sorted_prompts = sorted(self._prompts, key=sort_key)
sorted_prompts = sorted(self._prompts, key=self._prompt_sort_key)
contents = [p['content'] for p in sorted_prompts]
return join_with.join(contents)
def get_sorted_prompts_list(self) -> List[Dict]:
"""获取按正确顺序排列的提示词列表（用于 system 消息构建）。"""
return sorted(self._prompts, key=self._prompt_sort_key)
def get_prompts_list(self) -> List[Dict]:
"""获取所有提示词列表"""
"""获取所有提示词列表（原始加载顺序，已弃用，请用 get_sorted_prompts_list）。"""
return self._prompts.copy()
def get_skill_prompts(self) -> List[Dict]:
"""获取所有基础技能提示词（N_ 开头的文件）。"""
return [p for p in self._prompts if self._skill_sort_key(p['file']) is not None]
def reload(self):
"""重新加载所有提示词"""
self._prompts = []
self._load_prompts()
return len(self._prompts)
def get_prompt_count(self) -> int:
return len(self._prompts)
def get_prompts_dir_path(self) -> str:
return self._prompts_dir
@@ -6536,19 +6553,19 @@ class DeepSeekChat:
self._register_tools(exit_callback, command_executor)
def _init_system_prompts(self, system_prompt: str = None):
"""初始化系统提示词 - 从 PromptManager 加载"""
"""初始化系统提示词 - 从 PromptManager 加载（按 identity → 技能 → soul → status 顺序）"""
if system_prompt:
# 如果明确指定了 system_prompt，使用它
self._add_message("system", system_prompt)
self.logger.info("使用指定的系统提示词")
else:
# 从 PromptManager 加载所有提示词
prompts = self.prompt_manager.get_prompts_list()
# 从 PromptManager 加载所有提示词，按正确顺序排列
prompts = self.prompt_manager.get_sorted_prompts_list()
if prompts:
for prompt in prompts:
if prompt['content'].strip():
self._add_message("system", prompt['content'].strip())
self.logger.info(f"已加载 {len(prompts)} 个系统提示词")
self.logger.info(f"已加载 {len(prompts)} 个系统提示词（顺序: identity → 技能 → soul → status）")
else:
# 如果没有任何提示词，添加一个默认的
default_prompt = "你是一个智能AI助手。"
@@ -6847,8 +6864,15 @@ class DeepSeekChat:
# 默认情况下，如果记忆库存在，搜索一下也无妨
return True
def think_and_respond(self, input_text: str, output_callback: Callable = None) -> Optional[str]:
"""思考并回应 - 支持实时流式输出，支持思考模式"""
def think_and_respond(self, input_text: str, output_callback: Callable = None, is_continuation: bool = False) -> Optional[str]:
"""思考并回应 - 支持实时流式输出，支持思考模式。
Args:
input_text: 用户输入或继续提示
output_callback: 流式输出回调
is_continuation: 是否为工具调用后的递归继续；为 True 时不注入冗长的工作流指令，
因为 system prompt 中已有完整行为规范，继续时只需传递工具结果
"""
if not self.conversation_active:
self.logger.info("对话已结束，重置状态")
self.conversation_active = True
@@ -6880,54 +6904,14 @@ class DeepSeekChat:
self.stats["api_calls"] += 1
# 前置提示词
pre_prompt = """【系统指令 - 核心工作流程】
在开始任何任务之前，你必须严格按照以下优先级执行：
## 第一步：搜索可用技能（最高优先级）⭐
当用户提出任何任务、需求或问题时，**必须先使用 `search_skills` 工具**搜索相关技能：
- 使用 `search_skills keyword="[任务关键词]"` 搜索是否有现成的扩展技能可以完成该任务
- 例如：用户说"帮我查天气"，你应该先执行 `search_skills keyword="天气"`
- 例如：用户说"帮我计算BMI"，你应该先执行 `search_skills keyword="BMI"`
- 如果搜索结果中有匹配的技能，优先使用该技能完成任务
- 只有在没有合适技能的情况下，才考虑使用其他工具或方法
## 第二步：搜索相关记忆
在每次回答用户问题前，**必须**使用 `memory search` 工具搜索记忆库中与当前话题相关的记忆，确保回答的连贯性和准确性。
## 第三步：积极向用户提问
如果你有任何无法确定的事情，那么你需要用户提供额外信息，
请务必使用 `ask_user` 工具向用户提问，以便获得更准确的指引。
## 第四步：执行任务或回答问题
根据搜索结果，选择合适的工具或直接回答。
【重要提醒】
- 技能优先原则：有技能先用技能，不要自己硬编码实现
- 如果没有合适的技能，可以使用其他工具或直接回答
- 对话结束时，必须使用 `save_memory_and_end_conversation` 工具保存本次对话的重要信息并更新智能体状态
- 当了解到用户的身份信息时，使用 `update_identity` 工具更新 00_IDENTITY.md
- 当你的角色、性格、能力边界需要调整时，使用 `update_soul` 工具更新 99_SOUL.md 文件来重新定义自己
- 绝对不要直接打开大文件来读，应该使用额外的压缩工具
现在请处理用户的输入："""
# 后置提示词
post_prompt = """
【注意】
当本次对话的话题讨论完成时，请使用 `save_memory_and_end_conversation` 工具结束对话并保存重要信息到记忆库，同时更新你的状态到 100_STATUS.md。
【自我认知更新】
在对话过程中，如果你发现：
- 自己的角色定位需要调整
- 性格特点需要改变
- 能力边界需要重新定义
- 行为准则需要优化
请使用 `update_soul` 工具更新 99_SOUL.md 文件，这有助于你更好地理解自己并为用户提供更精准的服务。"""
enhanced_input = pre_prompt + "\n\n" + input_text + post_prompt
if is_continuation:
# 递归继续（工具调用后）：直接使用输入文本，不重复注入工作流指令
# system prompt 中已有完整行为规范，此处只需传递工具结果的上下文
enhanced_input = input_text
else:
# 首轮回话：附加简洁的工作流程提醒（不重复 system prompt 中已有的详细说明）
pre_prompt = "【工作流程提醒】\n1. 如果用户提出新任务，先用 search_skills 搜索是否有现成技能可用\n2. 根据话题相关性决定是否 memory search（简单问候/确认无需搜索）\n3. 信息不足时用 ask_user 提问，不要猜测\n4. 遇到大文件用分页读取（head/tail/start_line/end_line），不要直接打开大文件\n5. 话题完成时用 save_memory_and_end_conversation 保存记忆并结束"
enhanced_input = pre_prompt + "\n\n" + input_text
# 记录本轮开始前的消息长度，用于 API 失败时回滚，防止坏消息污染对话历史
msgs_before_turn = len(self.messages)
@@ -7146,9 +7130,8 @@ class DeepSeekChat:
if output_callback:
output_callback("line", i18n.get('analyzing_tool_results') + "\n")
# 【关键修复4】递归调用，让AI基于工具结果继续回答
# 使用一个新的输入提示，避免重复
return self.think_and_respond("请基于以上工具执行结果继续回答用户的问题。", output_callback)
# 递归调用，让AI基于工具结果继续回答；is_continuation=True 避免重复注入工作流指令
return self.think_and_respond("请基于以上工具执行结果继续回答。", output_callback, is_continuation=True)
elif full_response:
# 正常添加助手消息，包含 reasoning_content
confs/README.md
+2
View File
@@ -181,6 +181,7 @@ port = resp["data"]["value"]
- **扩展名白名单**：仅允许 `.ini/.conf/.cfg/.json/.yaml/.yml/.toml/.txt`，其余一律拒绝
- **路径遍历防护**：配置名必须等于 `os.path.basename(name)`，且 realpath 必须位于 `configs/` 目录前缀之下
- **文件大小限制**：单配置文件上限 1 MiB（`MAX_CONFIG_SIZE`），超过则拒绝读取并返回错误，防止超大文件导致内存溢出
- **仅监听本机**：默认 `127.0.0.1`，不对外暴露，仅供内部服务调用
- **无鉴权**：当前无身份校验，依赖"仅本机 + 内部网络"的信任边界；如需承载敏感配置（如 API 密钥），应额外评估鉴权方案
@@ -224,5 +225,6 @@ confs/
- 配置文件需手动放入 `configs/` 目录（或用 `--dir` 指定其他目录），服务不会自动生成业务配置
- 文件以 UTF-8 读取，非 UTF-8 文件会返回 `配置读取失败` 错误
- 配置名区分大小写（与文件系统一致）
- 单配置文件大小上限为 1 MiB（`MAX_CONFIG_SIZE`），超过此限制会返回 `配置文件过大` 错误，防止超大文件读入内存导致 OOM
- 缓存基于文件 mtime 判断变化，若同秒内修改且 mtime 精度不足（如某些 FAT 文件系统），可能无法及时重载——可手动调 `reload` 路由强制刷新
- 当前接入情况见上文"集成关系 → 已接入的服务"；新服务接入可参考 static 的三层 fail-safe 模式（内置默认 → confs 优先 → 本地回退）
confs/confs.py
+9 -1
View File
@@ -42,6 +42,8 @@ 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"}
# 单配置文件大小上限（1 MiB），防止超大文件读入内存导致 OOM
MAX_CONFIG_SIZE = 1 * 1024 * 1024
# ── 异常 ────────────────────────────────────────────────────────────────
@@ -103,13 +105,19 @@ class ConfigStore:
path = self._resolve(name)
with self._lock:
try:
mtime = os.path.getmtime(path)
st = os.stat(path)
mtime = st.st_mtime
size = st.st_size
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
if size > MAX_CONFIG_SIZE:
raise ConfigError(
f"配置文件过大: {name}（{size} 字节，上限 {MAX_CONFIG_SIZE} 字节）"
)
try:
with open(path, "r", encoding="utf-8") as f:
content = f.read()
confs/log_client.py
+244 -173
View File
@@ -1,173 +1,244 @@
"""统一日志客户端 — 各服务共用（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()
"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
设计要点：
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts，反映事件真实发生时间
用法：
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 超时（秒）
_QUEUE_MAXSIZE = 10000 # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
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(maxsize=_QUEUE_MAXSIZE)
self._stop = threading.Event()
self._flush_signal = threading.Event()
self._flush_done = 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):
"""推送一条日志（非阻塞，立即返回）。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
entry = {
'ts': _now_iso(), # 入队时即生成时间戳，反映真实事件时间
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
try:
self._queue.put_nowait(entry)
except queue.Full:
# 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
try:
self._queue.get_nowait()
except queue.Empty:
pass
try:
self._queue.put_nowait(entry)
except queue.Full:
pass # 极端竞争下仍失败则静默丢弃
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_done.clear()
self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch
# 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
self._flush_done.wait(timeout=self._flush_interval + 2.0)
# 主线程排空队列中剩余条目
self._drain_queue_to_db()
def _drain_queue_to_db(self):
"""排空主线程可见的队列中剩余条目，直接发送到 db。"""
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 时发送。
外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
都不会死亡，避免日志队列无限增长导致 OOM。
"""
batch = []
last_flush = time.monotonic()
while not self._stop.is_set():
try:
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:
try:
self._send(batch)
except Exception as e:
# _send 内部已有细粒度异常处理，此处是最终安全网
print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
batch = []
last_flush = time.monotonic()
if self._flush_signal.is_set():
self._flush_signal.clear()
self._flush_done.set()
except Exception as e:
# 最外层安全网：任何意外异常都不应杀死日志线程
print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
time.sleep(0.5) # 防止异常风暴导致 CPU 满载
# 退出前 flush 剩余
if batch:
try:
self._send(batch)
except Exception as e:
print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
self._flush_done.set()
def _send(self, batch):
"""通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
不抛异常给调用方。
"""
try:
client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
except Exception as e:
print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
return
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
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()
dnfs/log_client.py
+244 -173
View File
@@ -1,173 +1,244 @@
"""统一日志客户端 — 各服务共用（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()
"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
设计要点：
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts，反映事件真实发生时间
用法：
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 超时（秒）
_QUEUE_MAXSIZE = 10000 # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
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(maxsize=_QUEUE_MAXSIZE)
self._stop = threading.Event()
self._flush_signal = threading.Event()
self._flush_done = 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):
"""推送一条日志（非阻塞，立即返回）。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
entry = {
'ts': _now_iso(), # 入队时即生成时间戳，反映真实事件时间
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
try:
self._queue.put_nowait(entry)
except queue.Full:
# 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
try:
self._queue.get_nowait()
except queue.Empty:
pass
try:
self._queue.put_nowait(entry)
except queue.Full:
pass # 极端竞争下仍失败则静默丢弃
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_done.clear()
self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch
# 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
self._flush_done.wait(timeout=self._flush_interval + 2.0)
# 主线程排空队列中剩余条目
self._drain_queue_to_db()
def _drain_queue_to_db(self):
"""排空主线程可见的队列中剩余条目，直接发送到 db。"""
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 时发送。
外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
都不会死亡，避免日志队列无限增长导致 OOM。
"""
batch = []
last_flush = time.monotonic()
while not self._stop.is_set():
try:
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:
try:
self._send(batch)
except Exception as e:
# _send 内部已有细粒度异常处理，此处是最终安全网
print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
batch = []
last_flush = time.monotonic()
if self._flush_signal.is_set():
self._flush_signal.clear()
self._flush_done.set()
except Exception as e:
# 最外层安全网：任何意外异常都不应杀死日志线程
print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
time.sleep(0.5) # 防止异常风暴导致 CPU 满载
# 退出前 flush 剩余
if batch:
try:
self._send(batch)
except Exception as e:
print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
self._flush_done.set()
def _send(self, batch):
"""通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
不抛异常给调用方。
"""
try:
client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
except Exception as e:
print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
return
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
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()
smtp/log_client.py
+244 -173
View File
@@ -1,173 +1,244 @@
"""统一日志客户端 — 各服务共用（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()
"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
设计要点：
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts，反映事件真实发生时间
用法：
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 超时（秒）
_QUEUE_MAXSIZE = 10000 # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
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(maxsize=_QUEUE_MAXSIZE)
self._stop = threading.Event()
self._flush_signal = threading.Event()
self._flush_done = 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):
"""推送一条日志（非阻塞，立即返回）。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
entry = {
'ts': _now_iso(), # 入队时即生成时间戳，反映真实事件时间
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
try:
self._queue.put_nowait(entry)
except queue.Full:
# 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
try:
self._queue.get_nowait()
except queue.Empty:
pass
try:
self._queue.put_nowait(entry)
except queue.Full:
pass # 极端竞争下仍失败则静默丢弃
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_done.clear()
self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch
# 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
self._flush_done.wait(timeout=self._flush_interval + 2.0)
# 主线程排空队列中剩余条目
self._drain_queue_to_db()
def _drain_queue_to_db(self):
"""排空主线程可见的队列中剩余条目，直接发送到 db。"""
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 时发送。
外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
都不会死亡，避免日志队列无限增长导致 OOM。
"""
batch = []
last_flush = time.monotonic()
while not self._stop.is_set():
try:
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:
try:
self._send(batch)
except Exception as e:
# _send 内部已有细粒度异常处理，此处是最终安全网
print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
batch = []
last_flush = time.monotonic()
if self._flush_signal.is_set():
self._flush_signal.clear()
self._flush_done.set()
except Exception as e:
# 最外层安全网：任何意外异常都不应杀死日志线程
print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
time.sleep(0.5) # 防止异常风暴导致 CPU 满载
# 退出前 flush 剩余
if batch:
try:
self._send(batch)
except Exception as e:
print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
self._flush_done.set()
def _send(self, batch):
"""通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
不抛异常给调用方。
"""
try:
client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
except Exception as e:
print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
return
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
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()
static/log_client.py
+244 -173
View File
@@ -1,173 +1,244 @@
"""统一日志客户端 — 各服务共用（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()
"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。
设计要点：
- 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表
- 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务
- fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃
- 缓冲批量：队列满 50 条或超时 2 秒时自动 flush
- 时间戳：日志入队时即记录 ts，反映事件真实发生时间
用法：
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 超时（秒）
_QUEUE_MAXSIZE = 10000 # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
def _safe_json(obj):
"""安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。"""
try:
return json.dumps(obj, ensure_ascii=False, default=str)
except Exception:
return repr(obj)
def _now_iso():
"""生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。"""
return datetime.now(timezone.utc).isoformat()
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(maxsize=_QUEUE_MAXSIZE)
self._stop = threading.Event()
self._flush_signal = threading.Event()
self._flush_done = 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):
"""推送一条日志（非阻塞，立即返回）。
队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。
"""
entry = {
'ts': _now_iso(), # 入队时即生成时间戳，反映真实事件时间
'service': service,
'level': level,
'message': message,
'trace_id': trace_id,
'extra': extra,
}
try:
self._queue.put_nowait(entry)
except queue.Full:
# 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略）
try:
self._queue.get_nowait()
except queue.Empty:
pass
try:
self._queue.put_nowait(entry)
except queue.Full:
pass # 极端竞争下仍失败则静默丢弃
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_done.clear()
self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch
# 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量）
self._flush_done.wait(timeout=self._flush_interval + 2.0)
# 主线程排空队列中剩余条目
self._drain_queue_to_db()
def _drain_queue_to_db(self):
"""排空主线程可见的队列中剩余条目，直接发送到 db。"""
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 时发送。
外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
都不会死亡，避免日志队列无限增长导致 OOM。
"""
batch = []
last_flush = time.monotonic()
while not self._stop.is_set():
try:
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:
try:
self._send(batch)
except Exception as e:
# _send 内部已有细粒度异常处理，此处是最终安全网
print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
batch = []
last_flush = time.monotonic()
if self._flush_signal.is_set():
self._flush_signal.clear()
self._flush_done.set()
except Exception as e:
# 最外层安全网：任何意外异常都不应杀死日志线程
print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}",
file=_sys.stderr)
time.sleep(0.5) # 防止异常风暴导致 CPU 满载
# 退出前 flush 剩余
if batch:
try:
self._send(batch)
except Exception as e:
print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr)
self._flush_done.set()
def _send(self, batch):
"""通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。
整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe），
不抛异常给调用方。
"""
try:
client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log")
except Exception as e:
print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr)
return
for entry in batch:
extra_val = entry.get("extra")
try:
extra_json = _safe_json(extra_val) if extra_val is not None else None
except Exception:
extra_json = None
payload = {
"action": "insert",
"table": "Logs",
"values": {
"ts": entry.get("ts", _now_iso()),
"service": entry.get("service"),
"level": entry.get("level"),
"message": entry.get("message"),
"trace_id": entry.get("trace_id"),
"extra": extra_json,
},
}
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()