133 lines
4.2 KiB
Go
133 lines
4.2 KiB
Go
package jobs
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
log "github.com/go-admin-team/go-admin-core/logger"
|
|
"github.com/google/uuid"
|
|
"gorm.io/gorm"
|
|
|
|
"go-admin/app/jobs/models"
|
|
)
|
|
|
|
const (
|
|
executionErrorTargetMissing = "JOB_TARGET_NOT_FOUND"
|
|
executionErrorExecFailed = "JOB_EXECUTION_FAILED"
|
|
executionErrorHTTPFailed = "JOB_HTTP_FAILED"
|
|
executionErrorPanicked = "JOB_EXECUTION_PANICKED"
|
|
executionErrorInterrupted = "JOB_INTERRUPTED"
|
|
)
|
|
|
|
type executionFailure struct {
|
|
code string
|
|
message string
|
|
cause error
|
|
}
|
|
|
|
func (failure *executionFailure) Error() string {
|
|
if failure.cause != nil {
|
|
return failure.cause.Error()
|
|
}
|
|
return failure.message
|
|
}
|
|
|
|
func newExecutionFailure(code, message string, cause error) error {
|
|
return &executionFailure{code: code, message: message, cause: cause}
|
|
}
|
|
|
|
func runWithExecutionLog(db *gorm.DB, core JobCore, execute func() error) (executionErr error) {
|
|
startedAt := time.Now().UTC()
|
|
record := &models.SysJobExecutionLog{
|
|
ExecutionID: uuid.NewString(), JobID: core.JobId,
|
|
JobNameSnapshot: core.Name, InvokeTargetSnapshot: core.InvokeTarget,
|
|
TriggerType: models.JobTriggerScheduled, Status: models.JobExecutionRunning,
|
|
StartedAt: startedAt,
|
|
}
|
|
created := false
|
|
if db != nil {
|
|
if err := db.WithContext(context.Background()).Create(record).Error; err != nil {
|
|
log.Errorf("[Job] execution log create failed job_id=%d: %v", core.JobId, err)
|
|
} else {
|
|
created = true
|
|
}
|
|
}
|
|
|
|
defer func() {
|
|
if recovered := recover(); recovered != nil {
|
|
executionErr = newExecutionFailure(executionErrorPanicked, "任务执行异常中断", nil)
|
|
if created {
|
|
finishExecutionLog(db, core.JobId, record, startedAt, executionErr)
|
|
}
|
|
return
|
|
}
|
|
if created {
|
|
finishExecutionLog(db, core.JobId, record, startedAt, executionErr)
|
|
}
|
|
}()
|
|
executionErr = execute()
|
|
return executionErr
|
|
}
|
|
|
|
func finishExecutionLog(db *gorm.DB, jobID int, record *models.SysJobExecutionLog, startedAt time.Time, executionErr error) {
|
|
finishedAt := time.Now().UTC()
|
|
updates := map[string]any{
|
|
"status": models.JobExecutionSucceeded, "finished_at": finishedAt,
|
|
"duration_ms": finishedAt.Sub(startedAt).Milliseconds(), "error_code": "", "error_message": "",
|
|
}
|
|
if executionErr != nil {
|
|
code, message := publicExecutionFailure(executionErr)
|
|
updates["status"] = models.JobExecutionFailed
|
|
updates["error_code"] = code
|
|
updates["error_message"] = message
|
|
}
|
|
if err := db.WithContext(context.Background()).Model(&models.SysJobExecutionLog{}).
|
|
Where("id = ? AND status = ?", record.ID, models.JobExecutionRunning).Updates(updates).Error; err != nil {
|
|
log.Errorf("[Job] execution log finish failed job_id=%d execution_id=%s: %v", jobID, record.ExecutionID, err)
|
|
}
|
|
}
|
|
|
|
func publicExecutionFailure(err error) (string, string) {
|
|
var failure *executionFailure
|
|
if errors.As(err, &failure) {
|
|
return failure.code, failure.message
|
|
}
|
|
return executionErrorExecFailed, "任务执行失败，请查看受控服务日志"
|
|
}
|
|
|
|
// RecoverInterruptedExecutionLogs closes invocations left running by the
|
|
// previous process. Production currently runs one scheduler per database.
|
|
func RecoverInterruptedExecutionLogs(db *gorm.DB) error {
|
|
if db == nil {
|
|
return nil
|
|
}
|
|
now := time.Now().UTC()
|
|
var records []models.SysJobExecutionLog
|
|
if err := db.WithContext(context.Background()).Where("status = ?", models.JobExecutionRunning).Find(&records).Error; err != nil {
|
|
return err
|
|
}
|
|
return db.WithContext(context.Background()).Transaction(func(tx *gorm.DB) error {
|
|
for _, record := range records {
|
|
duration := now.Sub(record.StartedAt).Milliseconds()
|
|
if duration < 0 {
|
|
duration = 0
|
|
}
|
|
if err := tx.Model(&models.SysJobExecutionLog{}).
|
|
Where("id = ? AND status = ?", record.ID, models.JobExecutionRunning).Updates(map[string]any{
|
|
"status": models.JobExecutionInterrupted, "finished_at": now,
|
|
"duration_ms": duration, "error_code": executionErrorInterrupted,
|
|
"error_message": "服务重启前任务未完成",
|
|
}).Error; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func missingExecutionTarget(target string) error {
|
|
return newExecutionFailure(executionErrorTargetMissing, "任务调用目标未注册", fmt.Errorf("job target %q is not registered", target))
|
|
}
|