diff --git a/agent/agent.go b/agent/agent.go index d84f01c..28233a4 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -51,8 +51,9 @@ 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 { @@ -98,6 +99,10 @@ func (t *AgentTask) GetSchedule() string { return t.Cron } +func (t *AgentTask) GetRandomRange() int { + return t.RandomRange +} + type TaskResult struct { TaskID uint `json:"task_id"` LogID uint `json:"log_id"` @@ -674,7 +679,8 @@ func (a *Agent) updateTasks(tasks []AgentTask) { oldTask, exists := a.tasks[id] if !exists || oldTask.Schedule != task.Schedule || oldTask.Command != task.Command || oldTask.Enabled != task.Enabled || oldTask.Timeout != task.Timeout || - oldTask.WorkDir != task.WorkDir || oldTask.Envs != task.Envs { + oldTask.WorkDir != task.WorkDir || oldTask.Envs != task.Envs || + oldTask.RandomRange != task.RandomRange { if task.Enabled { err := a.cronManager.AddTask(task) if err != nil { diff --git a/internal/controllers/task_controller.go b/internal/controllers/task_controller.go index e821291..bb16b96 100644 --- a/internal/controllers/task_controller.go +++ b/internal/controllers/task_controller.go @@ -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 diff --git a/internal/executor/cron.go b/internal/executor/cron.go index 690052a..b79baf3 100644 --- a/internal/executor/cron.go +++ b/internal/executor/cron.go @@ -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 { diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 88e1932..b18a6d0 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -32,6 +32,7 @@ type CronTask interface { Task GetSchedule() string UseMise() bool + GetRandomRange() int } // Request 任务执行请求 diff --git a/internal/models/agent.go b/internal/models/agent.go index d34be0d..e878423 100644 --- a/internal/models/agent.go +++ b/internal/models/agent.go @@ -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 上报的任务执行结果 diff --git a/internal/models/task.go b/internal/models/task.go index 4e53018..e7d0143 100644 --- a/internal/models/task.go +++ b/internal/models/task.go @@ -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"` diff --git a/internal/models/vo/task_vo.go b/internal/models/vo/task_vo.go index 0316d63..a31e472 100644 --- a/internal/models/vo/task_vo.go +++ b/internal/models/vo/task_vo.go @@ -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, diff --git a/internal/services/agent_service.go b/internal/services/agent_service.go index ada9ca6..c657a14 100644 --- a/internal/services/agent_service.go +++ b/internal/services/agent_service.go @@ -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, } } diff --git a/internal/services/tasks/task_service.go b/internal/services/tasks/task_service.go index cb647f0..055e5b3 100644 --- a/internal/services/tasks/task_service.go +++ b/internal/services/tasks/task_service.go @@ -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 } diff --git a/web/src/api/index.ts b/web/src/api/index.ts index 126ff9d..04fb84e 100644 --- a/web/src/api/index.ts +++ b/web/src/api/index.ts @@ -286,6 +286,7 @@ export interface Task { envs: string retry_count: number retry_interval: number + random_range: number languages: { name: string; version: string }[] agent_id: number | null enabled: boolean diff --git a/web/src/views/tasks/RepoDialog.vue b/web/src/views/tasks/RepoDialog.vue index c86a464..c848727 100644 --- a/web/src/views/tasks/RepoDialog.vue +++ b/web/src/views/tasks/RepoDialog.vue @@ -327,10 +327,19 @@ async function save() {
- + +
+
+ + 秒 +
+
+
+
+
- + 分钟
@@ -353,12 +362,12 @@ async function save() {
- + 次
间隔 - + 秒
diff --git a/web/src/views/tasks/TaskDialog.vue b/web/src/views/tasks/TaskDialog.vue index 46f47be..497e7d0 100644 --- a/web/src/views/tasks/TaskDialog.vue +++ b/web/src/views/tasks/TaskDialog.vue @@ -572,11 +572,21 @@ async function save() {
+
+ +
+
+ + 秒 +
+

开启后，任务将在定时时间点后的 0 ~ {{ form.random_range || 0 }} 秒内随机执行。

+
+
- +
- + 分钟
@@ -609,12 +619,12 @@ async function save() {
- + 次
间隔 - + 秒
diff --git a/web/src/views/tasks/Tasks.vue b/web/src/views/tasks/Tasks.vue index 2ea9e6b..af21dda 100644 --- a/web/src/views/tasks/Tasks.vue +++ b/web/src/views/tasks/Tasks.vue @@ -111,13 +111,13 @@ function clearAgentFilter() { } function openCreate() { - editingTask.value = { name: '', command: '', type: TASK_TYPE.NORMAL, schedule: '0 * * * * *', timeout: 30, work_dir: '', enabled: true, clean_config: '', envs: '' } + editingTask.value = { name: '', command: '', type: TASK_TYPE.NORMAL, schedule: '0 * * * * *', timeout: 30, work_dir: '', enabled: true, clean_config: '', envs: '', random_range: 0 } isEdit.value = false showTaskDialog.value = true } function openCreateRepo() { - editingTask.value = { name: '', type: TASK_TYPE.REPO, schedule: '0 0 0 * * *', timeout: 30, enabled: true, clean_config: '', envs: '' } + editingTask.value = { name: '', type: TASK_TYPE.REPO, schedule: '0 0 0 * * *', timeout: 30, enabled: true, clean_config: '', envs: '', random_range: 0 } isEdit.value = false showRepoDialog.value = true }