diff --git a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load.py b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load.py index 2e03ba1ce63..5b9e3271f07 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load.py @@ -336,6 +336,16 @@ def _diagnose(load_stats, worker_load, anomaly_summary, sr_result, token_stats, } ) + id_mismatch_count = sr_result.get("id_consistency", {}).get("both_present_but_mismatch", 0) + if id_mismatch_count > 0: + diagnoses.append( + { + "severity": "MEDIUM", + "message": f"{id_mismatch_count} 个 select/release 在 FIFO 命中后 ID 不一致（疑似串流或日志错配）", + "source_layer": "FD 后端", + } + ) + # Token 计数器潜在泄漏 for t in token_stats: if t.get("alloc_count", 0) > t.get("release_count", 0): diff --git a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load_report.py b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load_report.py index 9d4e9b51496..3d636afd3b6 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load_report.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load_report.py @@ -25,10 +25,14 @@ def format_load_report(result): if result["diagnoses"]: sections.append("### 诊断") sections.append("") +<<<<<<< codex/modify-troubleshoot-for-skills-alignment-z2tpws + sections.append(f' 共 {len(result["diagnoses"])} 条诊断，见详情: [detail/load_diagnoses.md](../detail/load_diagnoses.md)') +======= sections.append( f' 共 {len(result["diagnoses"])} 条诊断，见详情: [detail/load_diagnoses.md](../detail/load_diagnoses.md)；' '匹配明细见 [detail/load_select_release.md](../detail/load_select_release.md)' ) +>>>>>>> develop sections.append("") detail_sections.append("## 诊断") detail_sections.append("") @@ -185,6 +189,36 @@ def format_load_report(result): detail_sections.append("## Select/Release Per-Worker") detail_sections.append("") +<<<<<<< codex/modify-troubleshoot-for-skills-alignment-z2tpws + id_consistency = sr.get("id_consistency", {}) + if id_consistency: + sections.append("### FIFO × ID 一致性校验") + sections.append("") + sections.append( + " matched={ok}, mismatch={mismatch}, select_only={so}, release_only={ro}, both_missing={bm}".format( + ok=id_consistency.get("both_present_and_equal", 0), + mismatch=id_consistency.get("both_present_but_mismatch", 0), + so=id_consistency.get("only_select_has_id", 0), + ro=id_consistency.get("only_release_has_id", 0), + bm=id_consistency.get("both_missing", 0), + ) + ) + sections.append("") + sections.append(" 说明: 主匹配按 worker FIFO，随后检查 matched 对中的 ID 是否一致。") + sections.append("") + detail_sections.append("## FIFO × ID 一致性") + detail_sections.append("") + detail_sections.append( + "- both_present_and_equal: select/release 都有可关联 ID 且相等\n" + "- both_present_but_mismatch: select/release 都有 ID 但不一致（需要重点排查）\n" + "- only_select_has_id: 仅 select 有 ID\n" + "- only_release_has_id: 仅 release 有 ID\n" + "- both_missing: 两边都没有可关联 ID" + ) + detail_sections.append("") + +======= +>>>>>>> develop if sr.get("worker_type_profile"): sections.append("### Worker URL 类型画像（基于 select）") sections.append("") @@ -251,6 +285,19 @@ def format_load_report(result): sections.append("") detail_sections.append("## Untracked selects（缺少可关联 ID）") detail_sections.append("") + + if sr.get("id_mismatched_matches"): + sections.append(f' ⚠ {len(sr["id_mismatched_matches"])} 个 FIFO 匹配对存在 ID 不一致') + sections.append(" > 完整列表见: [detail/load_select_release.md](../detail/load_select_release.md)") + sections.append("") + detail_sections.append("## FIFO 匹配但 ID 不一致（完整）") + detail_sections.append("") + for m in sr["id_mismatched_matches"]: + detail_sections.append( + f'- [{m.get("select_ts","")}] worker={_strip_scheme(m.get("worker",""))} ' + f'select_id={m.get("select_id","")} release_id={m.get("release_id","")} note={m.get("note","")}' + ) + detail_sections.append("") for u in sr["untracked_selects"]: detail_sections.append( f'- [{u.get("select_ts","")}] worker={_strip_scheme(u["worker"])} type={u["type"]} note={u.get("note","")}' diff --git a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/log_parser.py b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/log_parser.py index 548c29ebc29..10f62fc4b51 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/log_parser.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/log_parser.py @@ -501,6 +501,15 @@ def _normalize_worker_type(worker_type): return "unknown" +<<<<<<< codex/modify-troubleshoot-for-skills-alignment-z2tpws +def _normalize_worker_url_key(url): + if not url: + return "" + return re.sub(r"^https?://", "", str(url).strip().rstrip("/")) + + +======= +>>>>>>> develop def _infer_release_worker_type(release, selects, fallback_window_s=120): """为未显式标注 type 的 release 近似推断 worker type。 @@ -596,6 +605,7 @@ def match_select_release(lines, fallback_window_s=120): { "ts": ts, "worker": tm.group(2), + "worker_key": _normalize_worker_url_key(tm.group(2)), "type": _normalize_worker_type(tm.group(1)), "tags": tags, "tokens": int(tm.group(3)), @@ -612,6 +622,10 @@ def match_select_release(lines, fallback_window_s=120): { "ts": ts, "worker": trm.group(2), +<<<<<<< codex/modify-troubleshoot-for-skills-alignment-z2tpws + "worker_key": _normalize_worker_url_key(trm.group(2)), +======= +>>>>>>> develop # 文本默认按 prefill 记，再结合同 worker 邻近 select 做纠偏（mixed 场景） "type": f'{_normalize_worker_type(token_type or "prefill")}_tokens', "raw_token_type": token_type or "", @@ -628,6 +642,7 @@ def match_select_release(lines, fallback_window_s=120): { "ts": ts, "worker": sm.group(2), + "worker_key": _normalize_worker_url_key(sm.group(2)), "type": _normalize_worker_type(sm.group(1)), "tags": tags, "tokens": None, @@ -642,6 +657,7 @@ def match_select_release(lines, fallback_window_s=120): { "ts": ts, "worker": rm.group(2), + "worker_key": _normalize_worker_url_key(rm.group(2)), "type": _normalize_worker_type(rm.group(1)), "tags": tags, "tokens": None, @@ -653,28 +669,22 @@ def match_select_release(lines, fallback_window_s=120): if FAILED_SELECT_RE.search(line): failed_selects.append({"ts": ts, "tags": tags, "line": line_no}) - # Match by request_id / alt_id + # Match by worker FIFO（select -> 同 worker 下一条 release） matched = [] unmatched_selects = [] release_used = set() # 请求生命周期匹配只使用 request counter release（排除 token release） + # 说明：request_id 只用于覆盖率观测，不参与 select/release 配对条件。 counter_release_indexes = [i for i, r in enumerate(releases) if not str(r.get("type", "")).endswith("_tokens")] - release_by_key = defaultdict(list) - for i in counter_release_indexes: - r = releases[i] - _, key = _select_match_key(r.get("tags", {})) - if key: - release_by_key[key].append(i) - # 请求 ID 覆盖（按 select 事件近似请求数） total_req_est = len(selects) with_request_id = 0 with_alt_id = 0 without_any_id = 0 - pending_selects = [] untracked_selects = [] + pending_selects = [] for s in selects: key_type, key = _select_match_key(s.get("tags", {})) if key_type == "request_id": @@ -684,7 +694,6 @@ def match_select_release(lines, fallback_window_s=120): else: without_any_id += 1 - found = False if not key: # 没有任何可用 ID 时，不做退化匹配（只统计可观测信息） untracked_selects.append( @@ -696,53 +705,74 @@ def match_select_release(lines, fallback_window_s=120): "note": "no correlatable id (request_id/req_id/trace_id/session_id)", } ) - continue - - if key and key in release_by_key: - for ri in release_by_key[key]: - if ri not in release_used: - r = releases[ri] - matched.append( - { - "request_id": s["tags"].get("request_id", ""), - "worker": s["worker"], - "select_ts": s["ts"], - "release_ts": r["ts"], - "type": s["type"], - "match_method": key_type or "id", - } - ) - release_used.add(ri) - found = True - break - - if not found: - pending_selects.append(s) + pending_selects.append(s) + + # worker FIFO + ID 一致性联合校验： + # 1) 主匹配仍按 worker FIFO，保证在缺失 request_id 场景可工作 + # 2) 对已匹配对追加 ID 一致性检查（request_id/req_id/trace_id/session_id） + id_consistency = { + "both_present_and_equal": 0, + "both_present_but_mismatch": 0, + "only_select_has_id": 0, + "only_release_has_id": 0, + "both_missing": 0, + } + id_mismatched_matches = [] - # Fallback: 有 ID 但未匹配时，按 worker + 时间邻近匹配 for s in pending_selects: - sdt = _parse_ts_safe(s["ts"]) + sdt = _parse_ts_safe(s.get("ts")) best_idx = None - best_delta = None + best_ts = None for ri in counter_release_indexes: - r = releases[ri] if ri in release_used: continue - if r.get("worker") != s.get("worker"): + r = releases[ri] + if r.get("worker_key") != s.get("worker_key"): continue rdt = _parse_ts_safe(r.get("ts")) - if sdt and rdt: - delta = (rdt - sdt).total_seconds() - if delta < 0 or delta > fallback_window_s: - continue - else: - delta = 0 - if best_delta is None or delta < best_delta: - best_delta = delta + # 优先选择时间不早于 select 的最早 release；解析失败则按出现顺序 + if sdt and rdt and rdt < sdt: + continue + if best_idx is None: + best_idx = ri + best_ts = rdt + elif rdt and best_ts and rdt < best_ts: best_idx = ri + best_ts = rdt if best_idx is not None: r = releases[best_idx] + s_key_type, s_key = _select_match_key(s.get("tags", {})) + r_key_type, r_key = _select_match_key(r.get("tags", {})) + if s_key and r_key: + if s_key == r_key: + id_check = "match" + id_consistency["both_present_and_equal"] += 1 + else: + id_check = "mismatch" + id_consistency["both_present_but_mismatch"] += 1 + id_mismatched_matches.append( + { + "worker": s["worker"], + "select_ts": s["ts"], + "release_ts": r["ts"], + "select_id_key": s_key_type, + "select_id": s_key, + "release_id_key": r_key_type, + "release_id": r_key, + "note": "worker FIFO matched, but ID mismatched", + } + ) + elif s_key and not r_key: + id_check = "select_only" + id_consistency["only_select_has_id"] += 1 + elif (not s_key) and r_key: + id_check = "release_only" + id_consistency["only_release_has_id"] += 1 + else: + id_check = "both_missing" + id_consistency["both_missing"] += 1 + matched.append( { "request_id": s["tags"].get("request_id", ""), @@ -750,7 +780,8 @@ def match_select_release(lines, fallback_window_s=120): "select_ts": s["ts"], "release_ts": r["ts"], "type": s["type"], - "match_method": "worker_time_fallback", + "match_method": "worker_fifo", + "id_check": id_check, } ) release_used.add(best_idx) @@ -761,7 +792,7 @@ def match_select_release(lines, fallback_window_s=120): "select_ts": s["ts"], "type": s["type"], "tags": s["tags"], - "note": "no matching release found (request_id/worker-time)", + "note": "no matching release found (worker FIFO)", } ) @@ -770,14 +801,16 @@ def match_select_release(lines, fallback_window_s=120): per_worker = defaultdict(lambda: {"selects": 0, "releases": 0, "token_selects": 0, "token_releases": 0}) for s in selects: s_type = _normalize_worker_type(s.get("type")) - per_worker[s["worker"]]["selects"] += 1 + wkey = s.get("worker_key") or _normalize_worker_url_key(s.get("worker")) + per_worker[wkey]["selects"] += 1 if s_type in ("prefill", "mixed"): - per_worker[s["worker"]]["token_selects"] += 1 + per_worker[wkey]["token_selects"] += 1 for r in releases: + wkey = r.get("worker_key") or _normalize_worker_url_key(r.get("worker")) if str(r.get("type", "")).endswith("_tokens"): - per_worker[r["worker"]]["token_releases"] += 1 + per_worker[wkey]["token_releases"] += 1 else: - per_worker[r["worker"]]["releases"] += 1 + per_worker[wkey]["releases"] += 1 pw_result = {} for w, counts in per_worker.items(): @@ -792,7 +825,12 @@ def match_select_release(lines, fallback_window_s=120): # 基于 select 构建 worker URL -> dominant type 映射 per_worker_type_counts = defaultdict(lambda: defaultdict(int)) for s in selects: +<<<<<<< codex/modify-troubleshoot-for-skills-alignment-z2tpws + wkey = s.get("worker_key") or _normalize_worker_url_key(s.get("worker")) + per_worker_type_counts[wkey][_normalize_worker_type(s.get("type"))] += 1 +======= per_worker_type_counts[s["worker"]][_normalize_worker_type(s.get("type"))] += 1 +>>>>>>> develop worker_dominant_type = {} for w, counts in per_worker_type_counts.items(): worker_dominant_type[w] = sorted(counts.items(), key=lambda kv: -kv[1])[0][0] if counts else "unknown" @@ -804,7 +842,11 @@ def match_select_release(lines, fallback_window_s=120): if r_type_raw.endswith("_tokens"): base_t = _normalize_worker_type(r_type_raw.replace("_tokens", "")) # token release 按 worker URL 对应的 select 类型映射，不做邻近时间纠偏 +<<<<<<< codex/modify-troubleshoot-for-skills-alignment-z2tpws + mapped_t = worker_dominant_type.get(r.get("worker_key") or _normalize_worker_url_key(r.get("worker")), "unknown") +======= mapped_t = worker_dominant_type.get(r.get("worker", ""), "unknown") +>>>>>>> develop if mapped_t in ("prefill", "decode", "mixed"): base_t = mapped_t inferred_release_types[i] = f"{base_t}_tokens" @@ -890,6 +932,8 @@ def match_select_release(lines, fallback_window_s=120): "with_alt_id": with_alt_id, "without_any_id": without_any_id, }, + "id_consistency": id_consistency, + "id_mismatched_matches": id_mismatched_matches, "type_summary": dict(type_summary), "worker_type_profile": worker_type_profile, } @@ -1099,6 +1143,17 @@ def check(name, got, expected): msr = match_select_release(sample_lines) check("mixed token_releases inferred", msr["type_summary"].get("mixed", {}).get("token_releases", 0), 1) check("prefill token_releases remains 0", msr["type_summary"].get("prefill", {}).get("token_releases", 0), 0) +<<<<<<< codex/modify-troubleshoot-for-skills-alignment-z2tpws + check("id consistency exact match", msr["id_consistency"].get("both_present_and_equal", 0), 1) + + mismatch_lines = [ + "[INFO] 2026/04/12 10:01:00 logger.go:1: [request_id:r2] select worker (decode): http://10.0.0.2:9965, count: 1", + "[INFO] 2026/04/12 10:01:01 logger.go:1: [request_id:r3] release worker: http://10.0.0.2:9965, count: 0", + ] + mm = match_select_release(mismatch_lines) + check("id mismatch detected", mm["id_consistency"].get("both_present_but_mismatch", 0), 1) +======= +>>>>>>> develop print(f'\n{"=" * 40}') print(f"Results: {passed} passed, {failed} failed") diff --git a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/troubleshoot.py b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/troubleshoot.py index 641c5106bee..803bf6fba43 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/troubleshoot.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/troubleshoot.py @@ -24,6 +24,7 @@ import os import sys from datetime import datetime +from pathlib import Path # 确保能 import 同级模块 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) @@ -46,23 +47,56 @@ def determine_log_file(user_path=None): 3. fd-router.log（golang_router 根目录） """ if user_path: - if os.path.isfile(user_path): - return user_path + p = Path(user_path).expanduser() + if p.is_file(): + return str(p) print(f"ERROR: 文件不存在: {user_path}", file=sys.stderr) + print( + "提示: 若路径含空格/括号，请使用引号，例如: " + "python3 scripts/troubleshoot.py 'fastdeploy/golang_router/logs/fd-router (2).log' --load", + file=sys.stderr, + ) sys.exit(1) - # 尝试不同 CWD 下的候选路径 - candidates = [ - "logs/router.log", # CWD = golang_router/ - "fd-router.log", # CWD = golang_router/ - "fastdeploy/golang_router/logs/router.log", # CWD = 项目根 - "fastdeploy/golang_router/fd-router.log", # CWD = 项目根 + # 统一基于脚本位置与当前工作目录搜索，避免 CWD 差异导致找不到日志。 + script_dir = Path(__file__).resolve().parent + golang_router_dir = script_dir.parents[2] # .../fastdeploy/golang_router + cwd = Path.cwd() + + # 精确候选（优先常见命名） + exact_candidates = [ + golang_router_dir / "logs" / "router.log", + golang_router_dir / "fd-router.log", + cwd / "logs" / "router.log", + cwd / "fd-router.log", + cwd / "fastdeploy" / "golang_router" / "logs" / "router.log", + cwd / "fastdeploy" / "golang_router" / "fd-router.log", ] - for path in candidates: - if os.path.isfile(path): - return path + for p in exact_candidates: + if p.is_file(): + return str(p) + + # 模糊候选：支持 fd-router (2).log 等命名 + pattern_roots = [ + golang_router_dir / "logs", + golang_router_dir, + cwd / "logs", + cwd, + cwd / "fastdeploy" / "golang_router" / "logs", + cwd / "fastdeploy" / "golang_router", + ] + dynamic_candidates = [] + for root in pattern_roots: + if not root.is_dir(): + continue + dynamic_candidates.extend(sorted(root.glob("fd-router*.log"))) + dynamic_candidates.extend(sorted(root.glob("router*.log"))) + + if dynamic_candidates: + return str(dynamic_candidates[0]) print("ERROR: 未找到日志文件。请指定路径或检查 logs/ 目录。", file=sys.stderr) + print("已搜索: logs/router.log, fd-router.log, fd-router*.log, router*.log", file=sys.stderr) sys.exit(1)