737 lines
43 KiB
Python
737 lines
43 KiB
Python
#!/usr/bin/python3
|
||
"""LAN hardware dashboard with authenticated management. Python standard library."""
|
||
import argparse
|
||
import csv
|
||
from contextlib import contextmanager
|
||
import datetime as dt
|
||
import hashlib
|
||
import io
|
||
import json
|
||
import logging
|
||
import math
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import signal
|
||
import socket
|
||
from http.cookies import SimpleCookie, CookieError
|
||
import sqlite3
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from urllib.parse import parse_qs, urlparse
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
DATA = Path(os.environ.get('DASHBOARD_DATA', '/var/lib/hardware-dashboard'))
|
||
LOGS = Path(os.environ.get('HARDWARE_LOGS', '/var/log/hardware-watch'))
|
||
HWSTATE = Path(os.environ.get('HARDWARE_STATE', '/var/lib/hardware-watch/status.json'))
|
||
CONFIG = Path(os.environ.get('HARDWARE_CONFIG', '/etc/hardware-watch.json'))
|
||
INTERVAL = 10
|
||
KEEP_DAYS = 30
|
||
TZ = dt.timezone(dt.timedelta(hours=8))
|
||
STOP = threading.Event()
|
||
LOCK = threading.Lock()
|
||
LATEST = {}
|
||
TASK_SOCKET='/run/hardware-taskd/control.sock'
|
||
RANGES = {'1h': 3600, '6h': 21600, '24h': 86400, '7d': 604800, '30d': 2592000}
|
||
CATEGORY = {
|
||
'power': ('显卡辅助供电异常', '显卡检测不到必要的外接供电。', '正常关机并断电后，检查该卡供电线两端、转接头和退针情况；仍异常时换用匹配的已知正常供电线测试。'),
|
||
'gpu_init': ('显卡初始化失败', '系统发现了显卡，但驱动未能让它进入可用状态。', '先查看同一时段是否有供电或 PCIe 错误，再检查供电与插槽；不要直接反复重装驱动。'),
|
||
'gpu_driver': ('显卡状态读取失败', '管理接口没有返回全部预期显卡。', '对照供电、PCIe 和驱动日志排查；查看下方原始错误。'),
|
||
'gpu_missing': ('显卡未被发现', '实际发现的显卡少于预期，或内核报告未找到 NVIDIA GPU。', '先确认是否主动拆卡；否则正常关机断电后检查供电与插槽。'),
|
||
'gpu_identity': ('预期显卡缺失', '当前显卡身份与监控基准不一致。', '若主动更换了显卡，更新监控基准；否则检查对应显卡是否在线。'),
|
||
'gpu_temp': ('显卡温度偏高', '显卡温度超过配置的告警线。', '检查机箱风道和风扇，必要时降低计算负载；P100 依赖机箱送风。'),
|
||
'pcie_correctable': ('显卡与主板通信需要重试', '显卡与主板传数据时出现了系统可以自动纠正的通信错误。这条告警本身不表示显卡损坏。', '目前显卡在线时可以继续使用并观察。如果持续报错、任务变慢或显卡消失，安排空闲时正常关机并断电，检查显卡供电线和插槽，再做换线或换槽测试。'),
|
||
'pcie_fatal': ('PCIe 严重通信错误', '硬件连接出现不可纠正错误，可能导致设备不可用。', '先保存日志；正常关机断电后检查显卡供电、插槽及转接设备。'),
|
||
'gpu_xid': ('显卡运行错误', 'NVIDIA 驱动记录了 GPU 运行异常。', '结合原始 Xid 编号与同一时间的温度、供电、PCIe 日志排查。'),
|
||
'memory_cpu': ('CPU／内存硬件告警', '内核报告了机器检查或内存纠错事件。', '保留原始槽位与错误信息，由维护人员检查对应内存、CPU 和主板。'),
|
||
'storage': ('磁盘读写／通信异常', '内核记录了存储读写错误、超时或重置。', '先确认重要数据已有备份，再检查 SMART 状态、硬盘连接和供电。'),
|
||
'smart': ('磁盘健康告警', '磁盘 SMART 健康检查异常或无法读取。', '查看具体健康项；若出现介质错误或健康失败，优先备份并安排检修。'),
|
||
'disk_missing': ('预期磁盘缺失', '监控没有发现预期磁盘。', '确认是否主动移除；否则断电检查连接并安排维护。'),
|
||
'thermal': ('硬件温度告警', '内核报告过热或温度相关降速。', '检查散热器、风扇和机箱风道，必要时降低负载。'),
|
||
'sensor_temp': ('传感器温度告警', '硬件传感器温度达到告警阈值。', '检查对应 CPU、磁盘或机箱散热。'),
|
||
'clock': ('系统时间有变化', '历史系统时间曾跳变，原始时间可能不可靠。', '对照采集时间、启动 ID 和启动后经过时间；带“估算”的时刻不能当作精确时间。'),
|
||
'journal': ('日志采集状态', '硬件日志采集器记录了自身的运行状态。', '若持续采集失败或状态超过三分钟未更新，请维护人员检查 hardware-watch 服务。'),
|
||
'boot': ('系统启动记录', '采集器观察到了一个新的系统启动。', '如果不是计划内重启，查看此前是否有断电、温度或硬件异常。'),
|
||
'monitor': ('监控运行记录', '监控服务自身的状态变化。', '查看详细信息。'),
|
||
'web_monitor': ('网页采集异常', '网页无法取得最新硬件数据。', '检查监控服务是否正常，不要把旧数据当成实时状态。'),
|
||
'memory_pressure': ('可用内存偏少', '系统可用内存不足总量的 5%。', '检查正在运行的任务，减少不必要的并发或安排内存扩容。'),
|
||
}
|
||
|
||
def read(path, default=''):
|
||
try:
|
||
return Path(path).read_text().strip()
|
||
except OSError:
|
||
return default
|
||
|
||
def read_json(path, default=None):
|
||
try:
|
||
return json.loads(read(path))
|
||
except (ValueError, TypeError):
|
||
return {} if default is None else default
|
||
|
||
def iso(ts=None):
|
||
return dt.datetime.fromtimestamp(time.time() if ts is None else ts, TZ).isoformat(timespec='seconds')
|
||
|
||
def epoch(value):
|
||
try:
|
||
return dt.datetime.fromisoformat(value).timestamp()
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
def run(args, timeout=6):
|
||
try:
|
||
p = subprocess.run(args, capture_output=True, text=True, timeout=timeout,
|
||
env={**os.environ, 'LC_ALL': 'C'})
|
||
return p.returncode, p.stdout, p.stderr
|
||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||
return 124, '', str(exc)
|
||
|
||
@contextmanager
|
||
def connect():
|
||
db = sqlite3.connect(DATA / 'monitor.sqlite3', timeout=8)
|
||
db.row_factory = sqlite3.Row
|
||
try:
|
||
yield db
|
||
db.commit()
|
||
except Exception:
|
||
db.rollback()
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
def initialize():
|
||
DATA.mkdir(parents=True, exist_ok=True)
|
||
with connect() as db:
|
||
db.execute('PRAGMA journal_mode=WAL')
|
||
db.executescript('''
|
||
CREATE TABLE IF NOT EXISTS metrics (
|
||
ts INTEGER PRIMARY KEY, cpu REAL, memory REAL, cpu_temp REAL,
|
||
disk_read REAL, disk_write REAL, net_rx REAL, net_tx REAL,
|
||
gpus TEXT NOT NULL, source TEXT NOT NULL);
|
||
CREATE TABLE IF NOT EXISTS events (
|
||
id TEXT PRIMARY KEY, occurred REAL NOT NULL, observed REAL NOT NULL,
|
||
category TEXT, severity TEXT, raw TEXT NOT NULL);
|
||
CREATE INDEX IF NOT EXISTS event_time ON events(occurred DESC);
|
||
CREATE INDEX IF NOT EXISTS event_observed ON events(observed DESC);
|
||
CREATE INDEX IF NOT EXISTS event_category ON events(category, observed DESC);
|
||
CREATE INDEX IF NOT EXISTS metric_source ON metrics(source, ts);
|
||
CREATE TABLE IF NOT EXISTS imports (file TEXT PRIMARY KEY, inode TEXT, offset INTEGER);
|
||
CREATE TABLE IF NOT EXISTS boot_clock (boot TEXT PRIMARY KEY, anchor REAL, mono REAL);
|
||
CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT);
|
||
''')
|
||
|
||
def event_id(row):
|
||
if row.get('source') == 'kernel':
|
||
value = [row.get(k) for k in ('boot_id', 'category', 'device', 'first_monotonic_us', 'last_monotonic_us', 'detail')]
|
||
else:
|
||
value = [row.get(k) for k in ('observed_at', 'category', 'key', 'event', 'detail')]
|
||
return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()
|
||
|
||
def import_logs(db):
|
||
for path in sorted(LOGS.glob('*.jsonl')):
|
||
if not path.name.startswith(('events-', 'samples-')):
|
||
continue
|
||
stat = path.stat()
|
||
checkpoint = db.execute('SELECT inode,offset FROM imports WHERE file=?', (path.name,)).fetchone()
|
||
offset = checkpoint['offset'] if checkpoint and checkpoint['inode'] == str(stat.st_ino) and checkpoint['offset'] <= stat.st_size else 0
|
||
if offset == stat.st_size:
|
||
continue
|
||
with path.open('rb') as f:
|
||
f.seek(offset)
|
||
for _ in range(5000):
|
||
pos = f.tell()
|
||
line = f.readline()
|
||
if not line or not line.endswith(b'\n'):
|
||
f.seek(pos)
|
||
break
|
||
try:
|
||
row = json.loads(line)
|
||
except (ValueError, UnicodeDecodeError):
|
||
continue
|
||
if path.name.startswith('samples-'):
|
||
ts = epoch(row.get('checked_at'))
|
||
if ts is None or ts < time.time() - KEEP_DAYS * 86400:
|
||
continue
|
||
gpus = [{'uuid': x.get('uuid'), 'name': x.get('name'), 'pci': x.get('pci'),
|
||
'temperature_c': x.get('temperature_c'), 'power_w': x.get('power_w'),
|
||
'utilization': None, 'memory_used_mib': None,
|
||
'memory_total_mib': x.get('memory_mib')} for x in row.get('gpus', [])]
|
||
db.execute('INSERT OR IGNORE INTO metrics VALUES(?,NULL,NULL,NULL,NULL,NULL,NULL,NULL,?,?)',
|
||
(int(ts), json.dumps(gpus), 'legacy'))
|
||
else:
|
||
observed = epoch(row.get('observed_at'))
|
||
first = epoch(row.get('first_at')) or observed
|
||
if observed is None:
|
||
continue
|
||
last = epoch(row.get('last_at'))
|
||
mono = row.get('last_monotonic_us')
|
||
if last and mono and last <= observed + 60 and row.get('boot_id'):
|
||
m = int(mono) / 1e6
|
||
db.execute('''INSERT INTO boot_clock VALUES(?,?,?) ON CONFLICT(boot) DO UPDATE
|
||
SET anchor=excluded.anchor,mono=excluded.mono WHERE excluded.mono > boot_clock.mono''',
|
||
(row['boot_id'], last - m, m))
|
||
db.execute('INSERT OR IGNORE INTO events VALUES(?,?,?,?,?,?)',
|
||
(event_id(row), first, observed, row.get('category', 'monitor'),
|
||
row.get('severity', 'INFO'), json.dumps(row, ensure_ascii=False)))
|
||
offset = f.tell()
|
||
db.execute('INSERT OR REPLACE INTO imports VALUES(?,?,?)', (path.name, str(stat.st_ino), offset))
|
||
|
||
def temperatures():
|
||
result = []
|
||
for path in Path('/sys/class/hwmon').glob('hwmon*'):
|
||
chip = read(path / 'name')
|
||
for inp in path.glob('temp*_input'):
|
||
prefix = inp.name[:-6]
|
||
try:
|
||
result.append({'chip': chip, 'label': read(path / (prefix + '_label'), prefix),
|
||
'value': int(read(inp)) / 1000})
|
||
except ValueError:
|
||
pass
|
||
return result
|
||
|
||
def cpu_counters():
|
||
fields = list(map(int, read('/proc/stat').splitlines()[0].split()[1:9]))
|
||
return sum(fields), fields[3] + fields[4]
|
||
|
||
def counters():
|
||
net_rx = net_tx = disk_read = disk_write = 0
|
||
for line in read('/proc/net/dev').splitlines()[2:]:
|
||
name, values = line.split(':', 1)
|
||
if name.strip() == 'lo':
|
||
continue
|
||
fields = values.split()
|
||
net_rx += int(fields[0]); net_tx += int(fields[8])
|
||
disks = []
|
||
for path in Path('/sys/block').glob('*'):
|
||
if path.name.startswith(('loop', 'ram', 'zram', 'dm-', 'md')):
|
||
continue
|
||
fields = read(path / 'stat').split()
|
||
if len(fields) >= 7:
|
||
disk_read += int(fields[2]) * 512; disk_write += int(fields[6]) * 512
|
||
disks.append(path.name)
|
||
return {'net_rx': net_rx, 'net_tx': net_tx, 'disk_read': disk_read, 'disk_write': disk_write}, disks
|
||
|
||
def gpu_metrics():
|
||
fields = 'index,uuid,pci.bus_id,name,utilization.gpu,temperature.gpu,power.draw,power.limit,memory.used,memory.total'
|
||
code, out, err = run(['nvidia-smi', '--query-gpu=' + fields, '--format=csv,noheader,nounits'])
|
||
if code:
|
||
return [], (err or out or '无法读取显卡')[:1000]
|
||
result = []
|
||
for row in csv.reader(out.splitlines(), skipinitialspace=True):
|
||
if len(row) != 10:
|
||
continue
|
||
item = {'index': row[0], 'uuid': row[1], 'pci': row[2], 'name': row[3]}
|
||
for key, value in zip(('utilization', 'temperature_c', 'power_w', 'power_limit_w', 'memory_used_mib', 'memory_total_mib'), row[4:]):
|
||
try:
|
||
number = float(value)
|
||
item[key] = number if math.isfinite(number) else None
|
||
except ValueError:
|
||
item[key] = None
|
||
result.append(item)
|
||
return result, None if result else 'NVIDIA 管理接口没有返回可用显卡'
|
||
|
||
def issue(category, detail, level='WARN', device=''):
|
||
title, meaning, action = CATEGORY.get(category, CATEGORY['monitor'])
|
||
return {'category': category, 'title': title, 'meaning': meaning, 'action': action,
|
||
'detail': detail, 'severity': level, 'device': device}
|
||
|
||
def pci_address(value):
|
||
match = re.search(r'([0-9a-f]{4,8}):([0-9a-f]{2}):([0-9a-f]{2}\.[0-7])$', value or '', re.I)
|
||
if match:
|
||
return f'{int(match[1],16):04x}:{match[2].lower()}:{match[3].lower()}'
|
||
return value or ''
|
||
|
||
def device_map(gpus):
|
||
"""Map current GPU endpoints and their actual sysfs upstream bridges to one identity."""
|
||
result = {}
|
||
for index, gpu in enumerate(sorted(gpus,key=lambda x:x['uuid'])):
|
||
pci = pci_address(gpu['pci'])
|
||
label = f'第 {index+1} 张 P100' if 'P100' in gpu.get('name','') else f'第 {index+1} 张显卡'
|
||
entry = {'id':gpu['uuid'], 'label':label, 'pci':pci}
|
||
result[pci] = entry
|
||
path = Path('/sys/bus/pci/devices') / pci
|
||
if path.exists():
|
||
for parent in path.resolve().parents:
|
||
if re.fullmatch(r'[0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-7]',parent.name):
|
||
result.setdefault(parent.name,entry)
|
||
return result
|
||
|
||
def event_component(row,current):
|
||
address = pci_address(row.get('device',''))
|
||
same_boot = row.get('boot_id','').replace('-','') == current.get('boot_id','').replace('-','')
|
||
matched = current.get('device_map',{}).get(address) if same_boot else None
|
||
if matched:
|
||
return matched['id'], matched['label'], address
|
||
if row.get('category','').startswith('pcie'):
|
||
return address or 'unknown-link', '设备与主板的连接' + (f'（端口 {address[5:]}）' if address else ''), address
|
||
return address or row.get('key') or 'system', ('设备 '+address[5:]) if address else '这台电脑', address
|
||
|
||
def sample(previous):
|
||
now = time.time()
|
||
mono = time.monotonic()
|
||
total, idle = cpu_counters()
|
||
cpu = None
|
||
if previous and total > previous['cpu_total']:
|
||
cpu = max(0, min(100, 100 * (1 - (idle - previous['cpu_idle']) / (total - previous['cpu_total']))))
|
||
mem = {}
|
||
for line in read('/proc/meminfo').splitlines():
|
||
key, value = line.split(':', 1)
|
||
mem[key] = int(value.strip().split()[0]) * 1024
|
||
ram_total, ram_available = mem['MemTotal'], mem.get('MemAvailable', mem['MemFree'])
|
||
count, disks = counters()
|
||
rates = {key: None for key in count}
|
||
if previous and mono > previous['mono']:
|
||
rates = {key: max(0, (value - previous['counters'][key]) / (mono - previous['mono'])) / 1048576
|
||
for key, value in count.items()}
|
||
temps = temperatures()
|
||
cpu_temp = max([x['value'] for x in temps if x['chip'] == 'coretemp'], default=None)
|
||
disk_temp = max([x['value'] for x in temps if x['chip'] == 'nvme'], default=None)
|
||
gpu, gpu_error = gpu_metrics()
|
||
hw = read_json(HWSTATE)
|
||
cfg = read_json(CONFIG)
|
||
expected = cfg.get('expected_gpu_count', 2)
|
||
faults = []
|
||
if len(gpu) < expected:
|
||
faults.append(issue('gpu_missing', f'预期 {expected} 块，当前管理接口可用 {len(gpu)} 块。', 'ERROR'))
|
||
if gpu_error:
|
||
faults.append(issue('gpu_driver', gpu_error, 'ERROR'))
|
||
missing_uuid = set(cfg.get('expected_gpu_uuids', [])) - {x['uuid'] for x in gpu}
|
||
if missing_uuid and not gpu_error and len(gpu) >= expected:
|
||
faults.append(issue('gpu_identity', '当前显卡 UUID 与预期配置不同。', 'WARN'))
|
||
for item in gpu:
|
||
if item.get('temperature_c') is not None and item['temperature_c'] >= cfg.get('gpu_warn_c', 85):
|
||
faults.append(issue('gpu_temp', f'{item["temperature_c"]}°C，告警线 {cfg.get("gpu_warn_c",85)}°C', 'WARN', item['pci']))
|
||
if cpu_temp is not None and cpu_temp >= 85:
|
||
faults.append(issue('sensor_temp', f'CPU 最高温度 {cpu_temp}°C，网页关注线 85°C。'))
|
||
if ram_available / ram_total < .05:
|
||
faults.append(issue('memory_pressure', f'剩余 {ram_available/1073741824:.1f} GiB。'))
|
||
hw_age = now - (epoch(hw.get('checked_at')) or 0)
|
||
if hw_age > 180 or hw_age < -60:
|
||
faults.append(issue('journal', '硬件日志采集状态超过 3 分钟未更新，或系统时间异常。'))
|
||
else:
|
||
for value in hw.get('active', {}).values():
|
||
# Live GPU sampling uses the current configuration; the minute-level
|
||
# collector may still contain an alarm from before a settings change.
|
||
if value.get('category') in ('gpu_missing','gpu_identity'):
|
||
continue
|
||
matched = next((x for x in faults if x['category'] == value.get('category')), None)
|
||
if matched:
|
||
if value.get('severity') == 'ERROR':
|
||
matched['severity'] = 'ERROR'
|
||
matched['detail'] += ' ' + value.get('detail','')
|
||
continue
|
||
faults.append(issue(value.get('category', 'monitor'), value.get('detail', ''), value.get('severity', 'WARN')))
|
||
disk_usage = os.statvfs('/var/lib')
|
||
used = (disk_usage.f_blocks - disk_usage.f_bfree) * disk_usage.f_frsize
|
||
capacity = disk_usage.f_blocks * disk_usage.f_frsize
|
||
cpu_name = next((line.split(':', 1)[1].strip() for line in read('/proc/cpuinfo').splitlines()
|
||
if line.startswith('model name')), 'CPU')
|
||
cpu_sockets = len(set(line.split(':',1)[1].strip() for line in read('/proc/cpuinfo').splitlines()
|
||
if line.startswith('physical id')))
|
||
latest = {'ts': now, 'checked_at': iso(now), 'host': socket.gethostname(), 'address': '192.168.1.37',
|
||
'boot_id': read('/proc/sys/kernel/random/boot_id'), 'monotonic_s': mono,
|
||
'uptime_s': float(read('/proc/uptime').split()[0]), 'cpu': cpu, 'cpu_name': cpu_name,
|
||
'cpu_threads': os.cpu_count(), 'cpu_sockets': cpu_sockets, 'cpu_temp': cpu_temp, 'load': list(os.getloadavg()),
|
||
'memory': (1 - ram_available / ram_total) * 100, 'memory_total_gib': ram_total / 1073741824,
|
||
'memory_available_gib': ram_available / 1073741824, 'gpus': gpu,
|
||
'expected_gpu_count': expected, 'gpu_count_only': not bool(cfg.get('expected_gpu_uuids')),
|
||
'gpu_warn_c': cfg.get('gpu_warn_c',85),
|
||
'gpu_error': gpu_error, 'disk_temp': disk_temp, 'disks': disks, 'rates': rates,
|
||
'device_map': device_map(gpu),
|
||
'disk_used_gib': used / 1073741824, 'disk_total_gib': capacity / 1073741824,
|
||
'disk_usage_pct': used / capacity * 100, 'smart': hw.get('smart', {}),
|
||
'hardware_checked_at': hw.get('checked_at'), 'hardware_stale': hw_age > 180 or hw_age < -60,
|
||
'hardware_active': hw.get('active', {}) if 0 <= hw_age <= 180 else {},
|
||
'issues': faults, 'sensors': temps, 'interval': INTERVAL}
|
||
return latest, {'cpu_total': total, 'cpu_idle': idle, 'mono': mono, 'counters': count}
|
||
|
||
def sampler():
|
||
previous = None
|
||
last_clean = 0
|
||
while not STOP.is_set():
|
||
started = time.monotonic()
|
||
try:
|
||
current, previous = sample(previous)
|
||
with connect() as db:
|
||
import_logs(db)
|
||
db.execute('INSERT OR REPLACE INTO metrics VALUES(?,?,?,?,?,?,?,?,?,?)',
|
||
(int(current['ts']), current['cpu'], current['memory'], current['cpu_temp'],
|
||
current['rates']['disk_read'], current['rates']['disk_write'], current['rates']['net_rx'],
|
||
current['rates']['net_tx'], json.dumps(current['gpus']), 'live'))
|
||
db.execute('INSERT OR REPLACE INTO boot_clock VALUES(?,?,?)',
|
||
(current['boot_id'].replace('-', ''), current['ts'] - current['monotonic_s'], current['monotonic_s']))
|
||
if time.time() - last_clean >= 3600:
|
||
db.execute('DELETE FROM metrics WHERE ts < ?', (time.time() - KEEP_DAYS * 86400,))
|
||
db.execute('DELETE FROM events WHERE observed < ?', (time.time() - 90 * 86400,))
|
||
last_clean = time.time()
|
||
current['history_started_at'] = db.execute("SELECT MIN(ts) FROM metrics WHERE source='live'").fetchone()[0]
|
||
recent = db.execute('SELECT raw FROM events WHERE observed>? AND category IN (?,?,?)',
|
||
(time.time()-180, 'pcie_fatal', 'pcie_correctable', 'power')).fetchall()
|
||
for record in recent:
|
||
row = json.loads(record['raw'])
|
||
source_boot = row.get('boot_id', '').replace('-', '')
|
||
source_mono = row.get('last_monotonic_us')
|
||
if source_boot != current['boot_id'].replace('-', '') or source_mono is None:
|
||
continue
|
||
age = current['monotonic_s'] - int(source_mono) / 1e6
|
||
row_component = event_component(row,current)[0]
|
||
already_present = any(x['category'] == row['category'] and
|
||
event_component({'boot_id':current['boot_id'],'device':x.get('device',''),'category':x['category']},current)[0] == row_component
|
||
for x in current['issues'])
|
||
if 0 <= age <= 120 and not already_present:
|
||
severity = 'WARN' if row['category'] == 'pcie_correctable' else 'ERROR'
|
||
report = issue(row['category'], '最近两分钟仍有新的通信报告。' if row['category']=='pcie_correctable' else '最近两分钟仍有新报错。', severity, row.get('device',''))
|
||
report['component_label'] = event_component(row,current)[1]
|
||
current['issues'].append(report)
|
||
with LOCK:
|
||
LATEST.clear(); LATEST.update(current)
|
||
except Exception:
|
||
logging.exception('Sample failed')
|
||
with LOCK:
|
||
LATEST['collector_error'] = '网页采集失败，请查看服务日志。'
|
||
STOP.wait(max(.2, INTERVAL - (time.monotonic() - started)))
|
||
|
||
def snapshot():
|
||
with LOCK:
|
||
data = json.loads(json.dumps(LATEST))
|
||
data['server_ts'] = time.time()
|
||
data['stale'] = not data.get('ts') or time.time() - data.get('ts', 0) > 35
|
||
return data
|
||
|
||
def history(range_name):
|
||
seconds = RANGES.get(range_name, 86400)
|
||
end = int(time.time()); start = end - seconds
|
||
step = max(INTERVAL, math.ceil(seconds / 360))
|
||
with connect() as db:
|
||
rows = db.execute('''SELECT CAST((ts-?)/? AS INTEGER) AS b,AVG(cpu) cpu,AVG(memory) memory,
|
||
AVG(cpu_temp) cpu_temp,AVG(disk_read) disk_read,AVG(disk_write) disk_write,
|
||
AVG(net_rx) net_rx,AVG(net_tx) net_tx,SUM(source='live') live
|
||
FROM metrics WHERE ts>=? AND ts<=? GROUP BY b ORDER BY b''', (start,step,start,end)).fetchall()
|
||
gpu_rows = db.execute('''SELECT CAST((m.ts-?)/? AS INTEGER) AS b,
|
||
json_extract(j.value,'$.uuid') uuid, MAX(json_extract(j.value,'$.name')) name,
|
||
AVG(json_extract(j.value,'$.utilization')) utilization,
|
||
AVG(json_extract(j.value,'$.temperature_c')) temperature_c,
|
||
AVG(json_extract(j.value,'$.power_w')) power_w,
|
||
AVG(json_extract(j.value,'$.memory_used_mib')) memory_used_mib
|
||
FROM metrics m,json_each(m.gpus) j WHERE m.ts>=? AND m.ts<=?
|
||
GROUP BY b,uuid ORDER BY b''', (start,step,start,end)).fetchall()
|
||
points = {row['b']: {'ts': start + row['b'] * step, **{k: row[k] for k in row.keys() if k != 'b'}, 'gpus': {}} for row in rows}
|
||
devices = {}
|
||
for row in gpu_rows:
|
||
if row['b'] in points and row['uuid']:
|
||
points[row['b']]['gpus'][row['uuid']] = {key: row[key] for key in ('utilization','temperature_c','power_w','memory_used_mib')}
|
||
devices[row['uuid']] = row['name']
|
||
return {'start': start, 'end': end, 'step': step, 'range': range_name,
|
||
'points': list(points.values()), 'devices': devices}
|
||
|
||
def source_times(row, anchors):
|
||
observed = epoch(row.get('observed_at')) or time.time()
|
||
first = epoch(row.get('first_at')) or observed
|
||
last = epoch(row.get('last_at')) or first
|
||
adjusted = first > observed + 60 or last < first
|
||
note = ''
|
||
if adjusted:
|
||
boot = row.get('boot_id', '').replace('-', '')
|
||
if boot in anchors and row.get('first_monotonic_us') and row.get('last_monotonic_us'):
|
||
first = anchors[boot] + int(row['first_monotonic_us']) / 1e6
|
||
last = anchors[boot] + int(row['last_monotonic_us']) / 1e6
|
||
note = '原始系统时间曾异常；发生时间按同次启动的校时记录估算，原始时间保留如下。'
|
||
else:
|
||
first = last = observed
|
||
note = '原始系统时间不可靠，当前按监控读到记录的时间展示，无法确认精确发生时刻。'
|
||
return first, last, note
|
||
|
||
def merge_incidents(items):
|
||
"""Group reports, never claim a log line equals an independent hardware failure."""
|
||
buckets = {}
|
||
output = []
|
||
for item in sorted(items,key=lambda x:x['first_ts']):
|
||
if item['category'] in ('boot','clock','journal','monitor') or item['raw'].get('event') == 'RECOVERED':
|
||
item.update(record_count=1, signal_reports=0, raw_examples=[item['raw']])
|
||
output.append(item)
|
||
continue
|
||
key = (item['category'],item['raw'].get('boot_id','').replace('-',''),item['component_id'])
|
||
group = buckets.get(key)
|
||
if group is None or item['first_ts'] > group['last_ts'] + 300:
|
||
group = dict(item)
|
||
group.update(record_count=0, count=0, signal_reports=0, raw_examples=[],
|
||
observed_first_at=item['observed_at'], devices=[])
|
||
group['id'] = 'incident-' + hashlib.sha256(json.dumps([*key,item['first_ts']]).encode()).hexdigest()[:24]
|
||
buckets[key] = group
|
||
output.append(group)
|
||
group['first_ts'] = min(group['first_ts'],item['first_ts'])
|
||
group['first_at'] = iso(group['first_ts'])
|
||
if item['last_ts'] >= group['last_ts']:
|
||
for field in ('last_ts','last_at','raw'):
|
||
group[field] = item[field]
|
||
group['record_count'] += 1
|
||
group['count'] += item['count']
|
||
is_report = item['category'] != 'pcie_correctable' or 'PCIe Bus Error:' in item['raw'].get('detail','')
|
||
group['signal_reports'] += item['count'] if is_report else 0
|
||
group['observed_first_at'] = min(group['observed_first_at'],item['observed_at'])
|
||
group['observed_at'] = max(group['observed_at'],item['observed_at'])
|
||
if item['device'] and item['device'] not in group['devices']:
|
||
group['devices'].append(item['device'])
|
||
if len(group['raw_examples']) < 8 and not any(x.get('detail') == item['raw'].get('detail') for x in group['raw_examples']):
|
||
group['raw_examples'].append(item['raw'])
|
||
priority = {'ERROR':3,'WARN':2,'INFO':1}
|
||
if priority.get(item['severity'],0)>priority.get(group['severity'],0):
|
||
group['severity']=item['severity']
|
||
state_priority = {'unknown':6,'active':5,'related':4,'history':2,'recovered':1,'info':0}
|
||
if state_priority.get(item['state'],0)>state_priority.get(group['state'],0):
|
||
group['state']=item['state'];group['state_label']=item['state_label']
|
||
if item['time_note']:
|
||
group['time_note']=item['time_note']
|
||
return output
|
||
|
||
def explain_item(item,current):
|
||
online = len(current.get('gpus',[]))
|
||
expected = current.get('expected_gpu_count',2)
|
||
usable = online >= expected and not current.get('stale')
|
||
if item['category']=='pcie_correctable':
|
||
known_gpu=item.get('component_id','').lower().startswith('gpu')
|
||
device_word='显卡' if known_gpu else '设备'
|
||
item['title']=device_word+'连接反复重试' if item.get('signal_reports',0)>1 else device_word+'连接出现一次重试'
|
||
if not item.get('signal_reports'):
|
||
item['title']='设备连接出现重试记录'
|
||
item['meaning']='主板和'+device_word+'传数据时发现通信错误，系统会自动纠正或重发。这条记录不能证明'+device_word+'已经损坏。'
|
||
item['impact']=f'当前 {online}/{expected} 块显卡可被系统正常读取，暂未看到掉卡。' if usable else '当前未能确认全部显卡正常，请先看顶部的实时状态。'
|
||
item['urgency']='可以继续使用，建议安排空闲时检查连接' if usable else '先确认显卡是否在线，再安排维护'
|
||
item['next_steps']=['现在观察计算任务有没有报错、变慢或中断。',
|
||
'若通信重试持续出现，在业务空闲时正常关机并彻底断电。',
|
||
'检查显卡供电线两端和插槽；仍有问题时由维护人员换线、换槽对照测试。']
|
||
if item['state']=='active':
|
||
item['state_label']='仍在重试 · 显卡可用' if usable and known_gpu else '仍有连接重试 · 需要关注'
|
||
elif item['category']=='power':
|
||
item['meaning']='系统当时检测不到这张显卡需要的外接供电，所以显卡可能无法启动。线接触不良、线材、电源或卡上检测电路都可能引起这一现象。'
|
||
item['impact']=f'当前 {online}/{expected} 块显卡可用；这条是以前的供电异常记录。' if usable else '当前显卡状态不完整，请检查顶部告警。'
|
||
item['urgency']='历史记录，当前先观察是否再次出现' if usable else '需要检查辅助供电'
|
||
item['next_steps']=[item['action']]
|
||
else:
|
||
item['impact']='最近采样未再发现该问题。' if item['state']=='history' else item['state_label']
|
||
item['urgency']='查看说明，按建议处理'
|
||
item['next_steps']=[item['action']]
|
||
if item['state']=='history':
|
||
item['state_label']='以前的记录 · 当前未再发现'
|
||
if item['state']=='unknown':
|
||
item['impact']='实时数据已过期，暂时无法判断现在是否正常。'
|
||
item['urgency']='先恢复监控连接，再判断是否需要维修'
|
||
item['current_checked_at']=current.get('checked_at')
|
||
return item
|
||
|
||
def events(params):
|
||
now = time.time()
|
||
days = min(90, max(1, int(params.get('days', ['7'])[0])))
|
||
start = epoch(params.get('from', [''])[0]) or now - days * 86400
|
||
end = epoch(params.get('to', [''])[0]) or now + 60
|
||
if len(params.get('to', [''])[0]) == 10:
|
||
end += 86399
|
||
category = params.get('category', [''])[0]
|
||
severity = params.get('severity', [''])[0]
|
||
query = params.get('q', [''])[0][:120].lower()
|
||
page = max(0, min(10000, int(params.get('page', ['0'])[0])))
|
||
where = ['((occurred>=? AND occurred<=?) OR (observed>=? AND observed<=?))']
|
||
values = [start,end,start,end]
|
||
if category:
|
||
where.append('category=?'); values.append(category)
|
||
if severity=='problem':
|
||
where.append("severity IN ('ERROR','WARN')")
|
||
elif severity:
|
||
where.append('severity=?'); values.append(severity)
|
||
with connect() as db:
|
||
anchors = {x['boot'].replace('-', ''): x['anchor'] for x in db.execute('SELECT * FROM boot_clock')}
|
||
records = db.execute('SELECT * FROM events WHERE ' + ' AND '.join(where) + ' ORDER BY observed DESC LIMIT 15000', values).fetchall()
|
||
current = snapshot()
|
||
items = []
|
||
for record in records:
|
||
row = json.loads(record['raw'])
|
||
first,last,note = source_times(row,anchors)
|
||
if max(first,last) < start or min(first,last) > end:
|
||
continue
|
||
title,meaning,action = CATEGORY.get(row.get('category'), CATEGORY['monitor'])
|
||
ongoing = False
|
||
related = False
|
||
component_id,component_label,address = event_component(row,current)
|
||
for problem in current.get('issues', []):
|
||
if problem['category'] == row.get('category'):
|
||
other = event_component({'boot_id':current.get('boot_id',''),'device':problem.get('device',''),'category':problem['category']},current)[0]
|
||
related = component_id==other or not problem.get('device')
|
||
if related:
|
||
break
|
||
if row.get('key') and row['key'] in current.get('hardware_active', {}):
|
||
ongoing = row.get('event') != 'RECOVERED'
|
||
if not current.get('stale') and related and row.get('source') == 'kernel':
|
||
mono = row.get('last_monotonic_us')
|
||
same_boot = row.get('boot_id', '').replace('-','') == current.get('boot_id','').replace('-','')
|
||
if same_boot and mono:
|
||
ongoing = 0 <= current.get('monotonic_s',0) - int(mono)/1e6 <= 120
|
||
if current.get('stale'):
|
||
state,label = 'unknown','当前状态未知'
|
||
elif row.get('event') == 'RECOVERED':
|
||
state,label = 'recovered','已记录恢复'
|
||
elif ongoing:
|
||
state,label = 'active','当前仍有同类异常'
|
||
elif related:
|
||
state,label = 'related','历史记录 · 当前另有同类告警'
|
||
else:
|
||
state,label = 'history','历史记录 · 当前未复现'
|
||
if row.get('severity') == 'INFO' and row.get('event') != 'RECOVERED':
|
||
state,label = 'info','运行记录'
|
||
items.append({'id': record['id'], 'title': title, 'meaning': meaning, 'action': action,
|
||
'severity': row.get('severity','INFO'), 'category': row.get('category'),
|
||
'device': row.get('device',''), 'first_ts': first,'last_ts': last,
|
||
'first_at': iso(first),'last_at': iso(last),'observed_at': row.get('observed_at'),
|
||
'count': row.get('count',1),'time_note': note,'state': state,'state_label': label,'raw': row,
|
||
'component_id':component_id,'component_label':component_label})
|
||
raw_total=len(items)
|
||
view=params.get('view',['issues'])[0]
|
||
if view!='raw':
|
||
items=merge_incidents(items)
|
||
else:
|
||
for item in items:
|
||
item.update(record_count=1,signal_reports=item['count'] if item['category']!='pcie_correctable' or 'PCIe Bus Error:' in item['raw'].get('detail','') else 0,
|
||
raw_examples=[item['raw']])
|
||
for item in items:
|
||
explain_item(item,current)
|
||
if query:
|
||
items=[item for item in items if query in json.dumps(item,ensure_ascii=False).lower()]
|
||
raw_total=sum(item.get('record_count',1) for item in items)
|
||
items.sort(key=lambda x: x['last_ts'],reverse=True)
|
||
offset = page * 30
|
||
return {'items': items[offset:offset+30], 'total': len(items),'raw_total':raw_total,'view':view,'page': page,
|
||
'more': offset+30 < len(items), 'categories': [{'id':key,'name':value[0]} for key,value in CATEGORY.items()]}
|
||
|
||
def task_rpc(action,payload=None,token='',csrf='',client=''):
|
||
request={'action':action,'payload':payload or {},'token':token,'csrf':csrf,'client':client}
|
||
try:
|
||
with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as conn:
|
||
conn.settimeout(25);conn.connect(TASK_SOCKET)
|
||
conn.sendall(json.dumps(request,ensure_ascii=False).encode()+b'\n')
|
||
with conn.makefile('rb') as stream:
|
||
raw=stream.readline(1024*1024)
|
||
return json.loads(raw)
|
||
except (OSError,ValueError) as exc:
|
||
logging.warning('Task broker unavailable: %s',type(exc).__name__)
|
||
return {'ok':False,'status':503,'error':'定时任务管理服务暂不可用；设备监控仍可正常查看。'}
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = 'HardwareDashboard/1.0'
|
||
|
||
def log_message(self, fmt, *args):
|
||
if len(args)>1 and str(args[1]) not in ('200','304'):
|
||
logging.info('%s %s',self.client_address[0], fmt % args)
|
||
|
||
def send(self, body, content_type, status=200, attachment=None,extra_headers=None):
|
||
if isinstance(body,str):
|
||
body = body.encode('utf-8')
|
||
self.send_response(status)
|
||
self.send_header('Content-Type',content_type)
|
||
self.send_header('Content-Length',str(len(body)))
|
||
self.send_header('Cache-Control','no-store')
|
||
self.send_header('X-Content-Type-Options','nosniff')
|
||
self.send_header('X-Frame-Options','DENY')
|
||
self.send_header('Content-Security-Policy',"default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'")
|
||
if attachment:
|
||
self.send_header('Content-Disposition',f'attachment; filename="{attachment}"')
|
||
for key,value in (extra_headers or {}).items():self.send_header(key,value)
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def do_GET(self):
|
||
parsed = urlparse(self.path)
|
||
params = parse_qs(parsed.query)
|
||
try:
|
||
if parsed.path in ('/api/tasks/auth','/api/tasks','/api/tasks/item','/api/tasks/runs','/api/tasks/runlog','/api/tasks/history'):
|
||
action={'/api/tasks/auth':'auth.status','/api/tasks':'tasks.list','/api/tasks/item':'tasks.get',
|
||
'/api/tasks/runs':'tasks.runs','/api/tasks/runlog':'tasks.runlog','/api/tasks/history':'tasks.history'}[parsed.path]
|
||
payload={'id':params.get('id',[''])[0],'run_id':params.get('run_id',[''])[0]}
|
||
return self.task_response(action,payload)
|
||
elif parsed.path == '/api/status':
|
||
data = snapshot()
|
||
elif parsed.path == '/api/history':
|
||
data = history(params.get('range',['24h'])[0])
|
||
elif parsed.path == '/api/events':
|
||
data = events(params)
|
||
elif parsed.path == '/api/events.csv':
|
||
output = io.StringIO(); writer = csv.writer(output)
|
||
writer.writerow(['问题开始时间','最近发生时间','监控最近发现时间','原级别','问题','当前判断','设备','合并技术记录数','日志总行数','报错报告数（非独立故障次数）','维护建议','时间说明','原始日志样例'])
|
||
for row in events(params)['items']:
|
||
values = [row['first_at'],row['last_at'],row['observed_at'],row['severity'],row['title'],row['state_label'],row.get('component_label',row['device']),row.get('record_count',1),row['count'],row.get('signal_reports',0),row['action'],row['time_note'],row['raw'].get('detail','')]
|
||
writer.writerow(["'"+v if isinstance(v,str) and v.startswith(('=','+','-','@')) else v for v in values])
|
||
return self.send('\ufeff'+output.getvalue(),'text/csv; charset=utf-8',attachment='hardware-events.csv')
|
||
elif parsed.path in ('/','/app.js','/tasks.js','/style.css','/favicon.svg'):
|
||
filename = 'index.html' if parsed.path=='/' else parsed.path[1:]
|
||
mime = {'index.html':'text/html; charset=utf-8','app.js':'text/javascript; charset=utf-8','tasks.js':'text/javascript; charset=utf-8','style.css':'text/css; charset=utf-8','favicon.svg':'image/svg+xml'}
|
||
return self.send((ROOT/'static'/filename).read_bytes(),mime[filename])
|
||
else:
|
||
return self.send('Not found','text/plain',404)
|
||
self.send(json.dumps(data,ensure_ascii=False,allow_nan=False),'application/json; charset=utf-8')
|
||
except (ValueError,TypeError):
|
||
self.send('{"error":"参数格式不正确"}','application/json; charset=utf-8',400)
|
||
except (BrokenPipeError,ConnectionResetError):
|
||
pass
|
||
except Exception:
|
||
logging.exception('Request failed')
|
||
self.send('{"error":"读取失败，请稍后重试"}','application/json; charset=utf-8',500)
|
||
|
||
def task_response(self,action,payload):
|
||
token=''
|
||
try:
|
||
cookie=SimpleCookie();cookie.load(self.headers.get('Cookie',''))
|
||
if 'hw_task_session' in cookie:token=cookie['hw_task_session'].value
|
||
except CookieError:pass
|
||
result=task_rpc(action,payload,token,self.headers.get('X-Task-CSRF',''),self.client_address[0])
|
||
headers={}
|
||
if result.get('ok'):
|
||
data=result['data']
|
||
if action=='auth.login':
|
||
secret=data.pop('token')
|
||
headers['Set-Cookie']=f'hw_task_session={secret}; Path=/api/tasks; Max-Age={data["max_age"]}; HttpOnly; SameSite=Strict'
|
||
elif action in ('auth.logout','auth.password'):
|
||
headers['Set-Cookie']='hw_task_session=; Path=/api/tasks; Max-Age=0; HttpOnly; SameSite=Strict'
|
||
self.send(json.dumps(data,ensure_ascii=False),'application/json; charset=utf-8',extra_headers=headers)
|
||
else:
|
||
self.send(json.dumps({'error':result.get('error','任务操作失败。')},ensure_ascii=False),
|
||
'application/json; charset=utf-8',result.get('status',500))
|
||
|
||
def do_POST(self):
|
||
parsed=urlparse(self.path)
|
||
actions={'/api/tasks/login':'auth.login','/api/tasks/logout':'auth.logout','/api/tasks/password':'auth.password',
|
||
'/api/tasks/preview':'tasks.preview','/api/tasks/save':'tasks.save','/api/tasks/enable':'tasks.enable',
|
||
'/api/tasks/delete':'tasks.delete','/api/tasks/run':'tasks.run','/api/tasks/stop':'tasks.stop',
|
||
'/api/tasks/gpu-count':'monitor.gpu_count'}
|
||
if parsed.path not in actions:return self.send('Not found','text/plain',404)
|
||
origin=self.headers.get('Origin','')
|
||
host=self.headers.get('Host','')
|
||
if urlparse(origin).netloc!=host or urlparse(origin).scheme not in ('http','https') or urlparse(origin).hostname not in ('192.168.1.37','localhost','127.0.0.1'):
|
||
return self.send('{"error":"请从本机监控页面发起操作。"}','application/json; charset=utf-8',403)
|
||
try:
|
||
length=int(self.headers.get('Content-Length','0'))
|
||
if not 0<length<=131072:raise ValueError()
|
||
if not self.headers.get('Content-Type','').startswith('application/json'):raise ValueError()
|
||
payload=json.loads(self.rfile.read(length))
|
||
if not isinstance(payload,dict):raise ValueError()
|
||
except (ValueError,UnicodeDecodeError):
|
||
return self.send('{"error":"请求格式不正确或内容超过128KiB。"}','application/json; charset=utf-8',400)
|
||
self.task_response(actions[parsed.path],payload)
|
||
|
||
def main():
|
||
parser=argparse.ArgumentParser()
|
||
parser.add_argument('--host',default='0.0.0.0'); parser.add_argument('--port',type=int,default=8090)
|
||
args=parser.parse_args()
|
||
logging.basicConfig(level=logging.INFO,format='%(asctime)s %(levelname)s %(message)s')
|
||
initialize()
|
||
worker=threading.Thread(target=sampler,daemon=True); worker.start()
|
||
server=ThreadingHTTPServer((args.host,args.port),Handler)
|
||
server.daemon_threads=True
|
||
logging.info('Hardware dashboard listening on %s:%s',args.host,args.port)
|
||
try:
|
||
server.serve_forever(poll_interval=.5)
|
||
finally:
|
||
STOP.set(); server.server_close(); worker.join(timeout=10)
|
||
|
||
if __name__=='__main__':
|
||
main()
|