|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
from sqlalchemy import select, delete, func, text
|
|
|
from ..db import get_db
|
|
|
from ..models.clusters import Cluster
|
|
|
from ..models.nodes import Node
|
|
|
from ..deps.auth import get_current_user, PermissionChecker
|
|
|
from ..services.ssh_probe import check_ssh_connectivity, get_hdfs_cluster_id
|
|
|
from pydantic import BaseModel
|
|
|
from datetime import datetime, timezone
|
|
|
import uuid as uuidlib
|
|
|
from ..config import now_bj
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
|
|
|
def _get_username(u) -> str:
|
|
|
return getattr(u, "username", None) or (u.get("username") if isinstance(u, dict) else None)
|
|
|
|
|
|
|
|
|
class NodeCreateItem(BaseModel):
|
|
|
hostname: str
|
|
|
ip_address: str
|
|
|
ssh_user: str
|
|
|
ssh_password: str
|
|
|
description: str | None = None
|
|
|
|
|
|
class ClusterCreateRequest(BaseModel):
|
|
|
name: str
|
|
|
type: str
|
|
|
node_count: int
|
|
|
health_status: str
|
|
|
cpu_avg: float | None = None
|
|
|
memory_avg: float | None = None
|
|
|
description: str | None = None
|
|
|
namenode_ip: str | None = None
|
|
|
namenode_psw: str | None = None
|
|
|
rm_ip: str | None = None
|
|
|
rm_psw: str | None = None
|
|
|
nodes: list[NodeCreateItem]
|
|
|
|
|
|
|
|
|
@router.get("/clusters")
|
|
|
async def list_clusters(user=Depends(get_current_user), db: AsyncSession = Depends(get_db)):
|
|
|
"""按当前用户归属返回其可访问的集群列表。"""
|
|
|
try:
|
|
|
name = _get_username(user)
|
|
|
uid_res = await db.execute(text("SELECT id FROM users WHERE username=:un LIMIT 1"), {"un": name})
|
|
|
uid_row = uid_res.first()
|
|
|
if not uid_row:
|
|
|
return {"clusters": []}
|
|
|
ids_res = await db.execute(text("SELECT cluster_id FROM user_cluster_mapping WHERE user_id=:uid"), {"uid": uid_row[0]})
|
|
|
cluster_ids = [r[0] for r in ids_res.all()]
|
|
|
if not cluster_ids:
|
|
|
return {"clusters": []}
|
|
|
result = await db.execute(select(Cluster).where(Cluster.id.in_(cluster_ids)))
|
|
|
rows = result.scalars().all()
|
|
|
data = []
|
|
|
for c in rows:
|
|
|
data.append({
|
|
|
"uuid": str(c.uuid),
|
|
|
"name": c.name,
|
|
|
"type": c.type,
|
|
|
"node_count": c.node_count,
|
|
|
"health_status": c.health_status,
|
|
|
"cpu_avg": c.cpu_avg,
|
|
|
"memory_avg": c.memory_avg,
|
|
|
"namenode_ip": (str(c.namenode_ip) if c.namenode_ip else None),
|
|
|
"namenode_psw": c.namenode_psw,
|
|
|
"rm_ip": (str(c.rm_ip) if c.rm_ip else None),
|
|
|
"rm_psw": c.rm_psw,
|
|
|
"description": c.description,
|
|
|
})
|
|
|
return {"clusters": data}
|
|
|
except HTTPException:
|
|
|
raise
|
|
|
except Exception:
|
|
|
raise HTTPException(status_code=500, detail="server_error")
|
|
|
|
|
|
|
|
|
@router.post("/clusters")
|
|
|
async def create_cluster(
|
|
|
req: ClusterCreateRequest,
|
|
|
user=Depends(PermissionChecker(["cluster:register"])),
|
|
|
db: AsyncSession = Depends(get_db)
|
|
|
):
|
|
|
"""注册一个集群并建立当前用户的归属映射。"""
|
|
|
try:
|
|
|
name = _get_username(user)
|
|
|
# 移除硬编码的角色检查，PermissionChecker 已经处理了权限校验
|
|
|
|
|
|
# 参数校验：类型与状态
|
|
|
valid_types = {"hadoop", "spark", "kubernetes"}
|
|
|
valid_health = {"healthy", "warning", "error", "unknown"}
|
|
|
errors: list[dict] = []
|
|
|
if req.type not in valid_types:
|
|
|
errors.append({"field": "type", "message": "类型不合法，应为 hadoop/spark/kubernetes", "step": "参数校验"})
|
|
|
if req.health_status not in valid_health:
|
|
|
errors.append({"field": "health_status", "message": "状态不合法，应为 healthy/warning/error/unknown", "step": "参数校验"})
|
|
|
if req.node_count is None or req.node_count < 0:
|
|
|
errors.append({"field": "node_count", "message": "节点总数必须为非负整数", "step": "参数校验"})
|
|
|
if errors:
|
|
|
raise HTTPException(status_code=400, detail={"errors": errors})
|
|
|
|
|
|
# 1. 获取 HDFS 集群真实 UUID (从 NameNode 获取)
|
|
|
cluster_uuid, err = get_hdfs_cluster_id(str(req.namenode_ip), req.nodes[0].ssh_user, req.nodes[0].ssh_password)
|
|
|
if not cluster_uuid:
|
|
|
raise HTTPException(status_code=400, detail={"errors": [{"field": "namenode_ip", "message": f"无法获取集群ID: {err}"}]})
|
|
|
|
|
|
# 2. 检查该 UUID 是否已在数据库中
|
|
|
res = await db.execute(select(Cluster).where(Cluster.uuid == cluster_uuid).limit(1))
|
|
|
existing_cluster = res.scalars().first()
|
|
|
|
|
|
if existing_cluster:
|
|
|
# 集群已存在，仅建立用户映射
|
|
|
c = existing_cluster
|
|
|
new_uuid = cluster_uuid
|
|
|
else:
|
|
|
# 集群不存在，执行注册流程
|
|
|
# 检查集群名称是否已存在
|
|
|
name_exists = await db.execute(select(Cluster.id).where(Cluster.name == req.name).limit(1))
|
|
|
if name_exists.scalars().first():
|
|
|
raise HTTPException(status_code=400, detail={"errors": [{"field": "name", "message": "集群名称已存在"}]})
|
|
|
|
|
|
# SSH 连通性预检查
|
|
|
ssh_errors: list[dict] = []
|
|
|
for idx, n_req in enumerate(req.nodes):
|
|
|
ip = getattr(n_req, "ip_address", None) or getattr(n_req, "ip", None)
|
|
|
user_ = getattr(n_req, "ssh_user", None)
|
|
|
pwd_ = getattr(n_req, "ssh_password", None)
|
|
|
ok, conn_err = check_ssh_connectivity(str(ip), str(user_ or ""), str(pwd_ or ""))
|
|
|
if not ok:
|
|
|
ssh_errors.append({
|
|
|
"field": f"nodes[{idx}].ssh",
|
|
|
"message": "注册失败：SSH不可连接",
|
|
|
"step": "connect",
|
|
|
"detail": conn_err,
|
|
|
"hostname": getattr(n_req, "hostname", None),
|
|
|
"ip": str(ip) if ip is not None else None,
|
|
|
})
|
|
|
if ssh_errors:
|
|
|
raise HTTPException(status_code=400, detail={"errors": ssh_errors})
|
|
|
|
|
|
new_uuid = cluster_uuid
|
|
|
|
|
|
c = Cluster(
|
|
|
uuid=new_uuid,
|
|
|
name=req.name,
|
|
|
type=req.type,
|
|
|
node_count=req.node_count,
|
|
|
health_status=req.health_status,
|
|
|
cpu_avg=req.cpu_avg,
|
|
|
memory_avg=req.memory_avg,
|
|
|
namenode_ip=req.namenode_ip,
|
|
|
namenode_psw=req.namenode_psw,
|
|
|
rm_ip=req.rm_ip,
|
|
|
rm_psw=req.rm_psw,
|
|
|
description=req.description,
|
|
|
config_info={},
|
|
|
created_at=now_bj(),
|
|
|
updated_at=now_bj(),
|
|
|
)
|
|
|
db.add(c)
|
|
|
await db.flush() # 获取 c.id
|
|
|
|
|
|
# 插入节点
|
|
|
for n_req in req.nodes:
|
|
|
node_uuid = str(uuidlib.uuid4())
|
|
|
node = Node(
|
|
|
uuid=node_uuid,
|
|
|
cluster_id=c.id,
|
|
|
hostname=n_req.hostname,
|
|
|
ip_address=n_req.ip_address,
|
|
|
ssh_user=n_req.ssh_user,
|
|
|
ssh_password=n_req.ssh_password,
|
|
|
status="unknown",
|
|
|
created_at=now_bj(),
|
|
|
updated_at=now_bj(),
|
|
|
)
|
|
|
db.add(node)
|
|
|
|
|
|
# 3. 建立用户映射 (无论集群是新注册还是已存在)
|
|
|
uid_res = await db.execute(text("SELECT id FROM users WHERE username=:un LIMIT 1"), {"un": name})
|
|
|
uid_row = uid_res.first()
|
|
|
# 简化逻辑：如果是 admin 用户则赋予 admin 角色，否则赋予 operator 角色
|
|
|
role_key = "admin" if name == "admin" else "operator"
|
|
|
rid_res = await db.execute(text("SELECT id FROM roles WHERE role_key=:rk LIMIT 1"), {"rk": role_key})
|
|
|
rid_row = rid_res.first()
|
|
|
|
|
|
if uid_row and rid_row:
|
|
|
await db.execute(
|
|
|
text("INSERT INTO user_cluster_mapping(user_id, cluster_id, role_id) VALUES (:uid,:cid,:rid) ON CONFLICT (user_id, cluster_id) DO NOTHING"),
|
|
|
{"uid": uid_row[0], "cid": c.id, "rid": rid_row[0]}
|
|
|
)
|
|
|
|
|
|
await db.commit()
|
|
|
|
|
|
return {
|
|
|
"status": "success",
|
|
|
"message": "集群注册成功" if not existing_cluster else "集群已关联至当前用户",
|
|
|
"uuid": new_uuid
|
|
|
}
|
|
|
except HTTPException:
|
|
|
raise
|
|
|
except Exception as e:
|
|
|
import traceback
|
|
|
traceback.print_exc()
|
|
|
raise HTTPException(status_code=500, detail="server_error")
|
|
|
|
|
|
|
|
|
@router.delete("/clusters/{uuid}")
|
|
|
async def delete_cluster(
|
|
|
uuid: str,
|
|
|
user=Depends(PermissionChecker(["cluster:delete"])),
|
|
|
db: AsyncSession = Depends(get_db)
|
|
|
):
|
|
|
"""注销指定集群，并清理用户归属映射。"""
|
|
|
try:
|
|
|
name = _get_username(user)
|
|
|
# 移除硬编码的角色检查
|
|
|
try:
|
|
|
uo = uuidlib.UUID(uuid)
|
|
|
except Exception:
|
|
|
raise HTTPException(status_code=400, detail={"errors": [{"field": "uuid", "message": "UUID 格式不正确"}]})
|
|
|
res = await db.execute(select(Cluster).where(Cluster.uuid == str(uo)).limit(1))
|
|
|
c = res.scalars().first()
|
|
|
if not c:
|
|
|
return {"ok": True}
|
|
|
await db.execute(delete(Cluster).where(Cluster.id == c.id))
|
|
|
await db.execute(text("DELETE FROM user_cluster_mapping WHERE cluster_id=:cid"), {"cid": c.id})
|
|
|
await db.commit()
|
|
|
return {"ok": True}
|
|
|
except HTTPException:
|
|
|
raise
|
|
|
except Exception:
|
|
|
raise HTTPException(status_code=500, detail="server_error")
|
