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

feat: add pre and post command

Browse Source
This commit is contained in:
duorameng
2026-05-08 20:18:00 +08:00
parent 2f88fa0718
commit a3871e75b7
17 changed files with 255 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/services/agent_service.go
+21 -6
View File
@@ -314,13 +314,13 @@ func (s *AgentService) Heartbeat(token, ip, version, buildTime, hostname, osType
// GetTasks 获取 Agent 的任务列表
func (s *AgentService) GetTasks(agentID string) []models.AgentTask {
var tasks []models.Task
database.DB.Where("agent_id = ? AND enabled = ?", agentID, true).Find(&tasks)
var tasksList []models.Task
database.DB.Where("agent_id = ? AND enabled = ?", agentID, true).Find(&tasksList)
result := make([]models.AgentTask, len(tasks))
result := make([]models.AgentTask, len(tasksList))
envService := NewEnvService()
for i, task := range tasks {
for i, task := range tasksList {
// 加载环境配置
var envVars []string
@@ -343,14 +343,29 @@ func (s *AgentService) GetTasks(agentID string) []models.AgentTask {
}
envVarsStr := executor.FormatEnvVars(envVars)
command := string(task.Command)
preCommand := string(task.PreCommand)
postCommand := string(task.PostCommand)
workDir := task.WorkDir
// 仓库同步任务特殊处理：将配置转换为 reposync 命令行
if task.Type == constant.TaskTypeRepo {
command, workDir = tasks.BuildRepoCommand(&task)
// 仓库任务的前置/后置命令已作为参数传给 reposync 内部处理，此处清空防止重复执行
preCommand = ""
postCommand = ""
}
result[i] = models.AgentTask{
ID: task.ID,
Name: task.Name,
Command: string(task.Command),
Command: command,
PreCommand: preCommand,
PostCommand: postCommand,
Schedule: task.Schedule,
Timeout: task.Timeout,
WorkDir: task.WorkDir,
WorkDir: workDir,
Envs: envVarsStr,
Languages: []map[string]string(task.Languages),
RandomRange: task.RandomRange,
internal/services/tasks/executor_service.go
+29 -5
View File
@@ -404,7 +404,9 @@ func (es *ExecutorService) HandleTaskRetry(task *models.Task, req *executor.Exec
return &executor.ExecutionRequest{
TaskID: req.TaskID,
Name: latestTask.Name,
Command: string(latestTask.Command),
Command: string(latestTask.Command),
PreCommand: string(latestTask.PreCommand),
PostCommand: string(latestTask.PostCommand),
WorkDir: latestTask.WorkDir,
Envs: newEnvs,
Secrets: newSecrets,
@@ -454,6 +456,8 @@ func (es *ExecutorService) ExecuteDispatcher(ctx context.Context, req *executor.
// 解析路径变量 (如 $SCRIPTS_DIR$)
req.Command = es.ResolvePath(req.Command)
req.PreCommand = es.ResolvePath(req.PreCommand)
req.PostCommand = es.ResolvePath(req.PostCommand)
req.WorkDir = es.ResolvePath(req.WorkDir)
task := es.taskService.GetTaskByID(taskID)
@@ -478,10 +482,14 @@ func (es *ExecutorService) ExecuteDispatcher(ctx context.Context, req *executor.
req.Command = cmd
req.WorkDir = workDir
req.UseMise = false // 仓库同步任务不使用 mise，由系统原生执行
// 仓库任务的前置/后置命令已作为参数传给 reposync 内部处理，此处清空防止重复执行
req.PreCommand = ""
req.PostCommand = ""
// 强制脱敏并更新数据库日志
masks := append([]string{}, req.Secrets...)
masks = append(masks, utils.GetSystemSecrets()...)
// 补充仓库特有的 AuthToken
var repoCfg models.RepoConfig
if err := json.Unmarshal([]byte(task.Config), &repoCfg); err == nil && repoCfg.AuthToken != "" {
@@ -489,17 +497,20 @@ func (es *ExecutorService) ExecuteDispatcher(ctx context.Context, req *executor.
}
maskedCmd := utils.MaskSecrets(req.Command, masks)
// 更新数据库中的任务日志命令内容
if req.LogID != "" {
es.taskLogService.UpdateLogCommand(req.LogID, maskedCmd)
}
// 在控制台打印最终执行的脱敏命令
logger.Infof("[Executor] 仓库同步最终执行命令: %s", maskedCmd)
}
}
// 组合指令逻辑已移至 executor.ExecuteWithHooks 中，此处不再处理
// 以避免指令被重复组合。
// 远程任务
if task.AgentID != nil && *task.AgentID != "" {
// 将请求中已包含的环境变量（已合并）传递给 Agent
@@ -651,7 +662,9 @@ func (es *ExecutorService) ExecuteTask(taskID string, extraEnvs []string) *execu
req := &executor.ExecutionRequest{
TaskID: task.ID,
Name: task.Name,
Command: string(task.Command),
Command: string(task.Command),
PreCommand: string(task.PreCommand),
PostCommand: string(task.PostCommand),
WorkDir: task.WorkDir,
Envs: envs,
Secrets: secrets,
@@ -1039,6 +1052,11 @@ func (es *ExecutorService) HandleAgentResult(result *models.AgentTaskResult) err
// BuildRepoCommand 构建仓库同步任务的命令
func (es *ExecutorService) BuildRepoCommand(task *models.Task) (string, string) {
return BuildRepoCommand(task)
}
// BuildRepoCommand 构建仓库同步任务的命令（独立函数，方便 AgentService 调用）
func BuildRepoCommand(task *models.Task) (string, string) {
var config models.RepoConfig
if err := json.Unmarshal([]byte(task.Config), &config); err != nil {
return "", ""
@@ -1107,6 +1125,12 @@ func (es *ExecutorService) BuildRepoCommand(task *models.Task) (string, string)
if config.Extensions != "" {
args = append(args, "--extensions", config.Extensions)
}
if string(task.PreCommand) != "" {
args = append(args, "--pre-command", string(task.PreCommand))
}
if string(task.PostCommand) != "" {
args = append(args, "--post-command", string(task.PostCommand))
}
// 传递任务 ID，以便 reposync 内部直接处理脚本注册并输出日志
args = append(args, "--task-id", task.ID)
internal/services/tasks/task_service.go
+7 -2
View File
@@ -24,7 +24,7 @@ func (ts *TaskService) GetTaskBySourceID(sourceID string) *models.Task {
return &task
}
func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, workDir, cleanConfig, envs, taskType, config string, agentID *string, languages models.TaskLanguages, triggerType string, tags string, retryCount int, retryInterval int, randomRange int, sourceID string, pinType string) *models.Task {
func (ts *TaskService) CreateTask(name, command, preCommand, postCommand, schedule string, timeout int, workDir, cleanConfig, envs, taskType, config string, agentID *string, languages models.TaskLanguages, triggerType string, tags string, retryCount int, retryInterval int, randomRange int, sourceID string, pinType string) *models.Task {
if taskType == "" {
taskType = "task"
}
@@ -38,6 +38,8 @@ func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, w
ID: utils.GenerateID(),
Name: name,
Command: models.BigText(command),
PreCommand: models.BigText(preCommand),
PostCommand: models.BigText(postCommand),
PinType: pinType,
Tags: tags,
Type: taskType,
@@ -122,7 +124,7 @@ func (ts *TaskService) GetTaskByID(id string) *models.Task {
return &task
}
func (ts *TaskService) UpdateTask(id string, name, command, schedule string, timeout int, workDir, cleanConfig, envs string, enabled bool, taskType, config string, agentID *string, languages models.TaskLanguages, triggerType string, tags string, retryCount int, retryInterval int, randomRange int, sourceID string, pinType string) *models.Task {
func (ts *TaskService) UpdateTask(id string, name, command, preCommand, postCommand, schedule string, timeout int, workDir, cleanConfig, envs string, enabled bool, taskType, config string, agentID *string, languages models.TaskLanguages, triggerType string, tags string, retryCount int, retryInterval int, randomRange int, sourceID string, pinType string) *models.Task {
var task models.Task
res := database.DB.Where("id = ?", id).Limit(1).Find(&task)
if res.Error != nil || res.RowsAffected == 0 {
@@ -130,6 +132,8 @@ func (ts *TaskService) UpdateTask(id string, name, command, schedule string, tim
}
task.Name = name
task.Command = models.BigText(command)
task.PreCommand = models.BigText(preCommand)
task.PostCommand = models.BigText(postCommand)
task.PinType = pinType
task.Tags = tags
task.Schedule = schedule
@@ -159,6 +163,7 @@ func (ts *TaskService) UpdateTask(id string, name, command, schedule string, tim
"CleanConfig", "Envs", "Enabled", "AgentID", "Languages",
"RetryCount", "RetryInterval", "RandomRange", "Type",
"TriggerType", "Config", "SourceID", "PinType",
"PreCommand", "PostCommand",
).Updates(&task)
return &task