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

feat: add env task global setting

Browse Source
This commit is contained in:
engigu
2026-03-06 14:13:37 +08:00
parent beae79f448
commit 3effa5b326
14 changed files with 918 additions and 588 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/executor/cron.go
+6 -1
View File
@@ -99,7 +99,12 @@ func (m *CronManager) AddTask(task CronTask) error {
Type: TaskTypeCron,
Timeout: timeout,
WorkDir: workDir,
Envs: ParseEnvVars(envs),
Envs: func() []string {
if vars := task.GetEnvVars(); len(vars) > 0 {
return vars
}
return ParseEnvVars(envs)
}(),
Languages: languages,
UseMise: useMise,
}
internal/executor/executor.go
+28
View File
@@ -2,6 +2,7 @@ package executor
import (
"context"
"fmt"
"io"
"os"
"os/exec"
@@ -23,6 +24,7 @@ type Task interface {
GetTimeout() int
GetWorkDir() string
GetEnvs() string
GetEnvVars() []string
GetLanguages() []map[string]string
GetUseMise() bool
}
@@ -330,3 +332,29 @@ func ParseEnvVars(envStr string) []string {
return result
}
// FormatEnvVars 将环境变量列表格式化为逗号分隔的字符串 "KEY1=VALUE1,KEY2=VALUE2"
// 会对 , 和 = 进行转义
func FormatEnvVars(envs []string) string {
if len(envs) == 0 {
return ""
}
pairs := make([]string, 0, len(envs))
for _, pair := range envs {
// 寻找第一个等号
idx := strings.Index(pair, "=")
if idx == -1 {
continue
}
name := pair[:idx]
value := pair[idx+1:]
// 转义特殊字符
encodedValue := strings.ReplaceAll(value, ",", "{{COMMA}}")
encodedValue = strings.ReplaceAll(encodedValue, "=", "{{EQUAL}}")
pairs = append(pairs, fmt.Sprintf("%s=%s", name, encodedValue))
}
return strings.Join(pairs, ",")
}
internal/models/task.go
+7 -1
View File
@@ -27,7 +27,8 @@ type RepoConfig struct {
// TaskConfig 任务配置 RepoConfig+TaskConfig=task.config
type TaskConfig struct {
Concurrency int `json:"$task_concurrency"` // 0: disable concurrency, 1: enable concurrency
Concurrency int `json:"$task_concurrency"` // 0: disable concurrency, 1: enable concurrency
AllEnvs bool `json:"$task_all_envs"` // 开启则注入全部环境变量
}
// Task 代表一个计划任务
@@ -51,6 +52,7 @@ type Task struct {
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)
RuntimeEnvs []string `json:"-" gorm:"-"` // 运行时环境变量（非持久化）
LastRun *LocalTime `json:"last_run"`
NextRun *LocalTime `json:"next_run"`
CreatedAt LocalTime `json:"created_at"`
@@ -90,6 +92,10 @@ func (t *Task) GetLanguages() []map[string]string {
return t.Languages
}
func (t *Task) GetEnvVars() []string {
return t.RuntimeEnvs
}
func (t *Task) GetUseMise() bool {
return t.AgentID == nil || *t.AgentID == ""
}
internal/services/agent_service.go
+32 -33
View File
@@ -3,6 +3,7 @@ package services
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"os"
"path/filepath"
@@ -12,6 +13,7 @@ import (
"github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/database"
"github.com/engigu/baihu-panel/internal/logger"
"github.com/engigu/baihu-panel/internal/executor"
"github.com/engigu/baihu-panel/internal/models"
"github.com/engigu/baihu-panel/internal/services/tasks"
"github.com/engigu/baihu-panel/internal/utils"
@@ -310,18 +312,39 @@ func (s *AgentService) GetTasks(agentID string) []models.AgentTask {
database.DB.Where("agent_id = ? AND enabled = ?", agentID, true).Find(&tasks)
result := make([]models.AgentTask, len(tasks))
envService := NewEnvService()
for i, task := range tasks {
// 将环境变量 ID 转换为实际的环境变量键值对
envVarsStr := s.buildEnvVarsString(task.Envs)
// 加载环境配置
var envVars []string
// 检查全量注入模式
allEnvs := false
if task.Config != "" {
var config models.TaskConfig
if err := json.Unmarshal([]byte(task.Config), &config); err == nil {
if config.AllEnvs {
allEnvs = true
}
}
}
if allEnvs {
envVars = envService.GetAllEnvVars()
} else if task.Envs != "" {
envVars = envService.GetEnvVarsByIDs(task.Envs)
}
envVarsStr := executor.FormatEnvVars(envVars)
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" 格式
ID: task.ID,
Name: task.Name,
Command: task.Command,
Schedule: task.Schedule,
Timeout: task.Timeout,
WorkDir: task.WorkDir,
Envs: envVarsStr,
Languages: task.Languages,
RandomRange: task.RandomRange,
Enabled: task.Enabled,
@@ -331,30 +354,6 @@ func (s *AgentService) GetTasks(agentID string) []models.AgentTask {
return result
}
// buildEnvVarsString 将环境变量 ID 列表转换为键值对字符串
func (s *AgentService) buildEnvVarsString(envIDs string) string {
if envIDs == "" {
return ""
}
var envVars []models.EnvironmentVariable
ids := strings.Split(envIDs, ",")
database.DB.Where("id IN ?", ids).Find(&envVars)
if len(envVars) == 0 {
return ""
}
// 构建 "KEY1=VALUE1,KEY2=VALUE2" 格式
pairs := make([]string, 0, len(envVars))
for _, env := range envVars {
// 对值进行转义，避免特殊字符问题
encodedValue := strings.ReplaceAll(env.Value, ",", "{{COMMA}}")
encodedValue = strings.ReplaceAll(encodedValue, "=", "{{EQUAL}}")
pairs = append(pairs, fmt.Sprintf("%s=%s", env.Name, encodedValue))
}
return strings.Join(pairs, ",")
}
// ReportResult Agent 上报执行结果
func (s *AgentService) ReportResult(result *models.AgentTaskResult) error {
internal/services/env_service.go
+27 -6
View File
@@ -125,6 +125,32 @@ func (es *EnvService) GetEnvVarsByIDs(envIDs string) []string {
}
ids := splitEnvIDs(envIDs)
var envs []models.EnvironmentVariable
for _, id := range ids {
env := es.GetEnvVarByID(id)
if env != nil {
envs = append(envs, *env)
}
}
return es.formatEnvVars(envs)
}
// GetAllEnvVars 获取系统中所有的环境变量，并按 NAME=VALUE 格式返回（支持重名合并）
func (es *EnvService) GetAllEnvVars() []string {
var envs []models.EnvironmentVariable
if err := database.DB.Find(&envs).Error; err != nil {
return nil
}
return es.formatEnvVars(envs)
}
// formatEnvVars 将环境变量列表格式化为 NAME=VALUE 数组，并处理重名合并
func (es *EnvService) formatEnvVars(envs []models.EnvironmentVariable) []string {
if len(envs) == 0 {
return nil
}
type mergedEnv struct {
name string
values []string
@@ -132,12 +158,7 @@ func (es *EnvService) GetEnvVarsByIDs(envIDs string) []string {
var mergedList []mergedEnv
nameToIndex := make(map[string]int)
for _, id := range ids {
env := es.GetEnvVarByID(id)
if env == nil {
continue
}
for _, env := range envs {
if idx, ok := nameToIndex[env.Name]; ok {
mergedList[idx].values = append(mergedList[idx].values, env.Value)
} else {
internal/services/tasks/executor_service.go
+30 -13
View File
@@ -37,6 +37,7 @@ type SettingsService interface {
// EnvService 接口定义（避免循环依赖）
type EnvService interface {
GetEnvVarsByIDs(ids string) []string
GetAllEnvVars() []string
}
// Notifier 通知服务接口定义（避免循环依赖）
@@ -372,7 +373,7 @@ func (es *ExecutorService) HandleTaskRetry(task *models.Task, req *executor.Exec
return nil
}
newEnvs := es.loadEnvVars(latestTask.Envs)
newEnvs := es.loadEnvVars(latestTask.ID, latestTask.Envs)
return &executor.ExecutionRequest{
TaskID: req.TaskID,
Name: latestTask.Name,
@@ -444,14 +445,10 @@ func (es *ExecutorService) ExecuteDispatcher(ctx context.Context, req *executor.
}
}
// 加载环境变量
if task.Envs != "" {
req.Envs = append(req.Envs, es.loadEnvVars(task.Envs)...)
}
// 远程任务
if task.AgentID != nil && *task.AgentID != "" {
return es.ExecuteRemoteForScheduler(task, req.LogID)
// 将请求中已包含的环境变量（已合并）传递给 Agent
return es.ExecuteRemoteForScheduler(task, req.LogID, executor.FormatEnvVars(req.Envs))
}
// 本地任务
@@ -504,6 +501,9 @@ func (es *ExecutorService) AddCronTask(task *models.Task) error {
es.RemoveCronTask(task.ID) // 如果不是cron类型，确保从调度器移除
return nil
}
// 在加入调度器前，预先加载好环境信息
task.RuntimeEnvs = es.loadEnvVars(task.ID, task.Envs)
return es.cronManager.AddTask(task)
}
@@ -540,7 +540,7 @@ func (es *ExecutorService) loadCronTasks() {
}(task)
} else if task.TriggerType == constant.TriggerTypeCron && task.Schedule != "" && (task.AgentID == nil || *task.AgentID == "") {
// 只调度本地任务（agent_id 为空或 0）的定时任务
err := es.cronManager.AddTask(&task)
err := es.AddCronTask(&task)
if err != nil {
continue
}
@@ -583,7 +583,7 @@ func (es *ExecutorService) ExecuteTask(taskID string, extraEnvs []string) *execu
}
}
envs := es.loadEnvVars(task.Envs)
envs := es.loadEnvVars(task.ID, task.Envs)
if len(extraEnvs) > 0 {
envs = append(envs, extraEnvs...)
}
@@ -613,7 +613,7 @@ func (es *ExecutorService) ExecuteTask(taskID string, extraEnvs []string) *execu
// StopTaskExecution stops a running task execution by LogID
func (es *ExecutorService) StopTaskExecution(logID string) error {
var taskLog models.TaskLog
if err := database.DB.First(&taskLog, logID).Error; err != nil {
if err := database.DB.Where("id = ?", logID).First(&taskLog).Error; err != nil {
return fmt.Errorf("日志不存在")
}
@@ -819,7 +819,7 @@ func (es *ExecutorService) RemoveRunningGo(taskID string, goid int64) {
}
// ExecuteRemoteForScheduler 供 Scheduler 调用，执行远程任务并等待结果
func (es *ExecutorService) ExecuteRemoteForScheduler(task *models.Task, logID string) (*executor.Result, error) {
func (es *ExecutorService) ExecuteRemoteForScheduler(task *models.Task, logID string, envs string) (*executor.Result, error) {
agentID := *task.AgentID
logger.Infof("[Executor] 远程执行任务 #%s: %s (Agent #%s, LogID: %s)", task.ID, task.Name, agentID, logID)
@@ -843,6 +843,7 @@ func (es *ExecutorService) ExecuteRemoteForScheduler(task *models.Task, logID st
err := es.agentWSManager.SendToAgent(agentID, constant.WSTypeExecute, map[string]interface{}{
"task_id": task.ID,
"log_id": logID,
"envs": envs,
})
if err != nil {
return nil, fmt.Errorf("发送执行命令失败: %v", err)
@@ -936,8 +937,24 @@ func (es *ExecutorService) BuildRepoCommand(task *models.Task) (string, string)
return exePath + " " + strings.Join(args, " "), filepath.Dir(exePath)
}
// loadEnvVars 加载环境变量（支持重名合并）
func (es *ExecutorService) loadEnvVars(envIDs string) []string {
// loadEnvVars 加载环境变量，支持全局注入及重名合并
func (es *ExecutorService) loadEnvVars(taskID string, envIDs string) []string {
// 1. 检查是否开启了注入全部环境变量
if taskID != "" && es.taskService != nil {
task := es.taskService.GetTaskByID(taskID)
if task != nil && task.Config != "" {
var config models.TaskConfig
if err := json.Unmarshal([]byte(task.Config), &config); err == nil {
if config.AllEnvs {
if es.envService != nil {
return es.envService.GetAllEnvVars()
}
}
}
}
}
// 2. 否则按 ID 列表进行加载（支持合并逻辑在 envService 中处理）
if envIDs == "" {
return nil
}
internal/services/tasks/task_service.go
+1
View File
@@ -104,6 +104,7 @@ func (ts *TaskService) UpdateTask(id string, name, command, schedule string, tim
task.Enabled = enabled
task.AgentID = agentID
task.Languages = languages
task.Config = config
task.RetryCount = retryCount
task.RetryInterval = retryInterval
task.RandomRange = randomRange