diff --git a/agent/agent.go b/agent/agent.go index 28233a4..3e96a12 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -43,21 +43,21 @@ type WSMessage struct { } type AgentTask struct { - ID uint `json:"id"` - Name string `json:"name"` - Command string `json:"command"` - Schedule string `json:"schedule"` - Cron string `json:"cron"` - Timeout int `json:"timeout"` - WorkDir string `json:"work_dir"` - Envs string `json:"envs"` + ID string `json:"id"` + Name string `json:"name"` + Command string `json:"command"` + Schedule string `json:"schedule"` + Cron string `json:"cron"` + Timeout int `json:"timeout"` + WorkDir string `json:"work_dir"` + Envs string `json:"envs"` Languages []map[string]string `json:"languages"` RandomRange int `json:"random_range"` Enabled bool `json:"enabled"` } func (t *AgentTask) GetID() string { - return fmt.Sprintf("%d", t.ID) + return t.ID } func (t *AgentTask) GetName() string { @@ -104,9 +104,9 @@ func (t *AgentTask) GetRandomRange() int { } type TaskResult struct { - TaskID uint `json:"task_id"` - LogID uint `json:"log_id"` - AgentID uint `json:"agent_id"` // 仅用于 HTTP 上报时后端补充 + TaskID string `json:"task_id"` + LogID string `json:"log_id"` + AgentID string `json:"agent_id"` // 仅用于 HTTP 上报时后端补充 Command string `json:"command"` Output string `json:"output"` Error string `json:"error"` @@ -123,16 +123,16 @@ type Agent struct { machineID string scheduler *executor.Scheduler cronManager *executor.CronManager - tasks map[uint]*AgentTask // 本地任务缓存，用于执行 lookup + tasks map[string]*AgentTask // 本地任务缓存，用于执行 lookup lastTaskCount int mu sync.RWMutex client *http.Client wsConn *websocket.Conn wsMu sync.Mutex stopCh chan struct{} - wsStopCh chan struct{} // 用于停止当前 WebSocket 相关的 goroutine - taskLogs map[uint][]string // 记录最近的日志行，用于失败显示 - logMu sync.Mutex // taskLogs 的锁 + wsStopCh chan struct{} // 用于停止当前 WebSocket 相关的 goroutine + taskLogs map[string][]string // 记录最近的日志行，用于失败显示 + logMu sync.Mutex // taskLogs 的锁 } func NewAgent(config *Config, configFile string) *Agent { @@ -140,11 +140,11 @@ func NewAgent(config *Config, configFile string) *Agent { config: config, configFile: configFile, machineID: utils.GenerateMachineID(), - tasks: make(map[uint]*AgentTask), + tasks: make(map[string]*AgentTask), client: &http.Client{Timeout: 30 * time.Second}, stopCh: make(chan struct{}), lastTaskCount: -1, - taskLogs: make(map[uint][]string), + taskLogs: make(map[string][]string), } // 初始化调度器 @@ -171,7 +171,7 @@ type AgentHandler struct { func (h *AgentHandler) OnTaskScheduled(req *executor.ExecutionRequest) {} func (h *AgentHandler) OnTaskExecuting(req *executor.ExecutionRequest) (io.Writer, io.Writer, error) { - if req.LogID > 0 { + if req.LogID != "" { writer := &RealTimeLogWriter{agent: h.agent, logID: req.LogID} return writer, writer, nil } @@ -179,7 +179,7 @@ func (h *AgentHandler) OnTaskExecuting(req *executor.ExecutionRequest) (io.Write } func (h *AgentHandler) OnTaskHeartbeat(req *executor.ExecutionRequest, duration int64) { - if req.LogID > 0 { + if req.LogID != "" { h.agent.sendWSMessage(WSTypeTaskHeartbeat, map[string]interface{}{ "log_id": req.LogID, "duration": duration, @@ -196,11 +196,8 @@ func (h *AgentHandler) OnTaskHeartbeat(req *executor.ExecutionRequest, duration func (h *AgentHandler) OnTaskStarted(req *executor.ExecutionRequest) {} func (h *AgentHandler) OnTaskCompleted(req *executor.ExecutionRequest, result *executor.ExecutionResult) { - var taskID uint - fmt.Sscanf(req.TaskID, "%d", &taskID) - h.agent.sendTaskResult(&TaskResult{ - TaskID: taskID, + TaskID: req.TaskID, LogID: result.LogID, Command: req.Command, Output: result.Output, @@ -226,11 +223,8 @@ func (h *AgentHandler) OnTaskFailed(req *executor.ExecutionRequest, err error) { "content": errMsg, }) - var taskID uint - fmt.Sscanf(req.TaskID, "%d", &taskID) - h.agent.sendTaskResult(&TaskResult{ - TaskID: taskID, + TaskID: req.TaskID, LogID: req.LogID, Command: req.Command, Output: "", @@ -404,7 +398,7 @@ func (a *Agent) fetchTasks() { func (a *Agent) handleConnected(data json.RawMessage) { var resp struct { - AgentID uint `json:"agent_id"` + AgentID string `json:"agent_id"` Name string `json:"name"` IsNewAgent bool `json:"is_new_agent"` MachineID string `json:"machine_id"` @@ -413,9 +407,9 @@ func (a *Agent) handleConnected(data json.RawMessage) { json.Unmarshal(data, &resp) if resp.IsNewAgent { - logger.Infof("注册成功: Agent #%d, 机器码: %s", resp.AgentID, a.machineID[:16]+"...") + logger.Infof("注册成功: Agent #%s, 机器码: %s", resp.AgentID, a.machineID[:16]+"...") } else { - logger.Infof("连接成功: Agent #%d (已存在), 机器码: %s", resp.AgentID, a.machineID[:16]+"...") + logger.Infof("连接成功: Agent #%s (已存在), 机器码: %s", resp.AgentID, a.machineID[:16]+"...") } // 更新调度器配置 @@ -462,7 +456,7 @@ func (a *Agent) updateSchedulerConfig(config map[string]interface{}) { func (a *Agent) handleHeartbeatAck(data json.RawMessage) { var resp struct { - AgentID uint `json:"agent_id"` + AgentID string `json:"agent_id"` Name string `json:"name"` NeedUpdate bool `json:"need_update"` ForceUpdate bool `json:"force_update"` @@ -493,8 +487,8 @@ func (a *Agent) handleTasks(data json.RawMessage) { func (a *Agent) handleExecute(data json.RawMessage) { var req struct { - TaskID uint `json:"task_id"` - LogID uint `json:"log_id"` + TaskID string `json:"task_id"` + LogID string `json:"log_id"` } if err := json.Unmarshal(data, &req); err != nil { logger.Errorf("解析立即执行请求失败: %v", err) @@ -507,13 +501,13 @@ func (a *Agent) handleExecute(data json.RawMessage) { a.mu.RUnlock() if !exists { - logger.Warnf("任务 #%d 不存在，无法执行", req.TaskID) + logger.Warnf("任务 #%s 不存在，无法执行", req.TaskID) return } // 准备执行请求 execReq := &executor.ExecutionRequest{ - TaskID: fmt.Sprintf("%d", task.ID), + TaskID: task.ID, LogID: req.LogID, Name: task.Name, Command: task.Command, @@ -531,25 +525,25 @@ func (a *Agent) handleExecute(data json.RawMessage) { func (a *Agent) handleStop(data json.RawMessage) { var req struct { - LogID uint `json:"log_id"` + LogID string `json:"log_id"` } if err := json.Unmarshal(data, &req); err != nil { logger.Errorf("解析停止请求失败: %v", err) return } - logger.Infof("[Agent] 收到停止指令 LogID: %d", req.LogID) + logger.Infof("[Agent] 收到停止指令 LogID: %s", req.LogID) if a.scheduler.StopLog(req.LogID) { - logger.Infof("[Agent] 任务执行 #%d 已成功停止", req.LogID) + logger.Infof("[Agent] 任务执行 #%s 已成功停止", req.LogID) } else { - logger.Warnf("[Agent] 任务执行 #%d 停止失败（可能已完成或不在运行队列中）", req.LogID) + logger.Warnf("[Agent] 任务执行 #%s 停止失败（可能已完成或不在运行队列中）", req.LogID) } } // RealTimeLogWriter 实时日志写入器，通过 WebSocket 发送日志 type RealTimeLogWriter struct { agent *Agent - logID uint + logID string } func (w *RealTimeLogWriter) Write(p []byte) (n int, err error) { @@ -660,7 +654,7 @@ func (a *Agent) updateTasks(tasks []AgentTask) { a.mu.Lock() defer a.mu.Unlock() - newTasks := make(map[uint]*AgentTask) + newTasks := make(map[string]*AgentTask) for i := range tasks { newTasks[tasks[i].ID] = &tasks[i] } @@ -668,9 +662,9 @@ func (a *Agent) updateTasks(tasks []AgentTask) { // 1. 移除不再存在的任务 for id := range a.tasks { if _, exists := newTasks[id]; !exists { - a.cronManager.RemoveTask(fmt.Sprintf("%d", id)) + a.cronManager.RemoveTask(id) delete(a.tasks, id) - logger.Infof("移除调度任务 #%d", id) + logger.Infof("移除调度任务 #%s", id) } } @@ -684,13 +678,13 @@ func (a *Agent) updateTasks(tasks []AgentTask) { if task.Enabled { err := a.cronManager.AddTask(task) if err != nil { - logger.Errorf("添加调度任务 #%d 失败: %v", id, err) + logger.Errorf("添加调度任务 #%s 失败: %v", id, err) continue } - logger.Infof("已添加调度任务 #%d %s (%s)", id, task.Name, task.GetSchedule()) + logger.Infof("已添加调度任务 #%s %s (%s)", id, task.Name, task.GetSchedule()) } else { - a.cronManager.RemoveTask(fmt.Sprintf("%d", id)) - logger.Infof("调度任务 #%d 已禁用", id) + a.cronManager.RemoveTask(id) + logger.Infof("调度任务 #%s 已禁用", id) } a.tasks[id] = task } @@ -702,17 +696,17 @@ func (a *Agent) clearAllTasks() { defer a.mu.Unlock() for id := range a.tasks { - a.cronManager.RemoveTask(fmt.Sprintf("%d", id)) - logger.Infof("移除任务 #%d", id) + a.cronManager.RemoveTask(id) + logger.Infof("移除任务 #%s", id) } - a.tasks = make(map[uint]*AgentTask) + a.tasks = make(map[string]*AgentTask) a.lastTaskCount = 0 logger.Info("所有任务已清空") } -func (a *Agent) addTaskLog(logID uint, p []byte) { - if logID == 0 { +func (a *Agent) addTaskLog(logID string, p []byte) { + if logID == "" { return } a.logMu.Lock() @@ -727,8 +721,8 @@ func (a *Agent) addTaskLog(logID uint, p []byte) { } } -func (a *Agent) printLastLogs(logID uint) { - if logID == 0 { +func (a *Agent) printLastLogs(logID string) { + if logID == "" { return } a.logMu.Lock() @@ -739,15 +733,15 @@ func (a *Agent) printLastLogs(logID uint) { return } - logger.Errorf("--- 任务 #%d 失败日志预览 (最近 %d 行) ---", logID, len(lines)) + logger.Errorf("--- 任务 #%s 失败日志预览 (最近 %d 行) ---", logID, len(lines)) for _, line := range lines { fmt.Println(" " + line) } - logger.Errorf("--- 任务 #%d 结束 ---", logID) + logger.Errorf("--- 任务 #%s 结束 ---", logID) } -func (a *Agent) clearTaskLog(logID uint) { - if logID == 0 { +func (a *Agent) clearTaskLog(logID string) { + if logID == "" { return } a.logMu.Lock() diff --git a/agent/main.go b/agent/main.go index 91a2fbc..c7d7b50 100644 --- a/agent/main.go +++ b/agent/main.go @@ -373,7 +373,7 @@ func cmdTasks() { Code int `json:"code"` Msg string `json:"msg"` Data struct { - AgentID uint `json:"agent_id"` + AgentID string `json:"agent_id"` Tasks []AgentTask `json:"tasks"` } `json:"data"` } @@ -395,7 +395,7 @@ func cmdTasks() { fmt.Printf("共 %d 个任务:\n\n", len(tasks)) for i, task := range tasks { - fmt.Printf("[%d] ID: %d\n", i+1, task.ID) + fmt.Printf("[%d] ID: %s\n", i+1, task.ID) fmt.Printf(" 名称: %s\n", task.Name) fmt.Printf(" Cron: %s\n", task.Schedule) fmt.Printf(" 命令: %s\n", task.Command) diff --git a/internal/services/migration_v3.go b/internal/services/migration_v3.go index 2583a8a..b9afb98 100644 --- a/internal/services/migration_v3.go +++ b/internal/services/migration_v3.go @@ -116,7 +116,8 @@ func RunMigrationV3() error { if err != nil { return fmt.Errorf("自动备份失败，流程终止: %v", err) } - newPath := filepath.Join(backupDir, fmt.Sprintf("migration_v3_backup_%s.zip", filepath.Base(zipPath))) + baseName := strings.TrimSuffix(filepath.Base(zipPath), ".zip") + newPath := filepath.Join(backupDir, fmt.Sprintf("migration_v3_backup_%s.zip", baseName)) os.Rename(zipPath, newPath) logger.Infof("[MigrationV3] 备份成功: %s", newPath) } @@ -263,7 +264,7 @@ func performHardMigration(db *gorm.DB, mappings map[string]map[uint]string) erro if nid, exists := mappings[t.EntityName][uid]; exists { row["id"] = nid } else { - row["id"] = utils.GenerateID() + row["id"] = utils.GenerateID() } } @@ -328,13 +329,20 @@ func performHardMigration(db *gorm.DB, mappings map[string]map[uint]string) erro // 辅助函数：解析各种数字 ID func parseUint(val interface{}) uint { - if val == nil { return 0 } + if val == nil { + return 0 + } switch v := val.(type) { - case uint: return v - case int64: return uint(v) - case int: return uint(v) - case uint64: return uint(v) - case float64: return uint(v) + case uint: + return v + case int64: + return uint(v) + case int: + return uint(v) + case uint64: + return uint(v) + case float64: + return uint(v) case string: var u uint fmt.Sscanf(v, "%d", &u) @@ -346,12 +354,18 @@ func parseUint(val interface{}) uint { // 辅助函数：字段名转列名 func getColumnName(field string) string { switch field { - case "AgentID": return "agent_id" - case "TaskID": return "task_id" - case "UserID": return "user_id" - case "LogID": return "log_id" - case "Envs": return "envs" - default: return strings.ToLower(field) + case "AgentID": + return "agent_id" + case "TaskID": + return "task_id" + case "UserID": + return "user_id" + case "LogID": + return "log_id" + case "Envs": + return "envs" + default: + return strings.ToLower(field) } } @@ -361,7 +375,9 @@ func transformMultiIDs(oldStr string, parentEntity string, mappings map[string]m var result []string for _, p := range parts { p = strings.TrimSpace(p) - if p == "" { continue } + if p == "" { + continue + } if len(p) == 20 && !utils.IsNumeric(p) { result = append(result, p) // 已经是 xid，保留 continue @@ -390,7 +406,9 @@ func dropOldIndexes(db *gorm.DB, tableName string) { indexNames = append(indexNames, idx.Name) } case "mysql": - var indexes []struct{ KeyName string `gorm:"column:Key_name"` } + var indexes []struct { + KeyName string `gorm:"column:Key_name"` + } db.Raw("SHOW INDEX FROM `" + tableName + "`").Scan(&indexes) seen := make(map[string]bool) for _, idx := range indexes { @@ -400,7 +418,9 @@ func dropOldIndexes(db *gorm.DB, tableName string) { } } case "postgres": - var indexes []struct{ IndexName string `gorm:"column:indexname"` } + var indexes []struct { + IndexName string `gorm:"column:indexname"` + } db.Raw("SELECT indexname FROM pg_indexes WHERE tablename=?", tableName).Scan(&indexes) for _, idx := range indexes { if !strings.HasSuffix(idx.IndexName, "_pkey") {