feat: 初始化 GPU Monitor 项目
搭建基于 Flask + SocketIO 的 GPU 集群实时监控系统,包含远程采集、定时调度、Web 前端可视化及项目文档。
这个提交包含在:
@@ -0,0 +1,88 @@
|
||||
import paramiko
|
||||
import logging
|
||||
|
||||
# 配置日志
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger("GPUCollector")
|
||||
|
||||
class GPUCollector:
|
||||
"""
|
||||
负责通过 SSH 远程连接到服务器并采集 NVIDIA GPU 状态的类
|
||||
"""
|
||||
def __init__(self, server_config):
|
||||
self.alias = server_config.get('alias', 'Unknown')
|
||||
self.ip = server_config.get('ip')
|
||||
self.port = server_config.get('port', 22)
|
||||
self.username = server_config.get('username')
|
||||
self.password = server_config.get('password')
|
||||
self.timeout = 10
|
||||
|
||||
def fetch_gpu_data(self):
|
||||
"""
|
||||
执行远程 nvidia-smi 命令并解析结果
|
||||
返回: List[Dict] 包含每张显卡的详细信息,失败则返回 None
|
||||
"""
|
||||
ssh = paramiko.SSHClient()
|
||||
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||||
|
||||
try:
|
||||
# 建立连接
|
||||
ssh.connect(
|
||||
hostname=self.ip,
|
||||
port=self.port,
|
||||
username=self.username,
|
||||
password=self.password,
|
||||
timeout=self.timeout
|
||||
)
|
||||
|
||||
# 使用 bash -l -c 强制加载登录环境 (Login Shell)
|
||||
# 这样可以确保加载 /etc/profile 和 ~/.bash_profile,从而获取正确的 PATH
|
||||
raw_cmd = (
|
||||
"nvidia-smi --query-gpu=index,name,temperature.gpu,"
|
||||
"utilization.gpu,utilization.memory,memory.total,"
|
||||
"memory.used,memory.free --format=csv,noheader,nounits"
|
||||
)
|
||||
full_cmd = f'bash -l -c "{raw_cmd}"'
|
||||
|
||||
stdin, stdout, stderr = ssh.exec_command(full_cmd)
|
||||
output = stdout.read().decode('utf-8').strip()
|
||||
error = stderr.read().decode('utf-8').strip()
|
||||
|
||||
if error and not output:
|
||||
logger.error(f"[{self.alias}] SSH Command Error: {error}")
|
||||
return None
|
||||
|
||||
if not output:
|
||||
logger.warning(f"[{self.alias}] No output received from nvidia-smi")
|
||||
return None
|
||||
|
||||
# 解析 CSV 数据
|
||||
lines = output.split('\n')
|
||||
gpu_list = []
|
||||
for line in lines:
|
||||
if not line: continue
|
||||
parts = [p.strip() for p in line.split(',')]
|
||||
if len(parts) == 8:
|
||||
gpu_list.append({
|
||||
"index": int(parts[0]),
|
||||
"name": parts[1],
|
||||
"temp": int(parts[2]),
|
||||
"util": int(parts[3]),
|
||||
"mem_util": int(parts[4]),
|
||||
"mem_total": int(parts[5]),
|
||||
"mem_used": int(parts[6]),
|
||||
"mem_free": int(parts[7])
|
||||
})
|
||||
|
||||
return gpu_list
|
||||
|
||||
except paramiko.AuthenticationException:
|
||||
logger.error(f"[{self.alias}] SSH Authentication failed for {self.ip}")
|
||||
except paramiko.SSHException as e:
|
||||
logger.error(f"[{self.alias}] SSH Exception: {e}")
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.alias}] Unexpected error: {e}")
|
||||
finally:
|
||||
ssh.close()
|
||||
|
||||
return None
|
||||
@@ -0,0 +1,82 @@
|
||||
import yaml
|
||||
import logging
|
||||
from apscheduler.schedulers.background import BackgroundScheduler
|
||||
from .collector import GPUCollector
|
||||
|
||||
# 配置日志
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger("GPUScheduler")
|
||||
|
||||
class GPUScheduler:
|
||||
"""
|
||||
负责定时触发远程采集任务并将结果分发的调度类
|
||||
"""
|
||||
def __init__(self, config_path, socketio=None):
|
||||
self.config_path = config_path
|
||||
self.socketio = socketio # 可选:如果传入 socketio 实例,则直接推送数据到前端
|
||||
self.scheduler = BackgroundScheduler()
|
||||
self.last_data = {} # 存储各服务器最后一次采集的数据 {server_alias: [gpu_info]}
|
||||
|
||||
def load_config(self):
|
||||
"""加载本地配置文件"""
|
||||
try:
|
||||
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||||
return yaml.safe_load(f)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load config file: {e}")
|
||||
return None
|
||||
|
||||
def collect_all_servers(self):
|
||||
"""遍历所有服务器并采集数据"""
|
||||
config = self.load_config()
|
||||
if not config:
|
||||
return
|
||||
|
||||
servers = config.get('servers', [])
|
||||
all_results = {}
|
||||
|
||||
for s_conf in servers:
|
||||
alias = s_conf.get('alias', 'Unknown')
|
||||
logger.info(f"Collecting data from {alias}...")
|
||||
|
||||
collector = GPUCollector(s_conf)
|
||||
data = collector.fetch_gpu_data()
|
||||
|
||||
if data is not None:
|
||||
all_results[alias] = data
|
||||
else:
|
||||
all_results[alias] = None # 标记为采集失败
|
||||
|
||||
# 更新内存状态
|
||||
self.last_data = all_results
|
||||
|
||||
# 如果配置了 socketio,则实时推送给前端
|
||||
if self.socketio:
|
||||
self.socketio.emit('gpu_update', all_results)
|
||||
logger.info("GPU data broadcasted via SocketIO")
|
||||
|
||||
def start(self):
|
||||
"""启动定时任务"""
|
||||
config = self.load_config()
|
||||
if not config:
|
||||
logger.error("Cannot start scheduler: config file missing or invalid")
|
||||
return False
|
||||
|
||||
interval = config.get('settings', {}).get('interval', 5)
|
||||
|
||||
# 添加定时任务
|
||||
self.scheduler.add_job(
|
||||
self.collect_all_servers,
|
||||
'interval',
|
||||
seconds=interval,
|
||||
id='gpu_collection_job'
|
||||
)
|
||||
|
||||
self.scheduler.start()
|
||||
logger.info(f"Scheduler started. Interval: {interval}s")
|
||||
return True
|
||||
|
||||
def stop(self):
|
||||
"""停止调度器"""
|
||||
self.scheduler.shutdown()
|
||||
logger.info("Scheduler stopped")
|
||||
在新工单中引用
屏蔽一个用户