diff --git a/internal/database/migrate.go b/internal/database/migrate.go index 1a1a06b..6d58ad8 100644 --- a/internal/database/migrate.go +++ b/internal/database/migrate.go @@ -1,8 +1,13 @@ package database import ( + "fmt" + "reflect" + "strings" + "github.com/engigu/baihu-panel/internal/logger" "github.com/engigu/baihu-panel/internal/models" + "gorm.io/gorm/schema" ) func Migrate() error { @@ -11,7 +16,7 @@ func Migrate() error { logger.Warnf("[Database] 自定义迁移警告: %v", err) } - return AutoMigrate( + allModels := []interface{}{ &models.User{}, &models.Task{}, &models.TaskLog{}, @@ -26,7 +31,96 @@ func Migrate() error { &models.Language{}, &models.NotifyWay{}, &models.NotifyBinding{}, - ) + } + + if err := AutoMigrate(allModels...); err != nil { + return err + } + + // MySQL 的 TEXT 类型最大 64KB，LONGTEXT 最大 4GB + // 模型统一使用 type:text 保持跨数据库兼容，这里针对 MySQL 自动升级为 LONGTEXT + mysqlUpgradeTextColumns(allModels...) + + return nil +} + +// mysqlUpgradeTextColumns 反射扫描所有模型，将 gorm tag 中 type:text 的字段在 MySQL 上升级为 LONGTEXT +func mysqlUpgradeTextColumns(allModels ...interface{}) { + if DB.Dialector.Name() != "mysql" { + return + } + + // 获取当前数据库名 + var dbName string + DB.Raw("SELECT DATABASE()").Scan(&dbName) + + ns := schema.NamingStrategy{} + + for _, model := range allModels { + typ := reflect.TypeOf(model) + if typ.Kind() == reflect.Ptr { + typ = typ.Elem() + } + + // 获取表名 + tableName := "" + if tabler, ok := model.(interface{ TableName() string }); ok { + tableName = tabler.TableName() + } else { + continue + } + + for i := 0; i < typ.NumField(); i++ { + field := typ.Field(i) + gormTag := field.Tag.Get("gorm") + if gormTag == "" || !hasGormTypeText(gormTag) { + continue + } + + // 从 gorm tag 获取列名，没有则用 GORM 命名策略转换 + columnName := parseGormColumn(gormTag) + if columnName == "" { + columnName = ns.ColumnName("", field.Name) + } + + // 检查当前列类型，已经是 longtext 则跳过 + var columnType string + DB.Raw("SELECT DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND COLUMN_NAME = ?", + dbName, tableName, columnName).Scan(&columnType) + if strings.EqualFold(columnType, "longtext") { + continue + } + + sql := fmt.Sprintf("ALTER TABLE `%s` MODIFY COLUMN `%s` LONGTEXT", tableName, columnName) + if err := DB.Exec(sql).Error; err != nil { + logger.Debugf("[Database] MySQL 升级 %s.%s 为 LONGTEXT: %v", tableName, columnName, err) + } + } + } +} + +// hasGormTypeText 检查 gorm tag 中是否包含 type:text +func hasGormTypeText(gormTag string) bool { + for _, part := range strings.Split(gormTag, ";") { + if kv := strings.SplitN(strings.TrimSpace(part), ":", 2); len(kv) == 2 { + if strings.TrimSpace(kv[0]) == "type" && strings.EqualFold(strings.TrimSpace(kv[1]), "text") { + return true + } + } + } + return false +} + +// parseGormColumn 从 gorm tag 中提取 column:xxx +func parseGormColumn(gormTag string) string { + for _, part := range strings.Split(gormTag, ";") { + if kv := strings.SplitN(strings.TrimSpace(part), ":", 2); len(kv) == 2 { + if strings.TrimSpace(kv[0]) == "column" { + return strings.TrimSpace(kv[1]) + } + } + } + return "" } // customMigrations 自定义迁移（处理 AutoMigrate 无法自动完成的变更） diff --git a/internal/executor/scheduler.go b/internal/executor/scheduler.go index 53a2e85..7343434 100644 --- a/internal/executor/scheduler.go +++ b/internal/executor/scheduler.go @@ -62,17 +62,17 @@ const ( // ExecutionRequest 执行请求（标准接口） type ExecutionRequest struct { - TaskID string // 任务 ID - LogID string // 日志 ID - Name string // 任务名称 - Type TaskType // 任务类型 - Command string // 命令 - WorkDir string // 工作目录 - Envs []string // 环境变量 - Timeout int // 超时时间（分钟） - Languages []map[string]string // 语言环境配置 - UseMise bool // 是否使用 mise - Metadata ExecutionMetadata // 额外元数据 + TaskID string // 任务 ID + LogID string // 日志 ID + Name string // 任务名称 + Type TaskType // 任务类型 + Command string // 命令 + WorkDir string // 工作目录 + Envs []string // 环境变量 + Timeout int // 超时时间（分钟） + Languages []map[string]string // 语言环境配置 + UseMise bool // 是否使用 mise + Metadata ExecutionMetadata // 额外元数据 } // ExecutionMetadata 执行额外元数据 @@ -403,28 +403,31 @@ func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error) } } - // 2. 准备输出缓冲区（使用合并缓冲区保证顺序） - var combinedBuf safeBuffer + // 2. 准备输出写入器 + // 注意：对于有 TinyLog 的任务（handler 返回了非 nil 的 stdout/stderr）， + // 不再使用 combinedBuf 在内存中缓存全部输出副本。 + // TinyLog 已经将完整输出写入临时文件，OnTaskCompleted 会从那里读取压缩后的数据。 + // combinedBuf 仅在没有 TinyLog 的场景（系统任务）下使用，作为输出的 fallback。 + var fallbackBuf *safeBuffer var stdoutWriter, stderrWriter io.Writer if stdout != nil && stdout == stderr { - // 如果 stdout 和 stderr 是同一个对象，合并成一个 MultiWriter - // 这样后面 ExecuteWithHooks 才能识别出它们是同一个，从而开启 PTY 模式 - mw := io.MultiWriter(&combinedBuf, stdout) - stdoutWriter = mw - stderrWriter = mw - } else { - if stdout != nil { - stdoutWriter = io.MultiWriter(&combinedBuf, stdout) - } else { - stdoutWriter = &combinedBuf - } - + // 如果 stdout 和 stderr 是同一个对象（TinyLog），直接使用 + // 这样 ExecuteWithHooks 才能识别出它们是同一个，从而开启 PTY 模式 + stdoutWriter = stdout + stderrWriter = stderr + } else if stdout != nil { + stdoutWriter = stdout if stderr != nil { - stderrWriter = io.MultiWriter(&combinedBuf, stderr) + stderrWriter = stderr } else { - stderrWriter = &combinedBuf + stderrWriter = stdout } + } else { + // 没有外部 writer（系统任务），使用内存缓冲区 + fallbackBuf = &safeBuffer{} + stdoutWriter = fallbackBuf + stderrWriter = fallbackBuf } // 3. 实际开始执行事件 (经过队列和速率限制之后) @@ -460,6 +463,14 @@ func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error) execResult, execErr := s.executor(ctx, req, stdoutWriter, stderrWriter) // 5. 构建结果 + // 注意：对于普通任务，output 不再在 result 中保留。 + // OnTaskCompleted 会从 TinyLog 获取压缩后的完整输出。 + // 仅系统任务（无 TinyLog）使用 fallbackBuf 保存输出。 + var outputStr string + if fallbackBuf != nil { + outputStr = fallbackBuf.String() + } + result := &ExecutionResult{ TaskID: req.TaskID, LogID: req.LogID, // 传递 LogID @@ -467,7 +478,7 @@ func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error) if execResult != nil { result.Success = execResult.Status == constant.TaskStatusSuccess - result.Output = combinedBuf.String() + result.Output = outputStr result.Status = execResult.Status result.Duration = execResult.Duration result.ExitCode = execResult.ExitCode @@ -479,7 +490,7 @@ func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error) result.StartTime = start result.EndTime = time.Now() result.Duration = result.EndTime.Sub(result.StartTime).Milliseconds() - result.Output = combinedBuf.String() + result.Output = outputStr } if execErr != nil { diff --git a/internal/models/task.go b/internal/models/task.go index f1d86ea..43357f8 100644 --- a/internal/models/task.go +++ b/internal/models/task.go @@ -28,28 +28,28 @@ type RepoConfig struct { // TaskConfig 任务配置 RepoConfig+TaskConfig=task.config type TaskConfig struct { Concurrency int `json:"$task_concurrency"` // 0: disable concurrency, 1: enable concurrency - AllEnvs bool `json:"$task_all_envs"` // 开启则注入全部环境变量 + AllEnvs bool `json:"$task_all_envs"` // 开启则注入全部环境变量 } // Task 代表一个计划任务 type Task struct { - ID string `json:"id" gorm:"primaryKey;size:20"` - Name string `json:"name" gorm:"size:255;not null"` - Command string `json:"command" gorm:"type:text"` // 普通任务的命令 - Tags string `json:"tags" gorm:"size:255;default:''"` // 标签，逗号分隔 - Type string `json:"type" gorm:"size:20;default:'task'"` // 任务类型: constant.TaskTypeNormal, constant.TaskTypeRepo - TriggerType string `json:"trigger_type" gorm:"size:25;default:'cron'"` // 触发类型: constant.TriggerTypeCron, constant.TriggerTypeBaihuStartup - Config string `json:"config" gorm:"type:text"` // 配置 JSON（仓库同步配置等） - Schedule string `json:"schedule" gorm:"size:100"` // cron 表达式 - Timeout int `json:"timeout" gorm:"default:30"` // 超时时间（分钟），默认30分钟 - WorkDir string `json:"work_dir" gorm:"size:255;default:''"` // 工作目录，为空则使用 scripts 目录 - CleanConfig string `json:"clean_config" gorm:"size:255;default:''"` // 清理配置 JSON - Envs string `json:"envs" gorm:"type:text"` // 环境变量ID列表，逗号分隔 - Languages []map[string]string `json:"languages" gorm:"serializer:json;type:text"` // 针对本地任务的语言配置列表 - AgentID *string `json:"agent_id" gorm:"size:20;index"` // Agent ID，为空表示本地执行 + ID string `json:"id" gorm:"primaryKey;size:20"` + Name string `json:"name" gorm:"size:255;not null"` + Command string `json:"command" gorm:"type:text"` // 普通任务的命令 + Tags string `json:"tags" gorm:"size:255;default:''"` // 标签，逗号分隔 + Type string `json:"type" gorm:"size:20;default:'task'"` // 任务类型: constant.TaskTypeNormal, constant.TaskTypeRepo + TriggerType string `json:"trigger_type" gorm:"size:25;default:'cron'"` // 触发类型: constant.TriggerTypeCron, constant.TriggerTypeBaihuStartup + Config string `json:"config" gorm:"type:text"` // 配置 JSON（仓库同步配置等） + Schedule string `json:"schedule" gorm:"size:100"` // cron 表达式 + Timeout int `json:"timeout" gorm:"default:30"` // 超时时间（分钟），默认30分钟 + WorkDir string `json:"work_dir" gorm:"size:255;default:''"` // 工作目录，为空则使用 scripts 目录 + CleanConfig string `json:"clean_config" gorm:"size:255;default:''"` // 清理配置 JSON + Envs string `json:"envs" gorm:"type:text"` // 环境变量ID列表，逗号分隔 + Languages []map[string]string `json:"languages" gorm:"serializer:json;type:text"` // 针对本地任务的语言配置列表 + AgentID *string `json:"agent_id" gorm:"size:20;index"` // Agent ID，为空表示本地执行 RetryCount int `json:"retry_count" gorm:"default:0"` // 失败重试次数 RetryInterval int `json:"retry_interval" gorm:"default:0"` // 失败重试间隔(秒) - RandomRange int `json:"random_range" gorm:"default:0"` // 随机延迟范围(秒) + 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:"-"` // 运行时环境变量（非持久化） @@ -118,7 +118,7 @@ type TaskLog struct { TaskID string `json:"task_id" gorm:"size:20;index"` AgentID *string `json:"agent_id" gorm:"size:20;index"` // Agent ID，为空表示本地执行 Command string `json:"command" gorm:"type:text"` - Output string `json:"-" gorm:"type:longtext"` // gzip+base64 压缩后的日志 + Output string `json:"-" gorm:"type:text"` // gzip+base64 压缩后的日志 Error string `json:"error" gorm:"type:text"` // 额外的系统错误信息 Status string `json:"status" gorm:"size:20;index"` // success, failed Duration int64 `json:"duration"` // 执行耗时（毫秒）