admin/TaskPool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add random delay

Browse Source
This commit is contained in:
engigu
2026-03-02 18:29:07 +08:00
parent 42ac0f4acb
commit d8f1b8b50c
13 changed files with 114 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/controllers/task_controller.go
+4 -2
View File
@@ -67,6 +67,7 @@ func (tc *TaskController) CreateTask(c *gin.Context) {
TriggerType string `json:"trigger_type"`
RetryCount int `json:"retry_count"`
RetryInterval int `json:"retry_interval"`
RandomRange int `json:"random_range"`
}
if err := c.ShouldBindJSON(&req); err != nil {
@@ -93,7 +94,7 @@ func (tc *TaskController) CreateTask(c *gin.Context) {
workDir = resolveWorkDir(req.WorkDir)
}
task := tc.taskService.CreateTask(req.Name, req.Command, req.Schedule, req.Timeout, workDir, req.CleanConfig, req.Envs, req.Type, req.Config, req.AgentID, req.Languages, req.TriggerType, req.Tags, req.RetryCount, req.RetryInterval)
task := tc.taskService.CreateTask(req.Name, req.Command, req.Schedule, req.Timeout, workDir, req.CleanConfig, req.Envs, req.Type, req.Config, req.AgentID, req.Languages, req.TriggerType, req.Tags, req.RetryCount, req.RetryInterval, req.RandomRange)
// 如果是 Agent 任务，通知 Agent；否则添加到本地 cron
if task.AgentID != nil && *task.AgentID > 0 {
@@ -172,6 +173,7 @@ func (tc *TaskController) UpdateTask(c *gin.Context) {
TriggerType string `json:"trigger_type"`
RetryCount int `json:"retry_count"`
RetryInterval int `json:"retry_interval"`
RandomRange int `json:"random_range"`
}
if err := c.ShouldBindJSON(&req); err != nil {
@@ -192,7 +194,7 @@ func (tc *TaskController) UpdateTask(c *gin.Context) {
workDir = resolveWorkDir(req.WorkDir)
}
task := tc.taskService.UpdateTask(id, req.Name, req.Command, req.Schedule, req.Timeout, workDir, req.CleanConfig, req.Envs, req.Enabled, req.Type, req.Config, req.AgentID, req.Languages, req.TriggerType, req.Tags, req.RetryCount, req.RetryInterval)
task := tc.taskService.UpdateTask(id, req.Name, req.Command, req.Schedule, req.Timeout, workDir, req.CleanConfig, req.Envs, req.Enabled, req.Type, req.Config, req.AgentID, req.Languages, req.TriggerType, req.Tags, req.RetryCount, req.RetryInterval, req.RandomRange)
if task == nil {
utils.NotFound(c, "任务不存在")
return
internal/executor/cron.go
+30 -15
View File
@@ -1,7 +1,9 @@
package executor
import (
"math/rand"
"sync"
"time"
"github.com/engigu/baihu-panel/internal/systime"
@@ -87,27 +89,40 @@ func (m *CronManager) AddTask(task CronTask) error {
m.logger.Errorf("[CronManager] 任务 #%s 执行过程中发生 Panic: %v", taskID, r)
}
}()
m.logger.Infof("[CronManager] 触发计划任务 #%s (%s)", taskID, name)
req := &ExecutionRequest{
TaskID: taskID,
Name: name,
Command: cmd,
Type: TaskTypeCron,
Timeout: timeout,
WorkDir: workDir,
Envs: ParseEnvVars(envs),
Languages: languages,
UseMise: useMise,
// 构造执行请求的 Builder
reqBuilder := func() *ExecutionRequest {
return &ExecutionRequest{
TaskID: taskID,
Name: name,
Command: cmd,
Type: TaskTypeCron,
Timeout: timeout,
WorkDir: workDir,
Envs: ParseEnvVars(envs),
Languages: languages,
UseMise: useMise,
}
}
// 如果有关联的 Scheduler，加入队列执行
if m.scheduler != nil {
m.scheduler.EnqueueOrExecute(req)
randomRange := task.GetRandomRange()
if randomRange > 0 && m.scheduler != nil {
// 生成 0 到 randomRange 之间的随机秒数
delaySeconds := rand.Intn(randomRange)
delay := time.Duration(delaySeconds) * time.Second
m.logger.Infof("[CronManager] 任务 #%s 将随机延迟 %v (范围: %ds) 后入队执行", taskID, delay, randomRange)
// 使用调度器的延时投递功能，不阻塞当前 Cron 协程
m.scheduler.EnqueueDelayed(delay, reqBuilder)
} else {
m.logger.Infof("[CronManager] 触发计划任务 #%s (%s)", taskID, name)
if m.scheduler != nil {
m.scheduler.EnqueueOrExecute(reqBuilder())
}
}
// 触发下次运行时间更新事件
m.triggerNextRunEvent(taskID, req)
m.triggerNextRunEvent(taskID, &ExecutionRequest{TaskID: taskID})
})
if err != nil {
internal/executor/executor.go
+1
View File
@@ -32,6 +32,7 @@ type CronTask interface {
Task
GetSchedule() string
UseMise() bool
GetRandomRange() int
}
// Request 任务执行请求
internal/models/agent.go
+25 -2
View File
@@ -1,6 +1,8 @@
package models
import (
"strconv"
"github.com/engigu/baihu-panel/internal/constant"
"gorm.io/gorm"
@@ -59,8 +61,29 @@ type AgentTask struct {
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
Envs string `json:"envs"`
Languages []map[string]string `json:"languages"`
Enabled bool `json:"enabled"`
Languages []map[string]string `json:"languages"`
RandomRange int `json:"random_range"`
Enabled bool `json:"enabled"`
}
func (t AgentTask) GetID() string {
return strconv.FormatUint(uint64(t.ID), 10)
}
func (t AgentTask) GetName() string {
return t.Name
}
func (t AgentTask) GetCommand() string {
return t.Command
}
func (t AgentTask) GetSchedule() string {
return t.Schedule
}
func (t AgentTask) GetRandomRange() int {
return t.RandomRange
}
// AgentTaskResult Agent 上报的任务执行结果
internal/models/task.go
+5
View File
@@ -50,6 +50,7 @@ type Task struct {
AgentID *uint `json:"agent_id" gorm:"index"` // Agent ID，为空表示本地执行
RetryCount int `json:"retry_count" gorm:"default:0"` // 失败重试次数
RetryInterval int `json:"retry_interval" gorm:"default:0"` // 失败重试间隔(秒)
RandomRange int `json:"random_range" gorm:"default:0"` // 随机延迟范围(秒)
Enabled bool `json:"enabled" gorm:"default:true"`
RunningGo string `json:"running_go" gorm:"type:text"` // 正在运行的 go routine id 数组 (JSON)
LastRun *LocalTime `json:"last_run"`
@@ -103,6 +104,10 @@ func (t *Task) GetSchedule() string {
return t.Schedule
}
func (t *Task) GetRandomRange() int {
return t.RandomRange
}
// TaskLog 代表任务执行的日志记录
type TaskLog struct {
ID uint `json:"id" gorm:"primaryKey"`
internal/models/vo/task_vo.go
+2
View File
@@ -24,6 +24,7 @@ type TaskVO struct {
Enabled bool `json:"enabled"`
RetryCount int `json:"retry_count"`
RetryInterval int `json:"retry_interval"`
RandomRange int `json:"random_range"`
LastRun *models.LocalTime `json:"last_run"`
NextRun *models.LocalTime `json:"next_run"`
CreatedAt models.LocalTime `json:"created_at"`
@@ -53,6 +54,7 @@ func ToTaskVO(task *models.Task) *TaskVO {
Enabled: task.Enabled,
RetryCount: task.RetryCount,
RetryInterval: task.RetryInterval,
RandomRange: task.RandomRange,
LastRun: task.LastRun,
NextRun: task.NextRun,
CreatedAt: task.CreatedAt,
internal/services/agent_service.go
+4 -3
View File
@@ -317,9 +317,10 @@ func (s *AgentService) GetTasks(agentID uint) []models.AgentTask {
Schedule: task.Schedule,
Timeout: task.Timeout,
WorkDir: task.WorkDir,
Envs: envVarsStr, // 传递 "KEY1=VALUE1,KEY2=VALUE2" 格式
Languages: task.Languages,
Enabled: task.Enabled,
Envs: envVarsStr, // 传递 "KEY1=VALUE1,KEY2=VALUE2" 格式
Languages: task.Languages,
RandomRange: task.RandomRange,
Enabled: task.Enabled,
}
}
internal/services/tasks/task_service.go
+4 -2
View File
@@ -12,7 +12,7 @@ func NewTaskService() *TaskService {
return &TaskService{}
}
func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, workDir, cleanConfig, envs, taskType, config string, agentID *uint, languages []map[string]string, triggerType string, tags string, retryCount int, retryInterval int) *models.Task {
func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, workDir, cleanConfig, envs, taskType, config string, agentID *uint, languages []map[string]string, triggerType string, tags string, retryCount int, retryInterval int, randomRange int) *models.Task {
if taskType == "" {
taskType = "task"
}
@@ -36,6 +36,7 @@ func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, w
Enabled: true,
RetryCount: retryCount,
RetryInterval: retryInterval,
RandomRange: randomRange,
}
if triggerType != constant.TriggerTypeCron {
task.NextRun = nil
@@ -83,7 +84,7 @@ func (ts *TaskService) GetTaskByID(id int) *models.Task {
return &task
}
func (ts *TaskService) UpdateTask(id int, name, command, schedule string, timeout int, workDir, cleanConfig, envs string, enabled bool, taskType, config string, agentID *uint, languages []map[string]string, triggerType string, tags string, retryCount int, retryInterval int) *models.Task {
func (ts *TaskService) UpdateTask(id int, name, command, schedule string, timeout int, workDir, cleanConfig, envs string, enabled bool, taskType, config string, agentID *uint, languages []map[string]string, triggerType string, tags string, retryCount int, retryInterval int, randomRange int) *models.Task {
var task models.Task
if err := database.DB.First(&task, id).Error; err != nil {
return nil
@@ -101,6 +102,7 @@ func (ts *TaskService) UpdateTask(id int, name, command, schedule string, timeou
task.Languages = languages
task.RetryCount = retryCount
task.RetryInterval = retryInterval
task.RandomRange = randomRange
if taskType != "" {
task.Type = taskType
}