diff --git a/agent/main.go b/agent/main.go index 6b28819..1be87f3 100644 --- a/agent/main.go +++ b/agent/main.go @@ -18,6 +18,7 @@ import ( "ezops/agent/collector" "ezops/agent/model" "ezops/agent/reporter" + "ezops/agent/task" ) // 编译时可注入的版本信息 @@ -65,6 +66,10 @@ func main() { portCollector := collector.NewPortCollector() rep := reporter.New(*server, *token, logger) + // 初始化任务执行器与任务客户端（与 reporter 复用相同的 server/token/hostname） + taskExecutor := task.New(logger) + taskClient := task.NewClient(*server, *token, hostname, logger) + // 监听 SIGINT / SIGTERM，支持优雅退出 ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() @@ -111,7 +116,59 @@ func main() { report.CPU.UsagePercent, report.Memory.UsagePercent, len(report.Disks), len(report.Ports)) } - // 启动后立即执行一次，再按 ticker 周期执行 + // pollAndExecute 执行单次任务轮询：拉取待执行任务 → 逐个执行 → 回传结果。 + // 网络错误只记录日志，不 panic，下一个周期继续重试。 + pollAndExecute := func() { + tasks, err := taskClient.Poll(ctx) + if err != nil { + // 上下文已取消（Agent 退出中）属于正常情况，不输出告警 + if ctx.Err() != nil { + return + } + logger.Printf("任务轮询失败: %v", err) + return + } + if len(tasks) == 0 { + return // 暂无任务，静默返回 + } + logger.Printf("拉取到 %d 个待执行任务", len(tasks)) + + for _, t := range tasks { + // Agent 退出时跳过后续任务，避免退出延迟 + if ctx.Err() != nil { + return + } + result := taskExecutor.ExecuteTask(ctx, t) + if err := taskClient.Submit(ctx, result); err != nil { + if ctx.Err() != nil { + return + } + logger.Printf("任务 %s 结果回传失败: %v", t.TaskID, err) + } else { + logger.Printf("任务 %s 结果回传成功 status=%s", t.TaskID, result.Status) + } + } + } + + // 启动任务轮询协程：每 5 秒轮询一次，与指标采集并行运行。 + // 使用独立的 5 秒固定间隔，不受 -interval 影响。 + const taskPollInterval = 5 * time.Second + go func() { + // 启动后立即轮询一次，再按固定周期执行 + pollAndExecute() + ticker := time.NewTicker(taskPollInterval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + pollAndExecute() + case <-ctx.Done(): + return + } + } + }() + + // 主协程：启动后立即执行一次指标采集，再按 ticker 周期执行 collectAndReport() for { select { diff --git a/agent/task/client.go b/agent/task/client.go new file mode 100644 index 0000000..d3e3460 --- /dev/null +++ b/agent/task/client.go @@ -0,0 +1,149 @@ +// Package task 的 client.go 实现与后端任务接口的 HTTP 交互。 +// 涵盖拉取待执行任务（poll）与回传执行结果（submit result）。 +// 仅依赖 Go 标准库，失败时返回错误，由调用方决定重试策略。 +// +// 署名 Logan.Li +package task + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "net/url" + "strings" + "time" +) + +const ( + pollPath = "/api/v1/agent/tasks/poll" // 任务轮询接口 + resultPathFmt = "/api/v1/agent/tasks/%s/result" // 结果回传接口（含 task_id 占位） + httpTimeout = 10 * time.Second // 单次 HTTP 请求超时 + clientUserAgent = "aiops-agent/1.0" // 标识 Agent 版本 +) + +// Client 负责与后端任务接口通信。 +type Client struct { + server string + token string + hostname string + httpClient *http.Client + logger *log.Logger +} + +// NewClient 创建任务客户端实例。 +// - server: AIOps 后端地址（如 http://localhost:8080） +// - token: Bearer 认证令牌 +// - hostname: 本机主机名，用于后端筛选该主机待执行的任务 +func NewClient(server, token, hostname string, logger *log.Logger) *Client { + return &Client{ + server: strings.TrimRight(server, "/"), + token: token, + hostname: hostname, + httpClient: &http.Client{Timeout: httpTimeout}, + logger: logger, + } +} + +// Poll 拉取当前主机待执行的任务列表。 +// 返回空切片表示暂无任务（后端返回空数组或 HTTP 204）。 +func (c *Client) Poll(ctx context.Context) ([]Task, error) { + // 构造带 hostname 查询参数的 URL + u := c.server + pollPath + "?hostname=" + url.QueryEscape(c.hostname) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return nil, fmt.Errorf("构造轮询请求失败: %w", err) + } + c.attachHeaders(req) + + // 上下文已取消则直接返回，避免无意义请求 + if err := ctx.Err(); err != nil { + return nil, err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("轮询请求失败: %w", err) + } + defer resp.Body.Close() + + // 204 No Content 或 200 空体均视为无任务 + if resp.StatusCode == http.StatusNoContent { + return nil, nil + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return nil, fmt.Errorf("轮询响应异常: HTTP %d, body=%s", resp.StatusCode, string(body)) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("读取轮询响应失败: %w", err) + } + + // 空响应体视为无任务 + if len(bytes.TrimSpace(body)) == 0 { + return nil, nil + } + + var wrapper struct { + Code int `json:"code"` + Message string `json:"message"` + Data []Task `json:"data"` + } + if err := json.Unmarshal(body, &wrapper); err != nil { + // 兼容裸数组格式 + var tasks []Task + if err2 := json.Unmarshal(body, &tasks); err2 != nil { + return nil, fmt.Errorf("解析轮询响应失败: %w", err) + } + return tasks, nil + } + return wrapper.Data, nil +} + +// Submit 将任务执行结果回传至后端。 +func (c *Client) Submit(ctx context.Context, res Result) error { + payload, err := json.Marshal(res) + if err != nil { + return fmt.Errorf("序列化任务结果失败: %w", err) + } + + u := c.server + fmt.Sprintf(resultPathFmt, url.PathEscape(res.TaskID)) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("构造回传请求失败: %w", err) + } + c.attachHeaders(req) + + if err := ctx.Err(); err != nil { + return err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("回传请求失败: %w", err) + } + defer resp.Body.Close() + io.Copy(io.Discard, resp.Body) + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("回传响应异常: HTTP %d", resp.StatusCode) + } + return nil +} + +// attachHeaders 统一设置公共请求头（鉴权、内容类型、UA）。 +func (c *Client) attachHeaders(req *http.Request) { + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", clientUserAgent) + if c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } +} diff --git a/agent/task/executor.go b/agent/task/executor.go new file mode 100644 index 0000000..6746664 --- /dev/null +++ b/agent/task/executor.go @@ -0,0 +1,140 @@ +// Package task 实现 AIOps Agent 的任务拉取与执行能力。 +// Agent 通过轮询接口从后端获取待执行任务（命令/脚本），本地执行后回传结果。 +// +// 署名 Logan.Li +package task + +import ( + "bytes" + "context" + "fmt" + "log" + "os/exec" + "time" +) + +// 任务类型常量 +const ( + TypeCommand = "command" // 单条 shell 命令 + TypeScript = "script" // 多行 bash 脚本 +) + +// Task 表示从后端拉取到的待执行任务。 +type Task struct { + TaskID string `json:"task_id"` // 任务唯一标识，回传结果时使用 + Type string `json:"type"` // 任务类型：command / script + Command string `json:"command"` // 命令内容（command 时为单条命令，script 时为脚本正文） + Timeout int `json:"timeout"` // 执行超时，单位：秒；<=0 表示不限制 +} + +// Result 表示任务执行结果，用于回传后端。 +type Result struct { + TaskID string `json:"task_id"` // 对应的任务 ID + ExitCode int `json:"exit_code"` // 进程退出码；超时/启动失败时为 -1 + Result string `json:"result"` // 合并后的 stdout+stderr 输出 + Status string `json:"status"` // 执行状态：success / failed / timeout +} + +// Executor 负责本地执行任务并产出结果。 +type Executor struct { + logger *log.Logger +} + +// New 创建 Executor 实例。 +func New(logger *log.Logger) *Executor { + return &Executor{logger: logger} +} + +// ExecuteTask 执行单个任务，返回执行结果。 +// 根据任务类型选择执行方式： +// - command: 通过 "sh -c " 执行单条命令 +// - script: 通过 "bash" 以标准输入喂入脚本正文执行 +// +// 超时控制通过 context 实现；网络/信号导致的取消不会 panic。 +// 超时或执行失败均会返回 Result，不返回 error，便于调用方统一回传。 +func (e *Executor) ExecuteTask(ctx context.Context, t Task) Result { + res := Result{TaskID: t.TaskID} + + // 构造子进程：根据任务类型选择解释器 + var cmd *exec.Cmd + switch t.Type { + case TypeCommand: + // 单条命令：交给 /bin/sh -c 解释执行 + cmd = exec.CommandContext(ctx, "sh", "-c", t.Command) + case TypeScript: + // 脚本：通过 bash 标准输入执行，便于支持多行与 heredoc 等语法 + cmd = exec.CommandContext(ctx, "bash") + cmd.Stdin = bytes.NewReader([]byte(t.Command)) + default: + // 未知任务类型，直接标记失败，避免误执行 + res.ExitCode = -1 + res.Status = "failed" + res.Result = fmt.Sprintf("未知的任务类型: %q", t.Type) + e.logger.Printf("任务 %s 类型非法: %q", t.TaskID, t.Type) + return res + } + + // 合并 stdout 与 stderr，输出顺序与终端一致，便于排查问题 + var buf bytes.Buffer + cmd.Stdout = &buf + cmd.Stderr = &buf + + // 设置执行超时上下文（若任务指定了有效 timeout） + execCtx, cancel := e.withTimeout(ctx, t.Timeout) + defer cancel() + + startTime := time.Now() + err := cmd.Run() + elapsed := time.Since(startTime) + + // 优先判定超时：Run 返回的 deadline exceeded 与 ctx.Err() 一起识别 + if execCtx.Err() == context.DeadlineExceeded { + res.ExitCode = -1 + res.Status = "timeout" + res.Result = fmt.Sprintf("任务执行超时（%ds），已输出:\n%s", t.Timeout, buf.String()) + e.logger.Printf("任务 %s 执行超时（%ds）", t.TaskID, t.Timeout) + return res + } + + // 正常退出码提取 + res.ExitCode = exitCode(err) + res.Result = buf.String() + + // 根据退出码判定状态 + if res.ExitCode == 0 { + res.Status = "success" + e.logger.Printf("任务 %s 执行成功 exit=0 耗时=%s 输出=%dB", + t.TaskID, elapsed, buf.Len()) + } else { + res.Status = "failed" + e.logger.Printf("任务 %s 执行失败 exit=%d 耗时=%s 输出=%dB", + t.TaskID, res.ExitCode, elapsed, buf.Len()) + } + + return res +} + +// withTimeout 在父上下文基础上叠加超时控制。 +// 当 timeout <= 0 时直接返回父上下文，不做超时限制。 +// 返回的取消函数保证安全可调用（即使未创建新的超时上下文）。 +func (e *Executor) withTimeout(parent context.Context, timeout int) (context.Context, context.CancelFunc) { + if timeout <= 0 { + // 无超时要求：返回空取消函数，保持调用方代码统一 + return parent, func() {} + } + return context.WithTimeout(parent, time.Duration(timeout)*time.Second) +} + +// exitCode 从 exec.Run 返回的错误中提取退出码。 +// 成功时返回 0；无法识别的错误（如进程未启动）返回 -1。 +func exitCode(err error) int { + if err == nil { + return 0 + } + // 尝试断言为 *exec.ExitError 以获取 ExitCode + if exitErr, ok := err.(*exec.ExitError); ok { + return exitErr.ExitCode() + } + // 其他类型错误（如二进制不存在）统一记为 -1 + return -1 +} diff --git a/backend/agent/bin/aiops-agent-linux-amd64 b/backend/agent/bin/aiops-agent-linux-amd64 index 26f94fe..fdc04be 100755 Binary files a/backend/agent/bin/aiops-agent-linux-amd64 and b/backend/agent/bin/aiops-agent-linux-amd64 differ diff --git a/backend/agent/bin/aiops-agent-linux-arm64 b/backend/agent/bin/aiops-agent-linux-arm64 index c8d9efb..0ebfb50 100755 Binary files a/backend/agent/bin/aiops-agent-linux-arm64 and b/backend/agent/bin/aiops-agent-linux-arm64 differ diff --git a/backend/internal/api/handlers/agent_task.go b/backend/internal/api/handlers/agent_task.go new file mode 100644 index 0000000..8a73983 --- /dev/null +++ b/backend/internal/api/handlers/agent_task.go @@ -0,0 +1,220 @@ +package handlers + +import ( + "fmt" + "time" + + "ezops/internal/models" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "gorm.io/gorm" +) + +// AgentTaskHandler Agent 远程任务管理 // Logan.Li +type AgentTaskHandler struct { + db *gorm.DB +} + +func NewAgentTaskHandler(db *gorm.DB) *AgentTaskHandler { + return &AgentTaskHandler{db: db} +} + +// CreateTask 下发远程任务（前端调用，需 JWT） +// POST /api/v1/agent/tasks +func (h *AgentTaskHandler) CreateTask(c *gin.Context) { + var req struct { + Hostname string `json:"hostname"` + Type string `json:"type"` // command / script + Command string `json:"command"` + Timeout int `json:"timeout"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(400, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) + return + } + if req.Hostname == "" || req.Command == "" { + c.JSON(400, models.Response{Code: 400, Message: "hostname 和 command 不能为空"}) + return + } + if req.Timeout == 0 { + req.Timeout = 60 + } + if req.Type == "" { + req.Type = "command" + } + + username, _ := c.Get("username") + + task := models.AgentTask{ + TaskID: uuid.New().String(), + Hostname: req.Hostname, + Type: req.Type, + Command: req.Command, + Timeout: req.Timeout, + Status: "pending", + CreatedBy: fmt.Sprintf("%v", username), + } + + if err := h.db.Create(&task).Error; err != nil { + c.JSON(500, models.Response{Code: 500, Message: "创建失败: " + err.Error()}) + return + } + + c.JSON(200, models.Response{Code: 200, Message: "ok", Data: task}) +} + +// PollTasks Agent 轮询拉取任务（Agent 调用，无需 JWT） +// GET /api/v1/agent/tasks/poll?hostname=xxx +func (h *AgentTaskHandler) PollTasks(c *gin.Context) { + hostname := c.Query("hostname") + if hostname == "" { + c.JSON(400, models.Response{Code: 400, Message: "hostname 不能为空"}) + return + } + + // 查找该主机 pending 的任务，按时间排序，最多取 5 个 + var tasks []models.AgentTask + h.db.Where("hostname = ? AND status = ?", hostname, "pending"). + Order("created_at ASC"). + Limit(5). + Find(&tasks) + + // 标记为 running（防其他 agent 重复拉取） + now := time.Now() + for _, t := range tasks { + h.db.Model(&models.AgentTask{}).Where("id = ? AND status = ?", t.ID, "pending"). + Updates(map[string]interface{}{"status": "running", "picked_at": now}) + } + + c.JSON(200, models.Response{Code: 200, Data: tasks}) +} + +// SubmitResult Agent 回传执行结果（Agent 调用，无需 JWT） +// POST /api/v1/agent/tasks/:task_id/result +func (h *AgentTaskHandler) SubmitResult(c *gin.Context) { + taskID := c.Param("task_id") + + var req struct { + ExitCode int `json:"exit_code"` + Result string `json:"result"` + Status string `json:"status"` // success / failed / timeout + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(400, models.Response{Code: 400, Message: "参数错误"}) + return + } + + now := time.Now() + updates := map[string]interface{}{ + "exit_code": req.ExitCode, + "result": req.Result, + "status": req.Status, + "finished_at": now, + } + + if err := h.db.Model(&models.AgentTask{}).Where("task_id = ?", taskID).Updates(updates).Error; err != nil { + c.JSON(500, models.Response{Code: 500, Message: "更新失败"}) + return + } + + c.JSON(200, models.Response{Code: 200, Message: "ok"}) +} + +// ListTasks 查询任务列表（前端调用，需 JWT） +// GET /api/v1/agent/tasks?hostname=xxx&status=xxx +func (h *AgentTaskHandler) ListTasks(c *gin.Context) { + query := h.db.Model(&models.AgentTask{}) + + if hostname := c.Query("hostname"); hostname != "" { + query = query.Where("hostname = ?", hostname) + } + if status := c.Query("status"); status != "" { + query = query.Where("status = ?", status) + } + + var tasks []models.AgentTask + query.Order("created_at DESC").Limit(200).Find(&tasks) + + c.JSON(200, models.Response{Code: 200, Data: tasks}) +} + +// GetTask 获取单个任务详情 +// GET /api/v1/agent/tasks/:id +func (h *AgentTaskHandler) GetTask(c *gin.Context) { + var task models.AgentTask + if err := h.db.First(&task, c.Param("id")).Error; err != nil { + c.JSON(404, models.Response{Code: 404, Message: "任务不存在"}) + return + } + c.JSON(200, models.Response{Code: 200, Data: task}) +} + +// ===== Agent 告警规则 ===== + +// ListRules GET /api/v1/agent/rules +func (h *AgentTaskHandler) ListRules(c *gin.Context) { + var rules []models.AgentRule + h.db.Order("created_at DESC").Find(&rules) + c.JSON(200, models.Response{Code: 200, Data: rules}) +} + +// CreateRule POST /api/v1/agent/rules +func (h *AgentTaskHandler) CreateRule(c *gin.Context) { + var rule models.AgentRule + if err := c.ShouldBindJSON(&rule); err != nil { + c.JSON(400, models.Response{Code: 400, Message: err.Error()}) + return + } + rule.ID = 0 + rule.Enabled = true + if err := h.db.Create(&rule).Error; err != nil { + c.JSON(500, models.Response{Code: 500, Message: err.Error()}) + return + } + c.JSON(200, models.Response{Code: 200, Data: rule}) +} + +// UpdateRule PUT /api/v1/agent/rules/:id +func (h *AgentTaskHandler) UpdateRule(c *gin.Context) { + var rule models.AgentRule + if err := h.db.First(&rule, c.Param("id")).Error; err != nil { + c.JSON(404, models.Response{Code: 404, Message: "规则不存在"}) + return + } + var update models.AgentRule + if err := c.ShouldBindJSON(&update); err != nil { + c.JSON(400, models.Response{Code: 400, Message: err.Error()}) + return + } + h.db.Model(&rule).Updates(update) + c.JSON(200, models.Response{Code: 200, Data: rule}) +} + +// DeleteRule DELETE /api/v1/agent/rules/:id +func (h *AgentTaskHandler) DeleteRule(c *gin.Context) { + h.db.Delete(&models.AgentRule{}, c.Param("id")) + c.JSON(200, models.Response{Code: 200, Message: "ok"}) +} + +// ListAlerts GET /api/v1/agent/alerts +func (h *AgentTaskHandler) ListAlerts(c *gin.Context) { + query := h.db.Model(&models.AgentAlert{}) + if c.Query("resolved") == "false" { + query = query.Where("resolved = ?", false) + } + if hostname := c.Query("hostname"); hostname != "" { + query = query.Where("hostname = ?", hostname) + } + var alerts []models.AgentAlert + query.Order("created_at DESC").Limit(200).Find(&alerts) + c.JSON(200, models.Response{Code: 200, Data: alerts}) +} + +// ResolveAlert POST /api/v1/agent/alerts/:id/resolve +func (h *AgentTaskHandler) ResolveAlert(c *gin.Context) { + now := time.Now() + h.db.Model(&models.AgentAlert{}).Where("id = ?", c.Param("id")). + Updates(map[string]interface{}{"resolved": true, "resolved_at": now}) + c.JSON(200, models.Response{Code: 200, Message: "ok"}) +} diff --git a/backend/internal/api/handlers/alert_engine.go b/backend/internal/api/handlers/alert_engine.go new file mode 100644 index 0000000..9db10e3 --- /dev/null +++ b/backend/internal/api/handlers/alert_engine.go @@ -0,0 +1,288 @@ +package handlers + +import ( + "encoding/json" + "fmt" + "time" + + "ezops/internal/models" + + "gorm.io/gorm" +) + +// AlertEngine 告警引擎 — 定时扫描 Agent 上报数据并触发告警 // Logan.Li +type AlertEngine struct { + db *gorm.DB +} + +func NewAlertEngine(db *gorm.DB) *AlertEngine { + return &AlertEngine{db: db} +} + +// Check 每 30 秒调用一次 +func (e *AlertEngine) Check() { + e.checkResourceAlerts() + e.checkOfflineAlerts() + e.checkPortAlerts() + e.autoResolveAlerts() +} + +// 检查 CPU/内存/磁盘 阈值 +func (e *AlertEngine) checkResourceAlerts() { + var rules []models.AgentRule + e.db.Where("enabled = ? AND metric IN ?", true, []string{"cpu", "mem", "disk"}).Find(&rules) + + for _, rule := range rules { + // 获取目标主机最新上报 + var report models.AgentReport + query := e.db.Model(&models.AgentReport{}) + if rule.Hostname != "" { + query = query.Where("hostname = ?", rule.Hostname) + } + query.Order("id DESC").First(&report) + if report.ID == 0 { + continue + } + + var value float64 + switch rule.Metric { + case "cpu": + value = report.CPUUsage + case "mem": + value = report.MemUsage + case "disk": + var disks []models.DiskMetric + json.Unmarshal([]byte(report.DiskInfo), &disks) + for _, d := range disks { + if d.UsagePercent > value { + value = d.UsagePercent + } + } + } + + // 检查是否超阈值 + if !e.compareOperator(rule.Operator, value, rule.Threshold) { + continue + } + + // 避免重复告警：如果最近 10 分钟已有同规则未恢复告警，不重复 + var existing models.AgentAlert + tenMinAgo := time.Now().Add(-10 * time.Minute) + e.db.Where("rule_name = ? AND hostname = ? AND resolved = ? AND created_at > ?", + rule.Name, report.Hostname, false, tenMinAgo).First(&existing) + if existing.ID > 0 { + continue + } + + // 创建告警 + alert := models.AgentAlert{ + RuleName: rule.Name, + Hostname: report.Hostname, + Metric: rule.Metric, + Severity: "warning", + Message: fmt.Sprintf("%s: %s %.1f%% %s %.1f%%", report.Hostname, rule.Metric, value, rule.Operator, rule.Threshold), + Value: value, + Threshold: rule.Threshold, + } + if value > rule.Threshold*1.5 { + alert.Severity = "critical" + } + e.db.Create(&alert) + + // 更新规则最后触发时间 + now := time.Now() + e.db.Model(&rule).Update("last_triggered", now) + } +} + +// 检查主机离线（超过 3 分钟无上报）— 内置规则，无需手动创建 +func (e *AlertEngine) checkOfflineAlerts() { + offlineThreshold := 3 * time.Minute + + // 查所有曾上报的主机 + var hostnames []string + e.db.Model(&models.AgentReport{}).Distinct("hostname").Pluck("hostname", &hostnames) + + for _, hostname := range hostnames { + var report models.AgentReport + e.db.Where("hostname = ?", hostname).Order("id DESC").First(&report) + if report.ID == 0 { + continue + } + + isOffline := time.Since(report.ReportTime) > offlineThreshold + + // 检查是否有未恢复的离线告警 + var existing models.AgentAlert + e.db.Where("hostname = ? AND metric = ? AND resolved = ?", + hostname, "offline", false).First(&existing) + + if isOffline && existing.ID == 0 { + alert := models.AgentAlert{ + RuleName: "主机离线", + Hostname: hostname, + Metric: "offline", + Severity: "critical", + Message: fmt.Sprintf("主机 %s 已超过 3 分钟无数据上报，可能已离线", hostname), + Value: time.Since(report.ReportTime).Minutes(), + Threshold: 3, + } + e.db.Create(&alert) + } + } +} + +// 检查端口消失（之前有的监听端口现在没了） +func (e *AlertEngine) checkPortAlerts() { + var rules []models.AgentRule + e.db.Where("enabled = ? AND metric = ?", true, "port_down").Find(&rules) + + for _, rule := range rules { + if rule.Port == 0 { + continue + } + + // 获取目标主机最新上报 + var report models.AgentReport + query := e.db.Model(&models.AgentReport{}) + if rule.Hostname != "" { + query = query.Where("hostname = ?", rule.Hostname) + } + query.Order("id DESC").First(&report) + if report.ID == 0 { + continue + } + + // 检查端口是否在最新上报中 + var ports []models.PortMetric + json.Unmarshal([]byte(report.PortInfo), &ports) + + portUp := false + for _, p := range ports { + if p.Port == rule.Port { + portUp = true + break + } + } + + if portUp { + continue // 端口正常 + } + + // 端口消失了，检查是否已有未恢复告警 + var existing models.AgentAlert + e.db.Where("rule_name = ? AND hostname = ? AND resolved = ?", + rule.Name, report.Hostname, false).First(&existing) + if existing.ID > 0 { + continue + } + + alert := models.AgentAlert{ + RuleName: rule.Name, + Hostname: report.Hostname, + Metric: "port_down", + Severity: "critical", + Message: fmt.Sprintf("主机 %s 端口 %d 服务疑似下线", report.Hostname, rule.Port), + Value: float64(rule.Port), + Threshold: float64(rule.Port), + } + e.db.Create(&alert) + + now := time.Now() + e.db.Model(&rule).Update("last_triggered", now) + } +} + +// 自动恢复告警（指标恢复正常） +func (e *AlertEngine) autoResolveAlerts() { + var alerts []models.AgentAlert + e.db.Where("resolved = ?", false).Find(&alerts) + + for _, alert := range alerts { + resolved := false + + switch alert.Metric { + case "offline": + // 主机重新上报了 + var report models.AgentReport + e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report) + if report.ID > 0 && time.Since(report.ReportTime) < 3*time.Minute { + resolved = true + } + case "cpu", "mem", "disk": + // 获取对应规则，检查是否恢复正常 + var rule models.AgentRule + e.db.Where("name = ?", alert.RuleName).First(&rule) + if rule.ID == 0 { + continue + } + var report models.AgentReport + if rule.Hostname != "" { + e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report) + } else { + e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report) + } + if report.ID == 0 { + continue + } + var value float64 + switch alert.Metric { + case "cpu": + value = report.CPUUsage + case "mem": + value = report.MemUsage + case "disk": + var disks []models.DiskMetric + json.Unmarshal([]byte(report.DiskInfo), &disks) + for _, d := range disks { + if d.UsagePercent > value { + value = d.UsagePercent + } + } + } + if !e.compareOperator(rule.Operator, value, rule.Threshold) { + resolved = true + } + case "port_down": + var rule models.AgentRule + e.db.Where("name = ?", alert.RuleName).First(&rule) + if rule.ID > 0 { + var report models.AgentReport + e.db.Where("hostname = ?", alert.Hostname).Order("id DESC").First(&report) + var ports []models.PortMetric + json.Unmarshal([]byte(report.PortInfo), &ports) + for _, p := range ports { + if p.Port == rule.Port { + resolved = true + break + } + } + } + } + + if resolved { + now := time.Now() + e.db.Model(&alert).Updates(map[string]interface{}{ + "resolved": true, + "resolved_at": now, + }) + } + } +} + +func (e *AlertEngine) compareOperator(op string, val, threshold float64) bool { + switch op { + case ">": + return val > threshold + case ">=": + return val >= threshold + case "<": + return val < threshold + case "<=": + return val <= threshold + case "==": + return val == threshold + default: + return val > threshold + } +} diff --git a/backend/internal/api/handlers/automation.go b/backend/internal/api/handlers/automation.go index aa67a47..02c26b6 100644 --- a/backend/internal/api/handlers/automation.go +++ b/backend/internal/api/handlers/automation.go @@ -1,8 +1,13 @@ package handlers import ( + "bytes" + "context" + "fmt" "net/http" + "os/exec" "strconv" + "time" "ezops/internal/models" @@ -10,6 +15,50 @@ import ( "gorm.io/gorm" ) +// runLocal executes a script or shell command locally on the AIOps host and +// returns the combined stdout+stderr output, the process exit code (0 = success) +// and the wall-clock duration in milliseconds. A timeout prevents runaway jobs. +func runLocal(scriptType, content string, timeoutSec int) (output string, exitCode int, durationMs int) { + start := time.Now() + timeout := time.Duration(timeoutSec) * time.Second + if timeout <= 0 { + timeout = 300 * time.Second + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + var cmd *exec.Cmd + switch scriptType { + case "python": + cmd = exec.CommandContext(ctx, "python3", "-c", content) + default: // shell / command / "" → run via bash + cmd = exec.CommandContext(ctx, "bash", "-c", content) + } + + var out bytes.Buffer + cmd.Stdout = &out + cmd.Stderr = &out + + err := cmd.Run() + durationMs = int(time.Since(start).Milliseconds()) + output = out.String() + + if ctx.Err() == context.DeadlineExceeded { + output += fmt.Sprintf("\n[超时] 执行超过 %d 秒被终止", timeoutSec) + return output, -1, durationMs + } + if err != nil { + if exitError, ok := err.(*exec.ExitError); ok { + exitCode = exitError.ExitCode() + } else { + exitCode = -1 + output += "\n" + err.Error() + } + return output, exitCode, durationMs + } + return output, 0, durationMs +} + // AutomationHandler 自动化运维处理器 type AutomationHandler struct { db *gorm.DB @@ -278,15 +327,28 @@ func (h *AutomationHandler) ExecuteCronJob(c *gin.Context) { return } + // Resolve the content to execute. For type=script, load the linked Script. + content := job.Content + scriptType := "shell" + if job.Type == "script" && job.ScriptID != nil { + var script models.Script + if err := h.db.First(&script, *job.ScriptID).Error; err == nil { + content = script.Content + scriptType = script.Type + h.db.Model(&script).UpdateColumn("run_count", script.RunCount+1) + } + } + + now := time.Now() execution := models.TaskExecution{ TaskType: "cron", TaskID: uint(id), TaskName: job.Name, - Status: "pending", + Status: "running", TargetType: job.TargetType, Targets: job.Targets, + StartedAt: &now, } - if err := h.db.Create(&execution).Error; err != nil { c.JSON(http.StatusInternalServerError, models.Response{ Code: 500, @@ -295,9 +357,46 @@ 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() + status := "success" + if exitCode != 0 { + status = "failed" + } + + h.db.Model(&execution).Updates(map[string]interface{}{ + "status": status, + "output": output, + "duration": duration, + "finished_at": finishedAt, + }) + + // Update cron job statistics. + statsUpdates := map[string]interface{}{ + "last_run_at": now, + "last_status": status, + "run_count": job.RunCount + 1, + } + if status == "success" { + statsUpdates["success_count"] = job.SuccessCount + 1 + } else { + statsUpdates["fail_count"] = job.FailCount + 1 + } + h.db.Model(&job).Updates(statsUpdates) + + execution.Output = output + execution.Status = status + execution.Duration = duration + execution.FinishedAt = &finishedAt + + msg := "执行成功" + if status == "failed" { + msg = "执行失败" + } c.JSON(http.StatusOK, models.Response{ Code: 200, - Message: "执行中", + Message: msg, Data: execution, }) } @@ -520,23 +619,19 @@ func (h *AutomationHandler) ExecuteScript(c *gin.Context) { return } + // The request body is optional — when omitted we execute locally. var req models.ExecuteScriptRequest - if err := c.ShouldBindJSON(&req); err != nil { - c.JSON(http.StatusBadRequest, models.Response{ - Code: 400, - Message: "请求参数错误: " + err.Error(), - }) - return - } + _ = c.ShouldBindJSON(&req) + now := time.Now() execution := models.TaskExecution{ TaskType: "script", TaskID: uint(id), TaskName: script.Name, - Status: "pending", + Status: "running", TargetType: req.TargetType, + StartedAt: &now, } - if err := h.db.Create(&execution).Error; err != nil { c.JSON(http.StatusInternalServerError, models.Response{ Code: 500, @@ -545,9 +640,74 @@ func (h *AutomationHandler) ExecuteScript(c *gin.Context) { return } + // Execute locally on the AIOps host. + output, exitCode, duration := runLocal(script.Type, script.Content, script.Timeout) + finishedAt := time.Now() + status := "success" + if exitCode != 0 { + status = "failed" + } + + h.db.Model(&execution).Updates(map[string]interface{}{ + "status": status, + "output": output, + "duration": duration, + "finished_at": finishedAt, + }) + h.db.Model(&script).UpdateColumn("run_count", script.RunCount+1) + + execution.Output = output + execution.Status = status + execution.Duration = duration + execution.FinishedAt = &finishedAt + + msg := "执行成功" + if status == "failed" { + msg = "执行失败" + } c.JSON(http.StatusOK, models.Response{ Code: 200, - Message: "执行中", + Message: msg, Data: execution, }) } + +// ListExecutions 获取任务执行记录列表 +// @Summary 获取任务执行记录 +// @Description 获取脚本/定时任务的执行历史记录（支持 task_type / task_id 筛选） +// @Tags 自动化运维 +// @Produce json +// @Security Bearer +// @Param page query int false "页码" +// @Param page_size query int false "每页数量" +// @Param task_type query string false "任务类型 (cron/script)" +// @Param task_id query string false "任务ID" +// @Success 200 {object} models.Response{data=models.ListResponse} +// @Router /task-executions [get] +func (h *AutomationHandler) ListExecutions(c *gin.Context) { + var query models.PageQuery + c.ShouldBindQuery(&query) + + db := h.db.Model(&models.TaskExecution{}) + if taskType := c.Query("task_type"); taskType != "" { + db = db.Where("task_type = ?", taskType) + } + if taskID := c.Query("task_id"); taskID != "" { + db = db.Where("task_id = ?", taskID) + } + + var total int64 + db.Count(&total) + + var execs []models.TaskExecution + db.Order("created_at DESC").Offset(query.GetOffset()).Limit(query.GetLimit()).Find(&execs) + + c.JSON(http.StatusOK, models.Response{ + Code: 200, + Message: "success", + Data: models.ListResponse{ + List: execs, + Total: total, + }, + }) +} diff --git a/backend/internal/api/routes/routes.go b/backend/internal/api/routes/routes.go index 7dc8465..f95959a 100644 --- a/backend/internal/api/routes/routes.go +++ b/backend/internal/api/routes/routes.go @@ -8,6 +8,7 @@ import ( "github.com/gin-gonic/gin" "go.uber.org/zap" "gorm.io/gorm" + "time" ) // Register 注册所有路由 @@ -260,6 +261,9 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config) { scripts.POST("/:id/execute", automationHandler.ExecuteScript) } + // 任务执行记录（脚本/定时任务执行历史） + authorized.GET("/task-executions", automationHandler.ListExecutions) + // 审批流程路由 approvalHandler := handlers.NewApprovalHandler(db) approvals := authorized.Group("/approvals") @@ -306,16 +310,44 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config) { audit.POST("/logs/clear", auditHandler.ClearAuditLogs) } - // Agent 上报路由（无需 JWT，使用 Agent Token 鉴权） + // Agent 上报路由（无需 JWT） agentHandler := handlers.NewAgentHandler(db) + taskHandler := handlers.NewAgentTaskHandler(db) agentGroup := api.Group("/agent") { agentGroup.POST("/report", agentHandler.ReceiveReport) + agentGroup.GET("/tasks/poll", taskHandler.PollTasks) + agentGroup.POST("/tasks/:task_id/result", taskHandler.SubmitResult) } - // Agent 数据查询（需要 JWT） + // Agent 数据查询 + 任务管理（需要 JWT） agentQuery := authorized.Group("/agent") { agentQuery.GET("/latest", agentHandler.GetLatestReports) agentQuery.GET("/history/:hostname", agentHandler.GetHostHistory) + + // 远程任务 + agentQuery.POST("/tasks", taskHandler.CreateTask) + agentQuery.GET("/tasks", taskHandler.ListTasks) + agentQuery.GET("/tasks/:id", taskHandler.GetTask) + + // 告警规则 + agentQuery.GET("/rules", taskHandler.ListRules) + agentQuery.POST("/rules", taskHandler.CreateRule) + agentQuery.PUT("/rules/:id", taskHandler.UpdateRule) + agentQuery.DELETE("/rules/:id", taskHandler.DeleteRule) + + // 告警事件 + agentQuery.GET("/alerts", taskHandler.ListAlerts) + agentQuery.POST("/alerts/:id/resolve", taskHandler.ResolveAlert) } + + // 启动告警引擎（每 30 秒检查一次） + alertEngine := handlers.NewAlertEngine(db) + go func() { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for range ticker.C { + alertEngine.Check() + } + }() } diff --git a/backend/internal/models/agent_task.go b/backend/internal/models/agent_task.go new file mode 100644 index 0000000..71fc463 --- /dev/null +++ b/backend/internal/models/agent_task.go @@ -0,0 +1,71 @@ +package models + +import ( + "time" + + "gorm.io/gorm" +) + +// AgentTask Agent 远程任务 // Logan.Li +type AgentTask struct { + ID uint `gorm:"primarykey" json:"id"` + TaskID string `gorm:"size:36;uniqueIndex;not null" json:"task_id"` // UUID + Hostname string `gorm:"size:255;index;not null" json:"hostname"` // 目标主机 + Type string `gorm:"size:32;not null" json:"type"` // command / script / file + Command string `gorm:"type:text" json:"command"` // 命令或脚本内容 + Timeout int `gorm:"default:60" json:"timeout"` // 超时秒数 + Status string `gorm:"size:20;default:pending;index" json:"status"` // pending/running/success/failed/timeout + Result string `gorm:"type:text" json:"result"` // 执行输出 + ExitCode int `gorm:"default:-1" json:"exit_code"` // 退出码 + CreatedBy string `gorm:"size:255" json:"created_by"` // 创建人 + CreatedAt time.Time `json:"created_at"` + PickedAt *time.Time `json:"picked_at"` // Agent 拉取时间 + FinishedAt *time.Time `json:"finished_at"` // 完成时间 + UpdatedAt time.Time `json:"updated_at"` + DeletedAt gorm.DeletedAt `gorm:"index" json:"-"` +} + +func (AgentTask) TableName() string { + return "agent_tasks" +} + +// AgentRule 告警规则（基于 Agent 数据） // Logan.Li +type AgentRule struct { + ID uint `gorm:"primarykey" json:"id"` + Name string `gorm:"size:255;not null" json:"name"` + Hostname string `gorm:"size:255" json:"hostname"` // 空表示全部主机 + Metric string `gorm:"size:32;not null" json:"metric"` // cpu/mem/disk/offline/port_down + Operator string `gorm:"size:10;default:>" json:"operator"` // > >= < <= == + Threshold float64 `gorm:"type:decimal(10,2)" json:"threshold"` + Port int `json:"port"` // port_down 专用 + Enabled bool `gorm:"default:true" json:"enabled"` + LastTriggered *time.Time `json:"last_triggered"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + DeletedAt gorm.DeletedAt `gorm:"index" json:"-"` +} + +func (AgentRule) TableName() string { + return "agent_rules" +} + +// AgentAlert Agent 告警事件 // Logan.Li +type AgentAlert struct { + ID uint `gorm:"primarykey" json:"id"` + RuleName string `gorm:"size:255" json:"rule_name"` + Hostname string `gorm:"size:255;index" json:"hostname"` + Metric string `gorm:"size:32" json:"metric"` + Severity string `gorm:"size:20;default:warning" json:"severity"` // critical/warning/info + Message string `gorm:"type:text" json:"message"` + Value float64 `gorm:"type:decimal(10,2)" json:"value"` + Threshold float64 `gorm:"type:decimal(10,2)" json:"threshold"` + Resolved bool `gorm:"default:false;index" json:"resolved"` + ResolvedAt *time.Time `json:"resolved_at"` + CreatedAt time.Time `gorm:"index" json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + DeletedAt gorm.DeletedAt `gorm:"index" json:"-"` +} + +func (AgentAlert) TableName() string { + return "agent_alerts" +} diff --git a/backend/pkg/database/database.go b/backend/pkg/database/database.go index 762d1ff..6ac77ae 100644 --- a/backend/pkg/database/database.go +++ b/backend/pkg/database/database.go @@ -92,5 +92,8 @@ func AutoMigrate(db *gorm.DB) error { &models.CloudBilling{}, &models.BillingAlertRule{}, &models.AgentReport{}, + &models.AgentTask{}, + &models.AgentRule{}, + &models.AgentAlert{}, ) } diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index e0a95d0..34694b5 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -35,6 +35,8 @@ import LogManager from '@/components/monitor/logs/LogManager'; import AlertCenter from '@/components/monitor/alert-center/AlertCenter'; import TraceManager from '@/components/monitor/trace/TraceManager'; import GrafanaDashboard from '@/components/monitor/dashboard/GrafanaDashboard'; +import { AgentMonitor } from '@/components/agent/AgentMonitor'; +import { AgentAlerts } from '@/components/agent/AgentAlerts'; const queryClient = new QueryClient({ defaultOptions: { @@ -61,6 +63,7 @@ function App() { } /> } /> } /> + } /> } /> } /> } /> @@ -75,6 +78,7 @@ function App() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/frontend/src/components/agent/AgentAlerts.tsx b/frontend/src/components/agent/AgentAlerts.tsx new file mode 100644 index 0000000..16e07bf --- /dev/null +++ b/frontend/src/components/agent/AgentAlerts.tsx @@ -0,0 +1,542 @@ +import { useState, useEffect, useCallback } from 'react'; +import { toast } from 'sonner'; +import { useConfirm } from '@/components/ui/confirm-dialog'; +import { agentApi } from '@/services/api'; +import { Card, CardContent } from '@/components/ui/card'; +import { Button } from '@/components/ui/button'; +import { Input } from '@/components/ui/input'; +import { Label } from '@/components/ui/label'; +import { Badge } from '@/components/ui/badge'; +import { Switch } from '@/components/ui/switch'; +import { Tabs, TabsContent, TabsList, TabsTrigger } from '@/components/ui/tabs'; +import { + Dialog, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, +} from '@/components/ui/dialog'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '@/components/ui/select'; +import { + AlertTriangle, + CheckCircle, + Plus, + Pencil, + Trash2, + Loader2, + RefreshCw, + Bell, + Cpu, + MemoryStick, + HardDrive, + WifiOff, + Network, + Clock, +} from 'lucide-react'; + +// ============ 常量 ============ + +const METRIC_OPTIONS = [ + { value: 'cpu', label: 'CPU', icon: Cpu }, + { value: 'mem', label: '内存', icon: MemoryStick }, + { value: 'disk', label: '磁盘', icon: HardDrive }, + { value: 'offline', label: '主机离线', icon: WifiOff }, + { value: 'port_down', label: '端口不通', icon: Network }, +]; + +const OPERATOR_OPTIONS = [ + { value: '>', label: '> 大于' }, + { value: '<', label: '< 小于' }, + { value: '>=', label: '≥ 大于等于' }, + { value: '<=', label: '≤ 小于等于' }, + { value: '==', label: '= 等于' }, +]; + +const SEVERITY_OPTIONS = [ + { value: 'critical', label: '严重', color: 'bg-red-500/15 text-red-400 border-red-500/30' }, + { value: 'warning', label: '警告', color: 'bg-amber-500/15 text-amber-400 border-amber-500/30' }, + { value: 'info', label: '提示', color: 'bg-blue-500/15 text-blue-400 border-blue-500/30' }, +]; + +function getMetricLabel(metric: string) { + return METRIC_OPTIONS.find((m) => m.value === metric)?.label || metric; +} + +function asList(data: any): any[] { + if (!data) return []; + if (Array.isArray(data)) return data; + if (Array.isArray(data.list)) return data.list; + if (Array.isArray(data.items)) return data.items; + if (Array.isArray(data.data)) return data.data; + return []; +} + +// ============ 主组件 ============ + +export function AgentAlerts() { + const confirm = useConfirm(); + const [activeTab, setActiveTab] = useState('events'); + const [alerts, setAlerts] = useState([]); + const [rules, setRules] = useState([]); + const [loading, setLoading] = useState(true); + const [resolvedView, setResolvedView] = useState(false); + + // 规则编辑 + const [ruleOpen, setRuleOpen] = useState(false); + const [editingRule, setEditingRule] = useState(null); + const [ruleSaving, setRuleSaving] = useState(false); + const [ruleForm, setRuleForm] = useState({ + name: '', + metric: 'cpu', + operator: '>', + threshold: '90', + port: '', + severity: 'warning', + enabled: true, + duration: '60', + }); + + const loadData = useCallback(async () => { + setLoading(true); + try { + const [alertsRes, rulesRes] = await Promise.all([ + agentApi.listAlerts({ resolved: String(resolvedView) }), + agentApi.listRules(), + ]); + setAlerts(asList(alertsRes)); + setRules(asList(rulesRes)); + } catch (err) { + console.error(err); + toast.error('加载告警数据失败'); + } finally { + setLoading(false); + } + }, [resolvedView]); + + useEffect(() => { + loadData(); + }, [loadData]); + + // ============ 告警事件 ============ + const handleResolve = async (id: number) => { + const ok = await confirm({ title: '恢复告警', description: '确定将此告警标记为已恢复？', confirmText: '恢复' }); + if (!ok) return; + try { + await agentApi.resolveAlert(id); + toast.success('告警已恢复'); + loadData(); + } catch (err: any) { + toast.error(err?.message || '操作失败'); + } + }; + + // ============ 规则 CRUD ============ + const resetRuleForm = () => { + setRuleForm({ name: '', metric: 'cpu', operator: '>', threshold: '90', port: '', severity: 'warning', enabled: true, duration: '60' }); + setEditingRule(null); + }; + + const openCreateRule = () => { + resetRuleForm(); + setRuleOpen(true); + }; + + const openEditRule = (rule: any) => { + setEditingRule(rule); + setRuleForm({ + name: rule.name || '', + metric: rule.metric || 'cpu', + operator: rule.operator || '>', + threshold: String(rule.threshold ?? '90'), + port: rule.port ? String(rule.port) : '', + severity: rule.severity || 'warning', + enabled: rule.enabled !== false, + duration: rule.duration ? String(rule.duration) : '60', + }); + setRuleOpen(true); + }; + + const handleSaveRule = async () => { + if (!ruleForm.name.trim()) { + toast.error('请输入规则名称'); + return; + } + setRuleSaving(true); + try { + const payload: any = { + name: ruleForm.name.trim(), + metric: ruleForm.metric, + operator: ruleForm.operator, + threshold: Number(ruleForm.threshold) || 0, + severity: ruleForm.severity, + enabled: ruleForm.enabled, + duration: Number(ruleForm.duration) || 0, + }; + if (ruleForm.metric === 'port_down' && ruleForm.port) { + payload.port = Number(ruleForm.port); + } + if (editingRule) { + await agentApi.updateRule(editingRule.id, payload); + toast.success('规则已更新'); + } else { + await agentApi.createRule(payload); + toast.success('规则已创建'); + } + setRuleOpen(false); + resetRuleForm(); + loadData(); + } catch (err: any) { + toast.error(err?.message || '保存失败'); + } finally { + setRuleSaving(false); + } + }; + + const handleDeleteRule = async (id: number) => { + const ok = await confirm({ title: '删除规则', description: '确定删除此告警规则？', confirmText: '删除', variant: 'destructive' }); + if (!ok) return; + try { + await agentApi.deleteRule(id); + toast.success('规则已删除'); + loadData(); + } catch (err: any) { + toast.error(err?.message || '删除失败'); + } + }; + + const handleToggleRule = async (rule: any) => { + try { + await agentApi.updateRule(rule.id, { ...rule, enabled: !rule.enabled }); + loadData(); + } catch (err: any) { + toast.error(err?.message || '操作失败'); + } + }; + + const severityBadge = (sev: string) => { + const opt = SEVERITY_OPTIONS.find((s) => s.value === sev) || SEVERITY_OPTIONS[1]; + return ( + + + {opt.label} + + ); + }; + + return ( +
+ {/* 页头 */} +
+
+

Agent 告警

+

基于 Agent 上报指标的告警规则与事件

+
+ +
+ + + + 告警事件 + 告警规则 + + + {/* ========== 告警事件 ========== */} + +
+ + +
+ + {loading ? ( + + + + 加载中... + + + ) : alerts.length === 0 ? ( + + + +

{resolvedView ? '暂无已恢复的告警' : '当前无活动告警，一切正常'}

+ + + ) : ( +
+ {alerts.map((alert) => ( + + +
+
+
+ +
+
+
+ {severityBadge(alert.severity)} + {alert.rule_name || alert.name || alert.title || '告警'} +
+

{alert.message || alert.description || '触发告警阈值'}

+
+ + + {alert.hostname || '--'} + + + 指标: {getMetricLabel(alert.metric)} + + {alert.value != null && ( + + 当前值: {alert.value} + + )} + + + {alert.created_at || alert.triggered_at ? new Date(alert.created_at || alert.triggered_at).toLocaleString() : '--'} + +
+
+
+ {!resolvedView && ( + + )} +
+ + + ))} +
+ )} + + + {/* ========== 告警规则 ========== */} + +
+ +
+ + {loading ? ( + + + + 加载中... + + + ) : rules.length === 0 ? ( + + + +

暂无告警规则

+ + + + ) : ( +
+ {rules.map((rule) => { + const MetricIcon = METRIC_OPTIONS.find((m) => m.value === rule.metric)?.icon || Cpu; + return ( + + +
+
+
+ +
+
+
+ {rule.name} + {severityBadge(rule.severity)} + {rule.enabled === false && ( + 已禁用 + )} +
+
+ {getMetricLabel(rule.metric)} + {rule.metric !== 'offline' && ( + + {rule.operator} {rule.threshold}{rule.metric === 'port_down' ? '' : '%'} + + )} + {rule.metric === 'port_down' && rule.port ? ( + 端口 {rule.port} + ) : null} + {rule.duration ? 持续 {rule.duration}s : null} +
+
+
+ +
+ handleToggleRule(rule)} /> + + +
+
+ + + ); + })} +
+ )} + + + + {/* ========== 规则编辑弹窗 ========== */} + { if (!v) setRuleOpen(false); }}> + + + {editingRule ? '编辑规则' : '新建规则'} + 配置 Agent 指标告警规则 + + +
+
+ + setRuleForm({ ...ruleForm, name: e.target.value })} + placeholder="例如：CPU 使用率告警" + className="h-9 bg-slate-800 border-slate-700 text-slate-200" + /> +
+ +
+
+ + +
+
+ + +
+
+ + {ruleForm.metric !== 'offline' && ( +
+
+ + +
+
+ + setRuleForm({ ...ruleForm, threshold: e.target.value })} + className="h-9 bg-slate-800 border-slate-700 text-slate-200" + /> +
+
+ )} + + {ruleForm.metric === 'port_down' && ( +
+ + setRuleForm({ ...ruleForm, port: e.target.value })} + placeholder="例如：8080" + className="h-9 bg-slate-800 border-slate-700 text-slate-200" + /> +
+ )} + +
+ + setRuleForm({ ...ruleForm, duration: e.target.value })} + placeholder="指标持续超限多久后告警" + className="h-9 bg-slate-800 border-slate-700 text-slate-200" + /> +
+ +
+ + setRuleForm({ ...ruleForm, enabled: v })} /> +
+
+ + + + + + + +
+ ); +} + +export default AgentAlerts; diff --git a/frontend/src/components/agent/AgentMonitor.tsx b/frontend/src/components/agent/AgentMonitor.tsx new file mode 100644 index 0000000..6345550 --- /dev/null +++ b/frontend/src/components/agent/AgentMonitor.tsx @@ -0,0 +1,671 @@ +import { useState, useEffect, useRef, useCallback } from 'react'; +import { toast } from 'sonner'; +import { agentApi } from '@/services/api'; +import { AgentWizard } from './AgentWizard'; +import { Card, CardContent } from '@/components/ui/card'; +import { Button } from '@/components/ui/button'; +import { Input } from '@/components/ui/input'; +import { Label } from '@/components/ui/label'; +import { Textarea } from '@/components/ui/textarea'; +import { Badge } from '@/components/ui/badge'; +import { Progress } from '@/components/ui/progress'; +import { + Dialog, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, +} from '@/components/ui/dialog'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '@/components/ui/select'; +import { + Cpu, + MemoryStick, + HardDrive, + Wifi, + WifiOff, + RefreshCw, + Terminal, + Rocket, + Activity, + Network, + Loader2, + Clock, + Server, +} from 'lucide-react'; +import { + AreaChart, + Area, + XAxis, + YAxis, + CartesianGrid, + Tooltip, + ResponsiveContainer, +} from 'recharts'; + +// ============ 工具函数 ============ + +/** 在线判定：3 分钟内有上报视为在线 */ +function isOnline(lastReport: any): boolean { + if (!lastReport) return false; + const t = new Date(lastReport).getTime(); + if (Number.isNaN(t)) return false; + return Date.now() - t < 3 * 60 * 1000; +} + +function timeAgo(ts: any): string { + if (!ts) return '--'; + const t = new Date(ts).getTime(); + if (Number.isNaN(t)) return '--'; + const diff = Math.floor((Date.now() - t) / 1000); + if (diff < 60) return `${diff}秒前`; + if (diff < 3600) return `${Math.floor(diff / 60)}分钟前`; + if (diff < 86400) return `${Math.floor(diff / 3600)}小时前`; + return new Date(ts).toLocaleString(); +} + +function asList(data: any): any[] { + if (!data) return []; + if (Array.isArray(data)) return data; + if (Array.isArray(data.list)) return data.list; + if (Array.isArray(data.items)) return data.items; + if (Array.isArray(data.data)) return data.data; + return []; +} + +// ============ 组件 ============ + +export function AgentMonitor() { + const [hosts, setHosts] = useState([]); + const [loading, setLoading] = useState(true); + const [refreshing, setRefreshing] = useState(false); + const [selectedHost, setSelectedHost] = useState(null); + const [wizardOpen, setWizardOpen] = useState(false); + const timerRef = useRef | null>(null); + + // 远程命令 + const [cmdOpen, setCmdOpen] = useState(false); + const [cmdHost, setCmdHost] = useState(''); + const [cmdType, setCmdType] = useState('command'); + const [cmdCommand, setCmdCommand] = useState(''); + const [cmdTimeout, setCmdTimeout] = useState('30'); + const [cmdSubmitting, setCmdSubmitting] = useState(false); + const [cmdResult, setCmdResult] = useState(null); + const [cmdPolling, setCmdPolling] = useState(false); + const pollTimerRef = useRef | null>(null); + + const loadLatest = useCallback(async (silent = false) => { + if (!silent) setRefreshing(true); + try { + const data = await agentApi.getLatest(); + const list = asList(data); + setHosts(list); + // 同步选中主机的最新数据 + if (selectedHost) { + const fresh = list.find((h) => h.hostname === selectedHost.hostname); + if (fresh) setSelectedHost(fresh); + } + } catch (err) { + if (!silent) toast.error('加载主机列表失败'); + console.error(err); + } finally { + setLoading(false); + setRefreshing(false); + } + }, [selectedHost]); + + useEffect(() => { + loadLatest(); + timerRef.current = setInterval(() => loadLatest(true), 30000); + return () => { + if (timerRef.current) clearInterval(timerRef.current); + }; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, []); + + // 打开远程命令弹窗 + const openCommandDialog = (hostname: string) => { + setCmdHost(hostname); + setCmdType('command'); + setCmdCommand(''); + setCmdTimeout('30'); + setCmdResult(null); + setCmdOpen(true); + }; + + // 提交远程命令并轮询结果 + const submitCommand = async () => { + if (!cmdCommand.trim()) { + toast.error('请输入要执行的命令'); + return; + } + setCmdSubmitting(true); + setCmdResult(null); + try { + const res: any = await agentApi.createTask({ + hostname: cmdHost, + type: cmdType, + command: cmdCommand.trim(), + timeout: Number(cmdTimeout) || 30, + }); + const taskId = res?.id ?? res?.task_id; + toast.success('任务已下发'); + setCmdSubmitting(false); + + // 轮询任务结果 + if (taskId != null) { + setCmdPolling(true); + let attempts = 0; + pollTimerRef.current = setInterval(async () => { + attempts++; + try { + const tasks: any = await agentApi.listTasks({ hostname: cmdHost }); + const list = asList(tasks); + const t = list.find((x: any) => String(x.id) === String(taskId)); + if (t && t.status && t.status !== 'pending' && t.status !== 'running') { + setCmdResult(t); + setCmdPolling(false); + if (pollTimerRef.current) { + clearInterval(pollTimerRef.current); + pollTimerRef.current = null; + } + loadLatest(true); + } + } catch { + /* ignore */ + } + if (attempts > 15) { + // 30s 超时 + setCmdPolling(false); + setCmdResult({ status: 'timeout', result: '等待结果超时，请稍后在任务列表查看' }); + if (pollTimerRef.current) { + clearInterval(pollTimerRef.current); + pollTimerRef.current = null; + } + } + }, 2000); + } + } catch (err: any) { + toast.error(err?.message || '任务下发失败'); + setCmdSubmitting(false); + } + }; + + const closeCommandDialog = () => { + setCmdOpen(false); + if (pollTimerRef.current) { + clearInterval(pollTimerRef.current); + pollTimerRef.current = null; + } + setCmdPolling(false); + }; + + const onlineCount = hosts.filter((h) => isOnline(h.last_report || h.last_reported_at || h.updated_at)).length; + + const stats = [ + { label: '主机总数', value: hosts.length, icon: Server, color: 'text-blue-400' }, + { label: '在线', value: onlineCount, icon: Wifi, color: 'text-emerald-400' }, + { label: '离线', value: hosts.length - onlineCount, icon: WifiOff, color: 'text-red-400' }, + ]; + + return ( +
+ {/* 页头 */} +
+
+

Agent 监控

+

实时监控 Agent 上报的主机指标

+
+
+ + +
+
+ + {/* 统计卡片 */} +
+ {stats.map((s) => ( + + +
+ +
+
+
{s.value}
+
{s.label}
+
+ + + ))} +
+ + {/* 主机列表 */} + {loading ? ( + + + + 加载中... + + + ) : hosts.length === 0 ? ( + + + +

暂无 Agent 上报数据

+ + + + ) : ( +
+ {hosts.map((host) => { + const online = isOnline(host.last_report || host.last_reported_at || host.updated_at); + const cpu = Number(host.cpu_percent ?? host.cpu ?? 0); + const mem = Number(host.mem_percent ?? host.mem ?? 0); + const disk = Number(host.disk_percent ?? host.disk ?? 0); + const ports: any[] = Array.isArray(host.ports) ? host.ports : []; + return ( + setSelectedHost(host)} + > + +
+
+ + {host.hostname} +
+ {online ? ( + + + 在线 + + ) : ( + + + 离线 + + )} +
+ + {/* 指标条 */} +
+ + + +
+ + {/* 端口 & 时间 */} +
+ + + {ports.length} 个端口 + + + + {timeAgo(host.last_report || host.last_reported_at || host.updated_at)} + +
+ + {/* 操作 */} +
e.stopPropagation()}> + + +
+ + + ); + })} +
+ )} + + {/* 详情抽屉 */} + setSelectedHost(null)} + onCommand={(hn) => openCommandDialog(hn)} + /> + + {/* 远程命令弹窗 */} + { if (!v) closeCommandDialog(); }}> + + + + + 远程命令 + + + 在 {cmdHost} 上执行命令 + + + +
+
+
+ + setCmdHost(e.target.value)} + readOnly + className="h-9 bg-slate-800 border-slate-700 text-slate-200" + /> +
+
+ + +
+
+ +
+ +