|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
from sqlalchemy import select, text
|
|
|
from pydantic import BaseModel, Field
|
|
|
from datetime import datetime, timezone
|
|
|
import shlex
|
|
|
import uuid as uuidlib
|
|
|
import asyncio
|
|
|
|
|
|
from ..db import get_db
|
|
|
from ..deps.auth import get_current_user, PermissionChecker
|
|
|
from ..models.nodes import Node
|
|
|
from ..models.clusters import Cluster
|
|
|
from ..models.sys_exec_logs import SysExecLog
|
|
|
from ..models.hadoop_exec_logs import HadoopExecLog
|
|
|
from ..services.runner import run_remote_command
|
|
|
from ..ssh_utils import SSHClient
|
|
|
from ..config import now_bj
|
|
|
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
|
|
|
def _now() -> datetime:
|
|
|
"""返回当前 UTC 时间。"""
|
|
|
return now_bj()
|
|
|
|
|
|
|
|
|
def _get_username(u) -> str:
|
|
|
"""提取用户名。"""
|
|
|
return getattr(u, "username", None) or (u.get("username") if isinstance(u, dict) else None) or "system"
|
|
|
|
|
|
|
|
|
def _require_ops(u):
|
|
|
"""校验用户是否具有运维权限。"""
|
|
|
name = _get_username(u)
|
|
|
if name not in {"admin", "ops"}:
|
|
|
raise HTTPException(status_code=403, detail="not_allowed")
|
|
|
|
|
|
|
|
|
async def _find_accessible_node(db: AsyncSession, user_name: str, hostname: str) -> Node | None:
|
|
|
"""在用户可访问的集群中查找指定主机名的节点。"""
|
|
|
uid_res = await db.execute(text("SELECT id FROM users WHERE username=:un LIMIT 1"), {"un": user_name})
|
|
|
uid_row = uid_res.first()
|
|
|
if not uid_row:
|
|
|
return None
|
|
|
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 None
|
|
|
res = await db.execute(select(Node).where(Node.hostname == hostname, Node.cluster_id.in_(cluster_ids)).limit(1))
|
|
|
return res.scalars().first()
|
|
|
|
|
|
|
|
|
def _gen_exec_id() -> str:
|
|
|
"""生成执行记录ID。"""
|
|
|
return uuidlib.uuid4().hex[:32]
|
|
|
|
|
|
|
|
|
class ReadLogReq(BaseModel):
|
|
|
node: str = Field(..., description="目标节点主机名")
|
|
|
path: str = Field(..., description="日志文件路径")
|
|
|
lines: int = Field(200, ge=1, le=5000, description="读取行数")
|
|
|
pattern: str | None = Field(None, description="可选过滤正则")
|
|
|
sshUser: str | None = Field(None, description="SSH 用户名（可选）")
|
|
|
timeout: int = Field(20, ge=1, le=120, description="命令超时时间")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def _write_exec_log(db: AsyncSession, operation_id: str, description: str, user_id: int):
|
|
|
"""写入系统操作日志。"""
|
|
|
row = SysExecLog(
|
|
|
user_id=user_id,
|
|
|
description=description,
|
|
|
operation_time=_now()
|
|
|
)
|
|
|
db.add(row)
|
|
|
await db.flush()
|
|
|
await db.commit()
|
|
|
|
|
|
|
|
|
@router.post("/ops/read-log")
|
|
|
async def read_log(req: ReadLogReq, user=Depends(get_current_user), db: AsyncSession = Depends(get_db)):
|
|
|
"""读取远端日志文件内容，支持可选筛选。"""
|
|
|
try:
|
|
|
_require_ops(user)
|
|
|
uname = _get_username(user)
|
|
|
# 假设这里需要 user_id，从 user 对象获取或查询
|
|
|
user_id = getattr(user, "id", 1)
|
|
|
node = await _find_accessible_node(db, uname, req.node)
|
|
|
if not node:
|
|
|
raise HTTPException(status_code=404, detail="node_not_found")
|
|
|
path_q = shlex.quote(req.path)
|
|
|
cmd = f"tail -n {req.lines} {path_q}"
|
|
|
if req.pattern:
|
|
|
pat_q = shlex.quote(req.pattern)
|
|
|
cmd = f"{cmd} | grep -E {pat_q}"
|
|
|
|
|
|
start = _now()
|
|
|
code, out, err = await run_remote_command(str(getattr(node, "ip_address", "")), req.sshUser or "", cmd, timeout=req.timeout)
|
|
|
|
|
|
desc = f"Read log: {req.path} on {req.node} (Exit: {code})"
|
|
|
await _write_exec_log(db, None, desc, user_id)
|
|
|
|
|
|
if code != 0:
|
|
|
raise HTTPException(status_code=500, detail="exec_failed")
|
|
|
lines = [ln for ln in out.splitlines()]
|
|
|
return {"exitCode": code, "lines": lines}
|
|
|
except HTTPException:
|
|
|
raise
|
|
|
except Exception:
|
|
|
raise HTTPException(status_code=500, detail="server_error")
|
|
|
|
|
|
|
|
|
async def _write_hadoop_exec_log(db: AsyncSession, user_id: int, cluster_name: str, description: str, start_time: datetime, end_time: datetime):
|
|
|
"""写入 Hadoop 执行审计日志。"""
|
|
|
row = HadoopExecLog(
|
|
|
from_user_id=user_id,
|
|
|
cluster_name=cluster_name,
|
|
|
description=description,
|
|
|
start_time=start_time,
|
|
|
end_time=end_time
|
|
|
)
|
|
|
db.add(row)
|
|
|
await db.flush()
|
|
|
await db.commit()
|
|
|
|
|
|
|
|
|
@router.post("/ops/clusters/{cluster_uuid}/start")
|
|
|
async def start_cluster(
|
|
|
cluster_uuid: str,
|
|
|
user=Depends(PermissionChecker(["cluster:start"])),
|
|
|
db: AsyncSession = Depends(get_db)
|
|
|
):
|
|
|
"""启动集群：在 NameNode 执行 hsfsstart，在 ResourceManager 执行 yarnstart。"""
|
|
|
try:
|
|
|
# UUID 格式校验
|
|
|
try:
|
|
|
uuidlib.UUID(cluster_uuid)
|
|
|
except ValueError:
|
|
|
raise HTTPException(status_code=400, detail="invalid_uuid_format")
|
|
|
|
|
|
uname = _get_username(user)
|
|
|
user_id = getattr(user, "id", 1)
|
|
|
|
|
|
# 1. 查找集群
|
|
|
res = await db.execute(select(Cluster).where(Cluster.uuid == cluster_uuid).limit(1))
|
|
|
cluster = res.scalars().first()
|
|
|
if not cluster:
|
|
|
raise HTTPException(status_code=404, detail="cluster_not_found")
|
|
|
|
|
|
# 2. 获取 SSH 用户 (从关联节点中获取，默认为 hadoop)
|
|
|
node_res = await db.execute(select(Node).where(Node.cluster_id == cluster.id).limit(1))
|
|
|
node = node_res.scalars().first()
|
|
|
ssh_user = node.ssh_user if node and node.ssh_user else "hadoop"
|
|
|
|
|
|
start_time = _now()
|
|
|
logs = []
|
|
|
|
|
|
# 3. 在 NameNode 执行 start-dfs.sh
|
|
|
if cluster.namenode_ip and cluster.namenode_psw:
|
|
|
try:
|
|
|
def run_nn_start():
|
|
|
with SSHClient(str(cluster.namenode_ip), ssh_user, cluster.namenode_psw) as client:
|
|
|
return client.execute_command("start-dfs.sh")
|
|
|
out, err = await asyncio.to_thread(run_nn_start)
|
|
|
logs.append(f"NameNode ({cluster.namenode_ip}) start: {out} {err}")
|
|
|
except Exception as e:
|
|
|
logs.append(f"NameNode ({cluster.namenode_ip}) start failed: {str(e)}")
|
|
|
|
|
|
# 4. 在 ResourceManager 执行 start-yarn.sh
|
|
|
if cluster.rm_ip and cluster.rm_psw:
|
|
|
try:
|
|
|
def run_rm_start():
|
|
|
with SSHClient(str(cluster.rm_ip), ssh_user, cluster.rm_psw) as client:
|
|
|
return client.execute_command("start-yarn.sh")
|
|
|
out, err = await asyncio.to_thread(run_rm_start)
|
|
|
logs.append(f"ResourceManager ({cluster.rm_ip}) start: {out} {err}")
|
|
|
except Exception as e:
|
|
|
logs.append(f"ResourceManager ({cluster.rm_ip}) start failed: {str(e)}")
|
|
|
|
|
|
end_time = _now()
|
|
|
|
|
|
# 5. 更新集群状态 (仅当所有尝试都未抛出异常时)
|
|
|
# 改进：检查是否有失败日志
|
|
|
has_failed = any("failed" in log.lower() for log in logs)
|
|
|
if not has_failed:
|
|
|
cluster.health_status = "healthy"
|
|
|
else:
|
|
|
cluster.health_status = "error"
|
|
|
|
|
|
cluster.updated_at = end_time
|
|
|
await db.flush()
|
|
|
|
|
|
# 6. 记录日志
|
|
|
full_desc = " | ".join(logs)
|
|
|
await _write_hadoop_exec_log(db, user_id, cluster.name, f"Start Cluster: {full_desc}", start_time, end_time)
|
|
|
|
|
|
return {"status": "success", "logs": logs}
|
|
|
except HTTPException:
|
|
|
raise
|
|
|
except Exception as e:
|
|
|
print(f"Error starting cluster: {e}")
|
|
|
raise HTTPException(status_code=500, detail="server_error")
|
|
|
|
|
|
|
|
|
@router.post("/ops/clusters/{cluster_uuid}/stop")
|
|
|
async def stop_cluster(
|
|
|
cluster_uuid: str,
|
|
|
user=Depends(PermissionChecker(["cluster:stop"])),
|
|
|
db: AsyncSession = Depends(get_db)
|
|
|
):
|
|
|
"""停止集群：在 NameNode 执行 hsfsstop，在 ResourceManager 执行 yarnstop。"""
|
|
|
try:
|
|
|
# UUID 格式校验
|
|
|
try:
|
|
|
uuidlib.UUID(cluster_uuid)
|
|
|
except ValueError:
|
|
|
raise HTTPException(status_code=400, detail="invalid_uuid_format")
|
|
|
|
|
|
uname = _get_username(user)
|
|
|
user_id = getattr(user, "id", 1)
|
|
|
|
|
|
# 1. 查找集群
|
|
|
res = await db.execute(select(Cluster).where(Cluster.uuid == cluster_uuid).limit(1))
|
|
|
cluster = res.scalars().first()
|
|
|
if not cluster:
|
|
|
raise HTTPException(status_code=404, detail="cluster_not_found")
|
|
|
|
|
|
# 2. 获取 SSH 用户
|
|
|
node_res = await db.execute(select(Node).where(Node.cluster_id == cluster.id).limit(1))
|
|
|
node = node_res.scalars().first()
|
|
|
ssh_user = node.ssh_user if node and node.ssh_user else "hadoop"
|
|
|
|
|
|
start_time = _now()
|
|
|
logs = []
|
|
|
|
|
|
# 3. 在 NameNode 执行 stop-dfs.sh
|
|
|
if cluster.namenode_ip and cluster.namenode_psw:
|
|
|
try:
|
|
|
def run_nn_stop():
|
|
|
with SSHClient(str(cluster.namenode_ip), ssh_user, cluster.namenode_psw) as client:
|
|
|
return client.execute_command("stop-dfs.sh")
|
|
|
out, err = await asyncio.to_thread(run_nn_stop)
|
|
|
logs.append(f"NameNode ({cluster.namenode_ip}) stop: {out} {err}")
|
|
|
except Exception as e:
|
|
|
logs.append(f"NameNode ({cluster.namenode_ip}) stop failed: {str(e)}")
|
|
|
|
|
|
# 4. 在 ResourceManager 执行 stop-yarn.sh
|
|
|
if cluster.rm_ip and cluster.rm_psw:
|
|
|
try:
|
|
|
def run_rm_stop():
|
|
|
with SSHClient(str(cluster.rm_ip), ssh_user, cluster.rm_psw) as client:
|
|
|
return client.execute_command("stop-yarn.sh")
|
|
|
out, err = await asyncio.to_thread(run_rm_stop)
|
|
|
logs.append(f"ResourceManager ({cluster.rm_ip}) stop: {out} {err}")
|
|
|
except Exception as e:
|
|
|
logs.append(f"ResourceManager ({cluster.rm_ip}) stop failed: {str(e)}")
|
|
|
|
|
|
end_time = _now()
|
|
|
|
|
|
# 5. 更新集群状态
|
|
|
cluster.health_status = "unknown"
|
|
|
cluster.updated_at = end_time
|
|
|
await db.flush()
|
|
|
|
|
|
# 6. 记录日志
|
|
|
full_desc = " | ".join(logs)
|
|
|
await _write_hadoop_exec_log(db, user_id, cluster.name, f"Stop Cluster: {full_desc}", start_time, end_time)
|
|
|
|
|
|
return {"status": "success", "logs": logs}
|
|
|
except HTTPException:
|
|
|
raise
|
|
|
except Exception as e:
|
|
|
print(f"Error stopping cluster: {e}")
|
|
|
raise HTTPException(status_code=500, detail="server_error")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
