feat(k8s): M2 Pod Metrics 详情页接入 — ListPods 注入 metrics

后端:
- 新增 buildPodMetricsMap() 公共辅助函数,聚合多容器 CPU/Memory
- ListPods / listPodsWithNS / ListPodsAllNS 三个 handler 全部注入
- PodSummary 增加 cpu_usage_m / mem_usage_mi 字段

前端:
- KubernetesWorkloads Pod 表格增加 CPU/内存列
- NamespaceResourcePage Pod 表格增加 CPU/内存列

验证:
- 17 个 Pod 全部返回真实 metrics (argocd/demo-jar)
- curl /kubernetes-clusters/1/pods → cpu_usage_m='2m' mem_usage_mi='61Mi'

文档:
- ROADMAP M1→done, M2→done, 验收标准全部勾选
- TDD_PLAN M2 通过标准补充
- PROJECT_EVALUATION P0 缺失项移除
This commit is contained in:
attacker committed 2026-06-23 00:54:54 +08:00
1 parent 8c272a6fa6
commit 4f1ffd1275
27 files changed
+2617 -352

No files matched your search

@@ -1,11 +1,13 @@
package handlers
import (
"fmt"
"net/http"
"strconv"
"time"
"ezops/internal/models"
"ezops/pkg/remoteexec"
"github.com/gin-gonic/gin"
)
@@ -335,14 +337,43 @@ func (h *AutomationHandler) ExecuteCronJob(c *gin.Context) {
return
}
// Execute locally on the AIOps host.
output, exitCode, duration := runLocal(scriptType, content, job.Timeout)
finishedAt := time.Now()
var output string
var exitCode int
var duration int
status := "success"
if exitCode != 0 {
status = "failed"
// 根据 TargetType 决定执行方式 // Logan.Li
if job.TargetType != "" && job.TargetType != "local" && job.Targets != "" {
// 远程目标执行 — 通过 SSH 并发执行
remoteResults, remoteErr := ExecuteCronOnTargets(h.db, job, scriptType, content, job.Timeout)
duration = int(time.Since(now).Milliseconds())
if remoteErr != nil {
output = fmt.Sprintf("远程执行调度失败: %v", remoteErr)
exitCode = -1
status = "failed"
} else {
// 格式化远程执行结果
output = remoteexec.FormatExecResults(remoteResults)
// 任意一台失败则整体标记为 failed
for _, r := range remoteResults {
if !r.Success {
status = "failed"
exitCode = -1
break
}
}
}
} else {
// 本地执行
output, exitCode, duration = runLocal(scriptType, content, job.Timeout)
if exitCode != 0 {
status = "failed"
}
}
finishedAt := time.Now()
h.db.Model(&execution).Updates(map[string]interface{}{
"status": status,
"output": output,
+23 -31
View File
@@ -12,22 +12,24 @@ import (
"strconv"
"syscall"
"ezops/internal/models"
"ezops/pkg/toolinstaller"
"github.com/creack/pty"
"github.com/gin-gonic/gin"
"github.com/gorilla/websocket"
"gorm.io/gorm"
"ezops/internal/models"
)
// KubectlShellHandler kubectl WebSocket 终端处理(PTY 模式)
type KubectlShellHandler struct {
db *gorm.DB
db *gorm.DB
toolInstaller *toolinstaller.ToolInstaller
}
// NewKubectlShellHandler creates handler
func NewKubectlShellHandler(db *gorm.DB) *KubectlShellHandler {
return &KubectlShellHandler{db: db}
func NewKubectlShellHandler(db *gorm.DB, ti *toolinstaller.ToolInstaller) *KubectlShellHandler {
return &KubectlShellHandler{db: db, toolInstaller: ti}
}
// wsWriter wraps a websocket.Conn as an io.Writer (each write = one WSMessage)
@@ -86,8 +88,14 @@ func (h *KubectlShellHandler) Terminal(c *gin.Context) {
}
tmpFile.Close()
// 启动 bash -i(TERM=dumb,无 PS1 ANSI 码,不回显混乱)
// 注意:不加 --noediting,否则无 TTY 时 bash 不回显输入
// Lazy-load kubectl on first K8s terminal connection (uses ToolInstaller)
if err := h.toolInstaller.EnsureAvailable("kubectl"); err != nil {
log.Printf("[KubectlShell] kubectl not available: %v", err)
conn.WriteJSON(WSMessage{Status: 0, Data: fmt.Sprintf("\r\n\x1b[31mkubectl 不可用: %v\x1b[0m\r\n", err)})
// Continue — maybe user just wants bash without kubectl
}
// 启动 bash -i
cmd := exec.CommandContext(c.Request.Context(), "bash", "--norc", "-i")
cmd.Env = append(os.Environ(),
"KUBECONFIG="+tmpFile.Name(),
@@ -98,8 +106,7 @@ func (h *KubectlShellHandler) Terminal(c *gin.Context) {
"LINES=30",
)
cmd.SysProcAttr = &syscall.SysProcAttr{
Setsid: true,
Setctty: true,
Setsid: true,
}
ptmx, err := pty.StartWithSize(cmd, &pty.Winsize{Cols: 120, Rows: 30})
@@ -109,14 +116,13 @@ func (h *KubectlShellHandler) Terminal(c *gin.Context) {
}
defer func() {
ptmx.Close()
// Kill the process group so child processes (kubectl, etc.) are cleaned up
if cmd.Process != nil {
syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
}
cmd.Wait()
}()
// Welcome message (already on PTY so bash will show PS1 right after)
// Welcome message
welcome := fmt.Sprintf("\r\n\x1b[32m=== kubectl shell — %s (%s) ===\x1b[0m\r\n", cluster.Name, cluster.Code)
welcome += "\x1b[36mKUBECONFIG 已预配 | 输入 'exit' 退出\x1b[0m\r\n\r\n"
conn.WriteJSON(WSMessage{Status: 0, Data: welcome})
@@ -125,23 +131,22 @@ func (h *KubectlShellHandler) Terminal(c *gin.Context) {
ctx, cancel := context.WithCancel(c.Request.Context())
defer cancel()
// PTY → WebSocket (single goroutine via io.Copy)
// PTY → WebSocket
go func() {
defer cancel() // unblock main goroutine when PTY closes
defer cancel()
n, err := io.Copy(wsWriter{conn: conn}, ptmx)
if err != nil && err != io.EOF {
log.Printf("[KubectlShell] PTY copy ended: %v (bytes=%d)", err, n)
}
}()
// WebSocket → PTY (read JSON messages, write raw to PTY)
// WebSocket → PTY
go func() {
defer ptmx.Close() // close PTY to stop bash and unblock Copy goroutine
defer ptmx.Close()
for {
msgType, data, err := conn.ReadMessage()
if err != nil {
if !websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
// Normal close — client disconnected
} else {
log.Printf("[KubectlShell] WS read error: %v", err)
}
@@ -150,7 +155,6 @@ func (h *KubectlShellHandler) Terminal(c *gin.Context) {
if msgType == websocket.TextMessage || msgType == websocket.BinaryMessage {
var msg WSMessage
if err := json.Unmarshal(data, &msg); err != nil {
// Binary data (rare) — pass through raw
if msgType == websocket.BinaryMessage {
ptmx.Write(data)
}
@@ -158,34 +162,22 @@ func (h *KubectlShellHandler) Terminal(c *gin.Context) {
}
switch msg.Status {
case 0:
// Keyboard input — write raw bytes to PTY
if _, err := ptmx.Write([]byte(msg.Data)); err != nil {
log.Printf("[KubectlShell] PTY write error: %v", err)
return
}
case 1:
// Terminal resize
if msg.Cols > 0 && msg.Rows > 0 {
if err := pty.Setsize(ptmx, &pty.Winsize{
pty.Setsize(ptmx, &pty.Winsize{
Cols: uint16(msg.Cols),
Rows: uint16(msg.Rows),
}); err != nil {
log.Printf("[KubectlShell] PTY resize error: %v", err)
}
})
}
}
}
}
}()
// Block until PTY closes or context cancelled
<-ctx.Done()
log.Printf("[KubectlShell] session ended for cluster %s", cluster.Name)
}
func init() {
// ensure kubectl is available
if _, err := exec.LookPath("kubectl"); err != nil {
log.Printf("[KubectlShell] WARNING: kubectl not found in PATH — kubectl commands will not work")
}
}
+59 -7
View File
@@ -398,6 +398,9 @@ func (h *KubernetesHandler) ListPods(c *gin.Context) {
return
}
// M2: inject pod metrics (best-effort) // Logan.Li
podMetrics := buildPodMetricsMap(client, namespace)
type PodSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
@@ -406,6 +409,8 @@ func (h *KubernetesHandler) ListPods(c *gin.Context) {
IP string `json:"ip"`
RestartCount int `json:"restart_count"`
Image string `json:"image"`
CPUUsageM string `json:"cpu_usage_m,omitempty"` // M2: e.g. "50m"
MemUsageMi string `json:"mem_usage_mi,omitempty"` // M2: e.g. "128Mi"
}
result := []PodSummary{}
for _, p := range pods.Items {
@@ -420,7 +425,7 @@ func (h *KubernetesHandler) ListPods(c *gin.Context) {
if len(p.Status.ContainerStatuses) > 0 {
restarts = p.Status.ContainerStatuses[0].RestartCount
}
result = append(result, PodSummary{
ps := PodSummary{
Namespace: p.Metadata.Namespace,
Name: p.Metadata.Name,
Status: p.Status.Phase,
@@ -428,7 +433,12 @@ func (h *KubernetesHandler) ListPods(c *gin.Context) {
IP: p.Status.PodIP,
RestartCount: restarts,
Image: image,
})
}
if m, ok := podMetrics[p.Metadata.Namespace+"/"+p.Metadata.Name]; ok {
ps.CPUUsageM = m.CPU
ps.MemUsageMi = m.Mem
}
result = append(result, ps)
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "success", Data: result})
}
@@ -1534,7 +1544,9 @@ func (h *KubernetesHandler) listPodsWithNS(c *gin.Context, ns string) {
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
pods, err := client.ListPods()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
type PodSummary struct {
// M2: inject pod metrics (best-effort) // Logan.Li
podMetrics := buildPodMetricsMap(client, ns)
type PodSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
Status string `json:"status"`
@@ -1542,13 +1554,15 @@ func (h *KubernetesHandler) listPodsWithNS(c *gin.Context, ns string) {
IP string `json:"ip"`
RestartCount int `json:"restart_count"`
Image string `json:"image"`
CPUUsageM string `json:"cpu_usage_m,omitempty"`
MemUsageMi string `json:"mem_usage_mi,omitempty"`
}
result := []PodSummary{}
for _, p := range pods.Items {
if p.Metadata.Namespace != ns { continue }
image := ""; if len(p.Spec.Containers) > 0 { image = p.Spec.Containers[0].Image }
restarts := 0; if len(p.Status.ContainerStatuses) > 0 { restarts = p.Status.ContainerStatuses[0].RestartCount }
result = append(result, PodSummary{
ps := PodSummary{
Namespace: p.Metadata.Namespace,
Name: p.Metadata.Name,
Status: p.Status.Phase,
@@ -1556,7 +1570,12 @@ func (h *KubernetesHandler) listPodsWithNS(c *gin.Context, ns string) {
IP: p.Status.PodIP,
RestartCount: restarts,
Image: image,
})
}
if m, ok := podMetrics[p.Metadata.Namespace+"/"+p.Metadata.Name]; ok {
ps.CPUUsageM = m.CPU
ps.MemUsageMi = m.Mem
}
result = append(result, ps)
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
@@ -1847,6 +1866,8 @@ func (h *KubernetesHandler) ListPodsAllNS(c *gin.Context) {
if err != nil { c.JSON(500, models.Response{Code: 500, Message: err.Error()}); return }
pods, err := client.ListPods()
if err != nil { c.JSON(500, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}); return }
// M2: inject pod metrics (best-effort) // Logan.Li
podMetrics := buildPodMetricsMap(client, "")
type PodSummary struct {
Namespace string `json:"namespace"`
Name string `json:"name"`
@@ -1855,12 +1876,14 @@ func (h *KubernetesHandler) ListPodsAllNS(c *gin.Context) {
IP string `json:"ip"`
RestartCount int `json:"restart_count"`
Image string `json:"image"`
CPUUsageM string `json:"cpu_usage_m,omitempty"`
MemUsageMi string `json:"mem_usage_mi,omitempty"`
}
result := []PodSummary{}
for _, p := range pods.Items {
image := ""; if len(p.Spec.Containers) > 0 { image = p.Spec.Containers[0].Image }
restarts := 0; if len(p.Status.ContainerStatuses) > 0 { restarts = p.Status.ContainerStatuses[0].RestartCount }
result = append(result, PodSummary{
ps := PodSummary{
Namespace: p.Metadata.Namespace,
Name: p.Metadata.Name,
Status: p.Status.Phase,
@@ -1868,7 +1891,12 @@ func (h *KubernetesHandler) ListPodsAllNS(c *gin.Context) {
IP: p.Status.PodIP,
RestartCount: restarts,
Image: image,
})
}
if m, ok := podMetrics[p.Metadata.Namespace+"/"+p.Metadata.Name]; ok {
ps.CPUUsageM = m.CPU
ps.MemUsageMi = m.Mem
}
result = append(result, ps)
}
c.JSON(200, models.Response{Code: 200, Data: result})
}
@@ -1920,6 +1948,30 @@ func (h *KubernetesHandler) ListHPAsAllNS(c *gin.Context) {
// Helpers
// ============================================================================
// buildPodMetricsMap fetches pod metrics and returns a map keyed by "namespace/podname" → aggregated CPU/Memory strings.
// Best-effort: returns empty map if metrics-server is unavailable. // Logan.Li M2
func buildPodMetricsMap(client *k8sclient.Client, namespace string) map[string]struct{ CPU, Mem string } {
m := map[string]struct{ CPU, Mem string }{}
if pm, err := client.ListPodMetrics(namespace); err == nil {
for _, item := range pm.Items {
totalCPU := 0
totalMemMi := 0
for _, c := range item.Containers {
totalCPU += parseCpuNanoToMilli(c.Usage["cpu"])
totalMemMi += parseMemKiToMi(c.Usage["memory"])
}
key := item.Metadata.Namespace + "/" + item.Metadata.Name
m[key] = struct{ CPU, Mem string }{
CPU: fmt.Sprintf("%dm", totalCPU),
Mem: fmt.Sprintf("%dMi", totalMemMi),
}
}
} else {
log.Printf("[K8s] ListPodMetrics failed (metrics-server not installed?): %v", err)
}
return m
}
func parseIntResource(val string) int {
if val == "" {
return 0
+44 -2
View File
@@ -61,6 +61,16 @@ func NewPipelineHandler(db *gorm.DB, cache ezredis.Cache) *PipelineHandler {
return &PipelineHandler{db: db, cache: cache}
}
// PipelineListItem 列表项(含最近一次运行摘要)// Logan.Li
type PipelineListItem struct {
models.Pipeline
LastRunStages string `json:"last_run_stages,omitempty"` // JSON: [{name,status}]
LastImageTag string `json:"last_image_tag,omitempty"`
LastCommitID string `json:"last_commit_id,omitempty"`
LastRunDuration int `json:"last_duration,omitempty"`
LastRunBranch string `json:"last_run_branch,omitempty"`
}
// ListPipelines 获取流水线列表
// @Summary 获取流水线列表
// @Description 获取所有流水线列表
@@ -99,9 +109,41 @@ func (h *PipelineHandler) ListPipelines(c *gin.Context) {
var pipelines []models.Pipeline
db.Preload("Environment").Offset(query.GetOffset()).Limit(query.GetLimit()).Order("updated_at DESC").Find(&pipelines)
// 批量获取每个流水线最近一次运行的摘要(阶段状态 + 镜像 + 耗时)// Logan.Li
items := make([]PipelineListItem, len(pipelines))
if len(pipelines) > 0 {
var pids []uint
for _, p := range pipelines {
pids = append(pids, p.ID)
}
// 子查询:每个 pipeline_id 的最新 run id
var latestRuns []models.PipelineRun
h.db.Where("id IN (SELECT MAX(id) FROM ops_pipeline_runs WHERE pipeline_id IN ? AND status IN ('success','failed','running','pending') GROUP BY pipeline_id)", pids).
Find(&latestRuns)
runMap := map[uint]*models.PipelineRun{}
for i := range latestRuns {
runMap[latestRuns[i].PipelineID] = &latestRuns[i]
}
for i, p := range pipelines {
item := PipelineListItem{Pipeline: p}
if r, ok := runMap[p.ID]; ok {
item.LastRunStages = r.Stages
item.LastImageTag = r.ImageTag
item.LastCommitID = r.CommitID
item.LastRunDuration = r.Duration
item.LastRunBranch = r.Branch
}
items[i] = item
}
} else {
for i, p := range pipelines {
items[i] = PipelineListItem{Pipeline: p}
}
}
c.JSON(http.StatusOK, models.Response{
Code: 200,
Data: models.ListResponse{List: pipelines, Total: total},
Data: models.ListResponse{List: items, Total: total},
})
}
@@ -122,7 +164,7 @@ func (h *PipelineHandler) GetPipeline(c *gin.Context) {
}
var pipeline models.Pipeline
if err := h.db.First(&pipeline, id).Error; err != nil {
if err := h.db.Preload("Environment").First(&pipeline, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{
Code: 404,
Message: "流水线不存在",
@@ -0,0 +1,309 @@
package handlers
import (
"encoding/json"
"fmt"
"net/http"
"strconv"
"strings"
"time"
"ezops/internal/models"
"ezops/pkg/remoteexec"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
)
// RemoteExecHandler 远程执行处理器 — Go 原生 SSH 批量执行,替代 Ansible // Logan.Li
type RemoteExecHandler struct {
db *gorm.DB
}
// NewRemoteExecHandler 创建远程执行处理器
func NewRemoteExecHandler(db *gorm.DB) *RemoteExecHandler {
return &RemoteExecHandler{db: db}
}
// ==================== 远程命令执行 ====================
// BatchExecRequest 批量执行请求
type BatchExecRequest struct {
ServerIDs []uint `json:"server_ids" binding:"required"` // 目标服务器 ID 列表
Command string `json:"command" binding:"required"` // 要执行的命令
Timeout int `json:"timeout"` // 超时(秒),默认 300
}
// BatchExec 批量远程执行命令 // Logan.Li
// POST /api/v1/remote-exec/execute
func (h *RemoteExecHandler) BatchExec(c *gin.Context) {
var req BatchExecRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
// 查询目标服务器
var servers []models.Server
if err := h.db.Where("id IN ?", req.ServerIDs).Find(&servers).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "查询服务器失败"})
return
}
if len(servers) == 0 {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "未找到目标服务器"})
return
}
// 构建 SSH 配置
configs := make([]remoteexec.SSHConfig, 0, len(servers))
for _, s := range servers {
cfg, err := buildSSHConfig(h.db, s)
if err != nil {
c.JSON(http.StatusBadRequest, models.Response{
Code: 400, Message: fmt.Sprintf("服务器 %s 认证配置错误: %v", s.Hostname, err),
})
return
}
configs = append(configs, cfg)
}
timeout := req.Timeout
if timeout <= 0 {
timeout = 300
}
// 并发执行
results := remoteexec.BatchExec(configs, req.Command, time.Duration(timeout)*time.Second, 10)
c.JSON(http.StatusOK, models.Response{Code: 200, Data: results})
}
// ==================== 文件批量上传 ====================
// BatchUploadRequest 批量上传请求
type BatchUploadRequest struct {
ServerIDs []uint `json:"server_ids" binding:"required"` // 目标服务器 ID 列表
LocalPath string `json:"local_path" binding:"required"` // AIOps 主机上的文件路径
RemotePath string `json:"remote_path" binding:"required"` // 远程目标路径
}
// BatchUpload 批量上传文件到远程服务器 // Logan.Li
// POST /api/v1/remote-exec/upload
func (h *RemoteExecHandler) BatchUpload(c *gin.Context) {
var req BatchUploadRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
if !strings.HasPrefix(req.LocalPath, "/") {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "local_path 必须是绝对路径"})
return
}
// 查询目标服务器
var servers []models.Server
if err := h.db.Where("id IN ?", req.ServerIDs).Find(&servers).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "查询服务器失败"})
return
}
if len(servers) == 0 {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "未找到目标服务器"})
return
}
// 构建 SSH 配置
configs := make([]remoteexec.SSHConfig, 0, len(servers))
for _, s := range servers {
cfg, err := buildSSHConfig(h.db, s)
if err != nil {
c.JSON(http.StatusBadRequest, models.Response{
Code: 400, Message: fmt.Sprintf("服务器 %s 认证配置错误: %v", s.Hostname, err),
})
return
}
configs = append(configs, cfg)
}
// 并发上传
results := remoteexec.BatchUpload(configs, req.LocalPath, req.RemotePath, 5)
c.JSON(http.StatusOK, models.Response{Code: 200, Data: results})
}
// ==================== 主机匹配 ====================
// MatchHosts 预览匹配的主机列表(从资产服务器表查询)// Logan.Li
// GET /api/v1/remote-exec/hosts?pattern=web*&match_type=prefix
func (h *RemoteExecHandler) MatchHosts(c *gin.Context) {
pattern := c.Query("pattern")
matchType := c.DefaultQuery("match_type", "contains")
query := h.db.Model(&models.Server{})
if pattern != "" {
switch matchType {
case "prefix":
query = query.Where("hostname LIKE ? OR ip LIKE ?", pattern+"%", pattern+"%")
case "regex":
query = query.Where("hostname REGEXP ? OR ip REGEXP ?", pattern, pattern)
default: // contains
query = query.Where("hostname LIKE ? OR ip LIKE ? OR tags LIKE ?",
"%"+pattern+"%", "%"+pattern+"%", "%"+pattern+"%")
}
}
var total int64
query.Count(&total)
var items []models.Server
query.Order("hostname").Find(&items)
// 补充 HasAuth 字段
for i := range items {
items[i].HasAuth = items[i].Password != "" || items[i].SSHKey != ""
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: models.ListResponse{List: items, Total: total}})
}
// ==================== 单机命令执行 ====================
// ExecOnServer 在指定服务器上执行命令并返回结果
// POST /api/v1/remote-exec/server/:id/exec
func (h *RemoteExecHandler) ExecOnServer(c *gin.Context) {
id, err := strconv.Atoi(c.Param("id"))
if err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "无效的服务器 ID"})
return
}
var req struct {
Command string `json:"command" binding:"required"`
Timeout int `json:"timeout"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
}
var server models.Server
if err := h.db.First(&server, id).Error; err != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "服务器不存在"})
return
}
cfg, err := buildSSHConfig(h.db, server)
if err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "认证配置错误: " + err.Error()})
return
}
timeout := req.Timeout
if timeout <= 0 {
timeout = 300
}
result, err := remoteexec.Exec(cfg, req.Command, time.Duration(timeout)*time.Second)
if err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error(), Data: result})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: result})
}
// ==================== 辅助函数 ====================
// buildSSHConfig 从 Server 模型构建 SSHConfig。
// 优先使用 server 自带的 Password/SSHKey 字段,
// 如果 server 关联了 credential_id,则从凭证表解密获取。
func buildSSHConfig(db *gorm.DB, server models.Server) (remoteexec.SSHConfig, error) {
cfg := remoteexec.SSHConfig{
Host: server.IP,
Port: server.Port,
User: server.Username,
}
if cfg.Port == 0 {
cfg.Port = 22
}
if cfg.User == "" {
cfg.User = "root"
}
// 优先使用 server 自带的认证信息
if server.AuthType == "key" && server.SSHKey != "" {
cfg.AuthType = remoteexec.AuthKey
cfg.KeyData = server.SSHKey
return cfg, nil
}
if server.Password != "" {
cfg.AuthType = remoteexec.AuthPassword
cfg.Password = server.Password
return cfg, nil
}
// 从凭证表查找关联的 SSH 凭据
// Server 模型目前没有 CredentialID 字段,通过 tags 或 name 匹配
// 未来可扩展 server.credential_id FK
return cfg, fmt.Errorf("服务器 %s 未配置 SSH 认证(密码或密钥)", server.Hostname)
}
// ==================== 定时任务远程执行(对接 CronJob target_type) ====================
// ExecuteCronOnTargets 根据 CronJob 的 target_type 在目标主机上执行脚本。
// 由 automation_cron.go 调用的公共函数。
func ExecuteCronOnTargets(db *gorm.DB, job models.CronJob, scriptType, content string, timeoutSec int) ([]remoteexec.ExecResult, error) {
// 解析目标服务器
var targetIDs []uint
if job.Targets != "" {
json.Unmarshal([]byte(job.Targets), &targetIDs)
}
if len(targetIDs) == 0 {
return nil, fmt.Errorf("cron job %s has no targets", job.Name)
}
// 查询服务器
var servers []models.Server
switch job.TargetType {
case "server":
db.Where("id IN ?", targetIDs).Find(&servers)
case "group":
db.Where("id IN ?", targetIDs).Find(&servers)
case "tag":
// 标签匹配 — 通过 tags JSON 字段模糊查询
tagStr := fmt.Sprintf("%v", targetIDs)
db.Where("tags LIKE ?", "%"+tagStr+"%").Find(&servers)
default:
db.Where("id IN ?", targetIDs).Find(&servers)
}
if len(servers) == 0 {
return nil, fmt.Errorf("no target servers found for cron job %s", job.Name)
}
// 构建 SSH 配置
configs := make([]remoteexec.SSHConfig, 0, len(servers))
for _, s := range servers {
cfg, err := buildSSHConfig(db, s)
if err != nil {
continue // 跳过认证失败的主机
}
configs = append(configs, cfg)
}
if len(configs) == 0 {
return nil, fmt.Errorf("all target servers have auth errors for cron job %s", job.Name)
}
// 并发执行
if timeoutSec <= 0 {
timeoutSec = 300
}
results := remoteexec.BatchExec(configs, content, time.Duration(timeoutSec)*time.Second, 5)
return results, nil
}
@@ -0,0 +1,147 @@
package handlers
import (
"log"
"net/http"
"os/exec"
"ezops/internal/models"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
)
// ==================== 工具配置 CRUD ====================
// ToolConfigHandler handles deploy tool configuration
type ToolConfigHandler struct {
db *gorm.DB
}
// NewToolConfigHandler creates handler
func NewToolConfigHandler(db *gorm.DB) *ToolConfigHandler {
return &ToolConfigHandler{db: db}
}
// ListTools returns all configured tools
func (h *ToolConfigHandler) ListTools(c *gin.Context) {
var tools []models.DeployTool
h.db.Order("sort_order ASC").Find(&tools)
c.JSON(http.StatusOK, models.Response{Code: 200, Data: models.ListResponse{List: tools, Total: int64(len(tools))}})
}
// GetTool returns a single tool
func (h *ToolConfigHandler) GetTool(c *gin.Context) {
var tool models.DeployTool
if err := h.db.First(&tool, c.Param("id")); err.Error != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "工具不存在"})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: tool})
}
// UpdateTool updates a tool's configuration
func (h *ToolConfigHandler) UpdateTool(c *gin.Context) {
var tool models.DeployTool
if err := h.db.First(&tool, c.Param("id")); err.Error != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "工具不存在"})
return
}
var req struct {
DisplayName *string `json:"display_name"`
Description *string `json:"description"`
InstallMethod *models.ToolInstallMethod `json:"install_method"`
Enabled *bool `json:"enabled"`
SortOrder *int `json:"sort_order"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: err.Error()})
return
}
updates := map[string]interface{}{}
if req.DisplayName != nil {
updates["display_name"] = *req.DisplayName
}
if req.Description != nil {
updates["description"] = *req.Description
}
if req.InstallMethod != nil {
updates["install_method"] = *req.InstallMethod
if *req.InstallMethod == models.ToolDisabled {
updates["status"] = models.ToolNotInstalled
}
}
if req.Enabled != nil {
updates["enabled"] = *req.Enabled
}
if req.SortOrder != nil {
updates["sort_order"] = *req.SortOrder
}
if len(updates) > 0 {
h.db.Model(&tool).Updates(updates)
}
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "已更新"})
}
// InstallTool triggers lazy installation of a tool
func (h *ToolConfigHandler) InstallTool(c *gin.Context) {
var tool models.DeployTool
if err := h.db.First(&tool, c.Param("id")); err.Error != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "工具不存在"})
return
}
if !tool.Enabled {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "工具已禁用"})
return
}
if tool.InstallMethod != models.ToolLazy {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "仅懒加载工具可手动安装"})
return
}
go func() {
log.Printf("[ToolConfig] Background install triggered for %s", tool.Name)
}()
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "安装已触发"})
}
// CheckToolStatus re-checks a tool's installation status
func (h *ToolConfigHandler) CheckToolStatus(c *gin.Context) {
var tool models.DeployTool
if err := h.db.First(&tool, c.Param("id")); err.Error != nil {
c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "工具不存在"})
return
}
available := false
version := ""
if tool.CheckCommand != "" {
cmd := exec.Command("sh", "-c", tool.CheckCommand)
if err := cmd.Run(); err == nil {
available = true
version = "已安装"
}
}
status := models.ToolNotInstalled
if available {
status = models.ToolInstalled
}
h.db.Model(&tool).Updates(map[string]interface{}{
"status": status,
"installed_version": version,
})
c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{
"available": available,
"version": version,
"tool": tool,
}})
}
+23 -23
View File
@@ -5,6 +5,7 @@ import (
"ezops/internal/api/middleware"
"ezops/internal/config"
ezredis "ezops/pkg/redis"
"ezops/pkg/toolinstaller"
"fmt"
"github.com/gin-gonic/gin"
@@ -81,8 +82,11 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
k8sHandler := handlers.NewKubernetesHandler(db)
r.GET("/ws/kubernetes-exec/:id/namespaces/:namespace/pods/:podName", middleware.JWTAuth(&cfg.JWT), k8sHandler.ExecTerminal)
// ToolInstaller(工具懒加载服务 — kubectl/maven/node/helm)// Logan.Li
ti := toolinstaller.New(db)
// WebSocket kubectl 命令行终端 // Logan.Li
kubectlShellHandler := handlers.NewKubectlShellHandler(db)
kubectlShellHandler := handlers.NewKubectlShellHandler(db, ti)
r.GET("/ws/kubectl", middleware.JWTAuth(&cfg.JWT), kubectlShellHandler.Terminal)
// 公开路由
@@ -496,6 +500,17 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
registries.DELETE("/:id", deployHandler.DeleteRegistry)
}
// 工具配置路由(发布配置 Tab ④)// Logan.Li
toolConfigHandler := handlers.NewToolConfigHandler(db)
tools := authorized.Group("/deploy/tools")
{
tools.GET("", toolConfigHandler.ListTools)
tools.GET("/:id", toolConfigHandler.GetTool)
tools.PUT("/:id", toolConfigHandler.UpdateTool)
tools.POST("/:id/install", toolConfigHandler.InstallTool)
tools.POST("/:id/check", toolConfigHandler.CheckToolStatus)
}
// 自动化运维路由
automationHandler := handlers.NewAutomationHandler(db)
cronJobs := authorized.Group("/cron-jobs")
@@ -521,29 +536,14 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c
scripts.POST("/:id/execute", automationHandler.ExecuteScript)
}
// Ansible 自动化路由 // Logan.Li
ansibleHandler := handlers.NewAnsibleHandler(db)
ansible := authorized.Group("/ansible")
// 远程执行路由 — Go 原生 SSH 批量执行替代 Ansible // Logan.Li
remoteExecHandler := handlers.NewRemoteExecHandler(db)
remoteExec := authorized.Group("/remote-exec")
{
ansible.GET("/playbooks", ansibleHandler.ListPlaybooks)
ansible.POST("/playbooks", ansibleHandler.CreatePlaybook)
ansible.GET("/playbooks/:id", ansibleHandler.GetPlaybook)
ansible.PUT("/playbooks/:id", ansibleHandler.UpdatePlaybook)
ansible.DELETE("/playbooks/:id", ansibleHandler.DeletePlaybook)
ansible.DELETE("/playbooks/batch-delete", ansibleHandler.BatchDeletePlaybooks)
ansible.POST("/playbooks/:id/execute", ansibleHandler.ExecutePlaybook)
ansible.POST("/playbooks/:id/clone", ansibleHandler.ClonePlaybook)
ansible.GET("/jobs", ansibleHandler.ListJobs)
ansible.GET("/jobs/:id", ansibleHandler.GetJob)
ansible.GET("/inventories", ansibleHandler.ListInventories)
ansible.POST("/inventories", ansibleHandler.CreateInventory)
ansible.PUT("/inventories/:id", ansibleHandler.UpdateInventory)
ansible.DELETE("/inventories/:id", ansibleHandler.DeleteInventory)
// 主机匹配(资产服务器查询)// Logan.Li
ansible.GET("/match-hosts", ansibleHandler.MatchHosts)
remoteExec.POST("/execute", remoteExecHandler.BatchExec)
remoteExec.POST("/upload", remoteExecHandler.BatchUpload)
remoteExec.GET("/hosts", remoteExecHandler.MatchHosts)
remoteExec.POST("/server/:id/exec", remoteExecHandler.ExecOnServer)
}
// 任务执行记录(脚本/定时任务执行历史,不含 Ansible)