diff --git a/internal/controllers/dashboard_controller.go b/internal/controllers/dashboard_controller.go index e5b2069..bb6f5aa 100644 --- a/internal/controllers/dashboard_controller.go +++ b/internal/controllers/dashboard_controller.go @@ -7,18 +7,18 @@ import ( "baihu/internal/constant" "baihu/internal/database" "baihu/internal/models" - "baihu/internal/services" + "baihu/internal/services/tasks" "baihu/internal/utils" "github.com/gin-gonic/gin" ) type DashboardController struct { - cronService *services.CronService - executorService *services.ExecutorService + cronService *tasks.CronService + executorService *tasks.ExecutorService } -func NewDashboardController(cronService *services.CronService, executorService *services.ExecutorService) *DashboardController { +func NewDashboardController(cronService *tasks.CronService, executorService *tasks.ExecutorService) *DashboardController { return &DashboardController{ cronService: cronService, executorService: executorService, @@ -45,13 +45,29 @@ func (dc *DashboardController) GetStats(c *gin.Context) { today := time.Now().Format("2006-01-02") database.DB.Model(&models.SendStats{}).Where("day = ?", today).Select("COALESCE(SUM(num), 0)").Scan(&todayExecs) + // 调度统计:本地调度 + Agent 调度 + // 本地调度:agent_id 为 NULL 且 enabled = true 的任务 + localScheduled := dc.cronService.GetScheduledCount() + + // Agent 调度:agent_id 不为 NULL 且 enabled = true 的任务 + var agentScheduled int64 + database.DB.Model(&models.Task{}). + Where("agent_id IS NOT NULL AND enabled = ?", true). + Count(&agentScheduled) + + totalScheduled := localScheduled + int(agentScheduled) + + // 正在运行:目前只能统计本地运行的任务 + // Agent 端的运行状态需要通过心跳上报(未来优化) + running := dc.executorService.GetRunningCount() + stats := StatsResponse{ Tasks: taskCount, TodayExecs: todayExecs, Envs: envCount, Logs: logCount, - Scheduled: dc.cronService.GetScheduledCount(), - Running: dc.executorService.GetRunningCount(), + Scheduled: totalScheduled, + Running: running, } utils.Success(c, stats) diff --git a/internal/controllers/executor_controller.go b/internal/controllers/executor_controller.go index 3b8b7f7..5d7a608 100644 --- a/internal/controllers/executor_controller.go +++ b/internal/controllers/executor_controller.go @@ -3,17 +3,17 @@ package controllers import ( "strconv" - "baihu/internal/services" + "baihu/internal/services/tasks" "baihu/internal/utils" "github.com/gin-gonic/gin" ) type ExecutorController struct { - executorService *services.ExecutorService + executorService *tasks.ExecutorService } -func NewExecutorController(executorService *services.ExecutorService) *ExecutorController { +func NewExecutorController(executorService *tasks.ExecutorService) *ExecutorController { return &ExecutorController{executorService: executorService} } diff --git a/internal/controllers/settings_controller.go b/internal/controllers/settings_controller.go index 6ed7ab4..b47f7bb 100644 --- a/internal/controllers/settings_controller.go +++ b/internal/controllers/settings_controller.go @@ -9,6 +9,7 @@ import ( "baihu/internal/database" "baihu/internal/models" "baihu/internal/services" + "baihu/internal/services/tasks" "baihu/internal/utils" "fmt" "os" @@ -23,10 +24,10 @@ type SettingsController struct { settingsService *services.SettingsService loginLogService *services.LoginLogService backupService *services.BackupService - executorService *services.ExecutorService + executorService *tasks.ExecutorService } -func NewSettingsController(userService *services.UserService, loginLogService *services.LoginLogService, executorService *services.ExecutorService) *SettingsController { +func NewSettingsController(userService *services.UserService, loginLogService *services.LoginLogService, executorService *tasks.ExecutorService) *SettingsController { return &SettingsController{ userService: userService, settingsService: services.NewSettingsService(), diff --git a/internal/controllers/task_controller.go b/internal/controllers/task_controller.go index e4e5764..3e50893 100644 --- a/internal/controllers/task_controller.go +++ b/internal/controllers/task_controller.go @@ -6,18 +6,19 @@ import ( "baihu/internal/constant" "baihu/internal/services" + "baihu/internal/services/tasks" "baihu/internal/utils" "github.com/gin-gonic/gin" ) type TaskController struct { - taskService *services.TaskService - cronService *services.CronService + taskService *tasks.TaskService + cronService *tasks.CronService agentWSManager *services.AgentWSManager } -func NewTaskController(taskService *services.TaskService, cronService *services.CronService) *TaskController { +func NewTaskController(taskService *tasks.TaskService, cronService *tasks.CronService) *TaskController { return &TaskController{ taskService: taskService, cronService: cronService, diff --git a/internal/router/register.go b/internal/router/register.go index 9f1f798..462dc4b 100644 --- a/internal/router/register.go +++ b/internal/router/register.go @@ -4,9 +4,10 @@ import ( "baihu/internal/constant" "baihu/internal/controllers" "baihu/internal/services" + "baihu/internal/services/tasks" ) -var cronService *services.CronService +var cronService *tasks.CronService func RegisterControllers() *Controllers { // Initialize services @@ -17,13 +18,18 @@ func RegisterControllers() *Controllers { initService := services.NewInitService(settingsService) userService := initService.Initialize() - taskService := services.NewTaskService() + taskService := tasks.NewTaskService() envService := services.NewEnvService() scriptService := services.NewScriptService() - executorService := services.NewExecutorService(taskService) + sendStatsService := services.NewSendStatsService() + agentWSManager := services.GetAgentWSManager() + + // 创建任务执行服务(需要依赖注入) + taskExecutionService := tasks.NewTaskExecutionService(agentWSManager, sendStatsService) + executorService := tasks.NewExecutorService(taskService, taskExecutionService, settingsService, envService) // Initialize cron service - cronService = services.NewCronService(taskService, executorService) + cronService = tasks.NewCronService(taskService, executorService) cronService.Start() // Initialize and return controllers diff --git a/internal/services/agent_service.go b/internal/services/agent_service.go index 8c4f76c..da43d2e 100644 --- a/internal/services/agent_service.go +++ b/internal/services/agent_service.go @@ -5,6 +5,7 @@ import ( "baihu/internal/database" "baihu/internal/logger" "baihu/internal/models" + "baihu/internal/services/tasks" "crypto/rand" "encoding/hex" "fmt" @@ -350,7 +351,10 @@ func (s *AgentService) buildEnvVarsString(envIDs string) string { // ReportResult Agent 上报执行结果 func (s *AgentService) ReportResult(result *models.AgentTaskResult) error { - taskExecutionService := NewTaskExecutionService() + // 获取依赖的服务 + agentWSManager := GetAgentWSManager() + sendStatsService := NewSendStatsService() + taskExecutionService := tasks.NewTaskExecutionService(agentWSManager, sendStatsService) // 使用统一的结果处理流程 return taskExecutionService.ProcessAgentResult(result) diff --git a/internal/services/cron_service.go b/internal/services/tasks/cron_service.go similarity index 99% rename from internal/services/cron_service.go rename to internal/services/tasks/cron_service.go index ce2024d..364bfb1 100644 --- a/internal/services/cron_service.go +++ b/internal/services/tasks/cron_service.go @@ -1,4 +1,4 @@ -package services +package tasks import ( "sync" diff --git a/internal/services/executor_service.go b/internal/services/tasks/executor_service.go similarity index 90% rename from internal/services/executor_service.go rename to internal/services/tasks/executor_service.go index 4510692..846030f 100644 --- a/internal/services/executor_service.go +++ b/internal/services/tasks/executor_service.go @@ -1,4 +1,4 @@ -package services +package tasks import ( "baihu/internal/constant" @@ -17,6 +17,16 @@ import ( "time" ) +// SettingsService 接口定义(避免循环依赖) +type SettingsService interface { + Get(section, key string) string +} + +// EnvService 接口定义(避免循环依赖) +type EnvService interface { + GetEnvVarsByIDs(ids string) []string +} + // ExecutionResult represents the result of a task execution type ExecutionResult struct { TaskID int @@ -36,6 +46,8 @@ type taskJob struct { type ExecutorService struct { taskService *TaskService taskExecutionService *TaskExecutionService + settingsService SettingsService + envService EnvService results []ExecutionResult runningTasks map[int]bool mu sync.RWMutex @@ -50,9 +62,8 @@ type ExecutorService struct { } // NewExecutorService creates a new executor service -func NewExecutorService(taskService *TaskService) *ExecutorService { +func NewExecutorService(taskService *TaskService, taskExecutionService *TaskExecutionService, settingsService SettingsService, envService EnvService) *ExecutorService { // 从设置中读取调度配置 - settingsService := NewSettingsService() workerCount := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyWorkerCount, 4) queueSize := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyQueueSize, 100) rateInterval := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyRateInterval, 200) @@ -61,7 +72,9 @@ func NewExecutorService(taskService *TaskService) *ExecutorService { es := &ExecutorService{ taskService: taskService, - taskExecutionService: NewTaskExecutionService(), + taskExecutionService: taskExecutionService, + settingsService: settingsService, + envService: envService, results: make([]ExecutionResult, 0, 100), runningTasks: make(map[int]bool), taskQueue: make(chan taskJob, queueSize), @@ -77,7 +90,7 @@ func NewExecutorService(taskService *TaskService) *ExecutorService { } // getIntSetting 从设置中获取整数值 -func getIntSetting(s *SettingsService, section, key string, defaultVal int) int { +func getIntSetting(s SettingsService, section, key string, defaultVal int) int { val := s.Get(section, key) if val == "" { return defaultVal @@ -128,10 +141,9 @@ func (es *ExecutorService) Reload() { logger.Info("[Executor] 已停止工作线程") // 从设置中读取新配置 - settingsService := NewSettingsService() - workerCount := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyWorkerCount, 4) - queueSize := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyQueueSize, 100) - rateInterval := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyRateInterval, 200) + workerCount := getIntSetting(es.settingsService, constant.SectionScheduler, constant.KeyWorkerCount, 4) + queueSize := getIntSetting(es.settingsService, constant.SectionScheduler, constant.KeyQueueSize, 100) + rateInterval := getIntSetting(es.settingsService, constant.SectionScheduler, constant.KeyRateInterval, 200) // 重建 channel 和配置 es.mu.Lock() @@ -228,8 +240,7 @@ func (es *ExecutorService) executeNormalTask(task *models.Task) *ExecutionResult } // 加载环境变量 - envService := NewEnvService() - envVars := envService.GetEnvVarsByIDs(task.Envs) + envVars := es.envService.GetEnvVarsByIDs(task.Envs) // 确定工作目录 workDir := task.WorkDir diff --git a/internal/services/task_execution_service.go b/internal/services/tasks/task_execution_service.go similarity index 92% rename from internal/services/task_execution_service.go rename to internal/services/tasks/task_execution_service.go index 15f4da2..f8be11f 100644 --- a/internal/services/task_execution_service.go +++ b/internal/services/tasks/task_execution_service.go @@ -1,4 +1,4 @@ -package services +package tasks import ( "baihu/internal/database" @@ -15,15 +15,22 @@ import ( "time" ) +// AgentWSManager 接口定义(避免循环依赖) +type AgentWSManager interface { + SendToAgent(agentID uint, msgType string, data interface{}) error +} + // TaskExecutionService 统一的任务执行服务 type TaskExecutionService struct { taskLogService *TaskLogService + agentWSManager AgentWSManager } // NewTaskExecutionService 创建任务执行服务 -func NewTaskExecutionService() *TaskExecutionService { +func NewTaskExecutionService(agentWSManager AgentWSManager, sendStatsService SendStatsService) *TaskExecutionService { return &TaskExecutionService{ - taskLogService: NewTaskLogService(), + taskLogService: NewTaskLogService(sendStatsService), + agentWSManager: agentWSManager, } } @@ -127,8 +134,10 @@ func (s *TaskExecutionService) executeRemote(req *TaskExecutionRequest) error { } // 通过 WebSocket 发送立即执行命令给 Agent - manager := GetAgentWSManager() - err := manager.SendToAgent(agentID, "execute", map[string]interface{}{ + if s.agentWSManager == nil { + return fmt.Errorf("AgentWSManager 未初始化") + } + err := s.agentWSManager.SendToAgent(agentID, "execute", map[string]interface{}{ "task_id": task.ID, }) if err != nil { diff --git a/internal/services/task_log_service.go b/internal/services/tasks/task_log_service.go similarity index 88% rename from internal/services/task_log_service.go rename to internal/services/tasks/task_log_service.go index 27adaa8..90b5e39 100644 --- a/internal/services/task_log_service.go +++ b/internal/services/tasks/task_log_service.go @@ -1,4 +1,4 @@ -package services +package tasks import ( "baihu/internal/database" @@ -9,12 +9,21 @@ import ( "time" ) +// SendStatsService 接口定义(避免循环依赖) +type SendStatsService interface { + IncrementStats(taskID uint, status string) error +} + // TaskLogService 任务日志服务 -type TaskLogService struct{} +type TaskLogService struct { + sendStatsService SendStatsService +} // NewTaskLogService 创建任务日志服务 -func NewTaskLogService() *TaskLogService { - return &TaskLogService{} +func NewTaskLogService(sendStatsService SendStatsService) *TaskLogService { + return &TaskLogService{ + sendStatsService: sendStatsService, + } } // CleanConfig 清理配置 @@ -37,8 +46,11 @@ func (s *TaskLogService) SaveTaskLog(taskLog *models.TaskLog) error { // UpdateTaskStats 更新任务统计 func (s *TaskLogService) UpdateTaskStats(taskID uint, status string) { - sendStatsService := NewSendStatsService() - err := sendStatsService.IncrementStats(taskID, status) + if s.sendStatsService == nil { + logger.Error("[TaskLog] SendStatsService 未初始化") + return + } + err := s.sendStatsService.IncrementStats(taskID, status) if err != nil { logger.Errorf("UpdateTaskStats err: %v", err) return diff --git a/internal/services/task_service.go b/internal/services/tasks/task_service.go similarity index 99% rename from internal/services/task_service.go rename to internal/services/tasks/task_service.go index 25d43cc..fcc6f55 100644 --- a/internal/services/task_service.go +++ b/internal/services/tasks/task_service.go @@ -1,4 +1,4 @@ -package services +package tasks import ( "baihu/internal/database"