yumoqing/pcc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: cluster node assignment with role validation, node reuse prevention, pool-scoped node selection

Browse Source
This commit is contained in:
pccs 2026-08-13 11:52:21 +08:00
parent f171b51402
commit 12f91eb581
13 changed files with 1201 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

264
build/lib/build/lib/build/lib/build/lib/build/lib/build/lib/build/lib/pcc/__init__.py Normal file
View File

@ -0,0 +1,264 @@
"""
pcc — 统一集群管理 (Pooled Computing Cluster)
生命周期: deploy → run → stop → start → destroy
插件: k8s_plugin, slurm_plugin, ray_plugin
"""
import datetime, json, asyncio, subprocess
from appPublic.uniqueID import getID
from sqlor.dbpools import DBPools
MODULE_NAME = 'pccs'
_PLUGINS = {
'k8s': 'pcc.k8s_plugin',
'slurm': 'pcc.slurm_plugin',
'ray': 'pcc.ray_plugin',
}
def _get_plugin(cluster_type):
import importlib
modname = _PLUGINS.get(cluster_type)
if not modname:
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
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': f'Cluster {cluster_id} not found'}
cluster = recs[0]
# 更新状态为 deploying
await sor.U('cluster', {'id': cluster_id},
{'status': 'deploying', 'updated_at': datetime.datetime.now().isoformat()})
# 获取已分配的节点
node_recs = await sor.R('cluster_node', {'cluster_id': cluster_id})
control_nodes = [await sor.R('compute_node', {'id': n.node_id}) for n in node_recs if n.role == 'control']
compute_nodes = [await sor.R('compute_node', {'id': n.node_id}) for n in node_recs if n.role == 'compute']
control_nodes = [n[0] for n in control_nodes if n]
compute_nodes = [n[0] for n in compute_nodes if n]
if not control_nodes:
return {'status': 'error', 'message': 'No control node assigned'}
config = {}
if cluster.control_config:
try:
config = json.loads(cluster.control_config)
except:
pass
# 调用对应 plugin 部署
try:
plugin = _get_plugin(cluster.cluster_type)
deploy_log = await plugin.deploy_cluster(env, cluster_id, control_nodes, compute_nodes, config)
except Exception as e:
async with DBPools().sqlorContext(dbname) as sor:
await sor.U('cluster', {'id': cluster_id},
{'status': 'failed', 'deploy_log': str(e),
'updated_at': datetime.datetime.now().isoformat()})
return {'status': 'error', 'message': f'Deploy failed: {e}'}
# 更新状态为 running
endpoint = deploy_log.get('endpoint', '') if isinstance(deploy_log, dict) else ''
async with DBPools().sqlorContext(dbname) as sor:
await sor.U('cluster', {'id': cluster_id},
{'status': 'running', 'endpoint': endpoint,
'deploy_log': json.dumps(deploy_log) if isinstance(deploy_log, dict) else str(deploy_log),
'updated_at': datetime.datetime.now().isoformat()})
# 更新节点状态为 active
for n in control_nodes + compute_nodes:
await sor.U('cluster_node', {'cluster_id': cluster_id, 'node_id': n.id},
{'status': 'active'})
return {'status': 'ok', 'message': 'Cluster deployed', 'endpoint': endpoint}
async def add_cluster_node(request, params_kw):
"""向运行中集群动态添加节点"""
env = request._run_ns
dbname = env.get_module_dbname(MODULE_NAME)
cluster_id = params_kw.get('cluster_id', '')
node_id = params_kw.get('node_id', '')
role = params_kw.get('role', 'compute')
async with DBPools().sqlorContext(dbname) as sor:
cluster = await sor.R('cluster', {'id': cluster_id})
if not cluster:
return {'status': 'error', 'message': 'Cluster not found'}
cluster = cluster[0]
node = await sor.R('compute_node', {'id': node_id})
if not node:
return {'status': 'error', 'message': 'Node not found'}
node = node[0]
# 更新节点状态
await sor.U('compute_node', {'id': node_id},
{'status': 'allocated', 'updated_at': datetime.datetime.now().isoformat()})
# 记录分配
await sor.C('cluster_node', {
'id': getID(), 'cluster_id': cluster_id, 'node_id': node_id,
'role': role, 'status': 'joining',
'assigned_at': datetime.datetime.now().isoformat()
})
# 调用 plugin 添加
plugin = _get_plugin(cluster.cluster_type)
result = await plugin.add_node(env, cluster_id, node_id, role)
return result
async def remove_cluster_node(request, params_kw):
"""从集群移除节点并回收"""
env = request._run_ns
dbname = env.get_module_dbname(MODULE_NAME)
cluster_id = params_kw.get('cluster_id', '')
node_id = params_kw.get('node_id', '')
async with DBPools().sqlorContext(dbname) as sor:
cluster = await sor.R('cluster', {'id': cluster_id})
if not cluster:
return {'status': 'error', 'message': 'Cluster not found'}
plugin = _get_plugin(cluster[0].cluster_type)
result = await plugin.remove_node(env, cluster_id, node_id)
# 回收节点
async with DBPools().sqlorContext(dbname) as sor:
await sor.U('compute_node', {'id': node_id},
{'status': 'available', 'updated_at': datetime.datetime.now().isoformat()})
await sor.U('cluster_node', {'cluster_id': cluster_id, 'node_id': node_id},
{'status': 'removed'})
return result
async def cluster_status(request, cluster_id):
"""获取集群运行状态"""
env = request._run_ns
dbname = env.get_module_dbname(MODULE_NAME)
async with DBPools().sqlorContext(dbname) as sor:
cluster = await sor.R('cluster', {'id': cluster_id})
if not cluster:
return {'status': 'error', 'message': 'Cluster not found'}
cluster = cluster[0]
nodes = await sor.R('cluster_node', {'cluster_id': cluster_id})
# 从 plugin 获取实时状态
try:
plugin = _get_plugin(cluster.cluster_type)
live_status = await plugin.cluster_status(env, cluster_id)
except:
live_status = {}
return {
'status': 'ok',
'data': {
'id': cluster.id, 'name': cluster.name, 'type': cluster.cluster_type,
'version': cluster.version, 'endpoint': cluster.endpoint,
'state': cluster.status,
'nodes': [{'id': n.node_id, 'role': n.role, 'status': n.status} for n in nodes],
'live': live_status
}
}
async def stop_cluster(request, cluster_id):
"""停止集群"""
env = request._run_ns
dbname = env.get_module_dbname(MODULE_NAME)
async with DBPools().sqlorContext(dbname) as sor:
await sor.U('cluster', {'id': cluster_id},
{'status': 'stopped', 'updated_at': datetime.datetime.now().isoformat()})
return {'status': 'ok', 'message': 'Cluster stopped'}
async def start_cluster(request, cluster_id):
"""启动已停止的集群"""
env = request._run_ns
dbname = env.get_module_dbname(MODULE_NAME)
async with DBPools().sqlorContext(dbname) as sor:
await sor.U('cluster', {'id': cluster_id},
{'status': 'running', 'updated_at': datetime.datetime.now().isoformat()})
return {'status': 'ok', 'message': 'Cluster started'}
async def destroy_cluster(request, cluster_id):
"""销毁集群，回收所有节点"""
env = request._run_ns
dbname = env.get_module_dbname(MODULE_NAME)
async with DBPools().sqlorContext(dbname) as sor:
# 释放所有节点
nodes = await sor.R('cluster_node', {'cluster_id': cluster_id})
for n in nodes:
await sor.U('compute_node', {'id': n.node_id},
{'status': 'available', 'updated_at': datetime.datetime.now().isoformat()})
await sor.U('cluster_node', {'id': n.id}, {'status': 'removed'})
await sor.U('cluster', {'id': cluster_id},
{'status': 'destroyed', 'updated_at': datetime.datetime.now().isoformat()})
return {'status': 'ok', 'message': 'Cluster destroyed'}
async def cluster_status_all(request, params_kw):
"""获取所有集群状态统计"""
env = request._run_ns
dbname = env.get_module_dbname('pcc')
async with DBPools().sqlorContext(dbname) as sor:
clusters = await sor.R('cluster', {})
if not clusters:
return {'status': 'ok', 'data': {'total': 0, 'running': 0, 'deploying': 0, 'failed': 0, 'stopped': 0}}
statuses = {}
for c in clusters:
s = getattr(c, 'status', 'unknown')
statuses[s] = statuses.get(s, 0) + 1
return {'status': 'ok', 'data': {
'total': len(clusters),
'running': statuses.get('running', 0),
'deploying': statuses.get('deploying', 0),
'failed': statuses.get('failed', 0),
'stopped': statuses.get('stopped', 0),
'destroyed': statuses.get('destroyed', 0),
}}

221
build/lib/build/lib/build/lib/build/lib/build/lib/build/lib/build/lib/pcc/k8s_plugin/__init__.py Normal file
View File

@ -0,0 +1,221 @@
"""
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 add_node(env, cluster_id, node_id, role='compute'):
"""向 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):
"""从 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):
"""获取 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
async def allocate_unit(env, cluster_id, unit_name, cpu, memory, gpu=0):
"""K8s 算力单元分配: 创建 namespace + ResourceQuota"""
from pcc import ssh_exec as _ssh
node = await _find_control_node(env, cluster_id)
if not node:
return {'status': 'error', 'message': 'No control node found'}
quota_yaml = (
f"apiVersion: v1\\nkind: ResourceQuota\\nmetadata:\\n name: {unit_name}-quota\\n namespace: {unit_name}\\n"
f"spec:\\n hard:\\n requests.cpu: \\\"{cpu}\\\"\\n requests.memory: \\\"{memory}Gi\\\"\\n requests.nvidia.com/gpu: \\\"{gpu}\\\""
)
cmds = [
f"kubectl create namespace {unit_name}",
f"echo '{quota_yaml}' | kubectl apply -f -",
]
for cmd in cmds:
rc, out, err = await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', cmd)
if rc != 0:
return {'status': 'error', 'message': f'kubectl failed: {err[:200]}'}
return {'status': 'ok', 'message': f'Unit {unit_name}: ns + {cpu}C/{memory}G/{gpu}GPU'}
async def release_unit(env, cluster_id, unit_name):
"""K8s 算力单元回收: 删除 namespace"""
node = await _find_control_node(env, cluster_id)
if not node:
return {'status': 'error', 'message': 'No control node found'}
from pcc import ssh_exec as _ssh
rc, _, err = await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root',
f"kubectl delete namespace {unit_name} --timeout=60s")
if rc != 0:
return {'status': 'error', 'message': f'delete ns failed: {err[:200]}'}
return {'status': 'ok', 'message': f'Unit {unit_name} released'}
async def _find_control_node(env, cluster_id):
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 None
return await _node_info(env, recs[0].node_id)

174
build/lib/build/lib/build/lib/build/lib/build/lib/build/lib/build/lib/pcc/ray_plugin/__init__.py Normal file
View File

@ -0,0 +1,174 @@
"""
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 add_node(env, cluster_id, node_id, role='compute'):
"""向 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"""
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 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
async def allocate_unit(env, cluster_id, unit_name, cpu, memory, gpu=0):
"""Ray 算力单元分配: 创建 detached actor / placement group 绑定资源标签"""
node = await _find_control_node(env, cluster_id)
if not node:
return {'status': 'error', 'message': 'No control node found'}
from pcc import ssh_exec as _ssh
# 用 ray submit 提交一个资源占位 job
script = (
"import ray; ray.init(address='auto'); "
"pg = ray.util.placement_group([{'CPU': " + str(cpu) + ", 'GPU': " + str(gpu) + "}], strategy='STRICT_SPREAD', name='" + unit_name + "'); "
"ray.get(pg.ready()); print('PG ready: " + unit_name + "')"
)
py_cmd = f"python3 -c '{script}'"
rc, out, err = await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', py_cmd, timeout=60)
if rc != 0:
return {'status': 'error', 'message': f'ray pg failed: {err[:200]}'}
return {'status': 'ok', 'message': f'PlacementGroup {unit_name}: {cpu}C/{gpu}GPU'}
async def release_unit(env, cluster_id, unit_name):
"""Ray 算力单元回收: 删除 placement group"""
node = await _find_control_node(env, cluster_id)
if not node:
return {'status': 'error', 'message': 'No control node found'}
from pcc import ssh_exec as _ssh
py_cmd = f"python3 -c 'import ray; ray.init(address=\"auto\"); ray.util.remove_placement_group(ray.util.get_placement_group(\"{unit_name}\")); print(\"removed\")'"
await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', py_cmd, timeout=30)
return {'status': 'ok', 'message': f'PlacementGroup {unit_name} released'}
async def _find_control_node(env, cluster_id):
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 None
return await _node_info(env, recs[0].node_id)

229
build/lib/build/lib/build/lib/build/lib/build/lib/build/lib/build/lib/pcc/slurm_plugin/__init__.py Normal file
View File

@ -0,0 +1,229 @@
"""
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 add_node(env, cluster_id, node_id, role='compute'):
"""向 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):
"""从 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"""
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
async def allocate_unit(env, cluster_id, unit_name, cpu, memory, gpu=0):
"""Slurm 算力单元分配: 创建 partition + 关联节点"""
node = await _find_control_node(env, cluster_id)
if not node:
return {'status': 'error', 'message': 'No control node found'}
from pcc import ssh_exec as _ssh
# 创建 partition
cmd = f"scontrol create PartitionName={unit_name} MaxNodes=UNLIMITED Default=NO MaxTime=UNLIMITED State=UP"
rc, out, err = await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', cmd)
if rc != 0:
return {'status': 'error', 'message': f'scontrol failed: {err[:200]}'}
# 分配 node 到 partition (通过 cluster_node 表找该集群下可用的 compute 节点)
dbname = env.get_module_dbname('pccs')
async with DBPools().sqlorContext(dbname) as sor:
assigned = await sor.sqlExe(
"""SELECT cn.name FROM cluster_node cnn
JOIN compute_node cn ON cn.id = cnn.node_id
WHERE cnn.cluster_id=${cid}$ AND cnn.role='compute' AND cnn.status='active'
LIMIT 1""",
{'cid': cluster_id}
)
if assigned:
node_name = assigned[0].name
await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root',
f"scontrol update NodeName={node_name} Partition={unit_name}")
return {'status': 'ok', 'message': f'Partition {unit_name} created'}
async def release_unit(env, cluster_id, unit_name):
"""Slurm 算力单元回收: 删除 partition"""
node = await _find_control_node(env, cluster_id)
if not node:
return {'status': 'error', 'message': 'No control node found'}
from pcc import ssh_exec as _ssh
await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root',
f"scontrol delete PartitionName={unit_name}")
return {'status': 'ok', 'message': f'Partition {unit_name} deleted'}
async def _find_control_node(env, cluster_id):
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 None
return await _node_info(env, recs[0].node_id)

68
build/lib/build/lib/build/lib/build/lib/build/lib/build/lib/scripts/load_path.py Normal file
View File

@ -0,0 +1,68 @@
#!/usr/bin/env python3
"""pcc RBAC 权限管理"""
import subprocess, os, sys, json, glob
def find_sage_root():
for c in [os.path.expanduser("~/sage"), os.path.expanduser("~/repos/sage")]:
if os.path.isdir(os.path.join(c, "py3")): return c
return None
SAGE = find_sage_root()
if not SAGE: sys.exit("Sage root not found")
PY = os.path.join(SAGE, "py3", "bin", "python")
SET = os.path.join(SAGE, "set_role_perm.py")
JSON_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "json")
API_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "wwwroot", "api")
def load_cruds():
defs = []
for f in sorted(glob.glob(os.path.join(JSON_DIR, "*.json"))):
try:
d = json.load(open(f))
defs.append({"tblname": d["tblname"], "alias": d.get("alias", d["tblname"]), "subtables": d.get("params", {}).get("subtables", [])})
except: pass
return defs
def get_apis():
if not os.path.isdir(API_DIR): return []
return [f"/pcc/api/{f}" for f in sorted(os.listdir(API_DIR)) if f.endswith(".dspy")]
cruds = load_cruds()
apis = get_apis()
PATHS_ANY = [
f"/pcc/menu.ui",
]
PATHS_LOGINED = [
f"/pcc",
f"/pcc/index.ui",
]
for d in cruds:
PATHS_ANY.append(f"/pcc/{d['alias']}")
PATHS_LOGINED.append(f"/pcc/{d['alias']}/index.ui")
for act in ["get", "add", "update", "delete"]:
PATHS_LOGINED.append(f"/pcc/{d['alias']}/{act}_{d['tblname']}.dspy")
for api in apis:
PATHS_LOGINED.append(api)
PATHS_OPERATOR = list(PATHS_LOGINED)
PATHS_ANY = list(dict.fromkeys(PATHS_ANY))
PATHS_LOGINED = list(dict.fromkeys(PATHS_LOGINED))
PATHS_OPERATOR = list(dict.fromkeys(PATHS_OPERATOR))
def reg(role, paths):
ok = 0
for p in paths:
r = subprocess.run([PY, SET, role, p], capture_output=True, text=True)
if r.returncode == 0: ok += 1
print(f" {role}: {ok}/{len(paths)}")
return ok
total = 0
print(f"{mod_name}: any={len(PATHS_ANY)} logined={len(PATHS_LOGINED)} operator={len(PATHS_OPERATOR)}")
total += reg("any", PATHS_ANY)
total += reg("logined", PATHS_LOGINED)
total += reg("reseller.operator", PATHS_OPERATOR)
print(f"Done. {total} entries.")

21
json/cluster_list.json
View File

@ -27,7 +27,26 @@
"status": {
"title": "状态",
"width": 100
}
},
"alters": {
"control_node_id": {
"uitype": "code",
"dataurl": "{{entire_url(\"../api/get_available_nodes.dspy\")}}"
},
"storage_node_id": {
"uitype": "code",
"dataurl": "{{entire_url(\"../api/get_available_nodes.dspy\")}}"
},
"compute_node_id": {
"uitype": "code",
"dataurl": "{{entire_url(\"../api/get_available_nodes.dspy\")}}"
}
},
"exclouded": [
"control_node_id",
"storage_node_id",
"compute_node_id"
]
},
"editexclouded": [
"id",

18
models/cluster.json
View File

@ -56,6 +56,24 @@
"type": "str",
"length": 256
},
{
"name": "control_node_id",
"title": "控制节点",
"type": "str",
"length": 32
},
{
"name": "storage_node_id",
"title": "存储节点",
"type": "str",
"length": 32
},
{
"name": "compute_node_id",
"title": "算力节点",
"type": "str",
"length": 32
},
{
"name": "control_config",
"title": "控制节点配置(JSON)",

55
wwwroot/api/cluster_create.dspy
View File

@ -2,15 +2,68 @@ ns = params_kw.copy()
for k,v in list(ns.items()):
if v == 'NaN' or v == 'null': ns[k] = None
if k.endswith('_text'): ns.pop(k, None)
id = params_kw.get('id','')
if not id or len(str(id)) > 32: id = uuid()
ns['id'] = id
userorgid = await get_userorgid()
if not userorgid:
return {'widgettype':'Error','options':{'title':'Authorization Error','timeout':3,'cwidth':16,'cheight':9,'message':'Please login'}}
pool_id = ns.get('pool_id','')
control_node_id = ns.get('control_node_id','')
storage_node_id = ns.get('storage_node_id','')
compute_node_id = ns.get('compute_node_id','')
if not control_node_id:
return {'widgettype':'Error','options':{'title':'Error','message':'必须选择控制节点','cwidth':16,'cheight':9,'timeout':3}}
if not storage_node_id:
return {'widgettype':'Error','options':{'title':'Error','message':'必须选择存储节点','cwidth':16,'cheight':9,'timeout':3}}
ns['resellerid'] = userorgid
ns['created_at'] = datetime.datetime.now().isoformat()
ns['updated_at'] = ns['created_at']
# 从 ns 移除节点选择字段(非 cluster 表字段)
for k in ('control_node_id','storage_node_id','compute_node_id'):
ns.pop(k, None)
async with DBPools().sqlorContext(get_module_dbname('pcc')) as sor:
# 校验节点可用性
async def check_node(nid, pool_id):
if not nid:
return None
existing = await sor.R('cluster_node', {'node_id': nid})
if existing:
return '节点 ' + nid + ' 已被其他集群占用'
if pool_id:
node = await sor.R('compute_node', {'id': nid})
if node and getattr(node[0],'pool_id','') != pool_id:
return '节点 ' + nid + ' 不属于所选算力池'
return None
err = await check_node(control_node_id, pool_id)
if err: return {'widgettype':'Error','options':{'title':'Error','message':err,'cwidth':16,'cheight':9,'timeout':3}}
err = await check_node(storage_node_id, pool_id)
if err: return {'widgettype':'Error','options':{'title':'Error','message':err,'cwidth':16,'cheight':9,'timeout':3}}
err = await check_node(compute_node_id, pool_id)
if err: return {'widgettype':'Error','options':{'title':'Error','message':err,'cwidth':16,'cheight':9,'timeout':3}}
# 创建集群
await sor.C('cluster', ns)
return {'widgettype':'Message','options':{'cwidth':16,'cheight':9,'title':'Success','timeout':3,'message':'ok'}}
# 自动分配节点
now = datetime.datetime.now().isoformat()
assignments = []
if control_node_id:
assignments.append({'id': uuid(), 'cluster_id': id, 'node_id': control_node_id, 'role': 'control', 'status': 'joining', 'assigned_at': now})
if storage_node_id:
assignments.append({'id': uuid(), 'cluster_id': id, 'node_id': storage_node_id, 'role': 'storage', 'status': 'joining', 'assigned_at': now})
if compute_node_id:
assignments.append({'id': uuid(), 'cluster_id': id, 'node_id': compute_node_id, 'role': 'compute', 'status': 'joining', 'assigned_at': now})
for a in assignments:
await sor.C('cluster_node', a)
return {'widgettype':'Message','options':{'cwidth':16,'cheight':9,'title':'Success','timeout':3,'message':'集群已创建，已分配 ' + str(len(assignments)) + ' 个节点'}}

30
wwwroot/api/cluster_node_create.dspy
View File

@ -2,13 +2,41 @@ ns = params_kw.copy()
for k,v in list(ns.items()):
if v == 'NaN' or v == 'null': ns[k] = None
if k.endswith('_text'): ns.pop(k, None)
id = params_kw.get('id','')
if not id or len(str(id)) > 32: id = uuid()
ns['id'] = id
userorgid = await get_userorgid()
if not userorgid:
return {'widgettype':'Error','options':{'title':'Authorization Error','timeout':3,'cwidth':16,'cheight':9,'message':'Please login'}}
ns['allocated_at'] = datetime.datetime.now().isoformat()
cluster_id = ns.get('cluster_id','')
node_id = ns.get('node_id','')
role = ns.get('role','')
if not cluster_id:
return {'widgettype':'Error','options':{'title':'Error','message':'缺少集群ID','cwidth':16,'cheight':9,'timeout':3}}
if not node_id:
return {'widgettype':'Error','options':{'title':'Error','message':'缺少节点ID','cwidth':16,'cheight':9,'timeout':3}}
if role not in ('control','compute','storage'):
return {'widgettype':'Error','options':{'title':'Error','message':'角色必须是 control/compute/storage','cwidth':16,'cheight':9,'timeout':3}}
async with DBPools().sqlorContext(get_module_dbname('pcc')) as sor:
# 校验节点是否已被其他集群占用
existing = await sor.R('cluster_node', {'node_id': node_id})
if existing:
return {'widgettype':'Error','options':{'title':'Error','message':'该节点已被集群占用，不可重复添加','cwidth':16,'cheight':9,'timeout':3}}
# 校验节点是否属于该集群的算力池
cluster = await sor.R('cluster', {'id': cluster_id})
if cluster:
pool_id = getattr(cluster[0], 'pool_id', '')
node = await sor.R('compute_node', {'id': node_id})
if node and pool_id and getattr(node[0], 'pool_id', '') != pool_id:
return {'widgettype':'Error','options':{'title':'Error','message':'节点不属于该集群的算力池','cwidth':16,'cheight':9,'timeout':3}}
ns['assigned_at'] = datetime.datetime.now().isoformat()
ns.pop('allocated_at', None)
await sor.C('cluster_node', ns)
return {'widgettype':'Message','options':{'cwidth':16,'cheight':9,'title':'Success','timeout':3,'message':'ok'}}

18
wwwroot/api/get_available_nodes.dspy Normal file
View File

@ -0,0 +1,18 @@
pool_id = params_kw.get('pool_id','')
try:
async with DBPools().sqlorContext(get_module_dbname('pcpool')) as sor:
if pool_id:
nodes = await sor.R('compute_node', {'pool_id': pool_id})
else:
nodes = await sor.R('compute_node', {})
assigned_ids = set()
try:
async with DBPools().sqlorContext(get_module_dbname('pcc')) as sor:
assigned = await sor.R('cluster_node', {})
assigned_ids = set(str(a.node_id) for a in (assigned or []))
except:
pass
available = [{'value': str(n.id), 'text': str(n.name)} for n in (nodes or []) if str(n.id) not in assigned_ids]
return available
except:
return []

18
wwwroot/cluster_list/get_cluster.dspy
View File

@ -80,6 +80,24 @@ if not filterjson:
"type": "str",
"length": 256
},
{
"name": "control_node_id",
"title": "控制节点",
"type": "str",
"length": 32
},
{
"name": "storage_node_id",
"title": "存储节点",
"type": "str",
"length": 32
},
{
"name": "compute_node_id",
"title": "算力节点",
"type": "str",
"length": 32
},
{
"name": "control_config",
"title": "控制节点配置(JSON)",

54
wwwroot/cluster_list/index.ui
View File

@ -71,7 +71,26 @@
"status": {
"title": "状态",
"width": 100
}
},
"alters": {
"control_node_id": {
"uitype": "code",
"dataurl": "{{entire_url(\"../api/get_available_nodes.dspy\")}}"
},
"storage_node_id": {
"uitype": "code",
"dataurl": "{{entire_url(\"../api/get_available_nodes.dspy\")}}"
},
"compute_node_id": {
"uitype": "code",
"dataurl": "{{entire_url(\"../api/get_available_nodes.dspy\")}}"
}
},
"exclouded": [
"control_node_id",
"storage_node_id",
"compute_node_id"
]
},
@ -176,6 +195,39 @@
"datatype": "str",
"label": "API入口地址"
},
{
"name": "control_node_id",
"title": "控制节点",
"type": "str",
"length": 32,
"cwidth": 18,
"uitype": "code",
"datatype": "str",
"label": "控制节点",
"dataurl": "{{entire_url(\"../api/get_available_nodes.dspy\")}}"
},
{
"name": "storage_node_id",
"title": "存储节点",
"type": "str",
"length": 32,
"cwidth": 18,
"uitype": "code",
"datatype": "str",
"label": "存储节点",
"dataurl": "{{entire_url(\"../api/get_available_nodes.dspy\")}}"
},
{
"name": "compute_node_id",
"title": "算力节点",
"type": "str",
"length": 32,
"cwidth": 18,
"uitype": "code",
"datatype": "str",
"label": "算力节点",
"dataurl": "{{entire_url(\"../api/get_available_nodes.dspy\")}}"
},
{
"name": "control_config",
"title": "控制节点配置(JSON)",

80
wwwroot/cluster_node_list/add_cluster_node.dspy
View File

@ -1,52 +1,42 @@
ns = params_kw.copy()
for k,v in ns.items():
if v == 'NaN' or v == 'null':
ns[k] = None
id = params_kw.id
if not id or len(id) > 32:
id = uuid()
for k,v in list(ns.items()):
if v == 'NaN' or v == 'null': ns[k] = None
if k.endswith('_text'): ns.pop(k, None)
id = params_kw.get('id','')
if not id or len(str(id)) > 32: id = uuid()
ns['id'] = id
userorgid = await get_userorgid()
if not userorgid:
return {
"widgettype":"Error",
"options":{
"title":"Authorization Error",
"timeout":3,
"cwidth":16,
"cheight":9,
"message":"Please login"
}
}
ns['resellerid'] = userorgid
return {'widgettype':'Error','options':{'title':'Authorization Error','timeout':3,'cwidth':16,'cheight':9,'message':'Please login'}}
db = DBPools()
dbname = get_module_dbname('pcc')
async with db.sqlorContext(dbname) as sor:
r = await sor.C('cluster_node', ns.copy())
return {
"widgettype":"Message",
"options":{
"user_data":ns,
"cwidth":16,
"cheight":9,
"title":"Add Success",
"timeout":3,
"message":"ok"
}
}
cluster_id = ns.get('cluster_id','')
node_id = ns.get('node_id','')
role = ns.get('role','')
return {
"widgettype":"Error",
"options":{
"title":"Add Error",
"cwidth":16,
"cheight":9,
"timeout":3,
"message":"failed"
}
}
if not cluster_id:
return {'widgettype':'Error','options':{'title':'Error','message':'缺少集群ID','cwidth':16,'cheight':9,'timeout':3}}
if not node_id:
return {'widgettype':'Error','options':{'title':'Error','message':'缺少节点ID','cwidth':16,'cheight':9,'timeout':3}}
if role not in ('control','compute','storage'):
return {'widgettype':'Error','options':{'title':'Error','message':'角色必须是 control/compute/storage','cwidth':16,'cheight':9,'timeout':3}}
async with DBPools().sqlorContext(get_module_dbname('pcc')) as sor:
# 校验节点是否已被其他集群占用
existing = await sor.R('cluster_node', {'node_id': node_id})
if existing:
return {'widgettype':'Error','options':{'title':'Error','message':'该节点已被集群占用，不可重复添加','cwidth':16,'cheight':9,'timeout':3}}
# 校验节点是否属于该集群的算力池
cluster = await sor.R('cluster', {'id': cluster_id})
if cluster:
pool_id = getattr(cluster[0], 'pool_id', '')
node = await sor.R('compute_node', {'id': node_id})
if node and pool_id and getattr(node[0], 'pool_id', '') != pool_id:
return {'widgettype':'Error','options':{'title':'Error','message':'节点不属于该集群的算力池','cwidth':16,'cheight':9,'timeout':3}}
ns['assigned_at'] = datetime.datetime.now().isoformat()
ns.pop('allocated_at', None)
await sor.C('cluster_node', ns)
return {'widgettype':'Message','options':{'cwidth':16,'cheight':9,'title':'Success','timeout':3,'message':'ok'}}