// 定时任务执行引擎 — 供 HTTP 手动触发与后台 Cron 调度共用 // Logan.Li package controller import ( "fmt" "time" "ezops/model" "ezops/pkg/remoteexec" "ezops/pkg/scriptrunner" "gorm.io/gorm" ) // RunCronJobByID 执行指定定时任务并写入执行记录 // Logan.Li func RunCronJobByID(db *gorm.DB, id uint) (*model.TaskExecution, error) { var job model.CronJob if err := db.First(&job, id).Error; err != nil { return nil, fmt.Errorf("定时任务不存在") } return runCronJob(db, job) } // QueueCronJobByID 提交一次手动执行,立即返回执行记录,后台继续执行 // Logan.Li func QueueCronJobByID(db *gorm.DB, id uint) (*model.TaskExecution, error) { var job model.CronJob if err := db.First(&job, id).Error; err != nil { return nil, fmt.Errorf("定时任务不存在") } execution, startedAt, err := createCronExecution(db, job) if err != nil { return nil, err } go finishCronExecution(db, job, execution.ID, startedAt) return execution, nil } func runCronJob(db *gorm.DB, job model.CronJob) (*model.TaskExecution, error) { execution, startedAt, err := createCronExecution(db, job) if err != nil { return nil, err } finishCronExecution(db, job, execution.ID, startedAt) _ = db.First(execution, execution.ID).Error return execution, nil } func createCronExecution(db *gorm.DB, job model.CronJob) (*model.TaskExecution, time.Time, error) { startedAt := time.Now() execution := &model.TaskExecution{ TaskType: "cron", TaskID: job.ID, TaskName: job.Name, Status: "running", TargetType: job.TargetType, Targets: job.Targets, StartedAt: &startedAt, } if err := db.Create(execution).Error; err != nil { return nil, startedAt, fmt.Errorf("创建执行记录失败") } return execution, startedAt, nil } // finishCronExecution 执行定时任务主体并写回执行结果 // Logan.Li func finishCronExecution(db *gorm.DB, job model.CronJob, executionID uint, startedAt time.Time) { content := job.Content scriptType := job.Type if scriptType == "command" { scriptType = "shell" } if job.Type == "script" && job.ScriptID != nil { var script model.Script if err := db.First(&script, *job.ScriptID).Error; err == nil { content = script.Content scriptType = script.Type db.Model(&script).UpdateColumn("run_count", script.RunCount+1) } } var output string var exitCode int var duration int status := "success" if job.TargetType != "" && job.TargetType != "local" && job.Targets != "" { remoteResults, remoteErr := ExecuteCronOnTargets(db, job, scriptType, content, job.Timeout) duration = int(time.Since(startedAt).Milliseconds()) if remoteErr != nil { output = fmt.Sprintf("远程执行调度失败: %v", remoteErr) exitCode = -1 status = "failed" } else { output = remoteexec.FormatExecResults(remoteResults) for _, r := range remoteResults { if !r.Success { status = "failed" exitCode = -1 break } } } } else { rt := scriptrunner.DetectRuntime() if ok, reason := scriptrunner.CanRun(rt, scriptType); !ok { output = reason + "\n" exitCode = -1 duration = int(time.Since(startedAt).Milliseconds()) status = "failed" } else { output, exitCode, duration = runLocal(scriptType, content, job.Timeout) } if exitCode != 0 { status = "failed" } } finishedAt := time.Now() db.Model(&model.TaskExecution{}).Where("id = ?", executionID).Updates(map[string]interface{}{ "status": status, "output": output, "duration": duration, "finished_at": finishedAt, }) statsUpdates := map[string]interface{}{ "last_run_at": startedAt, "last_status": status, "run_count": job.RunCount + 1, } if status == "success" { statsUpdates["success_count"] = job.SuccessCount + 1 } else { statsUpdates["fail_count"] = job.FailCount + 1 } db.Model(&job).Updates(statsUpdates) _ = exitCode }