feat: 脚本/定时任务真实执行+输出捕获+执行历史 — 替换pending空壳为本地执行,采集stdout/stderr,更新统计;修复CronJobManager/Automation字段名不匹配(status→enabled,cron→cron_expr,command→content,language→type)
This commit is contained in:
1 parent
85bc2a8e05
commit
a4fd5fc178
19 files changed
+2715
-114
No files matched your search
Binary file not shown.
Binary file not shown.
@@ -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"})
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
@@ -92,5 +92,8 @@ func AutoMigrate(db *gorm.DB) error {
|
||||
&models.CloudBilling{},
|
||||
&models.BillingAlertRule{},
|
||||
&models.AgentReport{},
|
||||
&models.AgentTask{},
|
||||
&models.AgentRule{},
|
||||
&models.AgentAlert{},
|
||||
)
|
||||
}
|
||||
Reference in new issue
Block a user