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

feat: adjust comment

Browse Source
This commit is contained in:
engigu
2026-02-10 18:43:59 +08:00
parent 2aab7e976c
commit a8bd67dbd3
8 changed files with 71 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/controllers/file_controller.go
+6 -6
View File
@@ -37,8 +37,8 @@ type FileNode struct {
Children []*FileNode `json:"children,omitempty"`
}
// checkPath verify if the path is inside the workDir and safe to use.
// It returns the full absolute path and a boolean indicating if it's safe.
// checkPath 校验路径是否在工作目录内且安全。
// 它返回完整的绝对路径以及一个表示路径是否安全的布尔值。
func (fc *FileController) checkPath(path string, allowRoot bool) (string, bool) {
fullPath := filepath.Join(fc.workDir, filepath.Clean(path))
rel, err := filepath.Rel(fc.workDir, fullPath)
@@ -46,12 +46,12 @@ func (fc *FileController) checkPath(path string, allowRoot bool) (string, bool)
return "", false
}
// Basic traversal check
// 基础的目录穿越检查
if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
return "", false
}
// Root directory check
// 根目录检查
if !allowRoot && rel == "." {
return "", false
}
@@ -313,7 +313,7 @@ func (fc *FileController) RenameFile(c *gin.Context) {
utils.Success(c, nil)
}
// UploadArchive handles archive file upload and extraction
// UploadArchive 处理归档文件的上传和解压
func (fc *FileController) UploadArchive(c *gin.Context) {
targetDir := c.PostForm("path")
@@ -365,7 +365,7 @@ func (fc *FileController) UploadArchive(c *gin.Context) {
utils.SuccessMsg(c, "导入成功")
}
// UploadFiles handles multiple file uploads
// UploadFiles 处理多个文件的上传
func (fc *FileController) UploadFiles(c *gin.Context) {
targetDir := c.PostForm("path")
internal/executor/executor.go
+7 -7
View File
@@ -36,14 +36,14 @@ type Request struct {
Command string
WorkDir string
Envs []string
Timeout int // 分钟
Timeout int // 任务超时时间（分钟）
}
// Result 任务执行结果
type Result struct {
Output string
Error string
Status string // success, failed
Status string // 状态: success, failed
Duration int64 // 毫秒
ExitCode int
StartTime time.Time
@@ -153,7 +153,7 @@ func ExecuteWithHooks(ctx context.Context, req Request, stdout, stderr io.Writer
// 如果 stdout 和 stderr 指针不一致，但在逻辑上我们知道它们是同一个 MultiWriter，
// 这里会显示为 Pipe 模式。
if stdout != stderr && stdout != io.Discard {
logger.Debugf("[Executor] 任务 #%d stdout (%p) and stderr (%p) are different, falling back to Pipe mode.", logID, stdout, stderr)
logger.Debugf("[Executor] 任务 #%d stdout (%p) 和 stderr (%p) 不同，回退到 Pipe 模式。", logID, stdout, stderr)
}
logger.Infof("[Executor] 任务 #%d 启动于 Pipe 模式", logID)
if stdout != nil && stdout == stderr {
@@ -183,18 +183,18 @@ func ExecuteWithHooks(ctx context.Context, req Request, stdout, stderr io.Writer
if pipeWriter != nil {
pipeWriter.Close()
}
// Start 失败的处理
// 启动失败的处理
end := time.Now()
result := &Result{
Status: constant.TaskStatusFailed,
Duration: end.Sub(start).Milliseconds(),
ExitCode: 1,
StartTime: start, // 修正为 start
StartTime: start, // 记录开始时间
EndTime: end,
}
// 执行后钩子
if hooks != nil {
result.Output += "\n[System Error] " + err.Error()
result.Output += "\n[系统错误] " + err.Error()
hooks.PostExecute(ctx, logID, result)
}
return result, err
@@ -265,7 +265,7 @@ func ExecuteWithHooks(ctx context.Context, req Request, stdout, stderr io.Writer
if hooks != nil {
if hookErr := hooks.PostExecute(ctx, logID, result); hookErr != nil {
// 记录钩子错误但不影响执行结果
result.Output += "\n[Hook Error] " + hookErr.Error()
result.Output += "\n[钩子错误] " + hookErr.Error()
}
}
internal/models/agent.go
+3 -3
View File
@@ -71,10 +71,10 @@ type AgentTaskResult struct {
Output string `json:"output"`
Error string `json:"error"` // 额外的系统错误信息
Status string `json:"status"` // success, failed
Duration int64 `json:"duration"` // milliseconds
Duration int64 `json:"duration"` // 耗时（毫秒）
ExitCode int `json:"exit_code"`
StartTime int64 `json:"start_time"` // unix timestamp
EndTime int64 `json:"end_time"` // unix timestamp
StartTime int64 `json:"start_time"` // Unix 时间戳
EndTime int64 `json:"end_time"` // Unix 时间戳
}
// AgentRegisterRequest Agent 注册请求
internal/models/task.go
+5 -5
View File
@@ -32,14 +32,14 @@ type TaskConfig struct {
Concurrency int `json:"$task_concurrency"` // 0: disable concurrency, 1: enable concurrency
}
// Task represents a scheduled task
// Task 代表一个计划任务
type Task struct {
ID uint `json:"id" gorm:"primaryKey"`
Name string `json:"name" gorm:"size:255;not null"`
Command string `json:"command" gorm:"type:text"` // 普通任务的命令
Type string `json:"type" gorm:"size:20;default:'task'"` // 任务类型: constant.TaskTypeNormal, constant.TaskTypeRepo
Config string `json:"config" gorm:"type:text"` // 配置 JSON（仓库同步配置等）
Schedule string `json:"schedule" gorm:"size:100"` // cron expression
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
@@ -86,16 +86,16 @@ func (t *Task) GetSchedule() string {
return t.Schedule
}
// TaskLog represents a log entry for task execution
// TaskLog 代表任务执行的日志记录
type TaskLog struct {
ID uint `json:"id" gorm:"primaryKey"`
TaskID uint `json:"task_id" gorm:"index"`
AgentID *uint `json:"agent_id" gorm:"index"` // Agent ID，为空表示本地执行
Command string `json:"command" gorm:"type:text"`
Output string `json:"-" gorm:"type:longtext"` // gzip+base64 compressed
Output string `json:"-" gorm:"type:longtext"` // gzip+base64 压缩后的日志
Error string `json:"error" gorm:"type:text"` // 额外的系统错误信息
Status string `json:"status" gorm:"size:20;index"` // success, failed
Duration int64 `json:"duration"` // milliseconds
Duration int64 `json:"duration"` // 执行耗时（毫秒）
ExitCode int `json:"exit_code"`
StartTime *LocalTime `json:"start_time"`
EndTime *LocalTime `json:"end_time"`
internal/router/register.go
+2 -2
View File
@@ -10,7 +10,7 @@ import (
var executorService *tasks.ExecutorService
func RegisterControllers() *Controllers {
// Initialize services
// 初始化服务
settingsService := services.NewSettingsService()
loginLogService := services.NewLoginLogService()
@@ -37,7 +37,7 @@ func RegisterControllers() *Controllers {
// 启动计划任务
executorService.StartCron()
// Initialize and return controllers
// 初始化并返回控制器
return &Controllers{
Task: controllers.NewTaskController(taskService, executorService),
Auth: controllers.NewAuthController(userService, settingsService, loginLogService),
internal/router/router.go
+15 -15
View File
@@ -62,24 +62,24 @@ func Setup(c *Controllers) *gin.Engine {
root = router.Group("")
}
// Serve embedded Vue SPA static files with cache headers
// 静态资源服务（Vue SPA），带缓存头部
staticFS := static.GetFS()
assetsGroup := root.Group("/assets")
assetsGroup.Use(cacheControl("public, max-age=31536000, immutable")) // 1 year cache for hashed assets
assetsGroup.Use(cacheControl("public, max-age=31536000, immutable")) // 带哈希的资源缓存1年
assetsGroup.StaticFS("/", http.FS(mustSubFS(staticFS, "assets")))
// Serve logo.svg with short cache
// logo.svg 短缓存实现
root.GET("/logo.svg", func(ctx *gin.Context) {
data, err := static.ReadFile("logo.svg")
if err != nil {
ctx.Status(404)
return
}
ctx.Header("Cache-Control", "public, max-age=86400") // 1 day
ctx.Header("Cache-Control", "public, max-age=86400") // 缓存1天
ctx.Data(200, "image/svg+xml", data)
})
// API routes
// API 路由组
api := root.Group("/api/v1")
{
// Health check (无需认证)
@@ -105,13 +105,13 @@ func Setup(c *Controllers) *gin.Engine {
// 获取当前用户
authorized.GET("/auth/me", c.Auth.GetCurrentUser)
// Dashboard stats
// 仪表盘统计
authorized.GET("/stats", c.Dashboard.GetStats)
authorized.GET("/sentence", c.Dashboard.GetSentence)
authorized.GET("/sendstats", c.Dashboard.GetSendStats)
authorized.GET("/taskstats", c.Dashboard.GetTaskStats)
// Task routes
// 任务模块
tasks := authorized.Group("/tasks")
{
tasks.POST("", c.Task.CreateTask)
@@ -122,7 +122,7 @@ func Setup(c *Controllers) *gin.Engine {
tasks.POST("/stop/:logID", c.Task.StopTask)
}
// Task execution routes
// 任务执行模块
execution := authorized.Group("/execute")
{
execution.POST("/task/:id", c.Executor.ExecuteTask)
@@ -130,7 +130,7 @@ func Setup(c *Controllers) *gin.Engine {
execution.GET("/results", c.Executor.GetLastResults)
}
// Environment variable routes
// 环境变量模块
env := authorized.Group("/env")
{
env.POST("", c.Env.CreateEnvVar)
@@ -141,7 +141,7 @@ func Setup(c *Controllers) *gin.Engine {
env.DELETE("/:id", c.Env.DeleteEnvVar)
}
// Script routes
// 脚本模块
scripts := authorized.Group("/scripts")
{
scripts.POST("", c.Script.CreateScript)
@@ -151,7 +151,7 @@ func Setup(c *Controllers) *gin.Engine {
scripts.DELETE("/:id", c.Script.DeleteScript)
}
// File routes
// 文件管理模块
files := authorized.Group("/files")
{
files.GET("/tree", c.File.GetFileTree)
@@ -166,7 +166,7 @@ func Setup(c *Controllers) *gin.Engine {
files.POST("/uploadfiles", c.File.UploadFiles)
}
// Log routes
// 日志查看模块
logs := authorized.Group("/logs")
{
logs.GET("", c.Log.GetLogs)
@@ -174,11 +174,11 @@ func Setup(c *Controllers) *gin.Engine {
logs.GET("/:id", c.Log.GetLogDetail)
}
// Terminal routes
// 终端模块
authorized.GET("/terminal/ws", c.Terminal.HandleWebSocket)
authorized.POST("/terminal/exec", c.Terminal.ExecuteShellCommand)
// Settings routes
// 设置中心模块
settings := authorized.Group("/settings")
{
settings.POST("/password", c.Settings.ChangePassword)
@@ -241,7 +241,7 @@ func Setup(c *Controllers) *gin.Engine {
agentAPI.GET("/ws", c.Agent.WSConnect) // WebSocket 连接
}
// SPA fallback - serve index.html (no cache for HTML)
// SPA 兜底路由 - 返回 index.html（HTML禁用缓存以保证实时同步）
// 必须在最后注册，作为兜底路由
router.NoRoute(func(ctx *gin.Context) {
// 如果配置了前缀，只处理带前缀的路径
internal/services/agent_ws_service.go
+2 -2
View File
@@ -15,11 +15,11 @@ import (
// AgentWSManager WebSocket 连接管理器
type AgentWSManager struct {
connections map[uint]*AgentConnection // agentID -> connection
connections map[uint]*AgentConnection // Agent ID -> 连接对象
ipConnections map[string]int // IP -> 连接数
ipLastAttempt map[string]time.Time // IP -> 最后连接尝试时间
ipFailCount map[string]int // IP -> 连续失败次数
remoteWaiters map[uint]chan *models.AgentTaskResult // logID -> result channel
remoteWaiters map[uint]chan *models.AgentTaskResult // 日志 ID -> 结果通道
mu sync.RWMutex
}
internal/services/tasks/tiny_log.go
+31 -31
View File
@@ -14,7 +14,7 @@ import (
)
var (
// globalTinyLogManager keeps track of all active TinyLog instances
// globalTinyLogManager 跟踪所有活跃的 TinyLog 实例
globalTinyLogManager = &TinyLogManager{
logs: make(map[uint]*TinyLog),
}
@@ -43,12 +43,12 @@ func (m *TinyLogManager) Get(logID uint) *TinyLog {
return m.logs[logID]
}
// GetActiveLog returns an active TinyLog by its ID
// GetActiveLog 通过 ID 获取活跃的 TinyLog 实例
func GetActiveLog(logID uint) *TinyLog {
return globalTinyLogManager.Get(logID)
}
// TinyLog is a high-performance, low-memory log collector
// TinyLog 是一个高性能、低内存占用的日志收集器
type TinyLog struct {
LogID uint
mu sync.RWMutex
@@ -60,7 +60,7 @@ type TinyLog struct {
closed bool
}
// NewTinyLog creates a new TinyLog instance backed by a temporary file and registers it
// NewTinyLog 创建一个新的 TinyLog 实例（基于临时文件存储）并注册它
func NewTinyLog(logID uint) (*TinyLog, error) {
f, err := os.CreateTemp("", "task_log_*.log")
if err != nil {
@@ -78,7 +78,7 @@ func NewTinyLog(logID uint) (*TinyLog, error) {
return tl, nil
}
// Write implements io.Writer
// Write 实现 io.Writer 接口
func (l *TinyLog) Write(p []byte) (n int, err error) {
l.mu.Lock()
defer l.mu.Unlock()
@@ -87,7 +87,7 @@ func (l *TinyLog) Write(p []byte) (n int, err error) {
return 0, os.ErrClosed
}
// 1. Combine with remainder from previous call
// 1. 合并上次调用剩余的字节（可能是半个 UTF-8 字符）
originalInputLen := len(p)
payload := p
if len(l.remainder) > 0 {
@@ -95,13 +95,13 @@ func (l *TinyLog) Write(p []byte) (n int, err error) {
l.remainder = nil
}
// 2. Identify trailing partial UTF-8 sequence
// 2. 识别结尾不完整的 UTF-8 序列
lastSafe := len(payload)
// UTF-8 characters are max 4 bytes. Check the last few bytes.
// UTF-8 字符最多 4 字节，检查最后几个字节
for i := len(payload) - 1; i >= 0 && i >= len(payload)-4; i-- {
if utf8.RuneStart(payload[i]) {
if !utf8.FullRune(payload[i:]) {
// Indeed a partial rune at the end
// 发现末尾存在不完整字符
lastSafe = i
l.remainder = make([]byte, len(payload)-i)
copy(l.remainder, payload[i:])
@@ -110,29 +110,29 @@ func (l *TinyLog) Write(p []byte) (n int, err error) {
}
}
// If the entire payload is partial and not longer than a max UTF-8 char,
// keep it all for the next call.
// 如果整个负载都不完整且不超过一个 UTF-8 字符的最大长度，
// 则全部保留到下次调用
if lastSafe == 0 && len(l.remainder) > 0 {
return originalInputLen, nil
}
// 3. Convert only the complete part to UTF-8
// 3. 仅将完整的部分转换为 UTF-8
text := utils.ToUTF8(payload[:lastSafe])
data := []byte(text)
// 4. Write to file buffer
// 4. 写入文件缓冲区
_, err = l.writer.Write(data)
if err != nil {
return 0, err
}
// 5. Broadcast to subscribers
// 5. 广播给所有订阅者
if len(l.subscribers) > 0 {
for _, ch := range l.subscribers {
select {
case ch <- data:
default:
// Drop message if subscriber is too slow to avoid blocking writer
// 如果订阅者处理太慢，丢弃消息以避免阻塞写入
}
}
}
@@ -140,7 +140,7 @@ func (l *TinyLog) Write(p []byte) (n int, err error) {
return originalInputLen, nil
}
// Subscribe returns a channel that receives log chunks in real-time
// Subscribe 返回一个实时接收日志块的通道
func (l *TinyLog) Subscribe() chan []byte {
l.mu.Lock()
defer l.mu.Unlock()
@@ -150,7 +150,7 @@ func (l *TinyLog) Subscribe() chan []byte {
return ch
}
// Unsubscribe removes a subscriber
// Unsubscribe 移除订阅者
func (l *TinyLog) Unsubscribe(ch chan []byte) {
l.mu.Lock()
defer l.mu.Unlock()
@@ -164,7 +164,7 @@ func (l *TinyLog) Unsubscribe(ch chan []byte) {
}
}
// Close finishes writing and closes the file, and unregisters itself
// Close 完成写入，关闭文件并注销实例
func (l *TinyLog) Close() error {
l.mu.Lock()
defer l.mu.Unlock()
@@ -173,13 +173,13 @@ func (l *TinyLog) Close() error {
return nil
}
// Process any remaining bytes
// 处理剩余的字节
if len(l.remainder) > 0 {
text := utils.ToUTF8(l.remainder)
data := []byte(text)
_, _ = l.writer.Write(data)
// Also notify subscribers of the last bit
// 通知订阅者最后一部分内容
for _, ch := range l.subscribers {
select {
case ch <- data:
@@ -189,12 +189,12 @@ func (l *TinyLog) Close() error {
l.remainder = nil
}
// Flush buffer to file
// 将缓冲区刷新到文件
if err := l.writer.Flush(); err != nil {
return err
}
// Close all subscribers
// 关闭所有订阅者通道
for _, ch := range l.subscribers {
close(ch)
}
@@ -205,14 +205,14 @@ func (l *TinyLog) Close() error {
return l.file.Close()
}
// CompressAndCleanup reads the temporary file, compresses it, returns the result, and removes the file
// CompressAndCleanup 读取临时文件，进行压缩处理，返回结果并删除临时文件
func (l *TinyLog) CompressAndCleanup() (string, error) {
// Ensure closed
if !l.closed {
l.Close()
}
// Open temp file for reading
// 打开临时文件进行读取
f, err := os.Open(l.path)
if err != nil {
return "", err
@@ -222,17 +222,17 @@ func (l *TinyLog) CompressAndCleanup() (string, error) {
os.Remove(l.path) // Cleanup
}()
// Create buffer for compressed output
// 创建压缩输出缓冲区
var buf bytes.Buffer
b64Writer := base64.NewEncoder(base64.StdEncoding, &buf)
zlibWriter := zlib.NewWriter(b64Writer)
// Stream: File -> Zlib -> Base64 -> Buffer
// 流处理: 文件 -> Zlib -> Base64 -> 缓冲区
if _, err := io.Copy(zlibWriter, f); err != nil {
return "", err
}
// Close generic writers to flush data
// 关闭写入器以刷新数据
if err := zlibWriter.Close(); err != nil {
return "", err
}
@@ -243,12 +243,12 @@ func (l *TinyLog) CompressAndCleanup() (string, error) {
return buf.String(), nil
}
// ReadLastLines returns the last n lines of the log
// ReadLastLines 返回日志的最后 n 行
func (l *TinyLog) ReadLastLines(n int) ([]byte, error) {
l.mu.RLock()
defer l.mu.RUnlock()
// Flush writer to ensure file on disk is up to date
// 刷新写入器以确保磁盘上的文件是最新的
_ = l.writer.Flush()
stat, err := os.Stat(l.path)
@@ -257,7 +257,7 @@ func (l *TinyLog) ReadLastLines(n int) ([]byte, error) {
}
size := stat.Size()
var limit int64 = 65536 // Max 64KB for "last 100 lines" preview
var limit int64 = 65536 // 预览限制：最大 64KB
if size < limit {
limit = size
}
@@ -282,7 +282,7 @@ func (l *TinyLog) ReadLastLines(n int) ([]byte, error) {
return data, nil
}
// GetPath returns the temporary file path
// GetPath 返回临时文件路径
func (l *TinyLog) GetPath() string {
return l.path
}