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

fix: logging format

Browse Source
This commit is contained in:
engigu
2025-12-27 12:42:43 +08:00
parent adac81ef7a
commit 6a23b96271
6 changed files with 94 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/services/config_service.go
+6 -6
View File
@@ -2,7 +2,7 @@ package services
import (
"baihu/internal/constant"
"log"
"baihu/internal/logger"
"os"
"strconv"
@@ -78,7 +78,7 @@ func LoadConfig(path string) (*AppConfig, error) {
// 检查配置文件是否存在
if _, err := os.Stat(path); err == nil {
// 配置文件存在，从文件加载
log.Printf("[Config] Loading from file: %s", path)
logger.Infof("[Config] 从文件加载配置: %s", path)
cfg, err := ini.Load(path)
if err != nil {
return nil, err
@@ -88,7 +88,7 @@ func LoadConfig(path string) (*AppConfig, error) {
}
} else {
// 配置文件不存在，使用环境变量
log.Printf("[Config] File not found, loading from environment variables")
logger.Info("[Config] 配置文件不存在，从环境变量加载")
applyEnvOverrides()
}
@@ -106,12 +106,12 @@ func LoadConfig(path string) (*AppConfig, error) {
// 设置演示模式
if v := os.Getenv("BH_DEMO_MODE"); v == "true" || v == "1" {
constant.DemoMode = true
log.Printf("[Config] Demo mode enabled")
logger.Info("[Config] 演示模式已启用")
}
// 输出配置信息（隐藏敏感信息）
log.Printf("[Config] Server: %s:%d", Config.Server.Host, Config.Server.Port)
log.Printf("[Config] Database: type=%s, host=%s, port=%d, dbname=%s",
logger.Infof("[Config] 服务地址: %s:%d", Config.Server.Host, Config.Server.Port)
logger.Infof("[Config] 数据库: type=%s, host=%s, port=%d, dbname=%s",
Config.Database.Type, Config.Database.Host, Config.Database.Port, Config.Database.DBName)
return Config, nil
internal/services/cron_service.go
+25 -9
View File
@@ -37,31 +37,34 @@ func NewCronService(taskService *TaskService, executorService *ExecutorService)
func (cs *CronService) Start() {
cs.loadTasks()
cs.cron.Start()
logger.Info("Cron service started")
logger.Info("[Cron] 调度服务已启动")
}
// Stop stops the cron service
func (cs *CronService) Stop() {
ctx := cs.cron.Stop()
<-ctx.Done()
logger.Info("Cron service stopped")
logger.Info("[Cron] 调度服务已停止")
}
// loadTasks loads all enabled tasks from database
func (cs *CronService) loadTasks() {
tasks := cs.taskService.GetTasks()
count := 0
for _, task := range tasks {
if task.Enabled {
err := cs.AddTask(&task)
err := cs.addTask(&task, false)
if err != nil {
return
}
count++
}
}
logger.Infof("[Cron] 启动调度已加载 %d 个定时任务", count)
}
// AddTask adds a task to the cron scheduler
func (cs *CronService) AddTask(task *models.Task) error {
// addTask 内部添加任务方法，silent 控制是否打印日志
func (cs *CronService) addTask(task *models.Task, logEnabled bool) error {
cs.mu.Lock()
// 如果已存在，先移除
@@ -76,20 +79,27 @@ func (cs *CronService) AddTask(task *models.Task) error {
})
if err != nil {
cs.mu.Unlock()
logger.Errorf("Failed to add task %d: %v", task.ID, err)
logger.Errorf("[Cron] 添加任务失败 #%d: %v", task.ID, err)
return err
}
cs.entryMap[task.ID] = entryID
cs.mu.Unlock()
logger.Infof("Task %d (%s) scheduled with cron: %s", task.ID, task.Name, task.Schedule)
if logEnabled {
logger.Infof("[Cron] 任务已调度 #%d %s (%s)", task.ID, task.Name, task.Schedule)
}
// 更新下次运行时间
cs.updateNextRun(task.ID)
return nil
}
// AddTask adds a task to the cron scheduler
func (cs *CronService) AddTask(task *models.Task) error {
return cs.addTask(task, true)
}
// RemoveTask removes a task from the cron scheduler
func (cs *CronService) RemoveTask(taskID uint) {
cs.mu.Lock()
@@ -98,13 +108,19 @@ func (cs *CronService) RemoveTask(taskID uint) {
if entryID, exists := cs.entryMap[taskID]; exists {
cs.cron.Remove(entryID)
delete(cs.entryMap, taskID)
logger.Infof("Task %d removed from scheduler", taskID)
logger.Infof("[Cron] 任务已移除 #%d", taskID)
}
}
// runTask executes a task and updates its status
func (cs *CronService) runTask(taskID uint) {
logger.Infof("Running task %d", taskID)
// 获取任务信息用于日志
task := cs.taskService.GetTaskByID(int(taskID))
if task != nil {
logger.Infof("[Cron] 执行任务 #%d %s", taskID, task.Name)
} else {
logger.Infof("[Cron] 执行任务 #%d", taskID)
}
// 更新 last_run
now := time.Now()
internal/services/executor_service.go
+10 -10
View File
@@ -61,7 +61,7 @@ func NewExecutorService(taskService *TaskService) *ExecutorService {
queueSize := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyQueueSize, 100)
rateInterval := getIntSetting(settingsService, constant.SectionScheduler, constant.KeyRateInterval, 200)
logger.Infof("Executor service config: workers=%d, queue=%d, rate=%dms", workerCount, queueSize, rateInterval)
logger.Infof("[Executor] 配置: workers=%d, queue=%d, rate=%dms", workerCount, queueSize, rateInterval)
es := &ExecutorService{
taskService: taskService,
@@ -129,12 +129,12 @@ func (es *ExecutorService) Stop() {
// Reload 重新加载配置并重建 worker pool
func (es *ExecutorService) Reload() {
logger.Info("Reloading executor service...")
logger.Info("[Executor] 正在重载配置...")
// 停止现有 workers
close(es.stopCh)
es.wg.Wait()
logger.Info("Stopped executor service...")
logger.Info("[Executor] 已停止工作线程")
// 从设置中读取新配置
settingsService := NewSettingsService()
@@ -153,7 +153,7 @@ func (es *ExecutorService) Reload() {
// 启动新的 workers
es.startWorkers()
logger.Infof("Executor service reloaded: workers=%d, queue=%d, rate=%dms", workerCount, queueSize, rateInterval)
logger.Infof("[Executor] 配置已重载: workers=%d, queue=%d, rate=%dms", workerCount, queueSize, rateInterval)
}
// RegisterCallback 注册执行完成回调
@@ -186,7 +186,7 @@ func (es *ExecutorService) saveTaskLogCallback(taskID uint, command string, resu
compressed, err := utils.CompressToBase64(output)
if err != nil {
logger.Errorf("Failed to compress log: %v", err)
logger.Errorf("[Executor] 压缩日志失败: %v", err)
compressed = ""
}
@@ -204,7 +204,7 @@ func (es *ExecutorService) saveTaskLogCallback(taskID uint, command string, resu
}
if err := database.DB.Create(taskLog).Error; err != nil {
logger.Errorf("Failed to save task log: %v", err)
logger.Errorf("[Executor] 保存任务日志失败: %v", err)
}
}
@@ -216,7 +216,7 @@ func (es *ExecutorService) updateStatsCallback(taskID uint, _ string, result *Ex
}
sendStatsService := NewSendStatsService()
if err := sendStatsService.IncrementStats(taskID, status); err != nil {
logger.Errorf("Failed to update stats: %v", err)
logger.Errorf("[Executor] 更新统计失败: %v", err)
}
}
@@ -235,7 +235,7 @@ func (es *ExecutorService) cleanLogsCallback(taskID uint, _ string, _ *Execution
var config CleanConfig
if err := json.Unmarshal([]byte(task.CleanConfig), &config); err != nil {
logger.Errorf("Failed to parse clean config: %v", err)
logger.Errorf("[Executor] 解析清理配置失败: %v", err)
return
}
@@ -259,7 +259,7 @@ func (es *ExecutorService) cleanLogsCallback(taskID uint, _ string, _ *Execution
}
if deleted > 0 {
logger.Infof("Cleaned %d logs for task %d", deleted, taskID)
//logger.Infof("Cleaned %d logs for task %d", deleted, taskID)
}
}
@@ -270,7 +270,7 @@ func (es *ExecutorService) EnqueueTask(taskID int) {
// 成功入队
default:
// 队列满，直接执行（降级处理）
logger.Warnf("Task queue full, executing task %d directly", taskID)
logger.Warnf("[Executor] 任务队列已满，直接执行任务 #%d", taskID)
go es.executeTaskInternal(taskID)
}
}