From b643b6de0b07d46ac69ad017023875fc7fc4ba46 Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Fri, 7 Aug 2026 12:40:59 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20k8s/slurm/ray=20plugin=20SSH=E9=83=A8?=
 =?UTF-8?q?=E7=BD=B2=E5=AE=9E=E7=8E=B0=20=E2=80=94=20=E6=8E=A7=E5=88=B6?=
 =?UTF-8?q?=E8=8A=82=E7=82=B9=E9=83=A8=E7=BD=B2+=E7=AE=97=E5=8A=9B?=
 =?UTF-8?q?=E8=8A=82=E7=82=B9=E5=8A=A0=E5=85=A5+=E7=A7=BB=E9=99=A4+?=
 =?UTF-8?q?=E7=8A=B6=E6=80=81=E7=9B=91=E6=8E=A7?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pcc/__init__.py              |  35 ++++++-
 pcc/k8s_plugin/__init__.py   | 186 ++++++++++++++++++++++++++++++----
 pcc/ray_plugin/__init__.py   | 143 +++++++++++++++++++++++---
 pcc/slurm_plugin/__init__.py | 188 +++++++++++++++++++++++++++++++----
 4 files changed, 500 insertions(+), 52 deletions(-)

diff --git a/pcc/__init__.py b/pcc/__init__.py
index fc2430c..05f975d 100644
--- a/pcc/__init__.py
+++ b/pcc/__init__.py
@@ -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
diff --git a/pcc/k8s_plugin/__init__.py b/pcc/k8s_plugin/__init__.py
index 6e99b5d..148dece 100644
--- a/pcc/k8s_plugin/__init__.py
+++ b/pcc/k8s_plugin/__init__.py
@@ -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
diff --git a/pcc/ray_plugin/__init__.py b/pcc/ray_plugin/__init__.py
index 1bff9f6..972f976 100644
--- a/pcc/ray_plugin/__init__.py
+++ b/pcc/ray_plugin/__init__.py
@@ -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
diff --git a/pcc/slurm_plugin/__init__.py b/pcc/slurm_plugin/__init__.py
index a880993..ae487ad 100644
--- a/pcc/slurm_plugin/__init__.py
+++ b/pcc/slurm_plugin/__init__.py
@@ -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