diff --git a/fastdeploy/golang_router/.claude/skills/stat-cache-hitrate/scripts/log_parser.py b/fastdeploy/golang_router/.claude/skills/stat-cache-hitrate/scripts/log_parser.py index 0b7377b4865..d43d6909c64 100644 --- a/fastdeploy/golang_router/.claude/skills/stat-cache-hitrate/scripts/log_parser.py +++ b/fastdeploy/golang_router/.claude/skills/stat-cache-hitrate/scripts/log_parser.py @@ -125,7 +125,7 @@ def complete_time_arg(time_str, log_file, is_end=False): if m: mo, d = m.group(1).zfill(2), m.group(2).zfill(2) ts = _get_log_boundary_ts(log_file, "first") - year = ts[:4] if ts else "2026" + year = ts[:4] if ts else str(datetime.now().year) if m.group(3): # 有时间部分 h, mi = m.group(3).zfill(2), m.group(4) s = (m.group(5) or "00").zfill(2) @@ -139,7 +139,7 @@ def complete_time_arg(time_str, log_file, is_end=False): h, mi = m.group(1).zfill(2), m.group(2) s = (m.group(3) or "00").zfill(2) ts = _get_log_boundary_ts(log_file, "last") - date_part = ts[:10] if ts else "2026/01/01" + date_part = ts[:10] if ts else f"{datetime.now().year}/01/01" return f"{date_part} {h}:{mi}:{s}" # Fallback: 原样返回 @@ -204,9 +204,10 @@ def extract_tags(line): # Cache-Aware 策略行解析（类别 A） # ════════════════════════════════════════════════════════════════ +URL_RE = r"(?:https?://)?[A-Za-z0-9.-]+(?::\d+)?" STRATEGY_RE = re.compile(r"final strategy:\s*(\w+)") -SELECTED_RE = re.compile(r"selected=(http://\S+?)(?:,|\s|$)") -REASON_RE = re.compile(r"reason:\s*(.+?)(?:,\s*loads=|$)") +SELECTED_RE = re.compile(rf"selected=({URL_RE})(?:,|\s|$)") +REASON_RE = re.compile(r"reason:\s*(.+?)(?:,\s*loads=|\.?\s*ts_ms=|$)") def parse_cache_strategy_line(line): @@ -271,7 +272,7 @@ def parse_cache_strategy_line(line): # ════════════════════════════════════════════════════════════════ TOTAL_RUNNING_RE = re.compile(r"total_running=(\d+)") -WORKER_RUNNING_RE = re.compile(r"(http://[^:]+:\d+): running=(\d+)") +WORKER_RUNNING_RE = re.compile(rf"({URL_RE}): running=(\d+)") CACHE_HR_RE = re.compile(r"cache_hit_rate=([\d.]+)%\s*\(hits=(\d+)/total=(\d+)\)") diff --git a/fastdeploy/golang_router/.claude/skills/stat-cache-hitrate/scripts/stat_cache_hitrate.py b/fastdeploy/golang_router/.claude/skills/stat-cache-hitrate/scripts/stat_cache_hitrate.py index c193e99d47c..5487bc2cc96 100644 --- a/fastdeploy/golang_router/.claude/skills/stat-cache-hitrate/scripts/stat_cache_hitrate.py +++ b/fastdeploy/golang_router/.claude/skills/stat-cache-hitrate/scripts/stat_cache_hitrate.py @@ -14,6 +14,7 @@ import argparse import json import os +import re import subprocess import sys from collections import defaultdict @@ -32,6 +33,10 @@ ) from stats import compute_statistics, count_by, time_bucket + +def _strip_scheme(url): + return re.sub(r"^https?://", "", url) + # ════════════════════════════════════════════════════════════════ # Phase 1: 日志读取 # ════════════════════════════════════════════════════════════════ @@ -235,7 +240,7 @@ def compute_per_worker_stats(strategies): avg_hr = round(sum(data["hit_ratios"]) / len(data["hit_ratios"]), 1) if data["hit_ratios"] else 0 result.append( { - "Worker": worker.replace("http://", ""), + "Worker": _strip_scheme(worker), "Selected": data["selected_count"], "Select%": f"{round(data['selected_count'] / total_scoring * 100, 1)}%", "AvgHitRatio": f"{avg_hr}%", diff --git a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/cache.py b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/cache.py index 3a18b668a41..3fca296f4d6 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/cache.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/cache.py @@ -26,6 +26,10 @@ TOKENIZER_WARN_RE = re.compile(r"tokenizer failed, fallback to char tokens") +def _strip_scheme(url): + return re.sub(r"^https?://", "", url) + + def classify_fallback(record, tokenizer_degraded_ts=None): """对 process_tokens 策略行分类 fallback 原因。 @@ -210,9 +214,9 @@ def _analyze_suboptimal(records, hr_weight, lb_weight): suboptimal.append( { "ts": r.get("ts", ""), - "selected": selected.replace("http://", ""), + "selected": _strip_scheme(selected), "selected_hr": sel_hr, - "best_hr_worker": best_by_hr.replace("http://", ""), + "best_hr_worker": _strip_scheme(best_by_hr), "best_hr": max_hr, "reason": reason, } diff --git a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/errors.py b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/errors.py index 0817e280aa5..b8217a5ffa4 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/errors.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/errors.py @@ -44,6 +44,8 @@ ("No available", "FD 后端"), ("request failed", "FD 后端"), ("Removed unhealthy", "FD 后端"), + ("is not healthy", "FD 后端"), + ("is healthy", "FD 后端"), ("Backend request failed", "FD 后端"), ("Decode request failed", "FD 后端"), ("Prefill request failed", "FD 后端"), diff --git a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/health.py b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/health.py index d2d7ca77acb..ca01d718dbc 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/health.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/health.py @@ -21,11 +21,16 @@ # 健康事件解析 # ════════════════════════════════════════════════════════════════ -NOT_HEALTHY_RE = re.compile(r"(http://\S+)\s+is not healthy") -REMOVED_RE = re.compile(r"Removed unhealthy \w+ instance:\s*(http://\S+)") -IS_HEALTHY_RE = re.compile(r"(http://\S+)\s+is healthy") -COUNTER_PRESERVED_RE = re.compile(r"counter preserved.*?(http://\S+)") -CLEANUP_UNHEALTHY_RE = re.compile(r"cleanup unhealthy.*?(http://\S+)") +WORKER_URL_RE = r"((?:https?://)?[A-Za-z0-9.-]+(?::\d+)?)" +NOT_HEALTHY_RE = re.compile(rf"{WORKER_URL_RE}\s+is not healthy") +REMOVED_RE = re.compile(rf"Removed unhealthy \w+ instance:\s*{WORKER_URL_RE}") +IS_HEALTHY_RE = re.compile(rf"{WORKER_URL_RE}\s+is healthy") +COUNTER_PRESERVED_RE = re.compile(rf"counter preserved.*?{WORKER_URL_RE}") +CLEANUP_UNHEALTHY_RE = re.compile(rf"cleanup unhealthy.*?{WORKER_URL_RE}") + + +def _strip_scheme(url): + return re.sub(r"^https?://", "", url) def parse_health_event(line): @@ -110,7 +115,7 @@ def _build_worker_timelines(health_events, counter_events, register_events): # IP → worker URL 映射 ip_to_urls = defaultdict(set) for url in worker_urls: - ip_m = re.search(r"http://(\d+\.\d+\.\d+\.\d+)", url) + ip_m = re.search(r"(?:https?://)?(\d+\.\d+\.\d+\.\d+)", url) if ip_m: ip_to_urls[ip_m.group(1)].add(url) @@ -130,7 +135,7 @@ def _build_worker_timelines(health_events, counter_events, register_events): workers = {} for url in sorted(worker_urls): events = sorted(worker_events[url], key=lambda e: e["ts"] or "") - ip_m = re.search(r"http://(\d+\.\d+\.\d+\.\d+)", url) + ip_m = re.search(r"(?:https?://)?(\d+\.\d+\.\d+\.\d+)", url) worker_ip = ip_m.group(1) if ip_m else "" # 恢复检测：REMOVED 后有 register @@ -237,7 +242,7 @@ def _diagnose(workers): ) for url, w in workers.items(): - s = url.replace("http://", "") + s = _strip_scheme(url) if w["down_count"] > 3: diagnoses.append( { @@ -326,7 +331,7 @@ def format_health_report(result): ) table_data.append( { - "Worker": url.replace("http://", ""), + "Worker": _strip_scheme(url), "在线率": f'{w["uptime_pct"]}%', "下线次数": str(w["down_count"]), "平均下线时长": avg_down or "-", @@ -358,7 +363,7 @@ def format_health_report(result): for url, w in sorted(result["workers"].items()): if w["events"]: has_events = True - detail_parts.append(f'## {url.replace("http://", "")}') + detail_parts.append(f"## {_strip_scheme(url)}") detail_parts.append("") for evt in w["events"]: detail_parts.append(f' [{evt["ts"]}] {evt["type"]}') 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 e712011d932..9be82357494 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/load.py @@ -21,14 +21,19 @@ # Counter 异常检测正则 # ════════════════════════════════════════════════════════════════ -DOUBLE_RELEASE_RE = re.compile(r"release worker:\s*(http://\S+)\s+skipped.*?double-release") -COUNTER_CLEANED_RE = re.compile(r"release worker:\s*(http://\S+)\s+skipped.*?counter already cleaned up") -COUNTER_PRESERVED_RE = re.compile(r"counter preserved.*?(http://\S+)") -TOKEN_PRESERVED_RE = re.compile(r"token counter preserved.*?(http://\S+)") +URL_RE = r"((?:https?://)?[A-Za-z0-9.-]+(?::\d+)?)" +DOUBLE_RELEASE_RE = re.compile(rf"release worker:\s*{URL_RE}\s+skipped.*?double-release") +COUNTER_CLEANED_RE = re.compile(rf"release worker:\s*{URL_RE}\s+skipped.*?counter already cleaned up") +COUNTER_PRESERVED_RE = re.compile(rf"counter preserved.*?{URL_RE}") +TOKEN_PRESERVED_RE = re.compile(rf"token counter preserved.*?{URL_RE}") # Token 事件 -SELECT_TOKENS_RE = re.compile(r"select worker \(prefill\):\s*(http://\S+),\s*tokens:\s*(\d+)") -RELEASE_TOKENS_RE = re.compile(r"release prefill tokens:\s*(http://\S+),\s*tokens:\s*(\d+)") +SELECT_TOKENS_RE = re.compile(rf"select worker \(prefill\):\s*{URL_RE},\s*tokens:\s*(\d+)") +RELEASE_TOKENS_RE = re.compile(rf"release prefill tokens:\s*{URL_RE},\s*tokens:\s*(\d+)") + + +def _strip_scheme(url): + return re.sub(r"^https?://", "", url) def parse_counter_anomaly(line): @@ -89,7 +94,7 @@ def analyze_load(log_file, tail=None): avg = sum(vals) / len(vals) if vals else 0 worker_load.append( { - "worker": w_url.replace("http://", ""), + "worker": _strip_scheme(w_url), "avg_running": round(avg, 1), "max_running": max(vals) if vals else 0, "samples": len(vals), @@ -121,9 +126,9 @@ def analyze_load(log_file, tail=None): # Select/Release 匹配 sr_result = ( - match_select_release(h3_lines) + match_select_release(h3_lines + h11_lines) if h3_lines - else {"matched": [], "unmatched_selects": [], "failed_selects": [], "per_worker": {}} + else {"matched": [], "unmatched_selects": [], "untracked_selects": [], "failed_selects": [], "per_worker": {}} ) # Token 统计 @@ -133,7 +138,7 @@ def analyze_load(log_file, tail=None): pileup = _detect_pileup(stats_records) # 诊断 - diagnoses = _diagnose(load_stats, worker_load, anomaly_summary, sr_result, pileup) + diagnoses = _diagnose(load_stats, worker_load, anomaly_summary, sr_result, token_stats, pileup) return { "load_stats": load_stats, @@ -170,7 +175,7 @@ def _analyze_tokens(h3_lines, h11_lines): releases = token_release.get(w, []) result.append( { - "worker": w.replace("http://", ""), + "worker": _strip_scheme(w), "alloc_count": len(allocs), "alloc_avg": round(sum(allocs) / len(allocs), 0) if allocs else 0, "release_count": len(releases), @@ -195,7 +200,7 @@ def _detect_pileup(stats_records): return max_consecutive >= 5 -def _diagnose(load_stats, worker_load, anomaly_summary, sr_result, pileup): +def _diagnose(load_stats, worker_load, anomaly_summary, sr_result, token_stats, pileup): """生成负载诊断。""" diagnoses = [] @@ -236,16 +241,20 @@ def _diagnose(load_stats, worker_load, anomaly_summary, sr_result, pileup): } ) - # Select/Release 不一致 - for w_url, pw in sr_result.get("per_worker", {}).items(): - if pw.get("delta", 0) > 0: - diagnoses.append( - { - "severity": "HIGH", - "message": f'{w_url.replace("http://","")} select-release 差值 {pw["delta"]}（请求泄漏/卡住）', - "source_layer": "FD 后端", - } - ) + id_cov = sr_result.get("id_coverage", {}) + has_correlatable_ids = (id_cov.get("with_request_id", 0) + id_cov.get("with_alt_id", 0)) > 0 + + # Select/Release 不一致（仅在存在可关联 ID 时启用，避免无 ID 场景误报） + if has_correlatable_ids: + for w_url, pw in sr_result.get("per_worker", {}).items(): + if pw.get("delta", 0) > 0: + diagnoses.append( + { + "severity": "HIGH", + "message": f'{_strip_scheme(w_url)} select-release 差值 {pw["delta"]}（请求泄漏/卡住）', + "source_layer": "FD 后端", + } + ) # 卡住的请求 if sr_result.get("unmatched_selects"): @@ -257,6 +266,17 @@ def _diagnose(load_stats, worker_load, anomaly_summary, sr_result, pileup): } ) + # Token 计数器潜在泄漏 + for t in token_stats: + if t.get("alloc_count", 0) > t.get("release_count", 0): + diagnoses.append( + { + "severity": "MEDIUM", + "message": f'{t["worker"]} token alloc/release 不平衡 ({t["alloc_count"]}/{t["release_count"]})', + "source_layer": "Router", + } + ) + return diagnoses @@ -316,23 +336,44 @@ def format_load_report(result): sections.append("### 计数器异常") sections.append("") for a in result["counter_anomalies"]: - workers_str = ", ".join(f'{w.replace("http://","")}({c})' for w, c in a["workers"].items()) + workers_str = ", ".join(f'{_strip_scheme(w)}({c})' for w, c in a["workers"].items()) sections.append(f' {a["type"]}: {a["total"]} 次 [{workers_str}]') sections.append("") + id_cov = result.get("select_release", {}).get("id_coverage", {}) + if id_cov: + sections.append("### 请求标识覆盖（基于 select 近似请求数）") + sections.append("") + sections.append( + " total={total} | with_request_id={with_rid} | without_request_id={without_rid} | " + "with_alt_id={with_alt} | without_any_id={without_any}".format( + total=id_cov.get("total_requests_estimated", 0), + with_rid=id_cov.get("with_request_id", 0), + without_rid=id_cov.get("without_request_id", 0), + with_alt=id_cov.get("with_alt_id", 0), + without_any=id_cov.get("without_any_id", 0), + ) + ) + if id_cov.get("without_any_id", 0) > 0: + sections.append(" ℹ 无 request/session/trace/req_id 时，不做退化匹配，仅统计为 untracked。") + sections.append("") + # Select/Release 匹配 sr = result.get("select_release", {}) if sr.get("per_worker"): sections.append("### Select/Release 匹配") sections.append("") + id_cov = sr.get("id_coverage", {}) + no_correlatable_id = (id_cov.get("with_request_id", 0) + id_cov.get("with_alt_id", 0)) == 0 table_data = [] for w_url, pw in sorted(sr["per_worker"].items()): + delta_display = "N/A" if no_correlatable_id else str(pw["delta"]) table_data.append( { - "Worker": w_url.replace("http://", ""), + "Worker": _strip_scheme(w_url), "Select": str(pw["selects"]), "Release": str(pw["releases"]), - "Delta": str(pw["delta"]), + "Delta": delta_display, } ) sections.append( @@ -343,11 +384,20 @@ def format_load_report(result): ) ) sections.append("") + if no_correlatable_id: + sections.append(" ℹ 当前样本无可关联 ID，Delta 不用于请求泄漏结论。") + sections.append("") if sr.get("unmatched_selects"): sections.append(f' ⚠ {len(sr["unmatched_selects"])} 个未匹配 select（疑似请求卡住）') for u in sr["unmatched_selects"][:5]: - sections.append(f' [{u.get("select_ts","")}] {u["worker"].replace("http://","")} ({u["type"]})') + sections.append(f' [{u.get("select_ts","")}] {_strip_scheme(u["worker"])} ({u["type"]})') + sections.append("") + + if sr.get("untracked_selects"): + sections.append(f' ℹ {len(sr["untracked_selects"])} 个 select 缺少可关联 ID，未参与卡住判定') + for u in sr["untracked_selects"][:5]: + sections.append(f' [{u.get("select_ts","")}] {_strip_scheme(u["worker"])} ({u["type"]})') sections.append("") # Token 统计 diff --git a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/trace.py b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/trace.py index 45a5056616e..6c9a0323724 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/trace.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/analyzers/trace.py @@ -25,21 +25,26 @@ # ════════════════════════════════════════════════════════════════ PARSING_COMPLETE_RE = re.compile(r"Parsing completed.*worker selection") -SELECT_WORKER_RE = re.compile(r"select worker\s*(?:\((\w+)\))?:\s*(http://\S+)") -RELEASE_WORKER_RE = re.compile(r"release worker\s*(?:\((\w+)\))?:\s*(http://\S+)") -RELEASE_TOKENS_RE = re.compile(r"release prefill tokens:\s*(http://\S+),\s*tokens:\s*(\d+)") +URL_RE = r"((?:https?://)?[A-Za-z0-9.-]+(?::\d+)?)" +SELECT_WORKER_RE = re.compile(rf"select worker\s*(?:\((\w+)\))?:\s*{URL_RE}") +RELEASE_WORKER_RE = re.compile(rf"release worker\s*(?:\((\w+)\))?:\s*{URL_RE}") +RELEASE_TOKENS_RE = re.compile(rf"release prefill tokens:\s*{URL_RE},\s*tokens:\s*(\d+)") REQUEST_COMPLETE_RE = re.compile(r"Request completed successfully") TS_MS_RE = re.compile(r"ts_ms=(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+)") # Prefill 事件 -PREFILL_FIRST_CHUNK_RE = re.compile(r"\[prefill\] first chunk received.*?(http://\S+)") -PREFILL_DONE_RE = re.compile(r"\[prefill\] non-stream prefill response done.*?(http://\S+)") -PREFILL_ERROR_RE = re.compile(r"\[prefill\] (scanner error|copy error).*?(http://\S+)") -PREFILL_DEFER_RE = re.compile(r"\[prefill\] release in defer.*?(http://\S+)") -PREFILL_ERR_PATH_RE = re.compile(r"\[prefill\] release in CommonCompletions defer \(error path\).*?(http://\S+)") +PREFILL_FIRST_CHUNK_RE = re.compile(rf"\[prefill\] first chunk received.*?{URL_RE}") +PREFILL_DONE_RE = re.compile(rf"\[prefill\] non-stream prefill response done.*?{URL_RE}") +PREFILL_ERROR_RE = re.compile(rf"\[prefill\] (scanner error|copy error).*?{URL_RE}") +PREFILL_DEFER_RE = re.compile(rf"\[prefill\] release in defer.*?{URL_RE}") +PREFILL_ERR_PATH_RE = re.compile(rf"\[prefill\] release in CommonCompletions defer \(error path\).*?{URL_RE}") FAILED_SELECT_RE = re.compile(r"Failed to select") +def _strip_scheme(url): + return re.sub(r"^https?://", "", url) + + # ════════════════════════════════════════════════════════════════ # 主分析函数 # ════════════════════════════════════════════════════════════════ @@ -342,7 +347,7 @@ def format_trace_report(result): for evt in trace["events"]: line = f' [{evt.get("ts","")}] {evt["type"]}' if evt.get("worker"): - line += f' → {evt["worker"].replace("http://","")}' + line += f' → {_strip_scheme(evt["worker"])}' if evt.get("status"): line += f' [{evt["status"]}]' if evt.get("latency_ms"): 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 2a90d39b632..44f5cdebd94 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/log_parser.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/log_parser.py @@ -14,7 +14,7 @@ import re import sys from collections import defaultdict -from datetime import datetime +from datetime import datetime, timedelta # ════════════════════════════════════════════════════════════════ # 通用解析原语 @@ -152,7 +152,7 @@ def complete_time_arg(time_str, log_file, is_end=False): if m: mo, d = m.group(1).zfill(2), m.group(2).zfill(2) ts = _get_log_boundary_ts(log_file, "first") - year = ts[:4] if ts else "2026" + year = ts[:4] if ts else str(datetime.now().year) if m.group(3): # 有时间部分 h, mi = m.group(3).zfill(2), m.group(4) s = (m.group(5) or "00").zfill(2) @@ -166,7 +166,7 @@ def complete_time_arg(time_str, log_file, is_end=False): h, mi = m.group(1).zfill(2), m.group(2) s = (m.group(3) or "00").zfill(2) ts = _get_log_boundary_ts(log_file, "last") - date_part = ts[:10] if ts else "2026/01/01" + date_part = ts[:10] if ts else f"{datetime.now().year}/01/01" return f"{date_part} {h}:{mi}:{s}" # Fallback: 原样返回 @@ -218,6 +218,30 @@ def filter_file_by_time_range(log_file, start_str=None, end_str=None): return (tmp.name, True) +def filter_file_by_recent_minutes(log_file, minutes): + """按日志末时间戳向前过滤最近 N 分钟日志。 + + Returns: + tuple: (file_path, is_temp) — is_temp=True 时调用方负责删除 + """ + if minutes is None or minutes <= 0: + return (log_file, False) + + last_ts = _get_log_boundary_ts(log_file, "last") + if not last_ts: + return (log_file, False) + + try: + end_dt = parse_ts(last_ts) + except ValueError: + return (log_file, False) + + start_dt = end_dt - timedelta(minutes=minutes) + start_str = start_dt.strftime("%Y/%m/%d %H:%M:%S") + end_str = end_dt.strftime("%Y/%m/%d %H:%M:%S") + return filter_file_by_time_range(log_file, start_str=start_str, end_str=end_str) + + # Context tag：[session_id:...], [request_id:...], [trace_id:...], [req_id:...] TAG_RE = re.compile(r"\[(session_id|request_id|trace_id|req_id):([^\]]+)\]") @@ -228,7 +252,7 @@ def extract_tags(line): # Log level -LEVEL_RE = re.compile(r"\[(INFO|ERROR|WARN)\]") +LEVEL_RE = re.compile(r"\[(INFO|ERROR|WARN|DEBUG)\]") def extract_level(line): @@ -294,9 +318,10 @@ def parse_http_line(line, inference_only=False): # Cache-Aware 策略行解析（类别 H6） # ════════════════════════════════════════════════════════════════ +URL_RE = r"(?:https?://)?[A-Za-z0-9.-]+(?::\d+)?" STRATEGY_RE = re.compile(r"final strategy:\s*(\w+)") -SELECTED_RE = re.compile(r"selected=(http://\S+?)(?:,|\s|$)") -REASON_RE = re.compile(r"reason:\s*(.+?)(?:,\s*loads=|$)") +SELECTED_RE = re.compile(rf"selected=({URL_RE})(?:,|\s|$)") +REASON_RE = re.compile(r"reason:\s*(.+?)(?:,\s*loads=|\.?\s*ts_ms=|$)") def parse_cache_strategy_line(line): @@ -351,7 +376,7 @@ def parse_cache_strategy_line(line): # ════════════════════════════════════════════════════════════════ TOTAL_RUNNING_RE = re.compile(r"total_running=(\d+)") -WORKER_RUNNING_RE = re.compile(r"(http://[^:]+:\d+): running=(\d+)") +WORKER_RUNNING_RE = re.compile(rf"({URL_RE}): running=(\d+)") CACHE_HR_RE = re.compile(r"cache_hit_rate=([\d.]+)%\s*\(hits=(\d+)/total=(\d+)\)") @@ -438,14 +463,37 @@ def parse_error_line(line): # Select/Release 事件匹配 # ════════════════════════════════════════════════════════════════ -SELECT_RE = re.compile(r"select worker\s*(?:\((\w+)\))?:\s*(http://[^,\s]+)") -RELEASE_RE = re.compile(r"release worker\s*(?:\((\w+)\))?:\s*(http://[^,\s]+)") +SELECT_RE = re.compile(rf"select worker\s*(?:\((\w+)\))?:\s*({URL_RE})") +RELEASE_RE = re.compile(rf"release worker\s*(?:\((\w+)\))?:\s*({URL_RE})") FAILED_SELECT_RE = re.compile(r"Failed to select") -SELECT_TOKENS_RE = re.compile(r"select worker \(prefill\):\s*(http://[^,\s]+),\s*tokens:\s*(\d+)") -RELEASE_TOKENS_RE = re.compile(r"release prefill tokens:\s*(http://[^,\s]+),\s*tokens:\s*(\d+)") +SELECT_TOKENS_RE = re.compile(rf"select worker \(prefill\):\s*({URL_RE}),\s*tokens:\s*(\d+)") +RELEASE_TOKENS_RE = re.compile(rf"release prefill tokens:\s*({URL_RE}),\s*tokens:\s*(\d+)") + + +def _parse_ts_safe(ts): + if not ts: + return None + try: + return parse_ts(ts) + except ValueError: + return None + + +def _select_match_key(tags): + """构建请求关联 key，优先 request_id，其次 req_id/trace_id/session_id。""" + if not tags: + return (None, None) + rid = tags.get("request_id") + if rid: + return ("request_id", f"request_id:{rid}") + for k in ("req_id", "trace_id", "session_id"): + v = tags.get(k) + if v: + return ("alt_id", f"{k}:{v}") + return (None, None) -def match_select_release(lines): +def match_select_release(lines, fallback_window_s=120): """匹配 select/release worker 事件对。 Args: @@ -523,31 +571,60 @@ def match_select_release(lines): if FAILED_SELECT_RE.search(line): failed_selects.append({"ts": ts, "tags": tags, "line": line_no}) - # Match by request_id + # Match by request_id / alt_id matched = [] unmatched_selects = [] release_used = set() - release_by_reqid = defaultdict(list) + release_by_key = defaultdict(list) for i, r in enumerate(releases): - rid = r["tags"].get("request_id", "") - if rid: - release_by_reqid[rid].append(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 = [] for s in selects: - rid = s["tags"].get("request_id", "") + key_type, key = _select_match_key(s.get("tags", {})) + if key_type == "request_id": + with_request_id += 1 + elif key_type == "alt_id": + with_alt_id += 1 + else: + without_any_id += 1 + found = False - if rid and rid in release_by_reqid: - for ri in release_by_reqid[rid]: + if not key: + # 没有任何可用 ID 时，不做退化匹配（只统计可观测信息） + untracked_selects.append( + { + "worker": s["worker"], + "select_ts": s["ts"], + "type": s["type"], + "tags": s["tags"], + "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": rid, + "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) @@ -555,13 +632,50 @@ def match_select_release(lines): break if not found: + pending_selects.append(s) + + # Fallback: 有 ID 但未匹配时，按 worker + 时间邻近匹配 + for s in pending_selects: + sdt = _parse_ts_safe(s["ts"]) + best_idx = None + best_delta = None + for ri, r in enumerate(releases): + if ri in release_used: + continue + if r.get("worker") != s.get("worker"): + 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 + best_idx = ri + + if best_idx is not None: + r = releases[best_idx] + 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": "worker_time_fallback", + } + ) + release_used.add(best_idx) + else: unmatched_selects.append( { "worker": s["worker"], "select_ts": s["ts"], "type": s["type"], "tags": s["tags"], - "note": "no matching release found", + "note": "no matching release found (request_id/worker-time)", } ) @@ -583,8 +697,16 @@ def match_select_release(lines): return { "matched": matched, "unmatched_selects": unmatched_selects, + "untracked_selects": untracked_selects, "failed_selects": failed_selects, "per_worker": pw_result, + "id_coverage": { + "total_requests_estimated": total_req_est, + "with_request_id": with_request_id, + "without_request_id": total_req_est - with_request_id, + "with_alt_id": with_alt_id, + "without_any_id": without_any_id, + }, } diff --git a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/troubleshoot.py b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/troubleshoot.py index 4e64a2092b3..5096c5b294a 100644 --- a/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/troubleshoot.py +++ b/fastdeploy/golang_router/.claude/skills/troubleshoot/scripts/troubleshoot.py @@ -34,7 +34,7 @@ from analyzers.latency import analyze_latency, format_latency_report from analyzers.load import analyze_load, format_load_report from analyzers.trace import analyze_trace, format_trace_report -from log_parser import complete_time_arg, filter_file_by_time_range +from log_parser import complete_time_arg, filter_file_by_recent_minutes, filter_file_by_time_range def determine_log_file(user_path=None): @@ -71,10 +71,8 @@ def parse_tail_arg(tail_str): if tail_str is None: return None if tail_str.endswith("m"): - # 分钟模式：转换为大致行数（假设 ~20 行/秒） - minutes = int(tail_str[:-1]) - return minutes * 60 * 20 - return int(tail_str) + return {"type": "minutes", "value": int(tail_str[:-1])} + return {"type": "lines", "value": int(tail_str)} def determine_status(results): @@ -265,6 +263,18 @@ def main(): log_file = filtered_path print(f'时间范围过滤: {start_ts or "..."} ~ {end_ts or "..."}', file=sys.stderr) + tail_arg = parse_tail_arg(args.tail) + tail = None + # --tail Nm 采用真实时间窗口过滤，再全量分析过滤后的临时文件 + if tail_arg and tail_arg["type"] == "minutes": + filtered_path, is_temp = filter_file_by_recent_minutes(log_file, tail_arg["value"]) + if is_temp: + atexit.register(lambda p=filtered_path: os.unlink(p) if os.path.exists(p) else None) + log_file = filtered_path + print(f"--tail {tail_arg['value']}m: 使用日志时间戳过滤最近窗口", file=sys.stderr) + elif tail_arg and tail_arg["type"] == "lines": + tail = tail_arg["value"] + # 确定分析模式 any_mode = args.errors or args.latency or args.health or args.cache or args.load or args.trace run_errors = args.errors or (not any_mode) @@ -274,8 +284,6 @@ def main(): run_cache = args.cache or (not any_mode) run_trace = bool(args.trace) # trace 需要指定 ID，全量扫描不自动调用 - tail = parse_tail_arg(args.tail) - results = {} step = 0 total_steps = sum([run_errors, run_latency, run_health, run_cache, run_load, run_trace])