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

feat: add multi env setting support

Browse Source
This commit is contained in:
engigu
2026-02-13 19:26:20 +08:00
parent 8a52be9241
commit 3ab1ddc03a
13 changed files with 322 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/controllers/task_controller.go
+25 -27
View File
@@ -52,18 +52,17 @@ func resolveWorkDir(workDir string) string {
func (tc *TaskController) CreateTask(c *gin.Context) {
var req struct {
Name string `json:"name" binding:"required"`
Command string `json:"command"`
Type string `json:"type"`
Config string `json:"config"`
Schedule string `json:"schedule" binding:"required"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
CleanConfig string `json:"clean_config"`
Envs string `json:"envs"`
Language string `json:"language"`
LangVersion string `json:"lang_version"`
AgentID *uint `json:"agent_id"`
Name string `json:"name" binding:"required"`
Command string `json:"command"`
Type string `json:"type"`
Config string `json:"config"`
Schedule string `json:"schedule" binding:"required"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
CleanConfig string `json:"clean_config"`
Envs string `json:"envs"`
Languages []map[string]string `json:"languages"`
AgentID *uint `json:"agent_id"`
}
if err := c.ShouldBindJSON(&req); err != nil {
@@ -88,7 +87,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.Language, req.LangVersion)
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)
// 如果是 Agent 任务，通知 Agent；否则添加到本地 cron
if task.AgentID != nil && *task.AgentID > 0 {
@@ -148,19 +147,18 @@ func (tc *TaskController) UpdateTask(c *gin.Context) {
}
var req struct {
Name string `json:"name"`
Command string `json:"command"`
Type string `json:"type"`
Config string `json:"config"`
Schedule string `json:"schedule"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
CleanConfig string `json:"clean_config"`
Envs string `json:"envs"`
Enabled bool `json:"enabled"`
Language string `json:"language"`
LangVersion string `json:"lang_version"`
AgentID *uint `json:"agent_id"`
Name string `json:"name"`
Command string `json:"command"`
Type string `json:"type"`
Config string `json:"config"`
Schedule string `json:"schedule"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
CleanConfig string `json:"clean_config"`
Envs string `json:"envs"`
Enabled bool `json:"enabled"`
Languages []map[string]string `json:"languages"`
AgentID *uint `json:"agent_id"`
}
if err := c.ShouldBindJSON(&req); err != nil {
@@ -181,7 +179,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.Language, req.LangVersion)
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)
if task == nil {
utils.NotFound(c, "任务不存在")
return
internal/executor/cron.go
+10 -12
View File
@@ -78,8 +78,7 @@ func (m *CronManager) AddTask(task CronTask) error {
timeout := task.GetTimeout()
workDir := task.GetWorkDir()
envs := task.GetEnvs()
language := task.GetLanguage()
langVersion := task.GetLangVersion()
languages := task.GetLanguages()
useMise := task.UseMise()
entryID, err := m.cron.AddFunc(task.GetSchedule(), func() {
@@ -91,16 +90,15 @@ func (m *CronManager) AddTask(task CronTask) error {
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),
Language: language,
LangVersion: langVersion,
UseMise: useMise,
TaskID: taskID,
Name: name,
Command: cmd,
Type: TaskTypeCron,
Timeout: timeout,
WorkDir: workDir,
Envs: ParseEnvVars(envs),
Languages: languages,
UseMise: useMise,
}
// 如果有关联的 Scheduler，加入队列执行
internal/executor/executor.go
+27 -16
View File
@@ -23,8 +23,7 @@ type Task interface {
GetTimeout() int
GetWorkDir() string
GetEnvs() string
GetLanguage() string
GetLangVersion() string
GetLanguages() []map[string]string
GetUseMise() bool
}
@@ -37,13 +36,12 @@ type CronTask interface {
// Request 任务执行请求
type Request struct {
Command string
WorkDir string
Envs []string
Timeout int // 任务超时时间（分钟）
Language string
LangVersion string
UseMise bool
Command string
WorkDir string
Envs []string
Timeout int // 任务超时时间（分钟）
Languages []map[string]string
UseMise bool
}
// Result 任务执行结果
@@ -104,7 +102,7 @@ func ExecuteWithHooks(ctx context.Context, req Request, stdout, stderr io.Writer
finalCommand := req.Command
if req.UseMise {
finalCommand = BuildLanguageCommand(req.Command, req.Language, req.LangVersion)
finalCommand = BuildLanguageCommand(req.Command, req.Languages)
}
shell, args := utils.GetShellCommand(finalCommand)
cmd := exec.CommandContext(execCtx, shell, args...)
@@ -283,15 +281,28 @@ func ExecuteWithHooks(ctx context.Context, req Request, stdout, stderr io.Writer
}
// BuildLanguageCommand 构建语言环境执行命令 (使用 mise)
func BuildLanguageCommand(command, language, version string) string {
if language == "" {
func BuildLanguageCommand(command string, languages []map[string]string) string {
if len(languages) == 0 {
return command
}
v := version
if v == "" {
v = "latest"
var builder strings.Builder
builder.WriteString("mise exec")
for _, lang := range languages {
name := lang["name"]
version := lang["version"]
if name == "" {
continue
}
if version == "" {
version = "latest"
}
builder.WriteString(" " + name + "@" + version)
}
return "mise exec " + language + "@" + v + " -- " + command
builder.WriteString(" -- " + command)
return builder.String()
}
// ParseEnvVars 解析环境变量字符串 "KEY1=VALUE1,KEY2=VALUE2"
internal/executor/scheduler.go
+18 -20
View File
@@ -61,18 +61,17 @@ const (
// ExecutionRequest 执行请求（标准接口）
type ExecutionRequest struct {
TaskID string // 任务 ID
LogID uint // 日志 ID
Name string // 任务名称
Type TaskType // 任务类型
Command string // 命令
WorkDir string // 工作目录
Envs []string // 环境变量
Timeout int // 超时时间（分钟）
Language string // 语言环境
LangVersion string // 语言版本
UseMise bool // 是否使用 mise
Metadata map[string]interface{} // 额外元数据
TaskID string // 任务 ID
LogID uint // 日志 ID
Name string // 任务名称
Type TaskType // 任务类型
Command string // 命令
WorkDir string // 工作目录
Envs []string // 环境变量
Timeout int // 超时时间（分钟）
Languages []map[string]string // 语言环境配置
UseMise bool // 是否使用 mise
Metadata map[string]interface{} // 额外元数据
}
// ExecutionResult 执行结果（标准接口）
@@ -197,13 +196,12 @@ func NewScheduler(config SchedulerConfig, handler SchedulerEventHandler) *Schedu
executor: func(ctx context.Context, req *ExecutionRequest, stdout, stderr io.Writer) (*Result, error) {
hooks := &schedulerHooksAdapter{handler: handler, req: req}
return ExecuteWithHooks(ctx, Request{
Command: req.Command,
WorkDir: req.WorkDir,
Envs: req.Envs,
Timeout: req.Timeout,
Language: req.Language,
LangVersion: req.LangVersion,
UseMise: req.UseMise,
Command: req.Command,
WorkDir: req.WorkDir,
Envs: req.Envs,
Timeout: req.Timeout,
Languages: req.Languages,
UseMise: req.UseMise,
}, stdout, stderr, hooks)
},
taskQueue: make(chan *ExecutionRequest, config.QueueSize),
@@ -317,7 +315,7 @@ func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error)
finalCommand := req.Command
if req.UseMise {
finalCommand = BuildLanguageCommand(req.Command, req.Language, req.LangVersion)
finalCommand = BuildLanguageCommand(req.Command, req.Languages)
}
s.logger.Infof("[Scheduler] 实际执行命令: %s", finalCommand)
internal/models/agent.go
+9 -10
View File
@@ -52,16 +52,15 @@ func (AgentToken) TableName() string {
// AgentTask Agent 任务配置（用于下发给 Agent）
type AgentTask struct {
ID uint `json:"id"`
Name string `json:"name"`
Command string `json:"command"`
Schedule string `json:"schedule"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
Envs string `json:"envs"`
Language string `json:"language"`
LangVersion string `json:"lang_version"`
Enabled bool `json:"enabled"`
ID uint `json:"id"`
Name string `json:"name"`
Command string `json:"command"`
Schedule string `json:"schedule"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
Envs string `json:"envs"`
Languages []map[string]string `json:"languages"`
Enabled bool `json:"enabled"`
}
// AgentTaskResult Agent 上报的任务执行结果
internal/models/task.go
+21 -26
View File
@@ -34,26 +34,25 @@ type TaskConfig struct {
// Task 代表一个计划任务
type Task struct {
ID uint `json:"id" gorm:"primaryKey"`
Name string `json:"name" gorm:"size:255;not null"`
Command string `json:"command" gorm:"type:text"` // 普通任务的命令
Type string `json:"type" gorm:"size:20;default:'task'"` // 任务类型: constant.TaskTypeNormal, constant.TaskTypeRepo
Config string `json:"config" gorm:"type:text"` // 配置 JSON（仓库同步配置等）
Schedule string `json:"schedule" gorm:"size:100"` // cron 表达式
Timeout int `json:"timeout" gorm:"default:30"` // 超时时间（分钟），默认30分钟
WorkDir string `json:"work_dir" gorm:"size:255;default:''"` // 工作目录，为空则使用 scripts 目录
CleanConfig string `json:"clean_config" gorm:"size:255;default:''"` // 清理配置 JSON
Envs string `json:"envs" gorm:"size:255;default:''"` // 环境变量ID列表，逗号分隔
Language string `json:"language" gorm:"size:50;default:''"` // 针对本地任务的语言名称 (mise)
LangVersion string `json:"lang_version" gorm:"size:50;default:''"` // 针对本地任务的语言版本 (mise)
AgentID *uint `json:"agent_id" gorm:"index"` // Agent ID，为空表示本地执行
Enabled bool `json:"enabled" gorm:"default:true"`
RunningGo string `json:"running_go" gorm:"type:text"` // 正在运行的 go routine id 数组 (JSON)
LastRun *LocalTime `json:"last_run"`
NextRun *LocalTime `json:"next_run"`
CreatedAt LocalTime `json:"created_at"`
UpdatedAt LocalTime `json:"updated_at"`
DeletedAt gorm.DeletedAt `json:"-" gorm:"index"`
ID uint `json:"id" gorm:"primaryKey"`
Name string `json:"name" gorm:"size:255;not null"`
Command string `json:"command" gorm:"type:text"` // 普通任务的命令
Type string `json:"type" gorm:"size:20;default:'task'"` // 任务类型: constant.TaskTypeNormal, constant.TaskTypeRepo
Config string `json:"config" gorm:"type:text"` // 配置 JSON（仓库同步配置等）
Schedule string `json:"schedule" gorm:"size:100"` // cron 表达式
Timeout int `json:"timeout" gorm:"default:30"` // 超时时间（分钟），默认30分钟
WorkDir string `json:"work_dir" gorm:"size:255;default:''"` // 工作目录，为空则使用 scripts 目录
CleanConfig string `json:"clean_config" gorm:"size:255;default:''"` // 清理配置 JSON
Envs string `json:"envs" gorm:"size:255;default:''"` // 环境变量ID列表，逗号分隔
Languages []map[string]string `json:"languages" gorm:"serializer:json;type:text"` // 针对本地任务的语言配置列表
AgentID *uint `json:"agent_id" gorm:"index"` // Agent ID，为空表示本地执行
Enabled bool `json:"enabled" gorm:"default:true"`
RunningGo string `json:"running_go" gorm:"type:text"` // 正在运行的 go routine id 数组 (JSON)
LastRun *LocalTime `json:"last_run"`
NextRun *LocalTime `json:"next_run"`
CreatedAt LocalTime `json:"created_at"`
UpdatedAt LocalTime `json:"updated_at"`
DeletedAt gorm.DeletedAt `json:"-" gorm:"index"`
}
func (Task) TableName() string {
@@ -84,12 +83,8 @@ func (t *Task) GetEnvs() string {
return t.Envs
}
func (t *Task) GetLanguage() string {
return t.Language
}
func (t *Task) GetLangVersion() string {
return t.LangVersion
func (t *Task) GetLanguages() []map[string]string {
return t.Languages
}
func (t *Task) GetUseMise() bool {
internal/models/vo/task_vo.go
+18 -20
View File
@@ -7,24 +7,23 @@ import (
// TaskVO 任务视图对象
type TaskVO struct {
ID uint `json:"id"`
Name string `json:"name"`
Command string `json:"command"`
Type string `json:"type"`
Config string `json:"config"`
Schedule string `json:"schedule"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
CleanConfig string `json:"clean_config"`
Envs string `json:"envs"`
Language string `json:"language"`
LangVersion string `json:"lang_version"`
AgentID *uint `json:"agent_id"`
Enabled bool `json:"enabled"`
LastRun *models.LocalTime `json:"last_run"`
NextRun *models.LocalTime `json:"next_run"`
CreatedAt models.LocalTime `json:"created_at"`
UpdatedAt models.LocalTime `json:"updated_at"`
ID uint `json:"id"`
Name string `json:"name"`
Command string `json:"command"`
Type string `json:"type"`
Config string `json:"config"`
Schedule string `json:"schedule"`
Timeout int `json:"timeout"`
WorkDir string `json:"work_dir"`
CleanConfig string `json:"clean_config"`
Envs string `json:"envs"`
Languages []map[string]string `json:"languages"`
AgentID *uint `json:"agent_id"`
Enabled bool `json:"enabled"`
LastRun *models.LocalTime `json:"last_run"`
NextRun *models.LocalTime `json:"next_run"`
CreatedAt models.LocalTime `json:"created_at"`
UpdatedAt models.LocalTime `json:"updated_at"`
}
// ToTaskVO 将 Task 模型转换为 TaskVO
@@ -43,8 +42,7 @@ func ToTaskVO(task *models.Task) *TaskVO {
WorkDir: task.WorkDir,
CleanConfig: task.CleanConfig,
Envs: task.Envs,
Language: task.Language,
LangVersion: task.LangVersion,
Languages: task.Languages,
AgentID: task.AgentID,
Enabled: task.Enabled,
LastRun: task.LastRun,
internal/services/agent_service.go
+9 -10
View File
@@ -311,16 +311,15 @@ func (s *AgentService) GetTasks(agentID uint) []models.AgentTask {
envVarsStr := s.buildEnvVarsString(task.Envs)
result[i] = models.AgentTask{
ID: task.ID,
Name: task.Name,
Command: task.Command,
Schedule: task.Schedule,
Timeout: task.Timeout,
WorkDir: task.WorkDir,
Envs: envVarsStr, // 传递 "KEY1=VALUE1,KEY2=VALUE2" 格式
Language: task.Language,
LangVersion: task.LangVersion,
Enabled: task.Enabled,
ID: task.ID,
Name: task.Name,
Command: task.Command,
Schedule: task.Schedule,
Timeout: task.Timeout,
WorkDir: task.WorkDir,
Envs: envVarsStr, // 传递 "KEY1=VALUE1,KEY2=VALUE2" 格式
Languages: task.Languages,
Enabled: task.Enabled,
}
}
internal/services/tasks/executor_service.go
+15 -14
View File
@@ -352,13 +352,12 @@ func (es *ExecutorService) ExecuteDispatcher(ctx context.Context, req *executor.
// 本地任务
hooks := &LocalTaskHooks{es: es, logID: req.LogID}
return executor.ExecuteWithHooks(ctx, executor.Request{
Command: req.Command,
WorkDir: req.WorkDir,
Envs: req.Envs,
Timeout: req.Timeout,
Language: task.Language,
LangVersion: task.LangVersion,
UseMise: true, // 本地业务任务启用 mise
Command: req.Command,
WorkDir: req.WorkDir,
Envs: req.Envs,
Timeout: req.Timeout,
Languages: task.Languages,
UseMise: true, // 本地业务任务启用 mise
}, stdout, stderr, hooks)
}
@@ -470,13 +469,15 @@ func (es *ExecutorService) ExecuteTask(taskID int, extraEnvs []string) *executor
}
req := &executor.ExecutionRequest{
TaskID: fmt.Sprintf("%d", task.ID),
Name: task.Name,
Command: task.Command,
WorkDir: task.WorkDir,
Envs: envs,
Timeout: task.Timeout,
Type: executor.TaskTypeManual,
TaskID: fmt.Sprintf("%d", task.ID),
Name: task.Name,
Command: task.Command,
WorkDir: task.WorkDir,
Envs: envs,
Timeout: task.Timeout,
Languages: task.Languages,
UseMise: task.UseMise(),
Type: executor.TaskTypeManual,
}
es.scheduler.EnqueueOrExecute(req)
internal/services/tasks/task_service.go
+4 -6
View File
@@ -11,7 +11,7 @@ func NewTaskService() *TaskService {
return &TaskService{}
}
func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, workDir, cleanConfig, envs, taskType, config string, agentID *uint, language, langVersion string) *models.Task {
func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, workDir, cleanConfig, envs, taskType, config string, agentID *uint, languages []map[string]string) *models.Task {
if taskType == "" {
taskType = "task"
}
@@ -25,8 +25,7 @@ func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, w
WorkDir: workDir,
CleanConfig: cleanConfig,
Envs: envs,
Language: language,
LangVersion: langVersion,
Languages: languages,
AgentID: agentID,
Enabled: true,
}
@@ -67,7 +66,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, language, langVersion string) *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) *models.Task {
var task models.Task
if err := database.DB.First(&task, id).Error; err != nil {
return nil
@@ -81,8 +80,7 @@ func (ts *TaskService) UpdateTask(id int, name, command, schedule string, timeou
task.Envs = envs
task.Enabled = enabled
task.AgentID = agentID
task.Language = language
task.LangVersion = langVersion
task.Languages = languages
if taskType != "" {
task.Type = taskType
}