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

fix: agent id type error

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