feat: k8s/slurm/ray plugin SSH部署实现 — 控制节点部署+算力节点加入+移除+状态监控
This commit is contained in:
parent
c3507f1e71
commit
b643b6de0b
@ -3,7 +3,7 @@ pcc — 统一集群管理 (Pooled Computing Cluster)
|
||||
生命周期: deploy → run → stop → start → destroy
|
||||
插件: k8s_plugin, slurm_plugin, ray_plugin
|
||||
"""
|
||||
import datetime, json
|
||||
import datetime, json, asyncio, subprocess
|
||||
from appPublic.uniqueID import getID
|
||||
from sqlor.dbpools import DBPools
|
||||
|
||||
@ -20,10 +20,41 @@ def _get_plugin(cluster_type):
|
||||
import importlib
|
||||
modname = _PLUGINS.get(cluster_type)
|
||||
if not modname:
|
||||
raise ValueError(f'Unknown cluster_type: {cluster_type}')
|
||||
raise ValueError('Unknown cluster_type: ' + cluster_type)
|
||||
return importlib.import_module(modname)
|
||||
|
||||
|
||||
async def ssh_exec(host, port, user, cmd, timeout=120):
|
||||
"""异步 SSH 远程执行命令，返回 (exit_code, stdout, stderr)"""
|
||||
ssh_cmd = [
|
||||
'ssh', '-o', 'StrictHostKeyChecking=no',
|
||||
'-o', 'ConnectTimeout=10', '-o', 'BatchMode=yes',
|
||||
'-p', str(port), '{user}@{host}', cmd
|
||||
]
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*ssh_cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE
|
||||
)
|
||||
try:
|
||||
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout)
|
||||
except asyncio.TimeoutError:
|
||||
proc.kill()
|
||||
return -1, '', f'SSH timeout after {timeout}s'
|
||||
return proc.returncode, stdout.decode(), stderr.decode()
|
||||
|
||||
|
||||
async def ssh_exec_bg(host, port, user, cmd):
|
||||
"""异步后台执行，不等待结果"""
|
||||
ssh_cmd = [
|
||||
'ssh', '-o', 'StrictHostKeyChecking=no',
|
||||
'-o', 'ConnectTimeout=10', '-o', 'BatchMode=yes',
|
||||
'-p', str(port), user + '@' + host,
|
||||
'nohup bash -c "' + cmd.replace('"', '\\"') + '" > /dev/null 2>&1 &'
|
||||
]
|
||||
subprocess.Popen(ssh_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
|
||||
|
||||
async def deploy_cluster(request, cluster_id):
|
||||
"""部署集群: 读取集群配置 → 分配节点 → 调用 plugin 部署"""
|
||||
env = request._run_ns
|
||||
|
||||
@ -1,24 +1,176 @@
|
||||
'''
|
||||
k8s_plugin - K8s + VM 集群管理
|
||||
- K8s 部署 (kubeadm/k3s + 控制平面 + worker)
|
||||
- VM 生命周期 (创建/快照/迁移/销毁)
|
||||
- 节点加入/退出集群
|
||||
- kubectl / k8s API 状态监控
|
||||
- 算力单元: namespace + ResourceQuota 分配回收
|
||||
''')
|
||||
"""
|
||||
k8s_plugin — K8s 集群管理 (基于 kubeadm)
|
||||
- 控制节点: kubeadm init + CNI(flannel)
|
||||
- 算力节点: kubeadm join
|
||||
- 移除: kubectl drain + delete + kubeadm reset
|
||||
- 状态: kubectl get nodes
|
||||
"""
|
||||
from pcc import ssh_exec
|
||||
|
||||
K8S_VERSION = '1.29'
|
||||
POD_CIDR = '10.244.0.0/16'
|
||||
SVC_CIDR = '10.96.0.0/12'
|
||||
|
||||
|
||||
async def _node_info(env, node_id):
|
||||
"""从 DB 获取节点信息"""
|
||||
from sqlor.dbpools import DBPools
|
||||
dbname = env.get_module_dbname('pccs')
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('compute_node', {'id': node_id})
|
||||
return recs[0] if recs else None
|
||||
|
||||
|
||||
async def _install_k8s(node):
|
||||
"""在节点上安装 K8s 基础组件 (containerd + kubeadm + kubelet + kubectl)"""
|
||||
script = (
|
||||
"apt-get update -qq && "
|
||||
"apt-get install -y -qq apt-transport-https ca-certificates curl gpg && "
|
||||
"curl -fsSL https://pkgs.k8s.io/core:/stable:/v{ver}/deb/Release.key | gpg --dearmor -o /etc/apt/keyrings/kubernetes-apt-keyring.gpg && "
|
||||
"echo 'deb [signed-by=/etc/apt/keyrings/kubernetes-apt-keyring.gpg] https://pkgs.k8s.io/core:/stable:/v{ver}/deb/ /' > /etc/apt/sources.list.d/kubernetes.list && "
|
||||
"apt-get update -qq && "
|
||||
"apt-get install -y -qq kubelet kubeadm kubectl containerd && "
|
||||
"apt-mark hold kubelet kubeadm kubectl && "
|
||||
"mkdir -p /etc/containerd && "
|
||||
"containerd config default > /etc/containerd/config.toml 2>/dev/null || true && "
|
||||
"sed -i 's/SystemdCgroup = false/SystemdCgroup = true/' /etc/containerd/config.toml && "
|
||||
"systemctl restart containerd && systemctl enable kubelet"
|
||||
).format(ver=K8S_VERSION)
|
||||
return await ssh_exec(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', script, timeout=300)
|
||||
|
||||
|
||||
async def _install_cni(control_node, pod_cidr):
|
||||
"""在控制节点安装 CNI 网络插件"""
|
||||
cmd = f"kubectl apply -f https://raw.githubusercontent.com/flannel-io/flannel/master/Documentation/kube-flannel.yml"
|
||||
return await ssh_exec(control_node.ip_address, control_node.ssh_port or 22,
|
||||
control_node.ssh_user or 'root', cmd, timeout=120)
|
||||
|
||||
|
||||
async def deploy_cluster(env, cluster_id, control_nodes, compute_nodes, config):
|
||||
"""K8s 集群部署：只部署控制节点。算力节点后续通过 add_node 加入。"""
|
||||
pod_cidr = config.get('pod_cidr', POD_CIDR)
|
||||
svc_cidr = config.get('svc_cidr', SVC_CIDR)
|
||||
results = {'control_nodes': {}, 'compute_nodes': None, 'endpoint': ''}
|
||||
|
||||
for node in control_nodes:
|
||||
# 1. 安装 K8s 基础组件
|
||||
rc, out, err = await _install_k8s(node)
|
||||
if rc != 0:
|
||||
results['control_nodes'][node.id] = {'error': f'install failed: {err}'}
|
||||
continue
|
||||
|
||||
# 2. kubeadm init（仅第一个控制节点）
|
||||
init_cmd = (
|
||||
f"kubeadm init --pod-network-cidr={pod_cidr} --service-cidr={svc_cidr} "
|
||||
f"--kubernetes-version=v{K8S_VERSION} --ignore-preflight-errors=all"
|
||||
)
|
||||
rc, out, err = await ssh_exec(node.ip_address, node.ssh_port or 22,
|
||||
node.ssh_user or 'root', init_cmd, timeout=300)
|
||||
if rc != 0:
|
||||
results['control_nodes'][node.id] = {'error': f'init failed: {err[:200]}'}
|
||||
continue
|
||||
|
||||
# 3. 配置 kubectl
|
||||
setup_cmd = "mkdir -p $HOME/.kube && cp -f /etc/kubernetes/admin.conf $HOME/.kube/config && chown $(id -u):$(id -g) $HOME/.kube/config"
|
||||
await ssh_exec(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', setup_cmd)
|
||||
|
||||
# 4. 安装 CNI
|
||||
await _install_cni(node, pod_cidr)
|
||||
|
||||
# 5. 获取 join token（供后续算力节点使用）
|
||||
rc2, token_out, _ = await ssh_exec(node.ip_address, node.ssh_port or 22,
|
||||
node.ssh_user or 'root',
|
||||
"kubeadm token create --print-join-command", timeout=30)
|
||||
join_cmd = token_out.strip() if rc2 == 0 else ''
|
||||
|
||||
results['control_nodes'][node.id] = {
|
||||
'status': 'ok',
|
||||
'join_command': join_cmd,
|
||||
'endpoint': f'https://{node.ip_address}:6443'
|
||||
}
|
||||
results['endpoint'] = f'https://{node.ip_address}:6443'
|
||||
|
||||
# 保存 join_command 到集群 config
|
||||
async with DBPools().sqlorContext(env.get_module_dbname('pccs')) as sor:
|
||||
await sor.U('cluster', {'id': cluster_id},
|
||||
{'control_config': json.dumps(results)})
|
||||
|
||||
return results
|
||||
|
||||
async def deploy_k8s_cluster(env, cluster_id, control_nodes, compute_nodes, config):
|
||||
'''部署 K8s 集群: 初始化控制平面 → join worker → 验证'''
|
||||
pass
|
||||
|
||||
async def add_node(env, cluster_id, node_id, role='compute'):
|
||||
'''向已有集群添加节点'''
|
||||
pass
|
||||
"""向 K8s 集群添加算力节点：安装 k8s → kubeadm join"""
|
||||
node = await _node_info(env, node_id)
|
||||
if not node:
|
||||
return {'status': 'error', 'message': f'Node {node_id} not found'}
|
||||
|
||||
# 读取 join_command
|
||||
from sqlor.dbpools import DBPools
|
||||
from pcc import MODULE_NAME
|
||||
dbname = env.get_module_dbname(MODULE_NAME)
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('cluster', {'id': cluster_id})
|
||||
if not recs:
|
||||
return {'status': 'error', 'message': 'Cluster not found'}
|
||||
config = json.loads(recs[0].control_config or '{}')
|
||||
join_cmd = ''
|
||||
for cn_data in config.get('control_nodes', {}).values():
|
||||
if cn_data.get('join_command'):
|
||||
join_cmd = cn_data['join_command']
|
||||
break
|
||||
if not join_cmd:
|
||||
return {'status': 'error', 'message': 'No join command found, cluster may not be deployed'}
|
||||
|
||||
# 1. 安装 K8s
|
||||
rc, out, err = await _install_k8s(node)
|
||||
if rc != 0:
|
||||
return {'status': 'error', 'message': f'k8s install failed: {err[:200]}'}
|
||||
|
||||
# 2. kubeadm join
|
||||
rc2, out2, err2 = await ssh_exec(node.ip_address, node.ssh_port or 22,
|
||||
node.ssh_user or 'root', join_cmd, timeout=120)
|
||||
if rc2 != 0:
|
||||
return {'status': 'error', 'message': f'join failed: {err2[:200]}'}
|
||||
|
||||
return {'status': 'ok', 'message': f'Node {node.name} joined K8s cluster'}
|
||||
|
||||
|
||||
async def remove_node(env, cluster_id, node_id):
|
||||
'''从集群移除节点并回收资源'''
|
||||
pass
|
||||
"""从 K8s 集群移除节点"""
|
||||
node = await _node_info(env, node_id)
|
||||
if not node:
|
||||
return {'status': 'error', 'message': 'Node not found'}
|
||||
|
||||
# drain + delete + reset
|
||||
cmds = [
|
||||
f"kubectl drain {node.name} --ignore-daemonsets --delete-emptydir-data --timeout=60s",
|
||||
f"kubectl delete node {node.name}",
|
||||
"kubeadm reset -f",
|
||||
]
|
||||
for cmd in cmds:
|
||||
rc, out, err = await ssh_exec(node.ip_address, node.ssh_port or 22,
|
||||
node.ssh_user or 'root', cmd, timeout=120)
|
||||
return {'status': 'ok', 'message': f'Node {node.name} removed'}
|
||||
|
||||
|
||||
async def cluster_status(env, cluster_id):
|
||||
'''获取集群状态 (节点数、Pod数、资源使用率)'''
|
||||
pass
|
||||
"""获取 K8s 集群状态"""
|
||||
from sqlor.dbpools import DBPools
|
||||
dbname = env.get_module_dbname('pccs')
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('cluster_node', {'cluster_id': cluster_id, 'role': 'control', 'status': 'active'})
|
||||
if not recs:
|
||||
return {'error': 'No active control node'}
|
||||
|
||||
control_id = recs[0].node_id
|
||||
cnode = await _node_info(env, control_id)
|
||||
if not cnode:
|
||||
return {'error': 'Control node not found'}
|
||||
|
||||
rc, out, _ = await ssh_exec(cnode.ip_address, cnode.ssh_port or 22,
|
||||
cnode.ssh_user or 'root', 'kubectl get nodes -o wide', timeout=30)
|
||||
return {'status': 'ok', 'kubectl_nodes': out if rc == 0 else 'kubectl failed'}
|
||||
|
||||
|
||||
import json
|
||||
from sqlor.dbpools import DBPools
|
||||
|
||||
@ -1,23 +1,134 @@
|
||||
'''
|
||||
ray_plugin - Ray 集群管理
|
||||
- Ray 集群部署 (ray start --head / --worker)
|
||||
- 自动扩缩容 (Ray Autoscaler)
|
||||
- Dashboard + Prometheus 监控
|
||||
- 算力单元: Ray resource 标签分配回收
|
||||
'''
|
||||
"""
|
||||
ray_plugin — Ray 集群管理
|
||||
- 控制节点: ray start --head
|
||||
- 算力节点: ray start --address=<head_ip>:6379
|
||||
- 移除: ray stop
|
||||
- 状态: ray status
|
||||
"""
|
||||
from pcc import ssh_exec
|
||||
|
||||
|
||||
async def _node_info(env, node_id):
|
||||
from sqlor.dbpools import DBPools
|
||||
dbname = env.get_module_dbname('pccs')
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('compute_node', {'id': node_id})
|
||||
return recs[0] if recs else None
|
||||
|
||||
|
||||
async def _install_ray(node):
|
||||
"""安装 Ray"""
|
||||
script = (
|
||||
"apt-get update -qq && apt-get install -y -qq python3 python3-pip && "
|
||||
"pip3 install -q ray[default] 2>/dev/null || pip install -q ray[default]"
|
||||
)
|
||||
return await ssh_exec(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', script, timeout=300)
|
||||
|
||||
|
||||
async def deploy_cluster(env, cluster_id, control_nodes, compute_nodes, config):
|
||||
"""Ray 集群部署：仅部署 head 节点"""
|
||||
head_port = config.get('head_port', 6379)
|
||||
dashboard_port = config.get('dashboard_port', 8265)
|
||||
results = {'control_nodes': {}, 'endpoint': ''}
|
||||
|
||||
for node in control_nodes:
|
||||
# 1. 安装 Ray
|
||||
rc, out, err = await _install_ray(node)
|
||||
if rc != 0:
|
||||
results['control_nodes'][node.id] = {'error': f'install failed: {err[:200]}'}
|
||||
continue
|
||||
|
||||
# 2. ray start --head
|
||||
cmd = (
|
||||
f"ray start --head --port={head_port} --dashboard-host=0.0.0.0 "
|
||||
f"--dashboard-port={dashboard_port} --num-cpus={node.cpu_cores or 0} "
|
||||
f"--num-gpus={node.gpu_count or 0}"
|
||||
)
|
||||
rc2, out2, err2 = await ssh_exec(node.ip_address, node.ssh_port or 22,
|
||||
node.ssh_user or 'root', cmd, timeout=120)
|
||||
if rc2 != 0:
|
||||
results['control_nodes'][node.id] = {'error': f'start failed: {err2[:200]}'}
|
||||
continue
|
||||
|
||||
results['control_nodes'][node.id] = {
|
||||
'status': 'ok',
|
||||
'head_address': f'{node.ip_address}:{head_port}',
|
||||
'dashboard': f'http://{node.ip_address}:{dashboard_port}'
|
||||
}
|
||||
results['endpoint'] = f'{node.ip_address}:{head_port}'
|
||||
|
||||
async with DBPools().sqlorContext(env.get_module_dbname('pccs')) as sor:
|
||||
await sor.U('cluster', {'id': cluster_id},
|
||||
{'control_config': json.dumps(results)})
|
||||
return results
|
||||
|
||||
async def deploy_ray_cluster(env, cluster_id, control_nodes, compute_nodes, config):
|
||||
'''部署 Ray 集群: head node → worker nodes'''
|
||||
pass
|
||||
|
||||
async def add_node(env, cluster_id, node_id, role='compute'):
|
||||
'''向 Ray 集群添加 worker'''
|
||||
pass
|
||||
"""向 Ray 集群添加 worker 节点"""
|
||||
node = await _node_info(env, node_id)
|
||||
if not node:
|
||||
return {'status': 'error', 'message': f'Node {node_id} not found'}
|
||||
|
||||
# 获取 head 节点
|
||||
from sqlor.dbpools import DBPools
|
||||
dbname = env.get_module_dbname('pccs')
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('cluster_node', {'cluster_id': cluster_id, 'role': 'control', 'status': 'active'})
|
||||
cluster_recs = await sor.R('cluster', {'id': cluster_id})
|
||||
if not recs:
|
||||
return {'status': 'error', 'message': 'No active control node'}
|
||||
head_node = await _node_info(env, recs[0].node_id)
|
||||
if not head_node:
|
||||
return {'status': 'error', 'message': 'Head node not found'}
|
||||
|
||||
# 从 cluster config 读 head 地址
|
||||
config = json.loads(cluster_recs[0].control_config or '{}') if cluster_recs else {}
|
||||
head_addr = config.get('endpoint', f'{head_node.ip_address}:6379')
|
||||
|
||||
# 1. 安装 Ray
|
||||
rc, _, err = await _install_ray(node)
|
||||
if rc != 0:
|
||||
return {'status': 'error', 'message': f'install failed: {err[:200]}'}
|
||||
|
||||
# 2. ray start --address
|
||||
cmd = (
|
||||
f"ray start --address={head_addr} --num-cpus={node.cpu_cores or 0} "
|
||||
f"--num-gpus={node.gpu_count or 0}"
|
||||
)
|
||||
rc2, out2, err2 = await ssh_exec(node.ip_address, node.ssh_port or 22,
|
||||
node.ssh_user or 'root', cmd, timeout=120)
|
||||
if rc2 != 0:
|
||||
return {'status': 'error', 'message': f'join failed: {err2[:200]}'}
|
||||
|
||||
return {'status': 'ok', 'message': f'Node {node.name} joined Ray cluster'}
|
||||
|
||||
|
||||
async def remove_node(env, cluster_id, node_id):
|
||||
'''从 Ray 集群移除 worker'''
|
||||
pass
|
||||
"""从 Ray 集群移除 worker"""
|
||||
node = await _node_info(env, node_id)
|
||||
if not node:
|
||||
return {'status': 'error', 'message': 'Node not found'}
|
||||
|
||||
rc, _, _ = await ssh_exec(node.ip_address, node.ssh_port or 22,
|
||||
node.ssh_user or 'root', 'ray stop', timeout=60)
|
||||
return {'status': 'ok', 'message': f'Node {node.name} removed from Ray'}
|
||||
|
||||
|
||||
async def cluster_status(env, cluster_id):
|
||||
'''Ray Dashboard API 状态'''
|
||||
pass
|
||||
"""ray status"""
|
||||
from sqlor.dbpools import DBPools
|
||||
dbname = env.get_module_dbname('pccs')
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('cluster_node', {'cluster_id': cluster_id, 'role': 'control', 'status': 'active'})
|
||||
if not recs:
|
||||
return {'error': 'No active control node'}
|
||||
cnode = await _node_info(env, recs[0].node_id)
|
||||
if not cnode:
|
||||
return {'error': 'Control node not found'}
|
||||
rc, out, _ = await ssh_exec(cnode.ip_address, cnode.ssh_port or 22,
|
||||
cnode.ssh_user or 'root', 'ray status', timeout=30)
|
||||
return {'status': 'ok', 'ray_status': out if rc == 0 else 'ray status failed'}
|
||||
|
||||
|
||||
import json
|
||||
from sqlor.dbpools import DBPools
|
||||
|
||||
@ -1,24 +1,178 @@
|
||||
'''
|
||||
slurm_plugin - Slurm 集群管理
|
||||
- Slurm 部署 (slurmctld + slurmd + munge + 共享存储)
|
||||
- Partition 管理 / QoS / TRES
|
||||
- 节点 drain/resume (算力回收/分配)
|
||||
- sinfo / squeue / scontrol 监控
|
||||
'''
|
||||
'')
|
||||
"""
|
||||
slurm_plugin — Slurm 集群管理
|
||||
- 控制节点: slurmctld + munge + NFS server
|
||||
- 算力节点: slurmd + munge + NFS client
|
||||
- 移除: drain node → scontrol delete → reset
|
||||
- 状态: sinfo
|
||||
"""
|
||||
from pcc import ssh_exec
|
||||
|
||||
|
||||
async def _node_info(env, node_id):
|
||||
from sqlor.dbpools import DBPools
|
||||
dbname = env.get_module_dbname('pccs')
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('compute_node', {'id': node_id})
|
||||
return recs[0] if recs else None
|
||||
|
||||
|
||||
async def _install_slurm_deps(node):
|
||||
"""安装 Slurm 依赖 (munge + slurm)"""
|
||||
script = (
|
||||
"apt-get update -qq && "
|
||||
"apt-get install -y -qq munge slurm-wlm slurm-client nfs-common nfs-kernel-server "
|
||||
"&& systemctl enable munge && systemctl start munge"
|
||||
)
|
||||
return await ssh_exec(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', script, timeout=300)
|
||||
|
||||
|
||||
async def _setup_munge(control_node, compute_node):
|
||||
"""复制 munge key 从控制节点到算力节点"""
|
||||
cmd = f"scp -o StrictHostKeyChecking=no /etc/munge/munge.key {compute_node.ssh_user or 'root'}@{compute_node.ip_address}:/etc/munge/munge.key"
|
||||
rc, _, err = await ssh_exec(control_node.ip_address, control_node.ssh_port or 22,
|
||||
control_node.ssh_user or 'root', cmd, timeout=30)
|
||||
if rc == 0:
|
||||
await ssh_exec(compute_node.ip_address, compute_node.ssh_port or 22,
|
||||
compute_node.ssh_user or 'root',
|
||||
"chown munge:munge /etc/munge/munge.key && chmod 400 /etc/munge/munge.key && systemctl restart munge", timeout=30)
|
||||
|
||||
|
||||
async def deploy_cluster(env, cluster_id, control_nodes, compute_nodes, config):
|
||||
"""Slurm 集群部署：控制节点安装 slurmctld + munge"""
|
||||
cluster_name = config.get('cluster_name', 'pccs-cluster')
|
||||
results = {'control_nodes': {}, 'endpoint': ''}
|
||||
|
||||
for node in control_nodes:
|
||||
# 1. 安装依赖
|
||||
rc, out, err = await _install_slurm_deps(node)
|
||||
if rc != 0:
|
||||
results['control_nodes'][node.id] = {'error': f'install failed: {err[:200]}'}
|
||||
continue
|
||||
|
||||
# 2. 生成 slurm.conf
|
||||
cpu = node.cpu_cores or 1
|
||||
mem = node.memory_gb or 4
|
||||
conf = f"""ClusterName={cluster_name}
|
||||
ControlMachine={node.ip_address}
|
||||
SlurmUser=root
|
||||
SlurmctldPort=6817
|
||||
SlurmdPort=6818
|
||||
AuthType=auth/munge
|
||||
StateSaveLocation=/var/spool/slurmctld
|
||||
SlurmdSpoolDir=/var/spool/slurmd
|
||||
ReturnToService=1
|
||||
SchedulerType=sched/backfill
|
||||
SelectType=select/cons_tres
|
||||
SelectTypeParameters=CR_Core
|
||||
AccountingStorageType=accounting_storage/none
|
||||
JobCompType=jobcomp/none
|
||||
NodeName={node.name} CPUs={cpu} RealMemory={mem * 1024} State=UNKNOWN
|
||||
PartitionName=debug Nodes={node.name} Default=YES MaxTime=INFINITE State=UP
|
||||
"""
|
||||
# 写配置到控制节点
|
||||
write_cmd = f"cat > /etc/slurm/slurm.conf << 'SLURM_EOF'\n{conf}\nSLURM_EOF"
|
||||
rc2, _, err2 = await ssh_exec(node.ip_address, node.ssh_port or 22,
|
||||
node.ssh_user or 'root', write_cmd, timeout=30)
|
||||
if rc2 != 0:
|
||||
results['control_nodes'][node.id] = {'error': f'config write failed: {err2}'}
|
||||
continue
|
||||
|
||||
# 3. 启动 slurmctld
|
||||
await ssh_exec(node.ip_address, node.ssh_port or 22,
|
||||
node.ssh_user or 'root',
|
||||
"mkdir -p /var/spool/slurmctld /var/spool/slurmd && "
|
||||
"systemctl enable slurmctld && systemctl start slurmctld", timeout=60)
|
||||
|
||||
results['control_nodes'][node.id] = {'status': 'ok', 'slurm_conf': conf}
|
||||
results['endpoint'] = node.ip_address
|
||||
|
||||
async with DBPools().sqlorContext(env.get_module_dbname('pccs')) as sor:
|
||||
await sor.U('cluster', {'id': cluster_id},
|
||||
{'control_config': json.dumps(results)})
|
||||
return results
|
||||
|
||||
async def deploy_slurm_cluster(env, cluster_id, control_nodes, compute_nodes, config):
|
||||
'''部署 Slurm 集群: 控制节点 slurmctld + munge, 算力节点 slurmd'''
|
||||
pass
|
||||
|
||||
async def add_node(env, cluster_id, node_id, role='compute'):
|
||||
'''向分区添加节点'''
|
||||
pass
|
||||
"""向 Slurm 集群添加算力节点：安装 slurmd + munge → 更新 slurm.conf"""
|
||||
node = await _node_info(env, node_id)
|
||||
if not node:
|
||||
return {'status': 'error', 'message': f'Node {node_id} not found'}
|
||||
|
||||
# 获取控制节点
|
||||
from sqlor.dbpools import DBPools
|
||||
dbname = env.get_module_dbname('pccs')
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('cluster_node', {'cluster_id': cluster_id, 'role': 'control', 'status': 'active'})
|
||||
if not recs:
|
||||
return {'status': 'error', 'message': 'No active control node'}
|
||||
cnode = await _node_info(env, recs[0].node_id)
|
||||
if not cnode:
|
||||
return {'status': 'error', 'message': 'Control node not found'}
|
||||
|
||||
# 1. 安装依赖
|
||||
rc, _, err = await _install_slurm_deps(node)
|
||||
if rc != 0:
|
||||
return {'status': 'error', 'message': f'install failed: {err[:200]}'}
|
||||
|
||||
# 2. 复制 munge key
|
||||
await _setup_munge(cnode, node)
|
||||
|
||||
# 3. 更新控制节点 slurm.conf 添加此节点
|
||||
cpu = node.cpu_cores or 1
|
||||
mem = node.memory_gb or 4
|
||||
add_line = f"NodeName={node.name} CPUs={cpu} RealMemory={mem * 1024} State=UNKNOWN"
|
||||
await ssh_exec(cnode.ip_address, cnode.ssh_port or 22, cnode.ssh_user or 'root',
|
||||
f"echo '{add_line}' >> /etc/slurm/slurm.conf && scontrol reconfigure", timeout=30)
|
||||
|
||||
# 4. 启动 slurmd
|
||||
await ssh_exec(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root',
|
||||
"systemctl enable slurmd && systemctl start slurmd", timeout=60)
|
||||
|
||||
return {'status': 'ok', 'message': f'Node {node.name} joined Slurm cluster'}
|
||||
|
||||
|
||||
async def remove_node(env, cluster_id, node_id):
|
||||
'''drain 节点后从集群移除'''
|
||||
pass
|
||||
"""从 Slurm 集群移除节点"""
|
||||
node = await _node_info(env, node_id)
|
||||
if not node:
|
||||
return {'status': 'error', 'message': 'Node not found'}
|
||||
|
||||
# 获取控制节点
|
||||
from sqlor.dbpools import DBPools
|
||||
dbname = env.get_module_dbname('pccs')
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('cluster_node', {'cluster_id': cluster_id, 'role': 'control'})
|
||||
if recs:
|
||||
cnode = await _node_info(env, recs[0].node_id)
|
||||
if cnode:
|
||||
# drain → remove from slurm.conf → scontrol reconfigure
|
||||
await ssh_exec(cnode.ip_address, cnode.ssh_port or 22, cnode.ssh_user or 'root',
|
||||
f"scontrol update NodeName={node.name} State=DOWN Reason=removing && "
|
||||
f"sed -i '/NodeName={node.name}/d' /etc/slurm/slurm.conf && "
|
||||
"scontrol reconfigure", timeout=30)
|
||||
|
||||
# stop slurmd on the node
|
||||
await ssh_exec(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root',
|
||||
"systemctl stop slurmd && systemctl disable slurmd", timeout=30)
|
||||
|
||||
return {'status': 'ok', 'message': f'Node {node.name} removed from Slurm'}
|
||||
|
||||
|
||||
async def cluster_status(env, cluster_id):
|
||||
'''sinfo + squeue 汇总'''
|
||||
pass
|
||||
"""sinfo"""
|
||||
from sqlor.dbpools import DBPools
|
||||
dbname = env.get_module_dbname('pccs')
|
||||
async with DBPools().sqlorContext(dbname) as sor:
|
||||
recs = await sor.R('cluster_node', {'cluster_id': cluster_id, 'role': 'control', 'status': 'active'})
|
||||
if not recs:
|
||||
return {'error': 'No active control node'}
|
||||
cnode = await _node_info(env, recs[0].node_id)
|
||||
if not cnode:
|
||||
return {'error': 'Control node not found'}
|
||||
rc, out, _ = await ssh_exec(cnode.ip_address, cnode.ssh_port or 22,
|
||||
cnode.ssh_user or 'root', 'sinfo', timeout=30)
|
||||
return {'status': 'ok', 'sinfo': out if rc == 0 else 'sinfo failed'}
|
||||
|
||||
|
||||
import json
|
||||
from sqlor.dbpools import DBPools
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user