Enhance log parsing and troubleshoot analysis: robust URL handling, tail-by-minutes, improved select/release matching and reporting

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+)\)")
 
 

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}%",

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,
             }

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 后端"),

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"]}')

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 统计