#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ hwdiag — GPU 服务器硬件诊断/压测一体化工具 (CLI + WebUI) 封装五类能力: 1. GPU 通道检查 nvidia-smi PCIe 链路 + lspci LnkSta + AER 错误按设备归属 2. 系统硬件错误日志 SEL(PERR/SERR 按日) / MCE / EDAC / Xid / OOM / 异常重启分析 3. 磁盘压测 Python 直接调用 fio，固定参数 20G / 300秒 / safe 4. CPU/内存压测 固定参数: CPU 80% / 内存 80% / 每阶段 600 秒 5. GPU 压测 使用 pressgpu 离线 benchmark，可选择测试显卡，默认运行 600 秒；缺失时自动安装 用法: 交互菜单 ./hwdiag.py (循环操作; 菜单[6] 浏览历次测试结果/摘要/实时跟踪) 子命令 ./hwdiag.py check gpu-link|hw-logs|all ./hwdiag.py stress cpumem ./hwdiag.py stress disk ./hwdiag.py stress gpu [--gpus 0,1|all] [--runtime SEC] ./hwdiag.py jobs 查看任务 ./hwdiag.py deps 安装/检查依赖 ./hwdiag.py stop <任务ID> 停止任务 ./hwdiag.py log <任务ID> 跟踪任务日志 WebUI ./hwdiag.py web [--port 18700] [--listen 127.0.0.1] [--token XXX] 默认只绑 127.0.0.1(SSH 隧道访问); 绑 0.0.0.0 必须带 token 部署: ./dtssh -cp hwdiag.py 1831-1840:/root/ (压测由 Python 直接调用 stress-ng/fio，只需分发这一个文件) 注意: DC 机器系统时区为 UTC, 本工具输出时间均为 UTC, +8 为北京时间。 """ import argparse import hashlib import json import os import re import shlex import shutil import signal import subprocess import sys import tempfile import threading import time import tarfile import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse, parse_qs from urllib.request import urlopen HOSTNAME = os.uname().nodename SCRIPT_DIR = Path(__file__).resolve().parent BASE_DIR = Path("/root/hwdiag") if os.geteuid() == 0 else Path.home() / ".hwdiag" JOBS_DIR = BASE_DIR / "jobs" DEFAULT_PORT = 18700 EXPECTED_GPU_WIDTH = 16 GPU_RUNTIME = 600 CPUMEM_CPU_PCT = 80 CPUMEM_MEM_PCT = 80 CPUMEM_RUNTIME = 600 DISK_SIZE = "20G" DISK_RUNTIME = 300 DISK_MODE = "safe" PRESSGPU_URL = "http://39.107.137.58/pressgpu.tar.gz" PRESSGPU_DIR = Path("/opt/pressgpu") PRESSGPU_INSTALL_LOCK = threading.Lock() C_RED = "\033[31m" C_GREEN = "\033[32m" C_YELLOW = "\033[33m" C_CYAN = "\033[36m" C_BOLD = "\033[1m" C_DIM = "\033[2m" C_OFF = "\033[0m" # ============================================================ # 基础设施 # ============================================================ def sh(cmd, timeout=40): """执行命令, 返回 (rc, stdout)。超时/缺命令返回空输出。""" try: p = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout) return p.returncode, (p.stdout or "") + (p.stderr or "") except subprocess.TimeoutExpired: return 124, "" except Exception: return 125, "" def have(cmd): return shutil.which(cmd) is not None # ============================================================ # 依赖软件 # ============================================================ DEP_GROUPS = [ ("系统/驱动基础", ["build-essential", "software-properties-common", "dkms", "ubuntu-drivers-common", "nvtop"]), ("CPU/内存压测", ["stress-ng"]), ("磁盘压测", ["fio", "smartmontools", "lvm2"]), ("GPU 压测", ["ocl-icd-libopencl1"]), ("硬件日志检查", ["ipmitool"]), ] JOB_DEPS = {"cpumem": ["stress-ng"], "disk": ["fio", "smartmontools", "lvm2"], "gpu": ["ocl-icd-libopencl1"]} def pkg_installed(pkg): rc, out = sh("dpkg-query -W -f='${Status}' " + pkg + " 2>/dev/null") return "install ok installed" in out def install_deps(pkgs=None): """安装缺失 apt 依赖, 返回 (installed, failed)。""" targets = pkgs or [p for _, ps in DEP_GROUPS for p in ps] todo = [p for p in targets if not pkg_installed(p)] if not todo: return [], [] rc, out = sh("apt-get install -y " + " ".join(todo), timeout=900) if rc != 0: sh("apt-get update -qq", timeout=600) rc, out = sh("apt-get install -y " + " ".join(todo), timeout=900) failed = [p for p in todo if not pkg_installed(p)] return [p for p in todo if p not in failed], failed def ensure_job_deps(kind): """压测任务启动前自动补装所需依赖, 失败则退出。""" inst, fail = install_deps(JOB_DEPS.get(kind, [])) if inst: print(f"已自动安装依赖: {' '.join(inst)}") if fail: sys.exit("依赖安装失败: " + " ".join(fail) + " (apt 源/网络问题, 可手跑: hwdiag.py deps)") def now_utc(): return time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime()) def strip_ansi(s): return re.sub(r"\x1b\[[0-9;]*m", "", s) def fmt_dur(sec): """秒 → '1h05m' / '5m12s' / '45s'""" sec = int(sec) h, m, s = sec // 3600, sec % 3600 // 60, sec % 60 if h: return f"{h}h{m:02d}m" return f"{m}m{s:02d}s" if m else f"{s}s" def save_check(kind, text): """检查类结果存档 → BASE_DIR/checks/_.txt (菜单[6]可回看)""" d = BASE_DIR / "checks" d.mkdir(parents=True, exist_ok=True) f = d / (time.strftime("%Y%m%d_%H%M%S") + "_" + kind + ".txt") f.write_text(strip_ansi(text).rstrip() + "\n") return f def hr(title=""): if title: print(f"\n{C_BOLD}{'=' * 62}{C_OFF}") print(f"{C_BOLD} {title}{C_OFF}") print(f"{C_BOLD}{'=' * 62}{C_OFF}") else: print("=" * 62) # ============================================================ # 检查 1: GPU 通道 (PCIe 链路 + AER 归属) # ============================================================ def bus_key(b): """归一化 PCIe 地址 → '61:00.0' (丢 domain; nvidia-smi 8位域 / AER 4位域 / lspci 无域 统一)""" return ":".join(b.lower().split(":")[-2:]) def get_gpus(): rc, out = sh("nvidia-smi --query-gpu=index,uuid,name,pci.bus_id,utilization.gpu," "memory.used,temperature.gpu,pcie.link.gen.current,pcie.link.gen.max," "pcie.link.width.current,pcie.link.width.max --format=csv,noheader,nounits") gpus = [] if rc != 0 or not out.strip(): return gpus for line in out.strip().splitlines(): f = [x.strip() for x in line.split(",")] if len(f) < 11: continue try: gpus.append({ "index": int(f[0]), "uuid": f[1], "name": f[2], "bus": bus_key(f[3]), "util": int(f[4]), "mem": int(f[5]), "temp": int(f[6]), "gen_cur": int(f[7]), "gen_max": int(f[8]), "w_cur": int(f[9]), "w_max": int(f[10]), }) except ValueError: continue return gpus def get_lnksta(bus): """lspci LnkSta: (speed, downgraded, width) """ rc, out = sh(f"lspci -vvs {bus} 2>/dev/null | grep -m1 'LnkSta:'") m = re.search(r"Speed (\S+)\s*(\(downgraded\))?", out) w = re.search(r"Width x(\d+)", out) if not m: return None return {"speed": m.group(1), "downgraded": bool(m.group(2)), "width": w.group(1) if w else "?"} def get_aer_counts(since=None): """内核日志 AER 按设备计数。since=None 只统计当前 boot (快)。""" scope = "-b 0" if since is None else f"--since '{since}'" if have("journalctl"): rc, out = sh(f"journalctl -k {scope} --no-pager 2>/dev/null | grep -E 'AER:'") else: rc, out = sh("dmesg 2>/dev/null | grep -E 'AER:'") counts = {} for dev in re.findall(r"([0-9A-Fa-f]{4}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}\.[0-9A-Fa-f])\s*:?\s*AER", out): counts[dev.lower()] = counts.get(dev.lower(), 0) + 1 return counts def check_gpu_link(since=None, as_text=True): gpus = get_gpus() if not gpus: msg = "nvidia-smi 不可用或无 GPU" return msg if as_text else {"error": msg} aer = {bus_key(d): c for d, c in get_aer_counts(since).items()} gpu_bus_parts = {g["bus"].split(".")[0] for g in gpus} # '61:00' (丢功能号, 同卡音频算同卡) state_file = BASE_DIR / "gpu_link_state.json" previous = {} if as_text and state_file.exists(): try: old = json.loads(state_file.read_text()) previous = {g["uuid"]: g for g in old.get("gpus", []) if g.get("uuid")} except Exception: previous = {} lines = [] data = [] issues = [] restored = [] snapshot = [] lines.append(f"GPU 通道检查 @ {HOSTNAME} {now_utc()}" f" {C_DIM}(AER 范围: {'当前boot' if since is None else since}){C_OFF}") lines.append(f"{'GPU':<4}{'PCIe地址':<12}{'当前链路':<20}{'正常宽度':<10}{'利用率':<8}{'AER':<6}判定") lines.append("-" * 86) worst = 0 # 0=ok 1=观察 2=异常 for g in gpus: lnk = get_lnksta(g["bus"]) current_width = g["w_cur"] speed_down = False if lnk: if lnk["width"] != "?": current_width = int(lnk["width"]) link_txt = f"{lnk['speed']} x{current_width}" if lnk["downgraded"]: speed_down = True link_txt += " 速率↓" else: link_txt = f"Gen{g['gen_cur']} x{current_width}" width_loss = current_width < EXPECTED_GPU_WIDTH n_aer = sum(c for d, c in aer.items() if d.split(".")[0] == g["bus"].split(".")[0]) old = previous.get(g["uuid"], {}) old_width = old.get("width") if width_loss: verdict = f"异常: x{current_width}，应x{EXPECTED_GPU_WIDTH}" color = C_RED worst = 2 issues.append({"gpu": g, "width": current_width, "old_width": old_width}) elif old_width is not None and int(old_width) < EXPECTED_GPU_WIDTH: verdict, color = "已恢复", C_GREEN restored.append({"gpu": g, "old_width": int(old_width), "width": current_width}) elif n_aer >= 50: verdict, color = "异常: AER过多", C_RED worst = 2 elif n_aer > 0 or speed_down: verdict, color = "观察", C_YELLOW worst = max(worst, 1) else: verdict, color = "正常", C_GREEN lines.append(f"{g['index']:<4}{g['bus']:<12}{link_txt:<20}{('x' + str(EXPECTED_GPU_WIDTH)):<10}" f"{str(g['util']) + '%':<8}{n_aer:<6}{color}{verdict}{C_OFF}") data.append({"index": g["index"], "bus": g["bus"], "name": g["name"], "util": g["util"], "temp": g["temp"], "link": link_txt, "downgraded": speed_down or width_loss, "width": current_width, "expected_width": EXPECTED_GPU_WIDTH, "aer": n_aer, "verdict": verdict}) snapshot.append({"uuid": g["uuid"], "index": g["index"], "bus": g["bus"], "width": current_width, "checked_at": now_utc()}) if issues: lines.append(f"\n{C_RED}{C_BOLD}发现 {len(issues)} 张显卡通道宽度不正常，需要重新处理：{C_OFF}") for item in issues: g, width = item["gpu"], item["width"] state = ("重新处理后仍异常" if item["old_width"] is not None and int(item["old_width"]) < EXPECTED_GPU_WIDTH else "通道异常") pct = int(width * 100 / EXPECTED_GPU_WIDTH) lines.append(f" {C_RED}[{state}] GPU {g['index']} / {g['bus']}: 当前 x{width}，" f"正常应为 x{EXPECTED_GPU_WIDTH}，通道宽度约为正常的 {pct}%{C_OFF}") lines.append("\n处理方法:") lines.append(" 1. 正常关机并彻底断开电源") lines.append(" 2. 重新插拔对应显卡，检查 PCIe 插槽、转接板和连接线") lines.append(" 3. 开机后再次运行 GPU 通道检查；恢复到 x16 才算处理完成") lines.append(" 4. 如果仍为 x8，交换显卡/槽位，判断问题跟随显卡还是留在通道") if restored: lines.append(f"\n{C_GREEN}{C_BOLD}处理后复检：{len(restored)} 张显卡通道已经恢复正常：{C_OFF}") for item in restored: g = item["gpu"] lines.append(f" {C_GREEN}[已修复] GPU {g['index']} / {g['bus']}: " f"x{item['old_width']} → x{item['width']}{C_OFF}") lines.append(" 建议继续运行 pressgpu 离线压测，再检查是否新增 AER/Xid 错误。") other = {d: c for d, c in aer.items() if d.split(".")[0] not in gpu_bus_parts} if other: lines.append(f"\n{C_YELLOW}非 GPU 设备也有 AER (可能为根端口/桥): {other}{C_OFF}") if all(g["util"] < 5 for g in gpus): lines.append(C_DIM + "注: GPU 空闲时链路降速属常见节能/BIOS 设置; " "但通道宽度 x8（本机标准 x16）仍按异常提示。" + C_OFF) if as_text: try: BASE_DIR.mkdir(parents=True, exist_ok=True) state_file.write_text(json.dumps({"host": HOSTNAME, "gpus": snapshot}, ensure_ascii=False, indent=2)) except OSError: pass text = "\n".join(lines) return text if as_text else {"gpus": data, "other_aer": other, "worst": worst} # ============================================================ # 检查 2: 系统硬件错误日志 # ============================================================ def check_hw_logs(days=3, as_text=True): out = {"sections": []} def sec(title, body, level="info"): out["sections"].append({"title": title, "body": body, "level": level}) # --- SEL 按日统计 --- sel_body = [] sel_events = [] if have("ipmitool"): rc, sel = sh("ipmitool sel list 2>/dev/null") sel_events = [l for l in sel.splitlines() if "|" in l] by_day = {} notable = [] for l in sel_events: parts = [p.strip() for p in l.split("|")] if len(parts) < 5: continue day, desc = parts[1], f"{parts[3]} {parts[4]}" key = ("PERR" if "PERR" in desc.upper() else "SERR" if "SERR" in desc.upper() else None) if key: by_day.setdefault(day, {}).setdefault(key, 0) by_day[day][key] += 1 elif not re.search(r"Unknown|Timestamp Clock Sync|Pre-Init", l): notable.append(l) if by_day: sel_body.append("PCIe 错误按日 (PERR/SERR):") for day in sorted(by_day): d = by_day[day] total = d.get("PERR", 0) + d.get("SERR", 0) mark = C_RED + " ⚠爆发" + C_OFF if total > 100 else "" sel_body.append(f" {day} PERR={d.get('PERR', 0):<5} SERR={d.get('SERR', 0):<5}{mark}") else: sel_body.append("SEL 中无 PERR/SERR 记录") if notable: sel_body.append("\n其他值得注意的 SEL 事件 (最近 8 条):") sel_body += [f" {l}" for l in notable[-8:]] else: sel_body.append("无 ipmitool") sec("BMC SEL (时间 UTC)", "\n".join(sel_body), "warn" if any("PERR" in str(b) for b in sel_body) else "info") # --- 内核硬件错误 / MCE / Xid / EDAC / OOM (近 N 天) --- since = f"-{days * 24}h" rc, klog = sh(f"journalctl -k --since '{since}' --no-pager 2>/dev/null") hw_err = {} for l in klog.splitlines(): if "[Hardware Error]" in l or "Machine check" in l: for dev in re.findall(r"([0-9A-Fa-f]{4}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}\.\d)", l): hw_err[dev.lower()] = hw_err.get(dev.lower(), 0) + 1 xids = {} for l in klog.splitlines(): m = re.search(r"NVRM: Xid (\d+)", l) if m: xids[m.group(1)] = xids.get(m.group(1), 0) + 1 edac = [] for mc in sorted(Path("/sys/devices/system/edac/mc").glob("mc*")): try: ce = (mc / "ce_count").read_text().strip() ue = (mc / "ue_count").read_text().strip() edac.append(f"{mc.name}: CE={ce} UE={ue}") except Exception: pass ooms = [l for l in klog.splitlines() if re.search(r"oom-kill|Out of memory", l)][:5] kern_lines = [f"范围: 近 {days} 天 (内核日志, {len(klog.splitlines())} 行)"] if hw_err: kern_lines.append("Hardware Error / MCE 按设备:") bus2gpu = {g["bus"]: g["index"] for g in get_gpus()} for d, c in sorted(hw_err.items(), key=lambda x: -x[1]): k = bus_key(d) tag = f" (GPU {bus2gpu[k]})" if k in bus2gpu else "" kern_lines.append(f" {k:<12} {c} 条{tag}") else: kern_lines.append("Hardware Error / MCE: 无") kern_lines.append("NVRM Xid: " + (", ".join(f"Xid{k}×{v}" for k, v in xids.items()) if xids else "无")) kern_lines.append("EDAC 内存: " + ("; ".join(edac) if edac else "无 EDAC 计数")) if ooms: kern_lines.append("OOM 事件 (最多5条):") kern_lines += [f" {ooms[i][:150]}" for i in range(len(ooms))] level = "error" if hw_err or xids else "info" sec("内核硬件错误汇总", "\n".join(kern_lines), level) # --- 重启分析: 有 reboot 无 shutdown = 异常复位 --- rc, rb = sh("last -x reboot 2>/dev/null | head -12") rc2, sd = sh("last -x shutdown 2>/dev/null | head -4") reboots = [l for l in rb.splitlines() if "reboot" in l and len(l.split()) > 5] shutdowns = [l for l in sd.splitlines() if "shutdown" in l] rb_body = [f"近列表内重启 {len(reboots)} 次 / 正常关机记录 {len(shutdowns)} 条"] rb_body += [f" {l[:90]}" for l in reboots[:8]] if len(reboots) >= 4 and len(shutdowns) < len(reboots) // 2: rb_body.append(C_RED + "大量 reboot 且几乎无 shutdown 记录 → 多为异常硬复位(挂死/掉电)" + C_OFF) sec(f"重启历史 (UTC, 近{days}天)", "\n".join(rb_body), "error" if len(reboots) >= 4 and len(shutdowns) < len(reboots) // 2 else "info") if not as_text: return out buf = [f"系统硬件错误日志 @ {HOSTNAME} {now_utc()}"] for s in out["sections"]: hr_flag = {"error": C_RED, "warn": C_YELLOW, "info": C_CYAN}[s["level"]] buf.append(f"\n{hr_flag}{C_BOLD}■ {s['title']}{C_OFF}") buf.append(s["body"]) return "\n".join(buf) # ============================================================ # 任务框架 (后台压测) # ============================================================ class Job: """一个后台任务: 独立进程组 + 日志文件 + meta.json (hwdiag 退出后任务继续)""" def __init__(self, jtype, label, cmd, env=None): self.id = time.strftime("%Y%m%d_%H%M%S") + "_" + uuid.uuid4().hex[:4] self.type, self.label, self.cmd = jtype, label, cmd self.dir = JOBS_DIR / self.id self.dir.mkdir(parents=True, exist_ok=True) self.logfile = self.dir / "run.log" e = dict(os.environ) if env: e.update(env) with open(self.logfile, "w") as f: self.proc = subprocess.Popen(["bash", "-c", cmd], stdout=f, stderr=f, cwd=str(self.dir), env=e, start_new_session=True) self.pid = self.proc.pid self.started = time.time() self._save() def _save(self): meta = {"id": self.id, "type": self.type, "label": self.label, "cmd": self.cmd, "pid": self.pid, "started": self.started, "cwd": str(self.dir)} if "ended" in self.__dict__: meta["ended"] = self.ended (self.dir / "meta.json").write_text(json.dumps(meta, ensure_ascii=False, indent=1)) def alive(self): # 本进程启动的任务用 poll()(waitpid 收尸, 否则自然退出后变僵尸, kill(0) 恒真); # 重启后恢复的历史任务 proc=None, 用 kill(0) 判断 (孤儿已由 init 收养) if self.proc is not None: if self.proc.poll() is None: return True if "ended" not in self.__dict__: # 首次发现退出 → 冻结耗时 self.ended = time.time() self._save() return False try: os.kill(self.pid, 0) return True except OSError: return False def stop(self): if not self.alive(): return False try: os.killpg(self.pid, signal.SIGTERM) for _ in range(15): if not self.alive(): return True time.sleep(0.2) os.killpg(self.pid, signal.SIGKILL) return True except OSError: return False def tail(self, n=200): try: return "\n".join(self.logfile.read_text(errors="replace").splitlines()[-n:]) except Exception: return "(无日志)" def summary_file(self): """压测脚本在任务目录子目录里生成的 summary.txt (磁盘/CPU内存压测都有)""" hits = sorted(self.dir.glob("*/summary.txt")) return hits[-1] if hits else None def summary(self): f = self.summary_file() try: return f.read_text(errors="replace").rstrip() if f else None except Exception: return None def info(self): running = self.alive() # 已结束且有 ended 记录 → 耗时冻结在实际运行时长; 历史旧任务无记录则保持至今 end = self.ended if (not running and "ended" in self.__dict__) else time.time() return {"id": self.id, "type": self.type, "label": self.label, "cmd": self.cmd, "pid": self.pid, "started": time.strftime("%m-%d %H:%M:%S", time.gmtime(self.started)), "elapsed": int(end - self.started), "running": running} def load_jobs(): jobs = [] if JOBS_DIR.exists(): for d in sorted(JOBS_DIR.iterdir()): meta = d / "meta.json" if meta.exists(): try: j = Job.__new__(Job) j.__dict__.update(json.loads(meta.read_text())) j.dir, j.logfile, j.proc = d, d / "run.log", None # CLI 启动的任务父进程早已退出, 无 ended 记录 → 用日志最后写入时间近似 if "ended" not in j.__dict__ and not j.alive() and j.logfile.exists(): j.ended = j.logfile.stat().st_mtime jobs.append(j) except Exception: continue return jobs def start_job(jtype, label, cmd, env=None): JOBS_DIR.mkdir(parents=True, exist_ok=True) return Job(jtype, label, cmd, env=env) def _worker_result_dir(prefix): path = Path.cwd() / f"{prefix}_{HOSTNAME}_{time.strftime('%Y%m%d_%H%M%S')}" path.mkdir(parents=True, exist_ok=False) return path def _worker_log(logfile, message): line = str(message) print(line, flush=True) with open(logfile, "a") as f: f.write(line + "\n") def _capture_command(argv, output, timeout=120): """运行只读采集命令，命令缺失/超时也写入文件并返回非零值。""" output = Path(output) with open(output, "w") as f: f.write("$ " + shlex.join([str(x) for x in argv]) + "\n") f.flush() try: return subprocess.run([str(x) for x in argv], stdout=f, stderr=subprocess.STDOUT, timeout=timeout).returncode except subprocess.TimeoutExpired: f.write(f"命令超时 ({timeout}s)\n") return 124 except OSError as e: f.write(f"命令失败: {e}\n") return 125 def _meminfo(): values = {} try: for line in Path("/proc/meminfo").read_text().splitlines(): key, _, rest = line.partition(":") if rest: values[key] = int(rest.strip().split()[0]) except (OSError, ValueError): pass return values def _available_cpus(): try: return len(os.sched_getaffinity(0)) except (AttributeError, OSError): return os.cpu_count() or 1 def _cpu_stress_command(nproc): return ["stress-ng", "--cpu", str(nproc), "--cpu-load", str(CPUMEM_CPU_PCT), "--timeout", f"{CPUMEM_RUNTIME}s", "--metrics-brief"] def _memory_stress_command(alloc_kb): jobs = min(16, max(1, _available_cpus())) return ["stress-ng", "--vm", str(jobs), "--vm-bytes", f"{alloc_kb}K", "--vm-method", "write64", "--vm-keep", "--timeout", f"{CPUMEM_RUNTIME}s", "--metrics-brief"] def _kernel_error_delta(before_file, after_file): pattern = re.compile( r"out of memory|oom-kill|machine check|hardware error|edac|ecc|" r"page allocation failure|segfault|killed process|nvrm.*xid|aer:|i/o error|timeout|reset", re.I, ) try: before = set(Path(before_file).read_text(errors="replace").splitlines()) after = Path(after_file).read_text(errors="replace").splitlines() return [line for line in after if line not in before and pattern.search(line)] except OSError: return [] def run_cpumem_worker(): result = _worker_result_dir("cpu_mem_stress") runlog = result / "run.log" summary = result / "summary.txt" _worker_log(runlog, f"CPU/内存压测: CPU={CPUMEM_CPU_PCT}% MEM={CPUMEM_MEM_PCT}% " f"每阶段={CPUMEM_RUNTIME}s") _capture_command(["lscpu"], result / "lscpu.log") _capture_command(["free", "-h"], result / "free_before.log") if have("sensors"): _capture_command(["sensors"], result / "sensors_before.log") _capture_command(["dmesg", "-T"], result / "dmesg_before.log") nproc = _available_cpus() cpu_cmd = _cpu_stress_command(nproc) _worker_log(runlog, "开始 CPU 压测: " + shlex.join(cpu_cmd)) with open(result / "stress_cpu.log", "w") as out: cpu_rc = subprocess.run(cpu_cmd, stdout=out, stderr=subprocess.STDOUT).returncode _worker_log(runlog, f"CPU 阶段完成，退出码 {cpu_rc}") time.sleep(15) mi = _meminfo() total_kb = mi.get("MemTotal", 0) avail_kb = mi.get("MemAvailable", 0) used_kb = max(0, total_kb - avail_kb) guard_kb = max(1048576, total_kb * 2 // 100) safety_kb = max(guard_kb, total_kb * 5 // 100) target_kb = total_kb * CPUMEM_MEM_PCT // 100 alloc_kb = min(max(0, target_kb - used_kb), max(0, avail_kb - safety_kb)) mem_rc, watchdog, mem_skipped = None, False, False if alloc_kb < 1048576: mem_skipped = True _worker_log(runlog, "安全可分配内存不足 1GiB，跳过内存阶段") else: mem_cmd = _memory_stress_command(alloc_kb) _worker_log(runlog, f"开始内存压测，分配约 {alloc_kb // 1048576}GiB: " + shlex.join(mem_cmd)) with open(result / "stress_mem.log", "w") as out: proc = subprocess.Popen(mem_cmd, stdout=out, stderr=subprocess.STDOUT) while proc.poll() is None: time.sleep(5) current = _meminfo().get("MemAvailable", 0) if current and current < guard_kb: watchdog = True _worker_log(runlog, f"内存 watchdog 触发: MemAvailable={current}KiB，终止压测") proc.terminate() try: proc.wait(timeout=15) except subprocess.TimeoutExpired: proc.kill() break mem_rc = proc.wait() _worker_log(runlog, f"内存阶段完成，退出码 {mem_rc}") _capture_command(["free", "-h"], result / "free_after.log") if have("sensors"): _capture_command(["sensors"], result / "sensors_after.log") _capture_command(["dmesg", "-T"], result / "dmesg_after.log") errors = _kernel_error_delta(result / "dmesg_before.log", result / "dmesg_after.log") if cpu_rc != 0 or watchdog or (mem_rc not in (None, 0)) or errors: verdict = "FAIL" elif mem_skipped: verdict = "UNKNOWN" else: verdict = "PASS" lines = [ "=" * 62, "CPU/内存压测摘要", "=" * 62, f"结论: {verdict}", f"CPU: {CPUMEM_CPU_PCT}% × {nproc} CPUs × {CPUMEM_RUNTIME}s，rc={cpu_rc}", f"内存: 目标 {CPUMEM_MEM_PCT}%，分配 {alloc_kb // 1048576}GiB × {CPUMEM_RUNTIME}s，rc={mem_rc}", f"内存 watchdog: {'触发' if watchdog else '未触发'}", "", "测试期间新增内核异常:", ] lines += errors[-30:] if errors else ["无"] summary.write_text("\n".join(lines) + "\n") print(summary.read_text(), end="", flush=True) return 0 if verdict == "PASS" else (2 if verdict == "UNKNOWN" else 1) def _disk_fio_commands(test_file, result): common = ["fio", "--filename=" + str(test_file), "--size=" + DISK_SIZE, "--ioengine=libaio", "--direct=1", "--runtime=" + str(DISK_RUNTIME), "--time_based", "--group_reporting", "--eta=never"] return [ ("顺序写", common + ["--name=seq_write", "--rw=write", "--bs=1M", "--iodepth=32", "--numjobs=1", "--output=" + str(result / "fio_01_seq_write.log")]), ("顺序读", common + ["--name=seq_read", "--rw=read", "--bs=1M", "--iodepth=32", "--numjobs=1", "--output=" + str(result / "fio_02_seq_read.log")]), ("随机混合读写", common + ["--name=randrw", "--rw=randrw", "--rwmixread=70", "--bs=4k", "--iodepth=32", "--numjobs=2", "--output=" + str(result / "fio_03_randrw.log")]), ] def _disk_devices(): try: p = subprocess.run(["lsblk", "-dn", "-o", "NAME,TYPE"], capture_output=True, text=True, timeout=30) return ["/dev/" + line.split()[0] for line in p.stdout.splitlines() if len(line.split()) >= 2 and line.split()[1] == "disk"] except (OSError, subprocess.TimeoutExpired): return [] def _capture_smart(result, suffix): if not have("smartctl"): return for dev in _disk_devices(): safe = Path(dev).name _capture_command(["smartctl", "-a", dev], result / f"smart_{suffix}_{safe}.log") def _smart_summary(result): pattern = re.compile( r"SMART overall-health|SMART Health Status|Critical Warning|Available Spare|" r"Percentage Used|Media and Data Integrity Errors|Error Information Log Entries|" r"Reallocated_Sector_Ct|Current_Pending_Sector|Offline_Uncorrectable|" r"UDMA_CRC_Error_Count|Unsafe Shutdowns", re.I, ) lines = [] for path in sorted(result.glob("smart_after_*.log")): try: for line in path.read_text(errors="replace").splitlines(): if pattern.search(line): lines.append(f"{path.name}: {line.strip()}") except OSError: pass return lines def run_disk_worker(): result = _worker_result_dir("disk_stress") runlog = result / "run.log" summary = result / "summary.txt" _worker_log(runlog, f"磁盘压测固定参数: {DISK_SIZE} / 每项 {DISK_RUNTIME}s / {DISK_MODE}") if os.geteuid() != 0: summary.write_text("结论: UNKNOWN\n磁盘压测需要 root 权限。\n") print(summary.read_text(), end="") return 2 _capture_command(["lsblk", "-o", "NAME,TYPE,SIZE,FSTYPE,MOUNTPOINTS,MODEL,SERIAL"], result / "lsblk.log") _capture_command(["df", "-hT"], result / "df_before.log") _capture_command(["findmnt", "-rno", "TARGET,SOURCE,FSTYPE,OPTIONS"], result / "findmnt.log") _capture_smart(result, "before") _capture_command(["dmesg", "-T"], result / "dmesg_before.log") size_bytes = 20 * 1024 ** 3 reserve_bytes = 5 * 1024 ** 3 free_bytes = shutil.disk_usage("/").free if free_bytes < size_bytes + reserve_bytes: summary.write_text(f"结论: UNKNOWN\n根分区可用 {free_bytes // 1024**3}GiB，" f"运行 20G 测试至少需要保留 25GiB。\n") print(summary.read_text(), end="") return 2 test_dir = Path(f"/.hwdiag_disk_stress_{os.getpid()}") test_file = test_dir / "fio_test_file" test_dir.mkdir(mode=0o700) results = [] try: for label, cmd in _disk_fio_commands(test_file, result): _worker_log(runlog, f"开始{label}: " + shlex.join(cmd)) try: rc = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT).returncode except OSError as e: _worker_log(runlog, f"{label}启动失败: {e}") rc = 125 results.append((label, rc)) _worker_log(runlog, f"{label}完成，退出码 {rc}") finally: try: test_file.unlink(missing_ok=True) test_dir.rmdir() except OSError as e: _worker_log(runlog, f"测试文件清理警告: {e}") _capture_smart(result, "after") _capture_command(["dmesg", "-T"], result / "dmesg_after.log") _capture_command(["df", "-hT"], result / "df_after.log") errors = _kernel_error_delta(result / "dmesg_before.log", result / "dmesg_after.log") fio_lines = [] for path in sorted(result.glob("fio_*.log")): try: for line in path.read_text(errors="replace").splitlines(): if re.search(r"READ:|WRITE:|read: IOPS|write: IOPS|err=", line): fio_lines.append(f"{path.name}: {line.strip()}") except OSError: pass smart_lines = _smart_summary(result) verdict = "FAIL" if any(rc != 0 for _, rc in results) or errors else "PASS" lines = ["=" * 62, "磁盘压测摘要", "=" * 62, f"结论: {verdict}", f"参数: {DISK_SIZE} / 每项 {DISK_RUNTIME}s / {DISK_MODE}", "阶段: " + ", ".join(f"{label}=rc{rc}" for label, rc in results), "", "fio 关键结果:"] lines += fio_lines[-40:] if fio_lines else ["未提取到结果"] lines += ["", "SMART 关键结果:"] lines += smart_lines[-40:] if smart_lines else ["未获取到 SMART 结果"] lines += ["", "测试期间新增内核异常:"] lines += errors[-30:] if errors else ["无"] summary.write_text("\n".join(lines) + "\n") print(summary.read_text(), end="", flush=True) return 0 if verdict == "PASS" else 1 def run_internal_worker(kind): def stop_worker(_signum, _frame): raise KeyboardInterrupt old_term = signal.signal(signal.SIGTERM, stop_worker) old_int = signal.signal(signal.SIGINT, stop_worker) try: return run_cpumem_worker() if kind == "cpumem" else run_disk_worker() except KeyboardInterrupt: print("压测收到终止信号，正在退出。", flush=True) return 130 except Exception as e: print(f"压测执行失败: {e}", flush=True) return 1 finally: signal.signal(signal.SIGTERM, old_term) signal.signal(signal.SIGINT, old_int) # ---- 三类压测启动器 ---- def start_cpumem(): """使用统一参数运行 CPU/内存压测，避免用户误填危险参数。""" ensure_job_deps("cpumem") cmd = [sys.executable, str(Path(__file__).resolve()), "_worker", "cpumem"] return start_job("cpumem", f"CPU{CPUMEM_CPU_PCT}%/MEM{CPUMEM_MEM_PCT}% 每阶段{CPUMEM_RUNTIME}s", shlex.join(cmd)) def start_disk(): """使用统一的 20G/300s/safe 参数运行根分区文件级压测。""" ensure_job_deps("disk") cmd = [sys.executable, str(Path(__file__).resolve()), "_worker", "disk"] return start_job("disk", f"磁盘 {DISK_SIZE} {DISK_RUNTIME}s {DISK_MODE}", shlex.join(cmd)) def find_pressgpu(): binary = PRESSGPU_DIR / "wildrig-multi" return str(binary) if binary.is_file() and os.access(binary, os.X_OK) else None def _safe_extract_pressgpu(archive, destination): """只允许普通文件/目录，拒绝路径穿越、链接、设备文件和异常大归档。""" with tarfile.open(archive, "r:gz") as tar: members = tar.getmembers() total = 0 for member in members: path = Path(member.name) if path.is_absolute() or ".." in path.parts: raise RuntimeError(f"压缩包包含不安全路径: {member.name}") if not (member.isfile() or member.isdir()): raise RuntimeError(f"压缩包包含不允许的文件类型: {member.name}") total += max(0, member.size) if total > 256 * 1024 * 1024: raise RuntimeError("pressgpu 解压后体积异常，拒绝安装") tar.extractall(destination, members=members) def install_pressgpu(): """从内网固定包下载安装 pressgpu，并在写入 /opt 前校验 SHA256。""" with PRESSGPU_INSTALL_LOCK: existing = find_pressgpu() if existing: return existing if os.geteuid() != 0: print("pressgpu 自动安装需要 root 权限") return None BASE_DIR.mkdir(parents=True, exist_ok=True) log_path = BASE_DIR / "pressgpu_install.log" log_path.touch(mode=0o600, exist_ok=True) log_path.chmod(0o600) tmp = Path(tempfile.mkdtemp(prefix="hwdiag_pressgpu_")) archive = tmp / "pressgpu.tar.gz" unpack = tmp / "unpack" unpack.mkdir() print(f"未找到 pressgpu，正在从 {PRESSGPU_URL} 自动安装，日志: {log_path}") try: digest = hashlib.sha256() size = 0 with urlopen(PRESSGPU_URL, timeout=180) as response, open(archive, "wb") as out: while True: chunk = response.read(1024 * 1024) if not chunk: break size += len(chunk) if size > 256 * 1024 * 1024: raise RuntimeError("pressgpu 下载体积超过 256MiB，拒绝安装") digest.update(chunk) out.write(chunk) actual_sha = digest.hexdigest() with open(log_path, "w") as log: log.write(f"url={PRESSGPU_URL}\nsize={size}\nsha256={actual_sha}\n") _safe_extract_pressgpu(archive, unpack) source = unpack / "pressgpu" required = ("wildrig-multi", "startpress.sh", "stoppress.sh") missing = [name for name in required if not (source / name).is_file()] if missing: raise RuntimeError("压缩包缺少文件: " + ", ".join(missing)) PRESSGPU_DIR.mkdir(parents=True, exist_ok=True) PRESSGPU_DIR.chmod(0o755) for name in required: temporary = PRESSGPU_DIR / f".{name}.{os.getpid()}.tmp" shutil.copyfile(source / name, temporary) temporary.chmod(0o755) os.replace(temporary, PRESSGPU_DIR / name) (PRESSGPU_DIR / "source.txt").write_text( f"url={PRESSGPU_URL}\nsha256={actual_sha}\ninstalled={now_utc()}\n") binary = find_pressgpu() check = subprocess.run([binary, "--version"], capture_output=True, text=True, timeout=30) if binary else None if not binary : raise RuntimeError("wildrig-multi 安装失败") with open(log_path, "a") as log: log.write("version=" + ((check.stdout or check.stderr).strip()) + "\n") print(f"pressgpu 已自动安装: {binary}") return binary except Exception as e: with open(log_path, "a") as log: log.write(f"安装失败: {e}\n") print(f"pressgpu 自动安装失败: {e}") return None finally: shutil.rmtree(tmp, ignore_errors=True) def start_gpu(gpus="all", runtime=GPU_RUNTIME): """用 pressgpu 的离线 benchmark 模式压测选中的 GPU。""" available = {str(g["index"]) for g in get_gpus()} if not available: sys.exit("nvidia-smi 无 GPU") if gpus == "all": ids = sorted(available, key=int) else: raw = str(gpus).replace(" ", "") if not re.fullmatch(r"\d+(,\d+)*", raw): sys.exit("GPU 卡号格式错误，应为 0、0,1 或 all") ids = list(dict.fromkeys(raw.split(","))) missing = [i for i in ids if i not in available] if missing: sys.exit("GPU 卡号不存在: " + ",".join(missing)) try: runtime = int(runtime) except (TypeError, ValueError): sys.exit("GPU 压测时长必须是整数秒") if not 10 <= runtime <= 86400: sys.exit("GPU 压测时长须为 10-86400 秒") ensure_job_deps("gpu") binary = find_pressgpu() or install_pressgpu() if not binary: sys.exit(f"pressgpu 自动安装失败，详情见 {BASE_DIR / 'pressgpu_install.log'}") dev = ",".join(ids) command = ["timeout", "-k", "30", str(runtime + 60), binary, "--algo", "pearlhash", "--benchmark", "--benchmark-timeout", str(runtime), "--opencl-platforms", "nvidia", "--gpu-list", dev, "--gpu-temp-limit", "85", "--watchdog", "--no-color", "--print-time", "10"] return start_job("gpu", f"GPU[{dev}] pressgpu离线压测 {runtime}s", shlex.join(command)) # ============================================================ # CLI 交互菜单 # ============================================================ def pick_gpus(): gpus = get_gpus() if not gpus: print("无 GPU") return "all" print("\nGPU 列表:") for g in gpus: print(f" [{g['index']}] {g['name']} {g['bus']} util {g['util']}% {g['temp']}°C") s = input("选择卡号(逗号分隔, 回车=全部): ").strip() return s if s else "all" def cmd_deps(): hr("依赖软件安装/检查") all_missing = [] for name, pkgs in DEP_GROUPS: miss = [p for p in pkgs if not pkg_installed(p)] if miss: print(f" {C_YELLOW}{name}: 缺 {', '.join(miss)}{C_OFF}") all_missing += miss else: print(f" {C_GREEN}{name}: 已装齐{C_OFF}") pressgpu = find_pressgpu() print(f" {C_GREEN}GPU 压测: pressgpu 已就绪 {C_OFF}" if pressgpu else f" {C_YELLOW}GPU 压测: 缺少 pressgpu，将自动安装{C_OFF}") installed, failed = [], [] if all_missing: print(f"\n安装缺失的 {len(all_missing)} 个 apt 包 (可能需要几分钟)...") installed, failed = install_deps() if installed: print(f"{C_GREEN}已安装: {' '.join(installed)}{C_OFF}") if failed: print(f"{C_RED}安装失败: {' '.join(failed)} — 检查 apt 源/网络后重试{C_OFF}") if not pressgpu: pressgpu = install_pressgpu() if pressgpu and not failed: print(f"{C_GREEN}全部依赖已就绪。{C_OFF}") elif not pressgpu: print(f"{C_RED}pressgpu 自动安装失败，详情见 " f"{BASE_DIR / 'pressgpu_install.log'}{C_OFF}") def menu(): while True: try: if not _menu_once(): break except (EOFError, KeyboardInterrupt): print() break def _menu_once(): """一轮菜单交互; 返回 False 表示退出""" print(f""" {C_BOLD}╔══════════════════════════════════════════════╗ ║ hwdiag 硬件诊断/压测 @ {HOSTNAME:<19} ║ ╚══════════════════════════════════════════════╝{C_OFF} [1] GPU 通道检查 [2] 系统硬件错误日志分析 [3] 磁盘压测 [4] CPU/内存压测 [5] GPU 压测 (可选卡) [6] 历史任务与测试结果 [7] 安装/检查依赖软件 [8] 启动 WebUI (默认 127.0.0.1:{DEFAULT_PORT}) [0] 退出""") c = input("选择: ").strip() try: if c == "1": out = check_gpu_link() print(out) print(f"{C_DIM}(结果已存档: {save_check('gpu-link', out)} — 菜单[6]可回看){C_OFF}") elif c == "2": out = check_hw_logs() print(out) print(f"{C_DIM}(结果已存档: {save_check('hw-logs', out)} — 菜单[6]可回看){C_OFF}") elif c == "3": print(f"使用固定参数: {DISK_SIZE} / {DISK_RUNTIME}秒 / {DISK_MODE}") j = start_disk() _after_start(j) elif c == "4": print(f"使用固定参数: CPU {CPUMEM_CPU_PCT}% / 内存 {CPUMEM_MEM_PCT}% / " f"每阶段 {CPUMEM_RUNTIME}秒") j = start_cpumem() _after_start(j) elif c == "5": gpus = pick_gpus() rt = input(f"GPU 离线压测时长秒 [{GPU_RUNTIME}]: ").strip() or str(GPU_RUNTIME) j = start_gpu(gpus, rt) _after_start(j) elif c == "6": jobs_browser() elif c == "7": cmd_deps() elif c == "8": run_web() return False elif c in ("0", "q", "quit", "exit"): return False except SystemExit as e: print(f"{C_RED}{e}{C_OFF}") except KeyboardInterrupt: # 子流程(浏览器/跟踪)里 Ctrl-C → 回主菜单; 主菜单提示符处 Ctrl-C → 退出 print("\n(返回主菜单)") return True def _after_start(j): """压测启动后的去向: 返回菜单 / 实时跟踪""" print(f"已启动 {j.info()}") c = input("回车=返回菜单, t=实时跟踪日志: ").strip().lower() if c == "t": follow_job(j) # ---- 任务与测试结果浏览器 ---- def follow_job(j, poll=1.0): """实时跟踪任务日志; Ctrl-C 返回(任务不受影响), 任务结束后自动退出跟踪""" print(f"\n{C_BOLD}--- 实时跟踪 {j.id} (Ctrl-C 返回, 任务继续后台运行) ---{C_OFF}") try: existing = j.logfile.read_text(errors="replace") if j.logfile.exists() else "" pos = len(existing) for ln in existing.splitlines()[-8:]: print(" " + ln) while True: alive = j.alive() text = j.logfile.read_text(errors="replace") if j.logfile.exists() else "" if len(text) > pos: sys.stdout.write(text[pos:]) sys.stdout.flush() pos = len(text) if not alive: print(f"\n{C_DIM}[任务已结束, 结果可用菜单 6 查看]{C_OFF}") break time.sleep(poll) except KeyboardInterrupt: print(f"\n{C_DIM}(停止跟踪, 任务继续后台运行){C_OFF}") def _job_detail(j): """单个任务的详情: 元信息 + 测试摘要 + 日志尾部; 可继续跟踪/停止""" i = j.info() state = C_GREEN + "RUNNING" + C_OFF if i["running"] else C_DIM + "已结束" + C_OFF print(f"\n{C_BOLD}■ {i['id']} [{state}]{C_OFF}") print(f" 类型: {i['type']} 描述: {i['label']}") print(f" 启动(UTC): {i['started']} 已耗时: {fmt_dur(i['elapsed'])} pid: {i['pid']}") print(f" 命令: {C_DIM}{i['cmd']}{C_OFF}") s = j.summary() if s: f = j.summary_file() print(f"\n{C_BOLD}── 测试摘要 ({f.parent.name}/{f.name}) ──{C_OFF}") print(s) print(f"\n{C_BOLD}── 日志尾部 ──{C_OFF}") print(j.tail(40)) c = input("\n[f=实时跟踪 s=停止 回车=返回列表]: ").strip().lower() if c == "f": follow_job(j) elif c == "s": print("已停止" if j.stop() else "停止失败/已退出") def jobs_browser(): """任务与测试结果浏览器: 序号选任务 → 摘要/日志/跟踪/停止""" while True: jobs = load_jobs() if not jobs: print(f"\n{C_DIM}暂无历史任务 — 先从菜单发起一次压测{C_OFF}") return show = jobs[-20:] n_run = sum(1 for j in show if j.alive()) print(f"\n{C_BOLD}── 任务与测试结果 (共 {len(jobs)} 个, 显示最近 {len(show)}, 运行中 {n_run}) ──{C_OFF}") print(f" {'#':<4}{'状态':<6}{'ID':<22}{'描述':<26}{'耗时':>8} 摘要") for n, j in enumerate(show, 1): i = j.info() mark = "RUN" if i["running"] else "end" has_sum = "有" if j.summary_file() else "-" print(f" {n:<4}{mark:<6}{i['id']:<22}{i['label'][:24]:<26}" f"{fmt_dur(i['elapsed']):>8} {has_sum}") c = input("\n序号=查看结果 s=停止 c=检查存档 回车=返回菜单: ").strip().lower() if not c: return if c == "s": cs = input("要停止的序号: ").strip() if cs.isdigit() and 1 <= int(cs) <= len(show): print("已停止" if show[int(cs) - 1].stop() else "停止失败/已退出") elif c == "c": checks_browser() elif c.isdigit() and 1 <= int(c) <= len(show): try: _job_detail(show[int(c) - 1]) except KeyboardInterrupt: print() def checks_browser(): """检查类结果存档浏览 (菜单 1/2 每次运行的输出)""" d = BASE_DIR / "checks" files = sorted(d.glob("*.txt")) if d.exists() else [] if not files: print(f"{C_DIM}暂无检查存档 (菜单 1/2 运行后会自动存档于 {d}){C_OFF}") return show = files[-20:] print(f"\n{C_BOLD}── 检查结果存档 (共 {len(files)}, 显示最近 {len(show)}) ──{C_OFF}") for n, f in enumerate(show, 1): print(f" {n:<4}{f.name}") c = input("序号=查看, 回车=返回: ").strip() if c.isdigit() and 1 <= int(c) <= len(show): lines = show[int(c) - 1].read_text(errors="replace").splitlines() print("\n".join(lines[-200:])) def jobs_show(stop=False): """非交互列表 (CLI jobs 子命令用)""" jobs = load_jobs() live = [j for j in jobs if j.alive()] print(f"\n任务共 {len(jobs)}, 运行中 {len(live)}:") for j in jobs[-15:]: i = j.info() mark = C_GREEN + "RUN " + C_OFF if i["running"] else C_DIM + "done" + C_OFF print(f" {mark} {i['id']} {i['label']:<28} pid={i['pid']} " f"{fmt_dur(i['elapsed'])} {C_DIM}{i['cmd'][:60]}{C_OFF}") if stop and live: jid = input("\n要停止的任务ID(回车跳过): ").strip() if jid: for j in jobs: if j.id == jid: print("已停止" if j.stop() else "停止失败/已退出") return # ============================================================ # WebUI # ============================================================ PAGE = r""" hwdiag @ __HOSTNAME__

hwdiag

__HOSTNAME__ · UTC (北京时间 +8) ·

快速检查

GPU 压测

检查 pressgpu…
加载中…
使用 pressgpu 离线 benchmark；默认 600 秒，可选择测试显卡；缺失时自动安装

CPU / 内存压测

固定参数：CPU 80% / 内存 80% / 每阶段 600 秒，带内存 watchdog

磁盘压测

固定参数：20G / 每项 300 秒 / safe 模式

任务

每 3s 刷新 · 任务在 hwdiag 退出后仍继续运行
状态ID类型描述 pid开始(UTC)耗时

检查结果

点击上方按钮运行检查…




"""


class Handler(BaseHTTPRequestHandler):
    token = None

    _gpu_cache = {"ts": 0, "data": None}  # /api/state 的 GPU 检查 30s 缓存, 避免 3s 轮询跑 nvidia-smi+lspci

    @staticmethod
    def _strip_ansi(s):
        return strip_ansi(s)

    def log_message(self, fmt, *args):
        pass

    # ---- helpers ----
    def _auth(self):
        if not self.token:
            return True
        q = parse_qs(urlparse(self.path).query)
        tok = q.get("token", [""])[0] or self.headers.get("X-Token", "")
        return tok == self.token

    def _send(self, code, body, ctype="application/json; charset=utf-8"):
        data = body.encode() if isinstance(body, str) else body
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def _json(self, obj):
        self._send(200, json.dumps(obj, ensure_ascii=False))

    # ---- routes ----
    def do_GET(self):
        if not self._auth():
            return self._send(403, '{"error": "token 错误"}')
        u = urlparse(self.path)
        if u.path in ("/", "/index.html"):
            html = PAGE.replace("__HOSTNAME__", HOSTNAME)
            return self._send(200, html, "text/html; charset=utf-8")
        if u.path == "/api/state":
            gpus = []
            try:
                if time.time() - self._gpu_cache["ts"] > 30:
                    self._gpu_cache.update(ts=time.time(),
                                           data=check_gpu_link(as_text=False).get("gpus", []))
                gpus = self._gpu_cache["data"]
            except Exception:
                pass
            return self._json({"now": time.strftime("%m-%d %H:%M:%S", time.gmtime()),
                               "hostname": HOSTNAME, "gpus": gpus,
                               "pressgpu": find_pressgpu(),
                               "jobs": [j.info() for j in load_jobs()]})
        m = re.match(r"^/api/check/(gpu-link|hw-logs)$", u.path)
        if m:
            text = (check_gpu_link() if m.group(1) == "gpu-link" else check_hw_logs())
            return self._json({"text": self._strip_ansi(text)})
        m = re.match(r"^/api/log/([0-9_a-f]+)/?$", u.path)
        if m:
            for j in load_jobs():
                if j.id == m.group(1):
                    n = int(parse_qs(u.query).get("tail", ["300"])[0])
                    return self._json({"text": self._strip_ansi(j.tail(n))})
            return self._send(404, '{"error": "no such job"}')
        self._send(404, '{"error": "not found"}')

    def do_POST(self):
        if not self._auth():
            return self._send(403, '{"error": "token 错误"}')
        u = urlparse(self.path)
        length = int(self.headers.get("Content-Length", 0))
        body = {}
        if length:
            try:
                body = json.loads(self.rfile.read(length))
            except Exception:
                return self._send(400, '{"error": "bad json"}')
        try:
            if u.path == "/api/deps":
                installed, failed = install_deps()
                pressgpu = find_pressgpu() or install_pressgpu()
                return self._json({"installed": installed, "failed": failed,
                                   "pressgpu": pressgpu})
            if u.path == "/api/start":
                t = body.get("type")
                if t == "cpumem":
                    job = start_cpumem()
                elif t == "disk":
                    job = start_disk()
                elif t == "gpu":
                    gpus = body.get("gpus") or "all"
                    job = start_gpu(gpus, body.get("runtime") or GPU_RUNTIME)
                else:
                    return self._send(400, '{"error": "unknown type"}')
                return self._json(job.info())
            m = re.match(r"^/api/stop/([0-9_a-f]+)/?$", u.path)
            if m:
                for j in load_jobs():
                    if j.id == m.group(1):
                        j.stop()
                        return self._json({"stopped": j.id})
                return self._send(404, '{"error": "no such job"}')
        except SystemExit as e:
            return self._send(400, json.dumps({"error": str(e)}, ensure_ascii=False))
        except Exception as e:
            return self._send(500, json.dumps({"error": str(e)}, ensure_ascii=False))
        self._send(404, '{"error": "not found"}')


def run_web(listen="127.0.0.1", port=DEFAULT_PORT, token=None):
    JOBS_DIR.mkdir(parents=True, exist_ok=True)
    Handler.token = token
    if listen == "0.0.0.0" and not token:
        token = uuid.uuid4().hex[:8]
        Handler.token = token
        print(f"{C_YELLOW}监听 0.0.0.0 已自动生成访问 token{C_OFF}")
    srv = ThreadingHTTPServer((listen, port), Handler)
    url = f"http://{listen}:{port}/" + (f"?token={token}" if token else "")
    print(f"{C_GREEN}hwdiag WebUI 已启动: {url}{C_OFF}")
    if listen == "127.0.0.1":
        print(f"{C_DIM}本机访问; 远程请用 SSH 隧道: ssh -L {port}:127.0.0.1:{port} <机器>{C_OFF}")
    try:
        srv.serve_forever()
    except KeyboardInterrupt:
        print("\nWebUI 已退出 (后台任务不受影响)")


# ============================================================
# main
# ============================================================

def main():
    sys.stdout.reconfigure(line_buffering=True)
    if len(sys.argv) == 3 and sys.argv[1] == "_worker" and sys.argv[2] in ("cpumem", "disk"):
        raise SystemExit(run_internal_worker(sys.argv[2]))
    if os.isatty(0) is False and len(sys.argv) == 1:
        # 无参数 + 非终端(如 dtssh 批量) → 打印简要帮助而非挂住
        print(__doc__)
        return
    ap = argparse.ArgumentParser(description="hwdiag 硬件诊断/压测 (CLI+WebUI)",
                                 formatter_class=argparse.RawDescriptionHelpFormatter,
                                 epilog="无参数运行 = 交互菜单")
    sub = ap.add_subparsers(dest="cmd")
    c = sub.add_parser("check", help="只读检查")
    c.add_argument("what", choices=["gpu-link", "hw-logs", "all"])
    c.add_argument("--since", default=None, help="AER 统计范围, 如 '48h'/'2026-08-20' (默认当前boot)")
    s = sub.add_parser("stress", help="启动压测任务")
    s.add_argument("kind", choices=["cpumem", "disk", "gpu"])
    s.add_argument("--gpus", default="all", help="gpu: 卡号如 0 或 0,1 或 all")
    s.add_argument("--runtime", type=int, default=GPU_RUNTIME,
                   help="gpu: pressgpu 离线 benchmark 运行秒数")
    sub.add_parser("jobs", help="查看任务")
    sub.add_parser("deps", help="安装/检查依赖软件")
    st = sub.add_parser("stop", help="停止任务")
    st.add_argument("job_id")
    lg = sub.add_parser("log", help="跟踪任务日志")
    lg.add_argument("job_id")
    lg.add_argument("-n", type=int, default=80)
    w = sub.add_parser("web", help="启动 WebUI")
    w.add_argument("--port", type=int, default=DEFAULT_PORT)
    w.add_argument("--listen", default="127.0.0.1")
    w.add_argument("--token", default=None)
    args = ap.parse_args()

    if not args.cmd:
        menu()
    elif args.cmd == "check":
        if args.what in ("gpu-link", "all"):
            print(check_gpu_link(since=args.since))
        if args.what in ("hw-logs", "all"):
            print(check_hw_logs())
    elif args.cmd == "stress":
        if args.kind == "cpumem":
            job = start_cpumem()
        elif args.kind == "disk":
            job = start_disk()
        else:
            job = start_gpu(args.gpus, args.runtime)
        i = job.info()
        print(f"已启动: {i['label']}\n  id={i['id']} pid={i['pid']} running={i['running']}")
        print(f"  cmd: {i['cmd']}")
        print(f"  日志: {job.logfile}")
        print(f"  跟踪: ./hwdiag.py log {i['id']}   停止: ./hwdiag.py stop {i['id']}")
    elif args.cmd == "jobs":
        jobs_show()
    elif args.cmd == "deps":
        cmd_deps()
    elif args.cmd == "stop":
        for j in load_jobs():
            if j.id == args.job_id or j.id.startswith(args.job_id):
                print("已停止" if j.stop() else "进程已不在")
                return
        print("未找到任务", args.job_id)
    elif args.cmd == "log":
        for j in load_jobs():
            if j.id == args.job_id or j.id.startswith(args.job_id):
                print(j.tail(args.n))
                return
        print("未找到任务", args.job_id)
    elif args.cmd == "web":
        run_web(args.listen, args.port, args.token)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print()