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

fix: pg text error #48

Browse Source
This commit is contained in:
engigu
2026-03-08 20:51:04 +08:00
parent 93ac2217f9
commit a444e9b17b
3 changed files with 153 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/database/migrate.go
+96 -2
View File
@@ -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 无法自动完成的变更）
internal/executor/scheduler.go
+40 -29
View File
@@ -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 {
internal/models/task.go
+17 -17
View File
@@ -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"` // 执行耗时（毫秒）